P2P 系统

Peer to peer 系统：

• 每个结点在连接上是互联的，在功能上是平等的，在行为上是自由的
• 每个结点既是服务使用者，也是服务提供者
• 通常构建有高效的覆盖网 (overlay)，允许结点动态地加入和离开
• 每个结点通过冗余机制、周期性检测等，提供容错性

P2P系统的特点：结点自治性，控制分散性，高容错性，高可伸缩性，高可用性，负载平衡

Functional requirements:

• Locate and communicate with any individual resource：对资源的定位、交互
• Add new resources or remove them at will：资源的加入、离开
• Add hosts or remove them at will：主机的加入、离开
• Provide simple APIs to store and find data：存储、查找数据
  • Typical DHT interface: key with a GUID
  • $put(key,value)$, $get(key)\to value$

Non-functional requirements:

• Global scalability：可伸缩性，可扩展性
• Load balancing：负载平衡
• Accommodating to highly dynamic host availability：动态的主机可用性
• Optimization for local interactions between neighboring peers：与邻居交互的优化
• Security of data in an environment with heterogeneous trust：数据安全
• Anonymity, deniability and resistance to censorship：匿名、否认、抵制审查

现有的一些 P2P Overlays：

• Server-driven，
  • Napster: a music file sharing system
  • BitTorrent: a file sharing system
• Unstructured，
  • Gnutella: a file sharing system
• Structured，
  • Pastry, Tapestry: DHT, Prefix routing
  • Chord: DHT, a ring (an artificial one-dimensional space)
  • CAN: DHT, d-dimensional Cartesian coordinate space
  • Kademlia (eMule, BitTorrent without tracker，Ethereum / Block Chain)

Napster

使用方式：

1. 资源提供方，

   • 用户连接到服务器集群中的一台服务器，把他愿意与其它用户共享的文件信息发送给服务器，服务器根据这些信息和用户的位置，建立索引并加入到原有的索引表中

2. 资源使用方，

   • 用户发查询请求 Q 给与其相连的服务器，该服务器收到请求后，与其他服务器协作处理查询消息 Q，回复用户一个表单，这个表单包含了所查到的所有匹配的文件索引

   • 用户收到回复 R 后，选择他想要的文件，根据文件索引中对应的位置与其他用户直接建立连接并下载文件

服务器的作用：

• 维护所有用户的共享文件索引

• 监控用户的状态

BitTorrent

BitTorrent 由 BT 网站、.torrent 文件服务器、跟踪器 (Tracker)、BT 用户组成。

• BitTorrent 中，一个文件分割成固定大小的块 (chunk)，对应一个 .torrent 文件(种子文件)

  • 文件的名字和长度，下载次数、种子数、上载文件的人
  • Tracker 的位置 (用一个 URL 指定)
  • 与每个块相关的校验和

• BT 网站 (有一部分种子文件)，供用户搜索 .torrent 文件列表

• Tracker 保存该文件的所有下载者 (downloader) 和种子 (seed) 的注册信息，同时管理多个文件的并发下载

使用流程：用户请求 $\to$ BT网站 $\to$ .torrent 文件服务器 $\to$ .torrent 文件 $\to$ Tracker $\to$ 返回下载该文件的 peer 信息，

• BT 用户与那些 peers 直接相连，进行文件下载 (barter for chunks of the file)

• 下载同一个文件的用户围绕 Tracker 形成一个独立的子网

BitTorrent 将文件分片 (piece)，分片又被划分成子分片，子分片进行流水作业

• Piece Selection: Rarest First（稀有的分片，首先被传递，防止 peer 退出后失去资源）
• Piece Selection: Endgame Mode（快完成的文件，被系统竭尽全力地传递，防止功亏一篑）

一报还一报 (tit-for-tat) 的激励机制，

• 要求 peer 合作，合作，意味着 peer 要上传。不合作的话，就阻塞 peer

• Peer 本地提供基于 Pareto-efficient 的阻塞算法：每隔一定周期 (10秒) 计算下载率，下载率高的 peer 就不阻塞

• Optimistic unchoking：每隔一定周期 (30秒) 随机选择一个用户实施不阻塞

在 P2P 系统上使用 Merkle Tree：

1. 校验网络数据的完整性，确保数据块没有损坏
2. 比较两台机器上文件的异同，发现不一致的副本，减少需传输的数据量

Gnutella

Gnutella协议包含下列消息：

• Ping，Pong：检查结点是否在线

• Query，QueryResponse：

  • 文件搜索的洪泛 (Flooding) 策略：每个结点给它的每个邻居转发请求，这些邻居结点再依次把请求传递给它们的邻居，一直到找到匹配的文件为止

  • 每条消息有一个 TTL (time-to-live)，以限制洪泛

  • 每个结点缓存最近路由的消息，以支持汇聚 (沿洪泛的反向路径回传消息) ，阻止不必要的重复广播

• Get，Put：请求获取文件

改进：

• （(改进1）引入了 Ultrapeer 超级结点：
  1. 超级结点间连接紧密 (每个都有超过32个连接)，其他一些对等结点承担叶子结点的角色，
  2. 这大大减少了进行彻底搜索所需求的最大跳数。
• （改进2）改进了QRP (Query Routing Protocol)：
  1. 结点生成 QRT (Query Routing Table)，它包含代表结点上的每个文件的 Hash 值；
  2. 接着，发送 QRT 给所有与它相连的超级结点，超级结点基于所有相连的叶子结点的所有项，加上自身包含的文件的项，形成它们自己的 QRT，并与其他相连的超级结点交换 QRT；
  3. 这样，超级结点能对一个给定的查询，决定哪个路径能提供一个有效的路由，从而大大减少了不必要的流量。

Chord

Consistent Hashing

一致性哈希：

1. 均衡性 (Balance)：哈希的结果能够尽可能分布到所有的缓冲中
2. 单调性 (Monotonicity)：当缓冲区大小变化时，应尽量保护已分配的内容不会被重新映射到新缓冲区（Hash 函数与 Buffer 大小无关，类似 $x\mapsto ax+b(\mathrm{mod}n)$ 这种差劲的函数就不要用了，应当使用密码学 Hash 函数！）
3. 分散性 (Spread)：避免由于不同终端所见的缓冲范围有可能不同，从而导致相同内容被映射到不同缓冲区
4. 负载 (Load)：对于一个特定的缓冲区而言，避免被不同的用户映射为不同的内容

将 Hash 空间，组织成虚拟的圆环（长度 $m=32$ 比特的无符号整数），

• Node：节点的 GUID 经过 Hash 函数，计算出节点的 Chord Key，这就是这个节点在 ring 上的位置
• Object：对象的 GUID 经过 Hash 函数，计算出数据的 key 对应的 Chord Key，把它存放在后继的最近的那个 node 上
• 加入、退出：需要与自己的后继 node 联系，调整这个所的负责的 objects

另外，为了解决数据倾斜，采取虚拟节点机制：

Chord 的特点：

• Simplicity, provable correctness, and provable performance：简单、可证明正确性、可证明性能
• 源于一致性哈希算法：
  • Load balance（负载均衡），分布式 Hash 函数，使得 keys 均匀分布在 nodes 上
  • Decentralization（去中心化），完全地分布式
  • Scalability（延展性），查找（lookup）的复杂度仅仅是关于节点数量的对数
  • Availability（可用性），会自动调整 internal tables，确保总是能找到负责某个 key 的节点
  • Flexible naming（灵活的命名），对 keys 的结构没有任何限制，灵活地将 keys 映射到 Chord keys 上

Chord Ring：

1. 使用 SHA-1 对 IP address 做运算，构建一个 $m$ 比特的 node’s identifier
2. 一个 Chord key $k$，被存放到（逻辑上）不比它小的第一个节点上，叫做后继节点 $successor(k)$
3. 一个 Chord key $k$，（逻辑上）比它小的第一个节点，叫做前任节点 $predecessor(k)$

Simple Key Location

一个节点 $n$ 查询 $id$ 的定位，直接询问自己的后继 $n.successor$，函数 $n.FindSuccessor(id)$：

1. if $id\in (n.id,successor.id]$
2. then return $successor$
3. else return $successor.FindSuccessor(id)$

正确但低效！

Scalable Key Location

为了加速查找，给予每个节点更多先验知识：Finger Table（将地址空间分区，大小依次为 $2^0,2^1,\cdots ,2^{m-1}$）

1. 表格 $finger[1\cdots m]$ 包含 $m$ 项，每一项形如 $(start,int,succ)$
2. 节点 $n$ 上，$finger[i]$ 记录了信息 $\left(n.id+2^{i-1}(\mathrm{mod}{2}^m),n.id+2^{i-1},successor(n.id+2^{i-1})\right)$
   • $start$ 记录第 $i$ 区间的起始地址
   • $succ$ 记录第 $i$ 区间里的第一个节点

算法为：

• 函数 $n.FindSuccessor(id)$：
  1. if $id\in (n.id,successor.id]$
  2. then return $successor$
  3. else
     1. $n^{\prime }:=ClosestPrecedingNode(id)$
     2. return $n^{\prime }.FindSuccessor(id)$
• 函数 $n.ClosestPrecedingNode(id)$：
  1. for $i=m,\cdots ,1$
  2. if $finger[i]\in (n.id,id)$ then return $finger[i]$
  3. return $n$

节点的加入（Join）、退出（Departure），都需要修改 Finger Table，保证记录的 $succ$ 是正确的。

故障恢复：Key step in failure recovery is maintaining correct successor pointers.

• 每个 node 维护一个后继列表（successor-list），包含 $r$ 个最近的后继节点
• 一旦发现自己的直接后继 fail 了，就使用后继列表中的下一个活结点（the first live entry ）取代它
• 自稳定算法：使得 finger table 以及 successor-list 重新成为正确的

性能：

1. 给定 $N$ 个 nodes 以及 $K$ 个 keys，每个节点负责大约 $N/K$ 个数据。
2. Lookup 的复杂度仅为 $O(\log N)$，因为消息每一跳都距离至少减半。
3. 当 node 离开或者加入时，只需要换手（keys change hands）大约 $N/K$ 个数据。
4. 当 node 离开或者加入时，调整 finger-table 以及 successor-list 还有 routing invariants，只需要发送大约 $O({\log }_{2}N)$ 个消息。

Kademlia

系统中，

1. 每个结点根据其IP地址及端口分配一个唯一的、随机的 $160$ bits 的整数 nodeID
2. 每个对象被分配一个 key，也是 $160$ bits 的整数。
3. 数据形如 $⟨key,value⟩$，被存放在距离 $key$ 最近的 nodeID 上。

下面，我们定义结点之间的距离：结点的 nodeID 是 $x,y$，距离定义为 $dist(x,y):=x\text{ XOR }y$

• 非负性：$dist(x,y)\ge 0,\forall x,y$
• 正定性：$dist(x,y)=0⟺x=y$
• 对称性：$dist(x,y)=dist(y,x),\forall x,y$
  • 异或运算是对称的，结点可以从它收到的查询请求中学习到有用的路由信息。
• 三角不等式：$dist(x,y)+dist(y,z)\ge dist(x,z)$
• 单向性：对于任意点 $x$ 和距离 $d>0$，有且仅有唯一的点 $y$，满足 $dist(x,y)=d$
  • 单向性保证了对相同数据对象的定位最终将收敛于相同的路径，所以，用 “沿路径缓存” 能提高查找效率、缓解热点。

Routing Table

将所有节点按照 nodeID 组织成一颗二叉树（世界），一个节点对应一个叶子。每个节点都按照距离大小，对世界树进行分块，成为一系列的连续的、不包含自己的子树。

我们以子树的根节点的路径为子树命名。例如，节点 $0011$，把世界划分为 $1,01,000,0010$ 四个分块。

然后，每个节点 $n$ 维护一个路由表，

1. 包含 $160$ 项，第 $i=0,1\cdots ,159$ 项对应一个世界分块，就是以 $(n\oplus (1\ll i))\gg i$ 为树根的那个小世界。
2. 每一项都是一个大小为 $K$ 的列表（$K$ 桶），存放这个小世界中的至多 $K$ 个随机节点的信息。
3. 表头：最不经常访问的节点
4. 表尾：刚访问过的节点

易知，这些 $K$ 桶无交叠地覆盖了整个世界。Kademlia 路由表确保每个节点知道各个子树的至少一个节点。

路由表的维护：捎带更新

1. 节点 $n$，每当收到来自其他节点 $m$ 的信息时，利用它的 nodeID 来更新自己的 $K$ 桶
2. 找到 $dist(n,m)\in [2^i,2^{i+1})$ 对应的 $i:=\lfloor {\log }_{2}dist(n,m)\rfloor$，然后维护第 $i$ 个桶
3. 如果桶里已经有 $m$ 了，那么就把 $m$ 移到尾部
4. 如果桶里没有 $m$，同时列表长度小于 $K$，那么把 $m$ 追加到尾部
5. 如果桶里没有 $m$，同时列表长度达到 $K$，那么联系表头节点
   1. 如果联系到了，那么把表头移动到到尾部
   2. 如果联系不到，那么删除表头，然后把 $m$ 追加到尾部

Kademlia’s RPC

Kademlia 协议的 $4$ 中 RPC 操作：

• $Ping(nodeID)$：检测节点是否在线
• $Store(key,value)$：把数据存储到 P2P 系统里，其中 $key$ 是 Hash 值
• $FindNode(ID)$：获取距离 $ID$ 最近的 $k$ 个节点 $⟨\text{IP address},\text{UDP port},\text{nodeID}⟩$
• $FindValue(key)$：若消息的接收者曾经收到过 $Store(key,\cdot )$ 指令，那么就返回 $value$，同时把 $(key,value)$ 存储到据它所知的离 $key$ 最近的、没有返回 $value$ 的节点中

$FindNode(ID)$ 算法，采取连续查询策略：

1. 查询发起者 $n$ 计算距离 $d:=dist(n,ID)$
2. 设置 $i:=\lfloor {\log }_{2}d\rfloor$，从第 $i$ 个桶中取出任意的 $\alpha$ 个节点 $m$（如果不足 $\alpha$ 个，就从附近的桶中选择一些距离接近 $d$ 的节点），分别执行 $m.FindNode(ID)$
3. 那些接收到指令的节点 $m$，首先判断自己是否是 $ID$ 对应的节点
   1. 如果是，那么回应自己就是最接近的节点
   2. 否则，从桶中选择 $\alpha$ 个节点，把这些节点的信息回复给 $n$
4. 节点 $n$ 把这些节点收集起来，找出最近的前 $k$ 个节点。如果节点 $n$ 没有收到某个节点 $m$ 的回复，那么就把它从桶中删除。
5. 节点 $n$ 对于收到的每个节点，继续执行 $m.FindNode(ID)$ 操作，直到无法获得更近的 $k$ 个节点为止。
6. 返回这 $k$ 个节点。

$Store(key,value)$ 算法，采取邻近复制策略：

1. 发布者 $n$ 首先执行 $FindNode(key)$，获得最近的 $k$ 个节点 $m$，
2. 然后在这些节点上执行 $m.Store(key,value)$，把数据存到这些节点上。
3. 上述的 $k$ 个节点每经过 $1$ 小时，就重新发布 $(key,value)$，保证数据持续可用。
4. 而发布者 $n$ 每经过 $24$ 小时，就重新发布 $(key,value)$，否则全部的数据副本都会过期。
5. 任意一个节点发现了另一个更近的节点 $u$ 时，就执行 $u.Store(key,value)$，同时并不删除自己保存的数据副本。

Adaptability

Kademlia 网络的自适应性：

1. 更新：如果路由表的某个 $K$ 桶，在 $1$ 小时内未被查询过，那么就随机选一个 nodeID 执行 $FindNode(nodeID)$，刷新这个 $K$ 桶。
2. 加入：
   1. 新节点 $u$ 加入 P2P 网络，那么联系一个活节点 $w$，把 $w$ 放入自己的 $K$ 桶
   2. 然后执行 $w.FindNode(u.id)$，查询离自己最近的 $k$ 个节点，来初始化自己的 $K$ 桶
   3. 最后，把自身状态告诉这些节点，以更新它们的路由表
3. 退出：
   1. 节点不需要发布任何消息
   2. 每个节点周期性地发布自己的数据，把这些数据存放到自己的 $k$ 个邻居上

Distributed Hash Table

分布式 Hash 表（DFT）：

• Object key 以及 GUID 使用密码学 Hash 函数来计算
• 给定带有 nodeID 的节点，把它们组织成一个 ID-Space（1D line/ring, 2D square, tree based on bits, hypercube）
• 定义规则（rule），把 key 和 nodeID 联系起来（例如把 key 放在最接近的 nodeID 上）
• 建立路由表，记录自己的邻居（ID-space neighbors）以及远点（farther-away nodes）
• 对于 $N$ 个 nodes 和 $K$ 个 keys，高概率负载平衡：
  • 每个节点负责至多 $(1+O(\log N))\cdot K/N$ 个数据；
  • 第 $N+1$ 个节点加入和离开时，只有 $O(K/N)$ 个数据换手。