Flushing Proxy Channels at CPU side upon reaching the Inflight Request Limit

include/mscclpp/proxy_channel.hpp

@@ -11,6 +11,8 @@
 
 namespace mscclpp {
 
+constexpr int MAX_INFLIGHT_REQUEST = 500;
+
 struct BaseProxyChannel;
 struct ProxyChannel;
 
@@ -72,6 +74,7 @@ class ProxyService : public BaseProxyService {
   std::vector<RegisteredMemory> memories_;
   std::shared_ptr<Proxy> proxy_;
   int deviceNumaNode;
+  int inflightRequests;
 
   void bindThread();
 

src/proxy_channel.cc

@@ -24,6 +24,7 @@ MSCCLPP_API_CPP ProxyService::ProxyService(size_t fifoSize)
   int cudaDevice;
   MSCCLPP_CUDATHROW(cudaGetDevice(&cudaDevice));
   deviceNumaNode = getDeviceNumaNode(cudaDevice);
+  inflightRequests = 0;
 }
 
 MSCCLPP_API_CPP SemaphoreId ProxyService::buildAndAddSemaphore(Communicator& communicator,
@@ -71,6 +72,13 @@ ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger triggerRaw) {
 
   auto result = ProxyHandlerResult::Continue;
 
+  inflightRequests++;
+  if (!(trigger->fields.type & TriggerSync) && inflightRequests > MAX_INFLIGHT_REQUEST) {
+    semaphore->connection()->flush();
+    result = ProxyHandlerResult::FlushFifoTailAndContinue;
+    inflightRequests = 0;
+  }
+
   if (trigger->fields.type & TriggerData) {
     RegisteredMemory& dst = memories_[trigger->fields.dstMemoryId];
     RegisteredMemory& src = memories_[trigger->fields.srcMemoryId];
@@ -85,6 +93,7 @@ ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger triggerRaw) {
   if (trigger->fields.type & TriggerSync) {
     semaphore->connection()->flush();
     result = ProxyHandlerResult::FlushFifoTailAndContinue;
+    inflightRequests = 0;
   }
 
   return result;