Add urgent flag to SendRPC in trace.proto

ppopth · Mar 1, 2024 · dfd81e8 · dfd81e8
1 parent 3e3d6d7
commit dfd81e8
Show file tree

Hide file tree

Showing 11 changed files with 138 additions and 93 deletions.
diff --git a/floodsub.go b/floodsub.go
@@ -97,7 +97,7 @@ func (fs *FloodSubRouter) Publish(msg *Message) {
 			// Drop it. The peer is too slow.
 			continue
 		}
-		fs.tracer.SendRPC(out, pid)
+		fs.tracer.SendRPC(out, pid, false)
 	}
 }
 

diff --git a/gossip_tracer.go b/gossip_tracer.go
@@ -167,17 +167,17 @@ func (gt *gossipTracer) ValidateMessage(msg *Message) {
 	gt.fulfillPromise(msg)
 }
 
-func (gt *gossipTracer) AddPeer(p peer.ID, proto protocol.ID) {}
-func (gt *gossipTracer) RemovePeer(p peer.ID)                 {}
-func (gt *gossipTracer) Join(topic string)                    {}
-func (gt *gossipTracer) Leave(topic string)                   {}
-func (gt *gossipTracer) Graft(p peer.ID, topic string)        {}
-func (gt *gossipTracer) Prune(p peer.ID, topic string)        {}
-func (gt *gossipTracer) DuplicateMessage(msg *Message)        {}
-func (gt *gossipTracer) RecvRPC(rpc *RPC)                     {}
-func (gt *gossipTracer) SendRPC(rpc *RPC, p peer.ID)          {}
-func (gt *gossipTracer) DropRPC(rpc *RPC, p peer.ID)          {}
-func (gt *gossipTracer) UndeliverableMessage(msg *Message)    {}
+func (gt *gossipTracer) AddPeer(p peer.ID, proto protocol.ID)     {}
+func (gt *gossipTracer) RemovePeer(p peer.ID)                     {}
+func (gt *gossipTracer) Join(topic string)                        {}
+func (gt *gossipTracer) Leave(topic string)                       {}
+func (gt *gossipTracer) Graft(p peer.ID, topic string)            {}
+func (gt *gossipTracer) Prune(p peer.ID, topic string)            {}
+func (gt *gossipTracer) DuplicateMessage(msg *Message)            {}
+func (gt *gossipTracer) RecvRPC(rpc *RPC)                         {}
+func (gt *gossipTracer) SendRPC(rpc *RPC, p peer.ID, urgent bool) {}
+func (gt *gossipTracer) DropRPC(rpc *RPC, p peer.ID)              {}
+func (gt *gossipTracer) UndeliverableMessage(msg *Message)        {}
 
 func (gt *gossipTracer) ThrottlePeer(p peer.ID) {
 	gt.Lock()

diff --git a/gossipsub.go b/gossipsub.go
@@ -1224,7 +1224,7 @@ func (gs *GossipSubRouter) doSendRPC(rpc *RPC, p peer.ID, q *rpcQueue, urgent bo
 		gs.doDropRPC(rpc, p, "queue full")
 		return
 	}
-	gs.tracer.SendRPC(rpc, p)
+	gs.tracer.SendRPC(rpc, p, urgent)
 }
 
 func fragmentRPC(rpc *RPC, limit int) ([]*RPC, error) {

diff --git a/pb/trace.pb.go b/pb/trace.pb.go
diff --git a/pb/trace.proto b/pb/trace.proto
@@ -78,6 +78,7 @@ message TraceEvent {
   message SendRPC {
     optional bytes sendTo = 1;
     optional RPCMeta meta = 2;
+    optional bool urgent = 3;
   }
 
   message DropRPC {

diff --git a/peer_gater.go b/peer_gater.go
@@ -446,7 +446,7 @@ func (pg *peerGater) ThrottlePeer(p peer.ID) {}
 
 func (pg *peerGater) RecvRPC(rpc *RPC) {}
 
-func (pg *peerGater) SendRPC(rpc *RPC, p peer.ID) {}
+func (pg *peerGater) SendRPC(rpc *RPC, p peer.ID, urgent bool) {}
 
 func (pg *peerGater) DropRPC(rpc *RPC, p peer.ID) {}
 

diff --git a/pubsub.go b/pubsub.go
@@ -925,7 +925,7 @@ func (p *PubSub) announce(topic string, sub bool) {
 			go p.announceRetry(pid, topic, sub)
 			continue
 		}
-		p.tracer.SendRPC(out, pid)
+		p.tracer.SendRPC(out, pid, false)
 	}
 }
 
@@ -968,7 +968,7 @@ func (p *PubSub) doAnnounceRetry(pid peer.ID, topic string, sub bool) {
 		go p.announceRetry(pid, topic, sub)
 		return
 	}
-	p.tracer.SendRPC(out, pid)
+	p.tracer.SendRPC(out, pid, false)
 }
 
 // notifySubs sends a given message to all corresponding subscribers.

diff --git a/randomsub.go b/randomsub.go
@@ -157,7 +157,7 @@ func (rs *RandomSubRouter) Publish(msg *Message) {
 			rs.tracer.DropRPC(out, p)
 			continue
 		}
-		rs.tracer.SendRPC(out, p)
+		rs.tracer.SendRPC(out, p, false)
 	}
 }
 

diff --git a/score.go b/score.go
@@ -830,7 +830,7 @@ func (ps *peerScore) ThrottlePeer(p peer.ID) {}
 
 func (ps *peerScore) RecvRPC(rpc *RPC) {}
 
-func (ps *peerScore) SendRPC(rpc *RPC, p peer.ID) {}
+func (ps *peerScore) SendRPC(rpc *RPC, p peer.ID, urgent bool) {}
 
 func (ps *peerScore) DropRPC(rpc *RPC, p peer.ID) {}
 

diff --git a/tag_tracer.go b/tag_tracer.go
@@ -251,9 +251,9 @@ func (t *tagTracer) RejectMessage(msg *Message, reason string) {
 	}
 }
 
-func (t *tagTracer) RemovePeer(peer.ID)                {}
-func (t *tagTracer) ThrottlePeer(p peer.ID)            {}
-func (t *tagTracer) RecvRPC(rpc *RPC)                  {}
-func (t *tagTracer) SendRPC(rpc *RPC, p peer.ID)       {}
-func (t *tagTracer) DropRPC(rpc *RPC, p peer.ID)       {}
-func (t *tagTracer) UndeliverableMessage(msg *Message) {}
+func (t *tagTracer) RemovePeer(peer.ID)                       {}
+func (t *tagTracer) ThrottlePeer(p peer.ID)                   {}
+func (t *tagTracer) RecvRPC(rpc *RPC)                         {}
+func (t *tagTracer) SendRPC(rpc *RPC, p peer.ID, urgent bool) {}
+func (t *tagTracer) DropRPC(rpc *RPC, p peer.ID)              {}
+func (t *tagTracer) UndeliverableMessage(msg *Message)        {}