Gossip one one (#240)

* allow multiple codecs per protocol (without breaking things)

* add 1.1 protocol to gossip

* explicit peering part 1

* explicit peering part 2

* explicit peering part 3

* PeerInfo and ControlPrune protocols

* fix encodePrune

* validated always, even explicit peers

* prune by score (score is stub still)

* add a way to pass parameters to gossip

* standard setup fixes

* take into account explicit direct peers in publish

* add floodPublish logic

* small fixes, publish still half broken

* make sure to waitsub in sparse test

* use var semantics to optimize table access

* wip... lvalues don't work properly sadly...

* big publish refactor, replenish and balance

* fix internal tests

* use g.peers for fanout (todo: don't include flood peers)

* exclude non gossip from fanout

* internal test fixes

* fix flood tests

* fix test's trypublish

* test interop fixes

* make sure to not remove peers from gossip table

* restore old replenishFanout

* cleanups

* restore utility module import

* restore trace vs debug in gossip

* improve fanout replenish behavior further

* triage publish nil peers (issue is on master too but just hidden behind a if/in)

* getGossipPeers fixes

* remove topics from pubsubpeer (was unused)

* simplify rebalanceMesh (following spec) and make it finally reach D_high

* better diagnostics

* merge new pubsubpeer, copy 1.1 to new module

* fix up merge

* conditional enable gossip11 module

* add back topics in peers, re-enable flood publish

* add more heartbeat locking to prevent races

* actually lock the heartbeat

* minor fixes

* with sugar

* merge 1.0

* remove assertion in publish

* fix multistream 1.1 multi proto

* Fix merge oops

* wip

* fix gossip 11 upstream

* gossipsub11 -> gossipsub

* support interop testing

* tests fixing

* fix directchat build

* control prune updates (pb)

* wip parameters

* gossip internal tests fixes

* parameters wip

* finishup with params

* cleanups/wip

* small sugar

* grafted and pruned procs

* wip updateScores

* wip

* fix logging issue

* pubsubpeer, chronicles explicit override

* fix internal gossip tests

* wip

* tables troubleshooting

* score wip

* score wip

* fixes

* fix test utils generateNodes

* don't delete while iterating in score update

* fix grafted defect

* add a handleConnect in subscribeTopic

* pruning improvements

* wip

* score fixes

* post merge - builds gossip tests

* further merge fixes

* rebalance improvements and opportunistic grafting

* fix test for now

* restore explicit peering

* implement peer exchange graft message

* add an hard cap to PX

* backoff time management

* IWANT cap/budget

* Adaptive gossip dissemination

* outbound mesh quota, internal tests fixing

* oversub prune score based, finish outbound quota

* finishup with score and ihave budget

* use go daemon 0.3.0

* import fixes

* byScore cleanup score sorting

* remove pointless scaling in `/` Duration operator

* revert using libp2p org for daemon

* interop fixes

* fixes and cleanup

* remove heartbeat assertion, minor debug fixes

* logging improvements and cleaning up

* (to revert) add some traces

* add explicit topic to gossip rpcs

* pubsub merge fixes and type fix in switch

* Revert "(to revert) add some traces"

This reverts commit 4663eaa.

* cleanup some now irrelevant todo

* shuffle peers anyway as score might be disabled

* add missing shuffle

* old merge fix

* more merge fixes

* debug improvements

* re-enable gossip internal tests

* add gossip10 fallback (dormant but tested)

* split gossipsub internal tests into 1.0 and 1.1

Co-authored-by: Dmitriy Ryajov <[email protected]>

.appveyor.yml

@@ -39,7 +39,7 @@ install:
   - CD ..
 
   # install and build go-libp2p-daemon
-  - bash scripts/build_p2pd.sh p2pdCache v0.2.4
+  - bash scripts/build_p2pd.sh p2pdCache v0.3.0
 
 build_script:
   - nimble install -y --depsOnly

.gitignore

@@ -10,3 +10,5 @@ build/
 *.exe
 *.dll
 .vscode/
+.DS_Store
+tests/pubsub/testgossipsub

.travis.yml

@@ -45,7 +45,7 @@ install:
   - export PATH="$PWD/Nim/bin:$GOPATH/bin:$PATH"
 
   # install and build go-libp2p-daemon
-  - bash scripts/build_p2pd.sh p2pdCache v0.2.4
+  - bash scripts/build_p2pd.sh p2pdCache v0.3.0
 
 script:
   - nimble install -y --depsOnly

azure-pipelines.yml

@@ -112,7 +112,7 @@ steps:
       echo "PATH=${PATH}"
 
       # we can't seem to be able to build a 32-bit p2pd
-      env PATH="/c/custom/mingw64/bin:${PATH}" bash scripts/build_p2pd.sh p2pdCache v0.2.4
+      env PATH="/c/custom/mingw64/bin:${PATH}" bash scripts/build_p2pd.sh p2pdCache v0.3.0
 
       # install dependencies
       nimble refresh

examples/directchat.nim

@@ -124,7 +124,7 @@ proc readWriteLoop(p: ChatProto) {.async.} =
   asyncCheck p.readAndPrint()
 
 proc newChatProto(switch: Switch, transp: StreamTransport): ChatProto =
-  var chatproto = ChatProto(switch: switch, transp: transp, codec: ChatCodec)
+  var chatproto = ChatProto(switch: switch, transp: transp, codecs: @[ChatCodec])
 
   # create handler for incoming connection
   proc handle(stream: Connection, proto: string) {.async.} =

libp2p.nimble

@@ -47,8 +47,12 @@ task testinterop, "Runs interop tests":
   runTest("testinterop")
 
 task testpubsub, "Runs pubsub tests":
+  runTest("pubsub/testgossipinternal", sign = false, verify = false, moreoptions = "-d:pubsub_internal_testing")
   runTest("pubsub/testpubsub")
   runTest("pubsub/testpubsub", sign = false, verify = false)
+  runTest("pubsub/testgossipinternal10", sign = false, verify = false, moreoptions = "-d:pubsub_internal_testing")
+  runTest("pubsub/testpubsub", moreoptions = "-d:fallback_gossipsub_10")
+  runTest("pubsub/testpubsub", sign = false, verify = false, moreoptions = "-d:fallback_gossipsub_10")
 
 task testfilter, "Run PKI filter test":
   runTest("testpkifilter",

libp2p/daemon/daemonapi.nim

@@ -8,8 +8,8 @@
 ## those terms.
 
 ## This module implementes API for `go-libp2p-daemon`.
-import os, osproc, strutils, tables, strtabs
-import chronos
+import std/[os, osproc, strutils, tables, strtabs]
+import chronos, chronicles
 import ../varint, ../multiaddress, ../multicodec, ../cid, ../peerid
 import ../wire, ../multihash, ../protobuf/minprotobuf
 import ../crypto/crypto
@@ -737,10 +737,12 @@ proc newDaemonApi*(flags: set[P2PDaemonFlags] = {},
       opt.add $address
     args.add(opt)
   args.add("-noise=true")
+  args.add("-quic=false")
   args.add("-listen=" & $api.address)
 
   # We are trying to get absolute daemon path.
   let cmd = findExe(daemon)
+  trace "p2pd cmd", cmd, args
   if len(cmd) == 0:
     raise newException(DaemonLocalError, "Could not find daemon executable!")
 

libp2p/multiaddress.nim

@@ -11,7 +11,7 @@
 
 {.push raises: [Defect].}
 
-import nativesockets
+import nativesockets, hashes
 import tables, strutils, stew/shims/net
 import chronos
 import multicodec, multihash, multibase, transcoder, vbuffer, peerid,
@@ -56,6 +56,13 @@ const
   IPPROTO_TCP = Protocol.IPPROTO_TCP
   IPPROTO_UDP = Protocol.IPPROTO_UDP
 
+proc hash*(a: MultiAddress): Hash =
+  var h: Hash = 0
+  h = h !& hash(a.data.buffer)
+  h = h !& hash(a.data.offset)
+  h = h !& hash(a.data.length)
+  !$h
+
 proc ip4StB(s: string, vb: var VBuffer): bool =
   ## IPv4 stringToBuffer() implementation.
   try:

libp2p/multistream.nim

@@ -7,7 +7,7 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
-import strutils
+import std/[strutils]
 import chronos, chronicles, stew/byteutils
 import stream/connection,
        vbuffer,
@@ -28,7 +28,7 @@ type
   Matcher* = proc (proto: string): bool {.gcsafe.}
 
   HandlerHolder* = object
-    proto*: string
+    protos*: seq[string]
     protocol*: LPProtocol
     match*: Matcher
 
@@ -147,7 +147,8 @@ proc handle*(m: MultistreamSelect, conn: Connection, active: bool = false) {.asy
         trace "handle: listing protos", conn
         var protos = ""
         for h in m.handlers:
-          protos &= (h.proto & "\n")
+          for proto in h.protos:
+            protos &= (proto & "\n")
         await conn.writeLp(protos)
       of Codec:
         if not handshaked:
@@ -159,9 +160,9 @@ proc handle*(m: MultistreamSelect, conn: Connection, active: bool = false) {.asy
           await conn.write(Na)
       else:
         for h in m.handlers:
-          if (not isNil(h.match) and h.match(ms)) or ms == h.proto:
+          if (not isNil(h.match) and h.match(ms)) or h.protos.contains(ms):
             trace "found handler", conn, protocol = ms
-            await conn.writeLp((h.proto & "\n"))
+            await conn.writeLp(ms & "\n")
             await h.protocol.handler(conn, ms)
             return
         debug "no handlers", conn, protocol = ms
@@ -176,26 +177,30 @@ proc handle*(m: MultistreamSelect, conn: Connection, active: bool = false) {.asy
   trace "Stopped multistream handler", conn
 
 proc addHandler*(m: MultistreamSelect,
-                 codec: string,
+                 codecs: seq[string],
                  protocol: LPProtocol,
                  matcher: Matcher = nil) =
-  ## register a protocol
-  trace "registering protocol", codec = codec
-  m.handlers.add(HandlerHolder(proto: codec,
+  trace "registering protocols", protos = codecs
+  m.handlers.add(HandlerHolder(protos: codecs,
                                protocol: protocol,
                                match: matcher))
 
+proc addHandler*(m: MultistreamSelect,
+                 codec: string,
+                 protocol: LPProtocol,
+                 matcher: Matcher = nil) =
+  addHandler(m, @[codec], protocol, matcher)
+
 proc addHandler*(m: MultistreamSelect,
                  codec: string,
                  handler: LPProtoHandler,
                  matcher: Matcher = nil) =
   ## helper to allow registering pure handlers
-
-  trace "registering proto handler", codec = codec
+  trace "registering proto handler", proto = codec
   let protocol = new LPProtocol
   protocol.codec = codec
   protocol.handler = handler
 
-  m.handlers.add(HandlerHolder(proto: codec,
+  m.handlers.add(HandlerHolder(protos: @[codec],
                                protocol: protocol,
                                match: matcher))

libp2p/muxers/mplex/mplex.nim

@@ -185,7 +185,7 @@ method handle*(m: Mplex) {.async, gcsafe.} =
           await channel.reset()
   except CancelledError:
     # This procedure is spawned as task and it is not part of public API, so
-    # there no way for this procedure to be cancelled implicitely.
+    # there no way for this procedure to be cancelled implicitly.
     debug "Unexpected cancellation in mplex handler", m
   except LPStreamEOFError as exc:
     trace "Stream EOF", m, msg = exc.msg

libp2p/protocols/identify.nim

@@ -87,7 +87,7 @@ proc decodeMsg*(buf: seq[byte]): Option[IdentifyInfo] =
       iinfo.protoVersion = some(protoVersion)
     if r6.get():
       iinfo.agentVersion = some(agentVersion)
-    trace "decodeMsg: decoded message", pubkey = ($pubKey).shortLog,
+    debug "decodeMsg: decoded message", pubkey = ($pubKey).shortLog,
           addresses = $iinfo.addrs, protocols = $iinfo.protos,
           observable_address = $iinfo.observedAddr,
           proto_version = $iinfo.protoVersion,

libp2p/protocols/protocol.nim

@@ -16,7 +16,16 @@ type
                           Future[void] {.gcsafe, closure.}
 
   LPProtocol* = ref object of RootObj
-    codec*: string
+    codecs*: seq[string]
     handler*: LPProtoHandler ## this handler gets invoked by the protocol negotiator
 
 method init*(p: LPProtocol) {.base, gcsafe.} = discard
+
+func codec*(p: LPProtocol): string = 
+  assert(p.codecs.len > 0, "Codecs sequence was empty!")
+  p.codecs[0]
+
+func `codec=`*(p: LPProtocol, codec: string) =
+  # always insert as first codec 
+  # if we use this abstraction
+  p.codecs.insert(codec, 0)