WIP: Test DHT with large number of nodes

README-DAS.md

@@ -0,0 +1,26 @@
+# DAS emulator
+
+Emulate DAS DHT behavior, with a few simple assumption
+- the block is populated in the DHT by the builder (node 0)
+- all nodes start sampling at the same time
+- 1-way latency is 50ms (configurable)
+- no losses in transmission (configurable)
+- scaled down numbers (nodes, blocksize, etc., all configrable)
+
+## Compilation
+
+```
+# install Nim 1.6
+
+# install Nimble 0.14+
+nimble install nimble
+
+# make sure the newly installed nimble is used
+export PATH=~/.nimble/bin:$PATH
+
+# install dependencies
+nimble install 
+
+# compile and run passing on various flags
+nimble run "-d:chronicles_sinks=textlines[stdout,nocolors]" -d:chronicles_log_level=INFO -d:release -d:asyncTimer=virtual das
+```

das.nim

@@ -0,0 +1,185 @@
+import
+  std/[random, math],
+  chronicles,
+  chronos,
+  libp2pdht/dht,
+  libp2pdht/discv5/crypto as dhtcrypto,
+  libp2pdht/discv5/protocol as discv5_protocol,
+  test_helper
+
+logScope:
+  topics = "DAS emulator"
+
+proc bootstrapNodes(
+    nodecount: int,
+    bootnodes: seq[SignedPeerRecord],
+    rng = newRng(),
+    delay: int = 0
+  ) : Future[seq[(discv5_protocol.Protocol, PrivateKey)]] {.async.} =
+
+  debug "---- STARTING BOOSTRAPS ---"
+  for i in 0..<nodecount:
+    try:
+      let privKey = PrivateKey.example(rng)
+      let node = initDiscoveryNode(rng, privKey, localAddress(20302 + i), bootnodes)
+      await node.start()
+      result.add((node, privKey))
+      if delay > 0:
+        await sleepAsync(chronos.milliseconds(delay))
+    except TransportOsError as e:
+      echo "skipping node ",i ,":", e.msg
+
+  #await allFutures(result.mapIt(it.bootstrap())) # this waits for bootstrap based on bootENode, which includes bonding with all its ping pongs
+
+proc bootstrapNetwork(
+    nodecount: int,
+    rng = newRng(),
+    delay: int = 0
+  ) : Future[seq[(discv5_protocol.Protocol, PrivateKey)]] {.async.} =
+
+  let
+    bootNodeKey = PrivateKey.fromHex(
+      "a2b50376a79b1a8c8a3296485572bdfbf54708bb46d3c25d73d2723aaaf6a617")
+      .expect("Valid private key hex")
+    bootNodeAddr = localAddress(20301)
+    bootNode = initDiscoveryNode(rng, bootNodeKey, bootNodeAddr, @[]) # just a shortcut for new and open
+
+  #waitFor bootNode.bootstrap()  # immediate, since no bootnodes are defined above
+
+  var res = await bootstrapNodes(nodecount - 1,
+                           @[bootnode.localNode.record],
+                           rng,
+                           delay)
+  res.insert((bootNode, bootNodeKey), 0)
+  return res
+
+proc toNodeId(data: openArray[byte]): NodeId =
+  readUintBE[256](keccak256.digest(data).data)
+
+proc segmentData(s: int, segmentsize: int) : seq[byte] =
+  result = newSeq[byte](segmentsize)
+  var
+    r = s
+    i = 0
+  while r > 0:
+    assert(i<segmentsize)
+    result[i] = byte(r mod 256)
+    r = r div 256
+    i+=1
+
+proc sample(s: Slice[int], len: int): seq[int] =
+    # random sample without replacement
+    # TODO: not the best for small len
+  assert s.a <= s.b
+  var all = s.b - s.a + 1
+  var count = len
+  if len >= all div 10: # add better algo selector
+    var generated = newSeq[bool](all) # Initialized to false.
+    while count != 0:
+      let n = rand(s)
+      if not generated[n - s.a]:
+        generated[n - s.a] = true
+        result.add n
+        dec count
+  else:
+    while count != 0:
+      let n = rand(s)
+      if not (n in result):
+        result.add n
+        dec count
+
+
+when isMainModule:
+  proc main() {.async.} =
+    let
+      nodecount = 100
+      delay_pernode = 10 # in millisec
+      delay_init = 15*1000 # in millisec
+      blocksize = 256
+      segmentsize = 2
+      samplesize = 3
+      upload_timeout = 5.seconds
+      sampling_timeout = 5.seconds
+    assert(log2(blocksize.float).ceil.int <= segmentsize * 8 )
+    assert(samplesize <= blocksize)
+
+    var
+      segmentIDs = newSeq[NodeId](blocksize)
+
+    # start network
+    let
+      rng = newRng()
+      nodes = await bootstrapNetwork(nodecount=nodecount, delay=delay_pernode)
+
+    # wait for network to settle
+    await sleepAsync(chronos.milliseconds(delay_init))
+
+    # generate block and push data
+    info "starting upload to DHT"
+    let startTime = Moment.now()
+    var futs = newSeq[Future[seq[Node]]]()
+    for s in 0 ..< blocksize:
+      let
+        segment = segmentData(s, segmentsize)
+        key = toNodeId(segment)
+
+      segmentIDs[s] = key
+
+      futs.add(nodes[0][0].addValue(key, segment))
+
+    let pass = await allFutures(futs).withTimeout(upload_timeout)
+    info "uploaded to DHT", by = 0, pass, time = Moment.now() - startTime
+
+    # sample
+    proc startSamplingDA(n: discv5_protocol.Protocol): seq[Future[DiscResult[seq[byte]]]] =
+      ## Generate random sample and start the sampling process
+      var futs = newSeq[Future[DiscResult[seq[byte]]]]()
+
+      let sample = sample(0 ..< blocksize, samplesize)
+      debug "starting sampling", by = n, sample
+      for s in sample:
+        let fut = n.getValue(segmentIDs[s])
+        futs.add(fut)
+      return futs
+
+    proc sampleDA(n: discv5_protocol.Protocol): Future[(bool, int, Duration)] {.async.} =
+      ## Sample and return detailed results of sampling
+      let startTime = Moment.now()
+      var futs = startSamplingDA(n)
+
+      # test is passed if all segments are retrieved in time
+      let pass = await allFutures(futs).withTimeout(sampling_timeout)
+      var passcount: int
+      for f in futs:
+        if f.finished():
+          passcount += 1
+
+      let time = Moment.now() - startTime
+      info "sample", by = n.localNode, pass, cnt = passcount, time
+      return (pass, passcount, time)
+
+    # all nodes start sampling in parallel
+    var samplings = newSeq[Future[(bool, int, Duration)]]()
+    for n in 1 ..< nodecount:
+      samplings.add(sampleDA(nodes[n][0]))
+    await allFutures(samplings)
+
+    # print statistics
+    var
+      passed = 0
+    for f in samplings:
+      if f.finished():
+        let (pass, passcount, time) = await f
+        passed += pass.int
+        debug "sampleStats", pass, cnt = passcount, time
+      else:
+        error "This should not happen!"
+    info "sampleStats", passed, total = samplings.len, ratio = passed/samplings.len
+
+  waitfor main()
+
+# proc teardownAll() =
+#     for (n, _) in nodes: # if last test is enabled, we need nodes[1..^1] here
+#       await n.closeWait()
+
+

libp2pdht.nimble

@@ -6,11 +6,12 @@ description   = "DHT based on the libp2p Kademlia spec"
 license       = "MIT"
 skipDirs      = @["tests"]
 
+bin = @["das"]
 
 # Dependencies
 requires "nim >= 1.2.0",
          "nimcrypto >= 0.5.4 & < 0.6.0",
-         "bearssl#head",
+         "bearssl#f4c4233de453cb7eac0ce3f3ffad6496295f83ab",
          "chronicles >= 0.10.2 & < 0.11.0",
          "chronos >= 3.0.11 & < 3.1.0",
          "libp2p#unstable",
@@ -21,7 +22,8 @@ requires "nim >= 1.2.0",
          "stint",
          "asynctest >= 0.3.1 & < 0.4.0",
          "https://github.com/status-im/nim-datastore#head",
-         "questionable"
+         "questionable",
+         "datastore"
 
 task coverage, "generates code coverage report":
   var (output, exitCode) = gorgeEx("which lcov")

libp2pdht/dht.nim

@@ -1,4 +1,6 @@
 import
-  ./dht/[providers_encoding, providers_messages]
+  ./dht/[providers_encoding, providers_messages],
+  ./dht/[value_encoding, value_messages]
 
-export providers_encoding, providers_messages
+export providers_encoding, providers_messages
+export value_encoding, value_messages

libp2pdht/dht/value_encoding.nim

@@ -0,0 +1,77 @@
+import
+  ../discv5/[node],
+  libp2p/protobuf/minprotobuf,
+  ./value_messages
+
+func getField(pb: ProtoBuffer, field: int,
+               nid: var NodeId): ProtoResult[bool] {.inline.} =
+  ## Read ``NodeId`` from ProtoBuf's message and validate it
+  var buffer: seq[byte]
+  let res = ? pb.getField(field, buffer)
+  if not(res):
+    ok(false)
+  else:
+    nid = readUintBE[256](buffer)
+    ok(true)
+
+func write(pb: var ProtoBuffer, field: int, nid: NodeId) =
+  ## Write NodeId value ``nodeid`` to object ``pb`` using ProtoBuf's encoding.
+  write(pb, field, nid.toBytesBE())
+
+proc decode*(
+  T: typedesc[AddValueMessage],
+  buffer: openArray[byte]): Result[AddValueMessage, ProtoError] =
+
+  let pb = initProtoBuffer(buffer)
+  var msg = AddValueMessage()
+
+  ? pb.getRequiredField(1, msg.cId)
+  ? pb.getRequiredField(2, msg.value)
+
+  ok(msg)
+
+proc encode*(msg: AddValueMessage): seq[byte] =
+  var pb = initProtoBuffer()
+
+  pb.write(1, msg.cId)
+  pb.write(2, msg.value)
+
+  pb.finish()
+  pb.buffer
+
+proc decode*(
+  T: typedesc[GetValueMessage],
+  buffer: openArray[byte]): Result[GetValueMessage, ProtoError] =
+
+  let pb = initProtoBuffer(buffer)
+  var msg = GetValueMessage()
+
+  ? pb.getRequiredField(1, msg.cId)
+
+  ok(msg)
+
+proc encode*(msg: GetValueMessage): seq[byte] =
+  var pb = initProtoBuffer()
+
+  pb.write(1, msg.cId)
+
+  pb.finish()
+  pb.buffer
+
+proc decode*(
+  T: typedesc[ValueMessage],
+  buffer: openArray[byte]): Result[ValueMessage, ProtoError] =
+
+  let pb = initProtoBuffer(buffer)
+  var msg = ValueMessage()
+  ? pb.getRequiredField(1, msg.value)
+
+  ok(msg)
+
+proc encode*(msg: ValueMessage): seq[byte] =
+  var pb = initProtoBuffer()
+
+  pb.write(1, msg.value)
+
+  pb.finish()
+  pb.buffer

libp2pdht/dht/value_messages.nim

@@ -0,0 +1,14 @@
+import
+  ../discv5/[node]
+
+type
+  AddValueMessage* = object
+    cId*: NodeId
+    value*: seq[byte]
+
+  GetValueMessage* = object
+    cId*: NodeId
+
+  ValueMessage* = object
+    #total*: uint32
+    value*: seq[byte]