Fluffy: Make the Kademlia concurrency factor configurable via a debug parameter (#2868)

fluffy/conf.nim

@@ -45,6 +45,7 @@ const
   defaultTableIpLimitDesc* = $defaultPortalProtocolConfig.tableIpLimits.tableIpLimit
   defaultBucketIpLimitDesc* = $defaultPortalProtocolConfig.tableIpLimits.bucketIpLimit
   defaultBitsPerHopDesc* = $defaultPortalProtocolConfig.bitsPerHop
+  defaultAlphaDesc* = $defaultPortalProtocolConfig.alpha
   defaultMaxGossipNodesDesc* = $defaultPortalProtocolConfig.maxGossipNodes
   defaultRpcApis* = @["eth", "portal"]
   defaultRpcApisDesc* = "eth,portal"
@@ -254,6 +255,14 @@ type
       name: "bits-per-hop"
     .}: int
 
+    alpha* {.
+      hidden,
+      desc: "The Kademlia concurrency factor",
+      defaultValue: defaultPortalProtocolConfig.alpha,
+      defaultValueDesc: $defaultAlphaDesc,
+      name: "debug-alpha"
+    .}: int
+
     maxGossipNodes* {.
       hidden,
       desc: "The maximum number of nodes to send content to during gossip",

fluffy/fluffy.nim

@@ -181,9 +181,9 @@ proc run(
   ## Portal node setup
   let
     portalProtocolConfig = PortalProtocolConfig.init(
-      config.tableIpLimit, config.bucketIpLimit, config.bitsPerHop, config.radiusConfig,
+      config.tableIpLimit, config.bucketIpLimit, config.bitsPerHop, config.alpha,
-      config.disablePoke, config.maxGossipNodes, config.contentCacheSize,
+      config.radiusConfig, config.disablePoke, config.maxGossipNodes,
-      config.disableContentCache, config.maxConcurrentOffers,
+      config.contentCacheSize, config.disableContentCache, config.maxConcurrentOffers,
     )
 
     portalNodeConfig = PortalNodeConfig(

fluffy/network/wire/portal_protocol.nim

@@ -116,7 +116,6 @@ logScope:
   topics = "portal_wire"
 
 const
-  alpha = 3 ## Kademlia concurrency factor
   enrsResultLimit* = 32 ## Maximum amount of ENRs in the total Nodes messages
   ## that will be processed
   refreshInterval = 5.minutes ## Interval of launching a random query to
@@ -1035,14 +1034,15 @@ proc lookup*(
   for node in closestNodes:
     seen.incl(node.id)
 
-  var pendingQueries = newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](alpha)
+  var pendingQueries =
+    newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](p.config.alpha)
   var requestAmount = 0'i64
 
   while true:
     var i = 0
-    # Doing `alpha` amount of requests at once as long as closer non queried
+    # Doing `p.config.alpha` amount of requests at once as long as closer non queried
     # nodes are discovered.
-    while i < closestNodes.len and pendingQueries.len < alpha:
+    while i < closestNodes.len and pendingQueries.len < p.config.alpha:
       let n = closestNodes[i]
       if not asked.containsOrIncl(n.id):
         pendingQueries.add(p.lookupWorker(n, target))
@@ -1156,17 +1156,18 @@ proc contentLookup*(
   for node in closestNodes:
     seen.incl(node.id)
 
-  var pendingQueries =
+  var pendingQueries = newSeqOfCap[
-    newSeqOfCap[Future[PortalResult[FoundContent]].Raising([CancelledError])](alpha)
+    Future[PortalResult[FoundContent]].Raising([CancelledError])
+  ](p.config.alpha)
   var requestAmount = 0'i64
 
   var nodesWithoutContent: seq[Node] = newSeq[Node]()
 
   while true:
     var i = 0
-    # Doing `alpha` amount of requests at once as long as closer non queried
+    # Doing `p.config.alpha` amount of requests at once as long as closer non queried
     # nodes are discovered.
-    while i < closestNodes.len and pendingQueries.len < alpha:
+    while i < closestNodes.len and pendingQueries.len < p.config.alpha:
       let n = closestNodes[i]
       if not asked.containsOrIncl(n.id):
         pendingQueries.add(p.findContent(n, target))
@@ -1277,8 +1278,9 @@ proc traceContentLookup*(
     metadata["0x" & $cn.id] =
       NodeMetadata(enr: cn.record, distance: p.distance(cn.id, targetId))
 
-  var pendingQueries =
+  var pendingQueries = newSeqOfCap[
-    newSeqOfCap[Future[PortalResult[FoundContent]].Raising([CancelledError])](alpha)
+    Future[PortalResult[FoundContent]].Raising([CancelledError])
+  ](p.config.alpha)
   var pendingNodes = newSeq[Node]()
   var requestAmount = 0'i64
 
@@ -1286,9 +1288,9 @@ proc traceContentLookup*(
 
   while true:
     var i = 0
-    # Doing `alpha` amount of requests at once as long as closer non queried
+    # Doing `p.config.alpha` amount of requests at once as long as closer non queried
     # nodes are discovered.
-    while i < closestNodes.len and pendingQueries.len < alpha:
+    while i < closestNodes.len and pendingQueries.len < p.config.alpha:
       let n = closestNodes[i]
       if not asked.containsOrIncl(n.id):
         pendingQueries.add(p.findContent(n, target))
@@ -1439,11 +1441,12 @@ proc query*(
   for node in queryBuffer:
     seen.incl(node.id)
 
-  var pendingQueries = newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](alpha)
+  var pendingQueries =
+    newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](p.config.alpha)
 
   while true:
     var i = 0
-    while i < min(queryBuffer.len, k) and pendingQueries.len < alpha:
+    while i < min(queryBuffer.len, k) and pendingQueries.len < p.config.alpha:
       let n = queryBuffer[i]
       if not asked.containsOrIncl(n.id):
         pendingQueries.add(p.lookupWorker(n, target))

fluffy/network/wire/portal_protocol_config.nim

@@ -38,6 +38,7 @@ type
   PortalProtocolConfig* = object
     tableIpLimits*: TableIpLimits
     bitsPerHop*: int
+    alpha*: int
     radiusConfig*: RadiusConfig
     disablePoke*: bool
     maxGossipNodes*: int
@@ -53,11 +54,13 @@ const
   defaultContentCacheSize* = 100
   defaultDisableContentCache* = false
   defaultMaxConcurrentOffers* = 50
+  defaultAlpha* = 3
   revalidationTimeout* = chronos.seconds(30)
 
   defaultPortalProtocolConfig* = PortalProtocolConfig(
     tableIpLimits: DefaultTableIpLimits,
     bitsPerHop: DefaultBitsPerHop,
+    alpha: defaultAlpha,
     radiusConfig: defaultRadiusConfig,
     disablePoke: defaultDisablePoke,
     maxGossipNodes: defaultMaxGossipNodes,
@@ -71,6 +74,7 @@ proc init*(
     tableIpLimit: uint,
     bucketIpLimit: uint,
     bitsPerHop: int,
+    alpha: int,
     radiusConfig: RadiusConfig,
     disablePoke: bool,
     maxGossipNodes: int,
@@ -82,6 +86,7 @@ proc init*(
     tableIpLimits:
       TableIpLimits(tableIpLimit: tableIpLimit, bucketIpLimit: bucketIpLimit),
     bitsPerHop: bitsPerHop,
+    alpha: alpha,
     radiusConfig: radiusConfig,
     disablePoke: disablePoke,
     maxGossipNodes: maxGossipNodes,