status-im
/
nim-libp2p
mirror of https://github.com/status-im/nim-libp2p.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

take into account explicit direct peers in publish

This commit is contained in:
Giovanni Petrantoni 2020-06-26 12:31:08 +09:00
parent 84988a147c
commit cf51da00b2
2 changed files with 30 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
libp2p/protocols/pubsub/gossipsub.nim
View File

@ -524,40 +524,37 @@ method publish*(g: GossipSub,
await procCall PubSub(g).publish(topic, data) await procCall PubSub(g).publish(topic, data)
trace "about to publish message on topic", name = topic, trace "about to publish message on topic", name = topic,
data = data.shortLog data = data.shortLog
# directly copy explicit peers
# as we will always publish to those
var peers = g.explicitPeers
var peers: HashSet[string]
# TODO: we probably don't need to try multiple times
if data.len > 0 and topic.len > 0: if data.len > 0 and topic.len > 0:
for _ in 0..<5: # try to get peers up to 5 times if topic in g.topics: # if we're subscribed to the topic attempt to build a mesh
if peers.len > 0: await g.rebalanceMesh(topic)
break peers.incl(g.mesh.getOrDefault(topic))
else: # send to fanout peers
if topic in g.topics: # if we're subscribed to the topic attempt to build a mesh await g.replenishFanout(topic)
await g.rebalanceMesh(topic) if topic in g.fanout:
peers = g.mesh.getOrDefault(topic) peers.incl(g.fanout.getOrDefault(topic))
else: # send to fanout peers # set the fanout expiry time
await g.replenishFanout(topic) g.lastFanoutPubSub[topic] = Moment.fromNow(GossipSubFanoutTTL)
if topic in g.fanout:
peers = g.fanout.getOrDefault(topic)
# set the fanout expiry time
g.lastFanoutPubSub[topic] = Moment.fromNow(GossipSubFanoutTTL)
# wait a second between tries
await sleepAsync(1.seconds)
let msg = newMessage(g.peerInfo, data, topic, g.sign) let msg = newMessage(g.peerInfo, data, topic, g.sign)
trace "created new message", msg trace "created new message", msg
trace "publishing on topic", name = topic
if msg.msgId notin g.mcache:
g.mcache.put(msg)
var sent: seq[Future[void]] var sent: seq[Future[void]]
for p in peers: for p in peers:
# avoid sending to self
if p == g.peerInfo.id: if p == g.peerInfo.id:
continue continue
let peer = g.peers.getOrDefault(p)
trace "publishing on topic", name = topic if not isNil(peer.peerInfo):
if msg.msgId notin g.mcache:
g.mcache.put(msg)
if p in g.peers:
sent.add(g.peers[p].send(@[RPCMsg(messages: @[msg])])) sent.add(g.peers[p].send(@[RPCMsg(messages: @[msg])]))
checkFutures(await allFinished(sent)) checkFutures(await allFinished(sent))
libp2p_pubsub_messages_published.inc(labelValues = [topic]) libp2p_pubsub_messages_published.inc(labelValues = [topic])

9
libp2p/protocols/pubsub/pubsubpeer.nim
View File

@ -27,6 +27,15 @@ declareCounter(libp2p_pubsub_skipped_received_messages, "number of received skip
declareCounter(libp2p_pubsub_skipped_sent_messages, "number of sent skipped messages", labels = ["id"]) declareCounter(libp2p_pubsub_skipped_sent_messages, "number of sent skipped messages", labels = ["id"])
type type
TopicScoreParams = object
topicWeight: float
timeInMeshWeight: float
timeInMeshQuantum: Duration
timeInMeshCap: float
PeerScoreParams = object
topics: Table[string, TopicScoreParams]
PubSubObserver* = ref object PubSubObserver* = ref object
onRecv*: proc(peer: PubSubPeer; msgs: var RPCMsg) {.gcsafe, raises: [Defect].} onRecv*: proc(peer: PubSubPeer; msgs: var RPCMsg) {.gcsafe, raises: [Defect].}
onSend*: proc(peer: PubSubPeer; msgs: var RPCMsg) {.gcsafe, raises: [Defect].} onSend*: proc(peer: PubSubPeer; msgs: var RPCMsg) {.gcsafe, raises: [Defect].}