CMU 15640 Distributed Systems F24 学习笔记

Lecture 1: Course Overview

1. Distributed system: A collection of independent computers that appears to its users as a single coherent system

   • Features:

     • No shared memory – message-based communication
     • Each runs its own local OS
     • Heterogeneity

   • Goal: to appear as a single system (usually)

   • 

Lecture 2&3: The Internet

1. refer to 18741 Lecture 2

2. Circuit switching vs. Packet switching

   • Circuit switching
     • +Fast, stable, predictable performance
     • -Resource inefficiency
   • Packet switching
     • +General for many apps
     • +Efficient, robust resource sharing
     • -Lack of isolation: congestion, variable delay, etc.

3. Two problems that a communication protocol needs to solve

   1. How do we identify nodes? - Addressing
   2. How do messages get routed from one node to another? - Routing

4. Flat addressing e.g. MAC address vs. Hierarchical addressing e,g, IP address

   • Flat addressing

     • +”Plug and play”

     • +Ease of assignment and use

     • -Not scalable, leads to large forwarding tables

   • Hierarchical addressing

     • +Scalable
     • -Assignment and management are trickier

5. IP address: network component (used for forwarding across networks) + host component (used for forwarding within a network)

   • class-based assignment:
   • classless: Classless Interdomain Routing (CIDR)
     • Informally, “slash 26” →128.23.9.128/26
     • Formally, prefix represented with a 32-bit mask: 255.255.255.192 with all network prefix bits set to1 and host suffix bits to 0
     • Routing: routers use longest prefix matching: Router looks up its table which table entry the packet matches, The packet may match multiple table entries, The router will pick the entry with the longest prefix and route that way

6. Transport layer: Process multiplexing on the same host (using ports); Provide common end-to-end services for app layer [optional]

7. Transport Control Protocol (TCP)

   • TCP provides a reliable, in-order, two-way byte stream service
   • Flow control: avoid that the sender outruns the receivers
   • Error Control: recover from packet loss, corruption, reordering
   • Congestion control: controls the transmit rate of the sender

Lecture 4: Synchronization

Lecture 5: Time Synchronization

1. Basic time synchromization techniques
   • Cristian’s Time Sync
     • 
     • (-) a single time server might fail
   • Berkeley Algorithm
     • The time daemon (master node) asks all the other machines (workers) for their clock values => the machines answer => The time daemon tells everyone how to adjust their clock
   • Network Time Protocol (NTP)
     • Uses a hierarchy of time servers
       • Class 1 servers have highly-accurate clocks - connected directly to atomic clocks, etc.
       • Class 2 servers get time from only Class 1 and Class 2 servers
       • Class 3 servers get time from any server
     • Synchronization: use multiple one-way messages instead of immediate round-trip
       • Each message bears timestamps of recent events: Local times of Send and Receive of previous message + Local times of Send of current message
       • RTT = wait_time_client – server_proc_time = (t3-t0) – (t2-t1)
       • Offset = t2 - t3 + RTT/2 = ((t1-t0) + (t2-t3))/2
2. Lamport clocks
   • -> happens before
   • || concurrent
   • Total-Order Lamport Clocks
     • $L(e) = M * L_i(e) + i$
   • shortcoming: L(e) < L(e’) does not imply e happened before e’
3. Vector clocks
   • Goal: Want ordering that matches causality V(e) < V(e’) if and only if e → e’
   • How do we do this? Label each event by vector V(e) [c1, c2 …, cn] where ci = # events in process i that causally precede e
     • e → e’ implies V(e)<V(e’)
     • e || e’ implies neither V(e) <= V(e’) nor V(e’) <= V(e)

Lecture 6: Distributed Mutual Exclusion

• Centralized Mutual Exclusion
  • Performance:
    • 3 messages per cycle (1 request, 1 grant, 1 release)
    • Lock server creates bottleneck
  • Issues: What happens when coordinator crashes or reboots?
• > Bully Leader Election
  • P sends an ELECTION message to all processes with higher numbers.
  • If no one responds, P wins the election and becomes coordinator.
  • If one of the higher-ups answers, it takes over. P’s job is done.
• Decentralized Mutual Exclusion
  • Correctness:
    • Majority ensures safety
    • Fairness depends on random chance.
  • Performance:
    • 2m + m messages per attempt to get majority
    • unbounded number of messages per cycle
  • Issues:
    • Node failures (forgetting vote on reboot)
    • Backoff and retry problem
    • Starvation
• > Totally-Ordered Multicast
  • Multicast messages + local timestamp-ordered queue
  • Multicasts an ACK to all other processes
  • Process only if both at queue head and ACK’ed
• > Lamport Mutual Exclusion
  • Based on Lamport TO-multicast ⇒ Simplified
  • ACK only to requestor
  • Release (to all) after finished
  • Performance:
    • Process i sends n-1 request messages
    • Process i receives n-1 reply messages
    • Process i sends n-1 release messages
• > Ricart & Agrawala Mutual Exclusion
  • when receiving an message
    • Receiver uninterested in resource: it sends back an OK message to the sender
    • Receiver already has access to resource: it does not reply, queues the request
    • Receiver wants access to resource, but doesn’t have it yet: If the incoming message has a lower timestamp, the receiver sends back an OK message. If its own message has a lower timestamp, the receiver queues the incoming request and sends nothing
  • when completed critical section: send OK message to all pending requests
  • correctness: deadlock free, starvation free
  • Performance:
    • Each cycle involves 2(n-1) messages
    • n-1 requests by i
    • n-1 replies to i
• > Token Ring Mutual Exclusion
  • Correctness: Clearly safe: Only one process can hold token
  • Fairness: Will pass around ring at most once before getting access
  • Performance:
    • Each cycle requires between 1 - ∞ messages
    • Latency of protocol between 0 & n-1
  • Issues: Lost token
• 

Lecture 7: Remote Procedure Call (RPC)

1. RPC occurs in the following steps
   1. client procedure calls the client stub in the normal way
   2. client stub builds a message and calls the local OS
   3. client OS sends the message to the remote OS
   4. the remote OS gives the message to the server stub
   5. server stub unpacks the parameters and calls the server
   6. server does the work and returns the result to the stub
   7. server stub packs the result in a message and calls its local OS
   8. server’s OS sends the message to the client’s OS
   9. client’s OS gives the message to the client stub
   10. client stub unpacks the result and returns it to the client
2. handle memory access: pass by value, pass by copy/restore for pointers
   • (-) inefficient and complex because the server need to know size of data to copy
   • how about data structures containing pointers?
3. handle failure: break transparency
   • At-least-once: Just keep retrying on client side until you get a response (as long as idempotent)
   • At-most-once
4. handle latency: asynchronous
   • 

Lecture 8: Distributed Transactions, Two-phase Locking, Two-Phase Commit

1. Transaction: ACID Properties
   1. Atomicity: Each transaction completes in its entirely, or is aborted. If aborted, should not have effect on the shared global state.
   2. Consistency: Each transaction preserves a set of invariants about global state.
   3. Isolation: Also means serializability. Each transaction executes as if it were the only one with the ability to read/write shared global state.
   4. Durability: Once a transaction has been completed, or “committed”, its effects will persist, even in the presence of failures.
2. single server case: 2-phase locking (serializability)
   • Phase 1: Acquire locks, no locks released
   • Phase 2: Release locks, no locks acquired
   • Commit: Update changes, release locks
   • Abort: Throw away changes, release locks
   • issue: possible deadlock: a transaction may not know all the locks it needs ahead of time
     • Lock manager builds a “wait-for” graph. On finding a cycle, choose offending transaction and force abort
     • Use timeouts: find transaction waiting for a lock and force abort
3. 2-phase commit
   • 1 coordinator, N participants
   • Prepare => VoteCommit/VoteAbort (must flush to disk before sending msg) => DoCommit (persist before sending msg) / DoAbort => Ack
     • Messages in first phase
       • 1A: Coordinator sends Prepare to participants
       • 1B: Participants write to disk and respond to coordinator: VoteCommitor VoteAbort
     • Messages in the second phase
       • 2A: Coordinator checks votes. If all VoteCommit: write to disk and send DoCommit; else write to disk and send DoAbort
       • 2B: Participants receive the final decision, write Commit/Abort todisk, and send Ackback to coordinator, coordinator logs END and ends the transaction

Lecture 9: Fault tolerance: Checkpointing, Logging, Replication

1. Backward recovery vs. Forward recovery
   • Backward recovery: resend msg
     • (+) Easier to implement
     • (-) expensive
   • Forward recovery: reconstruct from received packets
     • (-) Harder to implement (need clever algorithms)
     • (+) Recovery could be faster
2. Checkpointing: Chandy-Lamport snapshotting algorithm
   • Snapshot initiation by process P
     • P records its own state
     • P sends marker to all other processes on outgoing channels
     • P starts recording on its incoming channels
   • Snapshot propagation: when processes Pi receives marker message
     • If receiving marker for the first time
     • Record its own state
     • Mark that incoming channel as “empty”
     • Propagate marker to all other processes on outgoing channels
     • Start recording incoming messages
     • Else: stop recording on that incoming channel
   • Snapshot termination
     • All processes have received a marker on all incoming channels
3. Database recovery in practice
   • Frequent checkpoint too slow => Fine-grained logging of operations
   • Log grows unbounded => Periodic checkpoint to truncate logs
4. Consistency: ordering of operations across multiple processors
   • Strict consistency: writes instantaneously visible to everyone, Read always returns value from latest write
   • Sequential consistency: all operations in some global order
     • 
   • Causal consistency: all nodes see potentially causally related writes in same order
     • 
   • No consistency

Lecture 10: Distributed Replication

1. State machine replication (SMR): replicate a deterministic state machine across multiple servers (or nodes) to ensure that even in the presence of failures, the system can continue to operate correctly => ensure consistency and fault tolerance
   • For the system to behave correctly, all nodes must agree on the sequence of operations applied to the state machine - Consensus
2. Fischer-Lynch-Paterson (FLP) Impossibility Theorem: No deterministic 1-crash-robust consensus algorithm exists with asynchronous communication
3. Paxos: One or more servers propose values, Only a single value is chosen, Once a value is chosen, it will stay chosen
   • properties: correctness (safety), Liveness (termination) ((-) dueling proposers lead to liveness violation), Fault-tolerance (to handle f failures, need 2f + 1 replicas)
   • algorithm
     1. [Proposers] Choose new proposal number n (globally ordered, unequally numbered)
     2. [Proposers] Broadcast Prepare(n) to all servers
     3. [Acceptors] Respond to Prepare(n):
        • If n > minProposal: (1) minProposal= n (2) Prepare-OK(acceptedProposal, acceptedValue)
        • else: Prepare-REJECT()
     4. [Proposers] When responses received from majority (majority quorum):
        • If any acceptedValues returned, replace value with acceptedValue for highest acceptedProposal
     5. [Proposers] Broadcast Accept(n,value) to all servers
     6. [Acceptors] Respond to Accept(n,value):
        • If n ≥ minProposal: (1) acceptedProposal = minProposal = n (2) acceptedValue= value (3) Accept-OK()
        • else: Accept-REJECT()
     7. [Proposers] When responses received from majority:
        • If majority Accept-OK(), value is definitely chosen
        • Could retry if no majority accept

Lecture 11&12: Distributed File System

1. Caching Mechanisms: Client-side caching - stores copies of files on the client machine, improving performance by reducing the need to access the server frequently
   • Caching can introduce consistency problems, particularly when multiple clients access or modify the same files. Various approaches to handle cache staleness
     • broadcast invalidations: After an update, every possible cache location is notified
       • (-) lots of useless network communication (-) not scalable
     • check on use
       • (+) strict consistency (-) slow reads (-) useless network communication (-) not scalable
     • callbacks: Clients register with server that they have a copy of file, Server tells them “Invalidate” when the file changes
     • leases: Granting exclusive/shared control of the cached objects for a limited amount of time (lease period)
       • lease renewal: Before the lease expires, the client can request a renewal to extend its access. If the server agrees, the lease period is extended; otherwise, it expires
       • read lease vs. write lease: Multiple clients can hold read leases simultaneously; Only one client can hold a write lease
   • NFS v2’s caching method: in-memory caching on the client, with file attributes expiring after 60 seconds. Changes on one machine can take up to 60 seconds to be visible on another. Dirty data is buffered locally for up to 30 seconds or until the file is closed, but if the client crashes before syncing with the server, changes are lost
     • (+) reduced network traffic
     • (-) data consistency guarantee is poor
     • NFS’s Failure Handling
       • Stateless Server
       • operations are idempotent
       • write-through caching: When file is closed, all modified blocks are sent to the server.
2. Consistency
   • AFS: Session Semantics: A file write is immediately visible to processes on the same client, But only after the file is closed on other clients; When a file is closed, changes are visible to new opens, but they are not visible to “old” opens
3. AFS vs NFS
   • AFS has lower server load than NFS
     • More files cached on clients
     • Callbacks: server not busy if files are read-only (common case)
     • AFS more scalable! Could support 50 clients (NFS was limited to 20)
   • But maybe slower
     • Read/write performance potentially slower than NFS
   • For both
     • Central server is bottleneck: all reads and writes hit it at least once;
     • It is a single point of failure.
     • It is costly to make them fast, beefy, and reliable servers.
4. Naming
   • NFS: clients mount NFS volume where they want
   • AFS: name space consistent across clients (= global)
5. Disconnected operation: Pessimistic vs. Optimistic Replica Control
   • Pessimistic: requires the client C to acquire exclusive (RW) or shared ® control of cached objects before accessing them in disconnected mode
   • Optimistic: allows to access the replica in every disconnected mode => Coda detects inconsistencies when a client reconnects, It tries to automatically merge the changes or alerts the user who is then responsible for merging
6. Version control

Lecture 14: Distributed Fault Tolerant Storage

1. metrics
   • Mean time to failure (MTTF): average of TTFs
   • Mean time to repair (MTTR): average of TTRs
   • Mean time between failures (MTBF) = MTTF + MTTR
   • Availability= MTTF / (MTTF + MTTR)
2. RAID

3. Availability metric for RAID: Mean Time to First Data Loss (MTTDL)
   • RAID-4 has the best MTTDL with good capacity

Lecture 15: Cluster Computing: MPI, Actor Model, MapReduce

1. HPC Programming Model: Message Passing Model MPI: Processes communicate and synchronize via exchange of messages
   • 
   • Standardized communication protocol for programming parallel computers
   • Functionality
     • Virtual topology, e.g. Finding number of processes, processor identity for a process, neighboring processes in a logical topology
     • Synchronization: e.g. barrier
     • Communication
   • Standardized set of group communication methods
     • 
   • Typical HPC operation: long-lived processes; partitioning; hold all data in memory; high bandwidth communication
     • strengths: high utilization of resources; effective
     • weaknesses: requires careful tuning of application to resources; intolerant of any variability
   • HPC Fault Tolerance
     • HPC nature: Tightly coupled processes - Failure of one process prevents all others from progressing
     • checkpoint, restore, performance scaling
2. MPI at application layer: Actor model
   • 
   • messages are in order and point-to-point
3. Cluster Programming Model
   • 
4. Cluster Computing Model - MapReduce
   • Stage 1: map
     • Dynamically map input file blocks onto mappers
     • Each generates key/value pairs from its blocks
     • Each writes $R$ files on local file system => total $M * R$ local files
   • Stage 2: shuffle
     • Each reducer: Handles $1/R$ of the possible key values
     • Each reducer: Fetches its file from each of $M$ mappers
     • Each reducer: Sorts all of its entries to group values by keys
   • Stage 3: reduce
     • Each reducer: Executes reducer function for each key
     • Each reducer: Writes output values to cluster filesystem
   • MapReduce Implementation
     • Built on Top of Cluster Filesystem - Google: GFS, Hadoop: HDFS
     • Provides global naming
     • Reliability via replication
     • Input/Output set of files in reliable file system
   • MapReduce Execution
     • 
   • Fault Tolerance: reschedule failed task
   • Stragglers: when done with most tasks, reschedule any remaining executing tasks

Lecture 16: Fault Tolerant In-Memory Computation: Spark

1. Limitation of MapReduce: Disk IO makes huge overhead; does not work for iterative applications (e.g. distributed ML)
2. In-Memory Computation: keep and share data sets in main memory
3. Resilient Distributed Datasets RDD
   • Efficient fault recovery using lineage: Data is partitioned and each operation is applied to every partition, Recompute lost partitions on failure
   • RDDs are resilient because Spark stores the complete lineage of an RDD
   • RDDs are Immutable Objects
     • enables lineage: RDDs need to be deterministic functions of input
     • Simplifies consistency: Caching and sharing RDDs across Spark nodes
     • Compatibility with storage interface (HDFS chunks are append only)
   • RDD operations
     • Transformations (lazy): create new RDD from existing one e.g. map, filter, sample, groupByKey, sortByKey, union, join, cross
     • Actions (eager): return value to caller e.g. count, sum, reduce, save, collect
     • Persist RDD to memory
   • Lazy evaluation because
     • it enables Spark to optimize the required operations
     • it enables Spark to recover from failures and slow workers
   • Spark Deployment
     • Master server (“driver”): Lineage and scheduling
     • Cluster manager (not part of Spark): Resource allocation
     • Worker nodes: Executors isolate concurrent tasks & Caches persist RDDs
     • 
   • Spark is not a good fit for
     • Non-batch workloads e.g. Applications with fine-grained updates to shared state
     • Datasets that don’t fit into memory
     • If you need high efficiency, SIMD/GPU

Lecture 17: Cluster Filesystems: GFS & HDFS

1. GFS Operation Environment: data center
   • Hierarchy of Aggregation and Core Switches
   • Communicating within a rack: low latency, high bandwidth, less contention for bandwidth
   • Communicating across racks: higher latency, limited available bandwidth, more contention
   • 
2. GFS Workload Assumptions
   • Large files, >= 100 MB in size
   • Large, streaming reads (>= 1 MB in size)
   • Large, sequential writes that mostly append
   • Concurrent appends by multiple clients (e.g., files used as producer-consumer queues) - Want atomicity for appends without synchronization overhead among clients
3. GFS Design Goals
   • Maintain high data and system availability
     • Handle failures transparently (i.e., automatically)
   • Low synchronization overhead between entities of GFS
     • Exploit parallelism of numerous disks/servers
     • High throughput for individual reads/writes more important than low latency
     • Co-design filesystem and applications
4. GFS Architecture
   • One master server
     • Holds all metadata in RAM; very fast operations on file system metadata
     • Migrates chunks between chunkservers
     • Controls consistency management
     • Garbage collects orphaned chunks
   • Many chunk servers (1000s)
     • Chunk: fixed size (64 MB) portion of file, identified by 64-bit globally unique ID
     • store file chunks on local disk using Linux file system, each with version number and checksums
     • Chunks replicated on configurable number of chunkservers (default: 3)
     • No caching of file data (beyond standard Linux buffer cache)
     • Send periodic heartbeats to Master
   • Many clients accessing different files stored on same cluster
     • Issues control (metadata) requests to master server
     • Issues data requests directly to chunk servers
     • caches metadata; but no caching of data
     • supports: open, close, read, append (append as an atomic operation without having to lock a file), snapshot, …
   • 
5. Client Operations
   • Read
   • Write: daisy-chain
     • Data pushed linearly along a chain
     • Flow of data decoupled from flow of control
     • Helps to
       • fully utilize each machine’s network bandwidth
       • avoid network bottlenecks and high-latency links
       • minimize the latency to push through all the data
   • Record Append:
     • relaxed consistency model and guarantees atomicity only within each chunk replica, not across all replicas
     • at-least-once append rule
6. GFS Fault Tolerance
   • Availability: chunk replication & master (state of the master) replication
   • Data Integrity: checksum for each chunk
   • Chunk server: missing / stale chunks are detected by master and re-replicate will happen
   • Master: logs metadata update to disk sequentially (WAL)
7. GFS Consistency Model
   • Changes to namespace (i.e., metadata) are atomic
   • Changes to data are ordered by a primary
8. HDFS: Erasure Codes for fault tolerance => uses parity chunks

Lecture 18: Virtualization

1. Virtual Machines

   • A virtual machine monitor (VMM) aka “hypervisor” implements the VM abstraction

     • 

   • Types of system virtualization

     • Type 1: Native/Bare metal

       • 
       • Higher performance

     • Type 2: Hosted

       • 
       • Easier to install and use, cheaper
       • Leverage host’s device drivers
       • Aka “client hypervisors”

   • Properties of VMs

     • isolation: Fault isolation, performance isolation, software isolation
     • encapsulation and portability: independent of physical hardware, enables VM snapshots, clones, migration of live running VMs
     • interposition: security isolation, enables encryption, compression, …

2. Hardware that can be virtualized

   • CPU Virtualization
     • Privileged instructions from user mode
       • cannot be executed directly by the VM
       • Trap (VM exit) and Emulate to VMM which processes it on behalf of the VM
     • Non-privileged instructions: Run directly by the VM on native CPU
   • Memory Virtualization
     • Logical Pages => Physical Pages (managed by guest OS) => Machine Pages (managed by VMM)
   • I/O Virtualization

3. Container Virtualization

   • Motivation: Overhead associated with deploying on VMs: I/O overhead, OS-startup overhead per VM, Memory/Disk overhead
   • Containers
     • Multiple isolated instances of programs
     • Running in user-space (shared kernel)
     • Instances see only resources (files, devices) assigned to their container
   • Requirements on Containers
     • Isolation and encapsulation
       • Fault and performance isolation
       • Encapsulation of environment, libraries, etc.
     • Low overhead: Fast instantiation / startup; Small per-operation overhead (I/O, …)
     • Reduced Portability
     • NO Interposition (no hypervisor)
   • Implementation
     • Resource View Isolation: each process is assigned a “namespace”
     • Resource Usage Isolation: meter resource usage and enforce hard limits per container, by usage counters for gorups of processes
     • Filesystem Isolation: layering of filesystems (copy on write):
       • Read-write (“upper”) layer that keeps per-container file changes
       • Read-only (“lower”) layer for original files
   • Advantages
     • Fast boot times: 100s of milliseconds (10s-100s of seconds for VMs)
     • High density: 1000s of containers per machine
     • Very small I/O overhead
     • Require no CPU support
   • Limitations
     • Implementation Complexity
     • Less general than VMs
     • Harder to migrate than VMs
     • Large attach surface
     • weak security isolation

Lecture 19: Scaling Techniques & Architectures

1. scale up vs. scale out
2. when to scale out: a good heuristic is queue length
3. 3-tier web service architecture
4. Monolithic Architecture vs. Microservice Architecture
5. Microservices reliability: service encapsulation: break down reasoning about failure in large microservice deployments

Lecture 20: Publish-Subscribe

1. Direct vs. Indirect communication
   • Direct communication: sender and receiver are “coupled” - exist in the same “time” and “space”
   • Indirect communication: sender and receiver are uncoupled (or decoupled)
2. Time vs. Space Uncoupling
   • Time uncoupling: a sender can send a message even if the receiver is still not available - Easily deal with volatile environments, network partitions, etc.
   • Space uncoupling: a sender can send a message but does not know to whom it is sending nor if more than one, if anyone, will receive the message - Easily deal with changes in nodes: failure, addition, etc.
3. Publish / Subscribe
   • time- and space-uncoupled => indirect communication
   • producer - Publish: send messages regardless of whether someone is listening
   • consumer - Subscribe: receive messages if anyone is sending them regardless of who
   • subscription models
     • Topic based: Events are classified into predefined topics. Subscription indicates interest in a topic
     • Content based: Events are structured in form of multiple attributes. Subscriptions can define a complex function over multiple attributes
   • broker network - an overlay network that can route events
   • Event routing algorithms - e.g. Flooding, Filtering, Rendezvous
4. Kafka
   • Broker - a Kafka Cluster consists of many Kafka Brokers on many servers
   • Record - have a key (optional), value, and timestamp
   • Topic - a stream of records categorized by feed name. Records are categorized into topics
   • Partition - append-only logs (ordered + immutable). Each topic’s data is split into partitions
   • Replica - partitions are replicated into replicas
     • One leader replica, rest are followers. Writes are sent to leader replica, and leader replica replicates record. Follower replicas are never read by consumers, never written to by producers
     • tolerates numReplicas - 1 dead brokers
   • Producer chooses which partition to send a record to: Typically based on key of record
     • e.g. hash of keys , or a round-robin strategy if no key
     • order of records cannot be guaranteed across partitions
     • Multiple producers can write to the same topic
   • Send message:
     • Fire-and-forget (not really used)
     • Synchronous send: block on producer.send()
     • Asynchronous send: callback function called when there is response from brokers, higher throughput
   • Message batching: can improve throughput
   • Consume: pull => on average load, not peak load
     • track their read offsets for all partitions => can rewind reads to replay older messages
   • Consumer group: a collection of consumers that jointly consume the same topic
     • Each partition consumed only by a single consumer => Partition = smallest unit of parallelism
     • Different consumer groups track different offsets
   • Commit point for Kafka writes: when all in-sync replicas (ISR) have applied message to log
     • Consumers only read committed messages
     • Producers configure when write acks are returned via request.required.acks
       • 0: producer never waits for an ack from the broker.
       • 1: producer gets an ack after the leader replica has received the data.
       • -1: producer gets an ack after all ISR have received the data
   • Kafka read guarantees: at-least-once
     • A message is only ever read by a single consumer in a group
     • Kafka delivers each partition’s records in order, but no ordering across partitions of a topic

Lecture 21: Internet Content Delivery: DNS & CDN

1. DNS-based client routing for CDNs
   • Client does name lookup for service
   • Authoritative name server resolves to CDN name (e.g., using CNAME, e.g. news.com -> news.cdn1.com)
   • CDN high-level name server chooses appropriate CDN instance (e.g. news.cdn1.com -> useast.cdn1.com)
   • CDN low-level name server chooses specific caching server (e.g. useast.cdn1.com -> 192.0.2.10)

Lecture 22: Crypto basics, secure communication

1. Summary of crypto techniques

   • 

2. Certification authority (CA): Certificate contains E’s public key AND the CA’s signature of E’s public key

3. DH key exchange: provides forward secrecy

   • steps
     • Alice and Bob agree on a large prime $p$ , and a generator $g$ that is a primitive root of $p$
     • Alice and Bob choose private numbers $a$ and $b$ at random in $Z^*_{p}$
     • Alice calculates $g ^a \pmod p$ and sends it publicly to Bob
     • Bob calculates $g^ b \pmod p$ and sends it publicly to Alice
     • Alice computes $B^a \pmod p$
     • Bob computes $A^b \pmod p$
   • 
   • shared secret: $g^{ab}\mod p$

4. TLS

   1. Step 1: Exchange Hellos: freshness, prevent replay attack
   2. Step 2: Certificate
   3. Step 3: Premaster Secret (e.g. DHE)
   4. Step 4: Derive Symmetric Keys
   5. Step 5: Exchange MACs
   6. Step 6: Send Messages

Lecture 23: Byzantine Fault Tolerance

1. Byzantine broadcast setup & definitions

   • Honest nodes follow the protocol exactly; Byzantine nodes do not follow the protocol, can collude (controlled by a single adversary)
   • Don’t know which nodes are Byzantine
   • Maximum number of Byzantine nodes = $f$
   • properties
     • Termination: every honest node eventually halts with some output
     • Agreement: all honest nodes halt with the same output (whether or not the leader is honest)
     • Validity: if the leader is an honest node, then the common output of the honest nodes is the private input v* of the sender
   • SMR can be reduced to Byzantine Broadcast using a framework where each iteration involves rotating leadership and invoking a Byzantine Broadcast subroutine => handles consensus in the presence of Byzantine faults
   • Setup
     1. Network setting: Synchronous / Asynchronous / Partially synchronous
     2. Trusted setup
   • Feasibility under different setup
     • Synchronous network, PKI: $f < n$
     • Synchronous network, no PKI: $f < n/3$
     • Asynchronous network: $f < 1 (n >= 2)$
     • Partially synchronous network: $f < n/3$

2. Dolev-Strong Protocol

   • synchronous network, PKI
   • operate $f+1$ rounds, works for any $f < n$
   • 
   • Agreement: Whenever an honest node receives a valid value, it must propagate to other honest nodes
     1. send it out to all other honest nodes if not at the end of the last round, or
     2. some other node must have done it previously thanks to f+1 signatures (at least 1 honest node)
   • Generally optimized total communication = $O(n^2f)$

3. Practical Byzantine Fault Tolerance (PBFT)

   • partially synchronous network

   • settings

     • total $3f+1$ nodes: One distinct primary $p$ for a view $v$, multiple backups
       • uses supermajority quorum $2f + 1$
       • Nodes keep logs; different voting is done independently for each index
         • log entries: <index, op, status>. status = pre-prepared, prepared, committed
         • Once a value is committed at a particular index $i$ in the log, it stays committed forever
       • Voting: primary proposes value, all nodes vote on whether to commit value to specific index
         • pre-prepare message: $\{\sigma_p(\{ \text{status}, v, i, d \}), m\}$, status = pre-prepare; $v$ = view, $i$ = log index, $d$ = hash of message, $m$ = message
         • voting message: $\{\sigma_p(\{ \text{status}, v, i, d \}), m\}$, status = prepare / commit
       • 
       • 2 stages of voting: prepare & commit
         • first voting: safety within view; second: safety across view
         • voting idea: If a replica receives 2f + 1 total matching commit messages, it means f + 1 honest nodes are prepared

Lecture 24: Longest-chain consensus, blockchain

1. Permissioned vs. permissionless

   • Permissioned: Nodes running the protocol are known in advance
   • Permissionless: Anyone can join/leave, Has an unknown and possibly ever-changing set of nodes running consensus

2. Longest-chain consensus

   • Start with a hard-coded genesis block
   • In each “round”: Choose one node L as the leader of round r; Node L proposes a set of blocks, each specifying a single predecessor block
   • Blocks are gossiped to all nodes
   • Always pick the longest chain
   • Honest nodes extend the longest chain
   • can have multiple forks => honest nodes vote on which fork to pick by appending blocks
   • roll back: when longest chain change (small chance)

3. Consistency (safety) and liveness guarantee

   • byzantine nodes $f < n/2$
     • $k$ consecutive Byzantine leaders can roll back $k-1$ honest blocks
     • >50% Byzantine: can rollback arbitrary number of blocks by only working on their own fork
   • Genesis block exists (assumed in implementation)
   • Leader is randomly selected, leader selection can be easily verified and cannot be influenced, AND, Every block produced by round-r leader must claim as its predecessor some block that belongs to a previous round <= Proof-of-work
   • Consistency in terms of finality: If an honest node regards a block B as finalized at time t, then the block should never be rolled back
     • Whenever >50% of the nodes are honest, all blocks on the longest chain other than the last $k$ (defined by clients, likelihood of consecutive Byzantine leaders goes exponentially small as $k$ increase) can be considered finalized

4. Proof-of-work

   • Every node has a chance to be chosen as leader to extend chains
   • Chance is proportional to compute power <- Sybil attack does not work

5. Nakamoto consensus analysis