Large message handling combined#1234
Conversation
updating local branch
|
Pull requests titles must follow the Conventional Commits specification |
|
Commits must follow the Conventional Commits specification |
There was a problem hiding this comment.
Very interesting PR!
Is this PR also implementing this part of the spec?:
if the message ID is found in the ongoing_receives list,
the peer should postpone sending the IWANT request for a defer_interval.
The defer_interval may be based on the message download time.
| @@ -226,6 +253,7 @@ method init*(g: GossipSub) = | |||
| g.codecs &= GossipSubCodec_12 | |||
| g.codecs &= GossipSubCodec_11 | |||
| g.codecs &= GossipSubCodec_10 | |||
There was a problem hiding this comment.
Would this change result in a GossipSubCodec_14 as described in libp2p/specs#654 ?
| g.mcache.put(msgId, msg) | ||
|
|
||
| g.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true) | ||
| #g.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true) |
There was a problem hiding this comment.
✂️ ?
| #g.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true) |
| #g.broadcast(peers, RPCMsg(messages: @[msg]), isHighPriority = true) | ||
| let sem = newAsyncSemaphore(1) | ||
| var staggerPeers = toSeq(peers) | ||
| g.rng.shuffle(staggerPeers) |
There was a problem hiding this comment.
Why are the peers randomly sorted?
| #We send message immediately after sending preamble to each peer | ||
| proc sendToOne(p: PubSubPeer) {.async.} = | ||
| g.broadcast(@[p], RPCMsg(control: some(ControlMessage( | ||
| preamble: @[ControlIHave(topicID: topic, messageIDs: @[msgId])] | ||
| ))), isHighPriority = true) | ||
|
|
||
| await sem.acquire() | ||
| defer: sem.release() | ||
|
|
||
| g.broadcast(@[p], RPCMsg(messages: @[msg]), isHighPriority = false) | ||
|
|
||
| for p in staggerPeers: | ||
| asyncSpawn sendToOne(p) | ||
|
|
There was a problem hiding this comment.
Using the AsyncSemaphore(1) means that the preambles are being sent sequentially to each peer, right? Why is this the case? Shouldn't we try to get the preamble messages being sent concurrently to all peers?
| await sem.acquire() | ||
| defer: sem.release() | ||
|
|
||
| g.broadcast(@[p], RPCMsg(messages: @[msg]), isHighPriority = false) |
There was a problem hiding this comment.
No isHighPriority = true?
| return result | ||
| continue | ||
| let msg = g.mcache.get(mid).valueOr: | ||
| libp2p_gossipsub_received_iwants.inc(1, labelValues = ["unknown"]) | ||
| continue | ||
| libp2p_gossipsub_received_iwants.inc(1, labelValues = ["correct"]) | ||
| messages.add(msg) | ||
| return messages | ||
| result.messages.add(msg) | ||
| result.ids.add(mid) |
There was a problem hiding this comment.
It's probably better to define a messages and ids variables, instead of using result. See https://status-im.github.io/nim-style-guide/language.result.html
| if peer.heIsReceivings[^1].len > 1000: break | ||
| if messageId.len > 100: continue |
There was a problem hiding this comment.
When do messageIds exceed 100? I'm wondering if this condition should be added in other places as i haven't seen length checking for messageIds using 100. (Probably a good idea to extract this into a constant too)
| preamble*: seq[ControlIHave] | ||
| imreceiving*: seq[ControlIWant] |
There was a problem hiding this comment.
Shouldn't these have their own type? ControlPreamble and ControlImReceiving?
| #We send message immediately after sending preamble to each peer | ||
| proc sendToOne(p: PubSubPeer) {.async.} = | ||
| g.broadcast(@[p], RPCMsg(control: some(ControlMessage( | ||
| preamble: @[ControlIHave(topicID: topic, messageIDs: @[msgId])] |
There was a problem hiding this comment.
Sending the message length is still pending, right?
| await sem.acquire() | ||
| defer: sem.release() | ||
|
|
||
| g.broadcast(@[p], RPCMsg(messages: @[msg]), isHighPriority = false) |
There was a problem hiding this comment.
Reading in the specs, I see the following:
Adding a message preamble may increase control overhead for small messages.
Therefore, it is preferable to use it only for messages that exceed the preamble_threshold.
The way it is coded right now, it will send the preamble always, right? Is this because this PR is for experiments?
|
@ufarooqstatus should this PR be kept open? or can it be closed |
yes, it can be closed. it was an early PoC for v1.4 |
We use message preamble to detect incoming messages. This information is used to:
PoC implementation for shadow simulation. Not intended for merging
Reduces duplicates to less than 1.5 for large messages
More details here