Rui's Blog
  • Rui's Blog/Paper Reading Notes - Introduction
  • Personal Blog
    • Personal Blog - Index
      • How to Create Picture-in-Picture Effect / Video Overlay for a Presentation Video
      • How to Do Your Part to Protect the Environment in Wisconsin
      • How to Get a Driver's License in Wisconsin
      • How to Travel from the U.S. to China onboard AA127 in June 2021
      • How to Transfer Credits Back to UW-Madison
      • Resources on Learning Academic Writing (for Computer Science)
    • Towards applying to CS Ph.D. programs
  • Machine Learning Systems
    • Machine Learning Systems - Index
      • MLSys Papers - Short Notes
      • [2011 NSDI] Dominant Resource Fairness: Fair Allocation of Multiple Resource Types
      • [2014 OSDI] Scaling Distributed Machine Learning with the Parameter Server
      • [2018 OSDI] Gandiva: Introspective Cluster Scheduling for Deep Learning
      • [2018 SIGCOMM] Chameleon: Scalable Adaptation of Video Analytics via Temporal and Cross-camera ...
      • [2018 NIPS] Dynamic Space-Time Scheduling for GPU Inference
      • [2019 ATC] Analysis of Large-Scale Multi-Tenant GPU Clusters for DNN Training Workloads
      • [2019 NSDI] Tiresias: A GPU Cluster Manager for Distributed Deep Learning
      • [2019 SOSP] ByteScheduler: A Generic Communication Scheduler for Distributed DNN Training ...
      • [2019 SOSP] PipeDream: Generalized Pipeline Parallelism for DNN Training
      • [2019 SOSP] Parity Models: Erasure-Coded Resilience for Prediction Serving Systems
      • [2019 NIPS] GPipe: Efficient Training of Giant Neural Networks using Pipeline Parallelism
      • [2019 SC] ZeRO: memory optimizations toward training trillion parameter models
      • [2020 OSDI] Gavel: Heterogeneity-Aware Cluster Scheduling Policies for Deep Learning Workloads
      • [2020 OSDI] AntMan: Dynamic Scaling on GPU Clusters for Deep Learning
      • [2020 OSDI] BytePS: A High Performance and Generic Framework for Distributed DNN Training
      • [2020 SIGCOMM] Reducto: On-Camera Filtering for Resource-Efficient Real-Time Video Analytics
        • [2020 MLSys] Salus: Fine-Grained GPU Sharing Primitives for Deep Learning Applications
      • [2020 EuroSys] AlloX: Compute Allocation in Hybrid Clusters
      • [2020 VLDB] PyTorch Distributed: Experiences on Accelerating Data Parallel Training
      • [2020 NetAI] Is Network the Bottleneck of Distributed Training?
      • [2020 NSDI] Themis: Fair and Efficient GPU Cluster Scheduling
      • [2021 MLSys] Accordion: Adaptive Gradient Communication via Critical Learning Regime Identification
      • [2021 VLDB] Analyzing and Mitigating Data Stalls in DNN Training
      • [2021 FAST] CheckFreq: Frequent, Fine-Grained DNN Checkpointing
      • [2021 EuroMLSys] Interference-Aware Scheduling for Inference Serving
      • [2021 OSDI] Pollux: Co-adaptive Cluster Scheduling for Goodput-Optimized Deep Learning
      • [2021 MLSys] Wavelet: Efficient DNN Training with Tick-Tock Scheduling
      • [2021 NSDI] SwitchML: Scaling Distributed Machine Learning with In-Network Aggregation
    • Big Data Systems - Index
      • Big Data Systems Papers - Short Notes
      • [2003 SOSP] The Google File System
      • [2004 OSDI] MapReduce: Simplified Data Processing on Large Clusters
      • [2010 SIGMOD] Pregel: A System for Large-Scale Graph Processing
      • [2011 NSDI] Mesos: A Platform for Fine-Grained Resource Sharing in the Data Center
      • [2012 NSDI] Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster ...
      • [2012 OSDI] PowerGraph: Distributed Graph-Parallel Computation on Natural Graphs
      • [2019 FAST] DistCache: Provable Load Balancing for Large-Scale Storage Systems with Distributed...
      • [2021 HotOS] From Cloud Computing to Sky Computing
      • [2021 EuroSys] NextDoor: Accelerating graph sampling for graph machine learning using GPUs
  • Earlier Readings & Notes
    • High Performance Computing Course Notes
      • Lecture 1: Course Overview
      • Lecture 2: From Code to Instructions. The FDX Cycle. Instruction Level Parallelism.
      • Lecture 3: Superscalar architectures. Measuring Computer Performance. Memory Aspects.
      • Lecture 4: The memory hierarchy. Caches.
      • Lecture 5: Caches, wrap up. Virtual Memory.
      • Lecture 6: The Walls to Sequential Computing. Moore’s Law.
      • Lecture 7: Parallel Computing. Flynn's Taxonomy. Amdahl's Law.
      • Lecture 8: GPU Computing Intro. The CUDA Programming Model. CUDA Execution Configuration.
      • Lecture 9: GPU Memory Spaces
      • Lecture 10: GPU Scheduling Issues.
      • Lecture 11: Execution Divergence. Control Flow in CUDA. CUDA Shared Memory Issues.
      • Lecture 12: Global Memory Access Patterns and Implications.
      • Lecture 13: Atomic operations in CUDA. GPU ode optimization rules of thumb.
      • Lecture 14: CUDA Case Studies. (1) 1D Stencil Operation. (2) Vector Reduction in CUDA.
      • Lecture 15: CUDA Case Studies. (3) Parallel Prefix Scan on the GPU. Using Multiple Streams in CUDA.
      • Lecture 16: Streams, and overlapping data copy with execution.
      • Lecture 17: GPU Computing: Advanced Features.
      • Lecture 18: GPU Computing with thrust and cub.
      • Lecture 19: Hardware aspects relevant in multi-core, shared memory parallel computing.
      • Lecture 20: Multi-core Parallel Computing with OpenMP. Parallel Regions.
      • Lecture 21: OpenMP Work Sharing.
      • Lecture 22: OpenMP Work Sharing
      • Lecture 23: OpenMP NUMA Aspects. Caching and OpenMP.
      • Lecture 24: Critical Thinking. Code Optimization Aspects.
      • Lecture 25: Computing with Supercomputers.
      • Lecture 26: MPI Parallel Programming General Introduction. Point-to-Point Communication.
      • Lecture 27: MPI Parallel Programming Point-to-Point communication: Blocking vs. Non-blocking sends.
      • Lecture 28: MPI Parallel Programming: MPI Collectives. Overview of topics covered in the class.
    • Cloud Computing Course Notes
      • 1.1 Introduction to Clouds, MapReduce
      • 1.2 Gossip, Membership, and Grids
      • 1.3 P2P Systems
      • 1.4 Key-Value Stores, Time, and Ordering
      • 1.5 Classical Distributed Algorithms
      • 4.1 Spark, Hortonworks, HDFS, CAP
      • 4.2 Large Scale Data Storage
    • Operating Systems Papers - Index
      • CS 736 @ UW-Madison Fall 2020 Reading List
      • All File Systems Are Not Created Equal: On the Complexity of Crafting Crash-Consistent Applications
      • ARC: A Self-Tuning, Low Overhead Replacement Cache
      • A File is Not a File: Understanding the I/O Behavior of Apple Desktop Applications
      • Biscuit: The benefits and costs of writing a POSIX kernel in a high-level language
      • Data Domain: Avoiding the Disk Bottleneck in the Data Domain Deduplication File System
      • Disco: Running Commodity Operating Systems on Scalable Multiprocessors
      • FFS: A Fast File System for UNIX
      • From WiscKey to Bourbon: A Learned Index for Log-Structured Merge Trees
      • LegoOS: A Disseminated, Distributed OS for Hardware Resource Disaggregation
      • LFS: The Design and Implementation of a Log-Structured File System
      • Lottery Scheduling: Flexible Proportional-Share Resource Management
      • Memory Resource Management in VMware ESX Server
      • Monotasks: Architecting for Performance Clarity in Data Analytics Frameworks
      • NFS: Sun's Network File System
      • OptFS: Optimistic Crash Consistency
      • RAID: A Case for Redundant Arrays of Inexpensive Disks
      • RDP: Row-Diagonal Parity for Double Disk Failure Correction
      • Resource Containers: A New Facility for Resource Management in Server Systems
      • ReVirt: Enabling Intrusion Analysis through Virtual-Machine Logging and Replay
      • Scheduler Activations: Effective Kernel Support for the User-Level Management of Parallelism
      • SnapMirror: File-System-Based Asynchronous Mirroring for Disaster Recovery
      • The Linux Scheduler: a Decade of Wasted Cores
      • The Unwritten Contract of Solid State Drives
      • Venti: A New Approach to Archival Storage
    • Earlier Notes
      • How to read a paper
  • FIXME
    • Template for Paper Reading Notes
Powered by GitBook
On this page
  • Lesson 1: P2P Systems
  • P2P Systems Introduction
  • Napster
  • Gnutella
  • FastTrack and BitTorrent
  • Chord
  • Failures in Chord
  • Pastry
  • Kelips
  • Summary
  • One of the questions I in the discussion thread

Was this helpful?

  1. Earlier Readings & Notes
  2. Cloud Computing Course Notes

1.3 P2P Systems

Lesson 1: P2P Systems

P2P Systems Introduction

  • Why study P2P systems?

    • P2P systems are the first distributed systems that seriously focused on scalability (w.r.t #nodes)

    • P2P techniques abound in cloud computing systems. E.g., key-value stores use Chord p2p hashing (consistent hashing)

Napster

  • When users upload files, the files are stored at client machines ("peers")

  • The Napster servers store directory information (a list of <filename, ip_addr, port_num>)

  • Napster search

    • Client sends server keywords to search with

    • Server searches (using ternary tree algorithm) and returns a list of hosts <ip_addr, port_num> to client

    • Client pings each host in the list to find transfer rates

    • Client fetches file from best host

  • All communication uses TCP

  • Joining a P2P system

    • Send an HTTP request to a well-known URL for that P2P service

    • Message routed to introducer, a well-known server that keeps track of some recently joined nodes in P2P system

    • Introducer initializes new peer's neighbor table

  • Problems

    • Central servers: source of congestion/single point of failure

    • No security: plain messages and passwords

    • Indirect infringement: responsible for users' copyright violation

Gnutella

  • Different from Napster, Gnutella eliminates the servers and have clients act as servers (servents), such that client machines search and retrieve amongst themselves

  • In the overlay graph (overlay in the sense that it is overlayed on top of the internet), peers being neighbors means that they know about each other's ip addr and port num, and can send them messages

  • Gnutella routes different messages within the overlay

  • There are five main message types in the Gnutella protocol

    • Query (search)

      • Queries are flooded out (forwarded to all peers except the peer from which the Query was received), TTL-restricted, and are forwarded only once

    • QueryHit (response to query)

      • A QueryHit messages contains:

        • Info about responder: <port, ip_addr, speed>

        • Results: <fileindex, filename, fsize>

        • servent_id: Unique identifier of responder (a function of its ip addr)

      • QueryHits are reverse-routed: If A sends B a Query and B got a hit, B sends to A a QueryHit

    • Push (used to initiate file transfer)

      • After QueryHits are received, the requestor chooses the "best" responder, and then initiates HTTP request directly to responder's ip_addr:host

      • IRL, responders may be behind firewalls that rejects incoming connections

      • If a HTTP request fails, it routes a Push message via links in the overlay. The Push message contains ip_addr:host at which the requestor can accept incoming connections. When the peer receives this Push message, it can generate an outgoing TCP connection (sends GIV, receives GET)

      • If the requestor is also behind a firewall, Gnutella gives up

        • Alternative: use a modified version of Gnutella to transfer the file via the overlay links themselves (this might be slow)

    • Ping (to probe network for other peers)

      • Peers initiate Pings periodically, and Pings are flooded out

    • Pong (reply to ping, contains address of another peer)

      • Pongs are routed along reverse paths

      • Pongs are used to keep neighbor lists fresh in spite of peers joining, leaving, and failing

  • Problems

    • Ping/Pong constitute 50% of the traffic

      • Solutions: Multiplex, cache, and reduce frequency

    • Repeated searches with same keywords

      • Solutions: Cache query, QueryHits

    • Modem-connected hosts do not have enough bandwidth for passing Gnutella traffic

      • Solution: Use a central server to act as proxy for such peers

      • Another solution: FastTrack

    • Large number of freeloaders (only download files, never upload files)

      • In 2000, 70% of the users are freeloaders

    • Flooding causes excessive traffic

      • To maintain meta info about peers in order for more intelligent routing, use structures P2P systems (e.g., Chord)

FastTrack and BitTorrent

FastTrack

  • Hybrid between Napster and Gnutella, takes advantages of "healthier" participants in the system

  • Like Gnutella, but designate some peers as "supernodes"

    • A supernode stores a directory listing a subset of nearby <filename, peer pointer> (similar to Napster servers)

    • Supernode membership changes over time

    • Any node may become a supernode, provided it has earned enough reputation

      • E.g., reputation is affected by length of periods of connectivity and total number of uploads

    • A peer searches by contacting a nearby supernode

BitTorrent

  • Files are split into blocks (32KB - 256KB)

  • Download Local Rarest First block policy: Prefers early download of blocks that are least replicated among neighbors

  • Tit for tat bandwidth usage: Provide blocks to neighbors that provided it the best download rates

    • Incentivizes nodes to provide good download rates

  • Choking: Limit number of neighbors to which concurrent uploads <= a number (5), i.e. the best neighbors. Everyone else is choked

    • Prevents overloading of the upload bandwidth

    • Periodically (e.g., 10s) re-evaluate this set

    • Optimistic unchoke: Periodically (e.g., 30s) unchoke a random neighbor to keep the unchoked set fresh

Chord

  • Distributed hash tables: objects = files

    • Performance concerns

      • Load balancing

      • Fault tolerance

      • Efficiency of lookups and inserts

      • Locality

    • Napster, Gnutella, and FastTrack are all DHTs

  • Chord: Consistent hashing on nodes' addresses

    • SHA-1(ip_addr, port) -> 160-bit string, truncated to m bits -> peer id

  • Each node stores peer pointers

    • Successors

    • Finger tables

      • Used for routing queries quickly

  • Consistent hashing: With K keys and N peers, each peer stores O(K/N) keys

  • Storing files

    • Filenames are also mapped using the same consistent hash function

    • File is stored at first peer with id greater than or equal to its key (mod 2^m)

  • Searching files

    • Takes O(log(N)) time

Failures in Chord

  • Solution 1: Maintain multiple (2log(N)) successor entries

  • Solution 2: Replicate file/key at r successors and predecessors

  • Dealing with dynamic changes (P2P systems have a high rate of churn: peers joining, leaving, and failing)

    • Stabilization protocol is run by all nodes periodically (talk to neighbors to update finger table)

    • New peers may need to copy some files/keys from other nodes

    • A new peer affects O(log(N)) other finger entries in the system

      • Number of messages per peer join = O(log(N) * log(N))

    • Concurrent peer joins/leaves/failures

      • Chord peers periodically run a stabilization algorithm that checks and updates pointers and keys, which ensures non-loopiness

    • Hash can get non-uniform -> bad load balancing

      • Solution: Virtual nodes (treat each node as multiple virtual nodes behaving independently)

Pastry

  • Just like Chord, assigns ids to nodes using a virtual ring

  • Leaf set: Each node knows its successors and predecessors

  • Routing table: Instead of "n+2^i" rule in Chord, use prefix matching -> log(N)

    • Consider a peer with id 01110100101. It maintains a neighbor peer with an id matching each of the following prefixes: {*, 0*, 01*, 011*, ..., 0111010010*}

      • For each prefix, among all the potential neighbors, the neighbor with the shortest RTT is selected

      • Early hops/shorter prefixes have many more candidates -> likely to be closer -> hops are short, yet overall stretch (compared to direct Internet paths) stays short

    • When it needs to route to a peer (e.g., 01110111001), it forwards to a neighbor with the largest matching prefix (011101*)

  • Problems

    • O(log(N)) lookup hops may be high

Kelips

  • Constant lookup cost to DHT

  • Instead of virtual rings, we use k (~= sqrt(N)) affinity groups

  • Each node is hashed (mod k) to a group

  • A peer is neighbors with all other nodes in its affinity group

  • Files are stored at whichever node uploaded them

    • Kelips decouples file replication/location from querying

    • Each filename hashed to a group

    • All nodes in the group replicate pointer information (i.e., <filename, location>)

  • Lookup

    • Find affinity group

    • Go to your contact for the file affinity group

      • If fails, try another neighbor to find a contact

    • Lookup = 1 hop (or a few under failures)

    • Memory cost: O(sqrt(N))

      • 1.93MB for 100K nodes, 10M files

Summary

  • Chord & Pastry & Kelips

    • Range of tradeoffs (memory vs. lookup cost vs. background bandwidth (in order to keep neighbors fresh))

      • Chord & Pastry use O(log(N)) for both memory & lookup

      • Kelips uses more memory (O(N^2)) & background bandwidth to provide O(1) lookup

    • All of them have provable properties

One of the questions I in the discussion thread

Hi, I have a question regarding question 8 in HW 3. My reasoning is as follows. Going down, we have 3 at level 3, 9 at level 4, 27 at level 5, making a total of 39. Going up, we have 1 at level 1, 2 at level 2, 6 at level 3, making a total of 9. Adding them up, it should be 48. I would appreciate it if someone can point out the mistake in my reasoning. Thanks in advance!

To provide more context, the question is:

A Gnutella topology looks like a balanced ternary tree with 4 levels of nodes, i.e., peers, as shown in the picture below. Thus, there is 1 root at Level 1, which has 3 children at Level 2, which each have 3 children at Level 3, which in turn each have 3 children at Level 4 – thus, there are a total of 40 nodes. If a child of the root (i.e., a Level 2 node in the tree) sends a Query message with TTL=3, then what are the number of nodes receiving the Query message, not including the originating node? Enter your answer as a numeric value in the text box below. (1 point)

Previous1.2 Gossip, Membership, and GridsNext1.4 Key-Value Stores, Time, and Ordering

Last updated 3 years ago

Was this helpful?

Original paper
Napster structure
Gnuteella message header format
FastTrack structure
BitTorrent structure
Comparative performance
Finger tables
Chord searching
Yes, more math
Kelips structure
Kelips soft state