[TensorpipeDeprecation] remove unused arguments from wait_for_senders() (#5917)

python/dgl/distributed/rpc.py

@@ -172,7 +172,7 @@ def finalize_receiver():
     _CAPI_DGLRPCFinalizeReceiver()
 
 
-def wait_for_senders(ip_addr, port, num_senders, blocking=True):
+def wait_for_senders(ip_addr, port, num_senders):
     """Wait all of the senders' connections.
 
     This api will be blocked until all the senders connect to the receiver.
@@ -185,10 +185,8 @@ def wait_for_senders(ip_addr, port, num_senders, blocking=True):
         receiver's port
     num_senders : int
         total number of senders
-    blocking : bool
-        whether to wait blockingly
     """
-    _CAPI_DGLRPCWaitForSenders(ip_addr, int(port), int(num_senders), blocking)
+    _CAPI_DGLRPCWaitForSenders(ip_addr, int(port), int(num_senders))
 
 
 def connect_receiver(ip_addr, port, recv_id, group_id=-1):

python/dgl/distributed/rpc_client.py

@@ -206,7 +206,7 @@ def connect_to_server(
     for server_id in range(num_servers):
         rpc.send_request(server_id, register_req)
     # wait server connect back
-    rpc.wait_for_senders(client_ip, client_port, num_servers, blocking=True)
+    rpc.wait_for_senders(client_ip, client_port, num_servers)
     print(
         "Client [{}] waits on {}:{}".format(os.getpid(), client_ip, client_port)
     )

python/dgl/distributed/rpc_server.py

@@ -85,7 +85,7 @@ def start_server(
     print(
         "Server is waiting for connections on [{}:{}]...".format(ip_addr, port)
     )
-    rpc.wait_for_senders(ip_addr, port, num_clients, blocking=True)
+    rpc.wait_for_senders(ip_addr, port, num_clients)
     rpc.set_num_client(num_clients)
     recv_clients = {}
     while True:

src/rpc/network/socket_communicator.cc

@@ -202,10 +202,8 @@ void SocketSender::SendLoop(
 
 /////////////////////////////////////// SocketReceiver
 //////////////////////////////////////////////
-bool SocketReceiver::Wait(
-    const std::string& addr, int num_sender, bool blocking) {
+bool SocketReceiver::Wait(const std::string& addr, int num_sender) {
   CHECK_GT(num_sender, 0);
-  CHECK_EQ(blocking, true);
   std::vector<std::string> substring;
   std::vector<std::string> ip_and_port;
   SplitStringUsing(addr, "//", &substring);

src/rpc/network/socket_communicator.h

@@ -159,12 +159,11 @@ class SocketReceiver : public Receiver {
    * @brief Wait for all the Senders to connect
    * @param addr Networking address, e.g., 'tcp://127.0.0.1:50051', 'mpi://0'
    * @param num_sender total number of Senders
-   * @param blocking whether wait blockingly
    * @return True for success and False for fail
    *
    * Wait() is not thread-safe and only one thread can invoke this API.
    */
-  bool Wait(const std::string& addr, int num_sender, bool blocking = true);
+  bool Wait(const std::string& addr, int num_sender);
 
   /**
    * @brief Recv RPCMessage from Sender. Actually removing data from queue.

src/rpc/rpc.cc

@@ -90,11 +90,10 @@ DGL_REGISTER_GLOBAL("distributed.rpc._CAPI_DGLRPCWaitForSenders")
       std::string ip = args[0];
       int port = args[1];
       int num_sender = args[2];
-      bool blocking = args[3];
       std::string addr;
       addr = StringPrintf("tcp://%s:%d", ip.c_str(), port);
-      if (RPCContext::getInstance()->receiver->Wait(
-              addr, num_sender, blocking) == false) {
+      if (RPCContext::getInstance()->receiver->Wait(addr, num_sender) ==
+          false) {
         LOG(FATAL) << "Wait sender socket failed.";
       }
     });