fix: increase ping concurrency (#341)

After running an js-ipfs node for a week I noticed the lan routing table ping queue had 200k items in it (😱).  I think this is because we encounter lots of peers that are advertising their private addresses, they get added to the routing table, then we need to ping existing peers to see if we should evict them, which times out because the private addresses aren't routable.  We do this with `concurrency: 1` so it takes 10s to process each item in the queue but we are adding them to the queue at a much faster rate than that so it grows out of control.

Increasing the concurrency alleviates the queue pressure, so here we increase the default concurrency to 10 and expose it as a config option, as well as the timeout value to allow tuning by the user for their specific application.

We also add the ping queue size and current in-flight ping requests to the list of tracked metrics for the node.

Author: achingbrain

package.json

@@ -133,7 +133,7 @@
   },
   "dependencies": {
     "@libp2p/crypto": "^0.22.12",
-    "@libp2p/interfaces": "^2.0.1",
+    "@libp2p/interfaces": "^2.0.2",
     "@libp2p/logger": "^1.1.4",
     "@libp2p/peer-id": "^1.1.10",
     "@libp2p/record": "^1.0.4",
@@ -167,7 +167,7 @@
     "varint": "^6.0.0"
   },
   "devDependencies": {
-    "@libp2p/interface-compliance-tests": "^2.0.1",
+    "@libp2p/interface-compliance-tests": "^2.0.3",
     "@libp2p/peer-id-factory": "^1.0.9",
     "@libp2p/peer-store": "^1.0.11",
     "@types/lodash.random": "^3.2.6",

src/index.ts

@@ -3,13 +3,48 @@ import { DualKadDHT } from './dual-kad-dht.js'
 import type { Selectors, Validators } from '@libp2p/interfaces/dht'
 
 export interface KadDHTInit {
+  /**
+   * How many peers to store in each kBucket (default 20)
+   */
   kBucketSize?: number
+
+  /**
+   * Whether to start up as a DHT client or server
+   */
   clientMode?: boolean
+
+  /**
+   * Record selectors
+   */
   selectors?: Selectors
+
+  /**
+   * Record validators
+   */
   validators?: Validators
+
+  /**
+   * How often to query our own PeerId in order to ensure we have a
+   * good view on the KAD address space local to our PeerId
+   */
   querySelfInterval?: number
-  lan?: boolean
+
+  /**
+   * A custom protocol prefix to use (default: '/ipfs')
+   */
   protocolPrefix?: string
+
+  /**
+   * How long to wait in ms when pinging DHT peers to decide if they
+   * should be evicted from the routing table or not (default 10000)
+   */
+  pingTimeout?: number
+
+  /**
+   * How many peers to ping in parallel when deciding if they should
+   * be evicted from the routing table or not (default 10)
+   */
+  pingConcurrency?: number
 }
 
 export class KadDHT extends DualKadDHT {

src/kad-dht.ts

@@ -26,6 +26,13 @@ import { validators as recordValidators } from '@libp2p/record/validators'
 import { selectors as recordSelectors } from '@libp2p/record/selectors'
 import { symbol } from '@libp2p/interfaces/peer-discovery'
 
+export interface SingleKadDHTInit extends KadDHTInit {
+  /**
+   * Whether to start up in lan or wan mode
+   */
+  lan?: boolean
+}
+
 /**
  * A DHT implementation modelled after Kademlia with S/Kademlia modifications.
  * Original implementation in go: https://github.com/libp2p/go-libp2p-kad-dht.
@@ -56,7 +63,7 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
   /**
    * Create a new KadDHT
    */
-  constructor (init: KadDHTInit) {
+  constructor (init: SingleKadDHTInit) {
     super()
 
     const {
@@ -66,7 +73,9 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
       selectors,
       querySelfInterval,
       lan,
-      protocolPrefix
+      protocolPrefix,
+      pingTimeout,
+      pingConcurrency
     } = init
 
     this.running = false
@@ -77,7 +86,9 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
     this.clientMode = clientMode ?? true
     this.routingTable = new RoutingTable({
       kBucketSize,
-      lan: this.lan
+      lan: this.lan,
+      pingTimeout,
+      pingConcurrency
     })
 
     this.providers = new Providers()

src/network.ts

@@ -97,7 +97,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
 
     try {
       const connection = await this.components.getConnectionManager().openConnection(to, options)
-      const streamData = await connection.newStream(this.protocol)
+      const streamData = await connection.newStream(this.protocol, options)
       stream = streamData.stream
 
       const response = await this._writeReadMessage(stream, msg.serialize(), options)
@@ -134,7 +134,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
 
     try {
       const connection = await this.components.getConnectionManager().openConnection(to, options)
-      const data = await connection.newStream(this.protocol)
+      const data = await connection.newStream(this.protocol, options)
       stream = data.stream
 
       await this._writeMessage(stream, msg.serialize(), options)

src/routing-table/index.ts

@@ -37,11 +37,14 @@ export interface KBucketTree {
 }
 
 const METRIC_ROUTING_TABLE_SIZE = 'routing-table-size'
+const METRIC_PING_QUEUE_SIZE = 'ping-queue-size'
+const METRIC_PING_RUNNING = 'ping-running'
 
 export interface RoutingTableInit {
   lan: boolean
   kBucketSize?: number
   pingTimeout?: number
+  pingConcurrency?: number
 }
 
 /**
@@ -57,18 +60,38 @@ export class RoutingTable implements Startable, Initializable {
   private components: Components = new Components()
   private readonly lan: boolean
   private readonly pingTimeout: number
+  private readonly pingConcurrency: number
   private running: boolean
 
   constructor (init: RoutingTableInit) {
-    const { kBucketSize, pingTimeout, lan } = init
+    const { kBucketSize, pingTimeout, lan, pingConcurrency } = init
 
     this.log = logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:routing-table`)
     this.kBucketSize = kBucketSize ?? 20
     this.pingTimeout = pingTimeout ?? 10000
+    this.pingConcurrency = pingConcurrency ?? 10
     this.lan = lan
-    this.pingQueue = new Queue({ concurrency: 1 })
     this.running = false
 
+    const updatePingQueueSizeMetric = () => {
+      this.components.getMetrics()?.updateComponentMetric({
+        system: 'libp2p',
+        component: `kad-dht-${this.lan ? 'lan' : 'wan'}`,
+        metric: METRIC_PING_QUEUE_SIZE,
+        value: this.pingQueue.size
+      })
+      this.components.getMetrics()?.updateComponentMetric({
+        system: 'libp2p',
+        component: `kad-dht-${this.lan ? 'lan' : 'wan'}`,
+        metric: METRIC_PING_RUNNING,
+        value: this.pingQueue.pending
+      })
+    }
+
+    this.pingQueue = new Queue({ concurrency: this.pingConcurrency })
+    this.pingQueue.addListener('add', updatePingQueueSizeMetric)
+    this.pingQueue.addListener('next', updatePingQueueSizeMetric)
+
     this._onPing = this._onPing.bind(this)
   }
 
@@ -127,12 +150,14 @@ export class RoutingTable implements Startable, Initializable {
             try {
               timeoutController = new TimeoutController(this.pingTimeout)
 
-              this.log('pinging old contact %p', oldContact.peer)
-              const connection = await this.components.getConnectionManager().openConnection(oldContact.peer, {
+              const options = {
                 signal: timeoutController.signal
-              })
-              const { stream } = await connection.newStream(PROTOCOL_DHT)
-              await stream.close()
+              }
+
+              this.log('pinging old contact %p', oldContact.peer)
+              const connection = await this.components.getConnectionManager().openConnection(oldContact.peer, options)
+              const { stream } = await connection.newStream(PROTOCOL_DHT, options)
+              stream.close()
               responded++
             } catch (err: any) {
               if (this.running && this.kb != null) {

src/rpc/index.ts

@@ -93,9 +93,9 @@ export class RPC implements Initializable {
       const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
       await pipe(
-        stream.source,
+        stream,
         lp.decode(),
-        source => (async function * () {
+        async function * (source) {
           for await (const msg of source) {
             // handle the message
             const desMessage = Message.deserialize(msg)
@@ -107,9 +107,9 @@ export class RPC implements Initializable {
               yield res.serialize()
             }
           }
-        })(),
+        },
         lp.encode(),
-        stream.sink
+        stream
       )
     })
       .catch(err => {