Adding testRecv and testSend for unbounded_buffer

gloo/CMakeLists.txt

@@ -43,6 +43,7 @@ list(APPEND GLOO_HDRS
   "${CMAKE_CURRENT_SOURCE_DIR}/broadcast_one_to_all.h"
   "${CMAKE_CURRENT_SOURCE_DIR}/context.h"
   "${CMAKE_CURRENT_SOURCE_DIR}/gather.h"
+  "${CMAKE_CURRENT_SOURCE_DIR}/gatherv.h"
   "${CMAKE_CURRENT_SOURCE_DIR}/math.h"
   "${CMAKE_CURRENT_SOURCE_DIR}/pairwise_exchange.h"
   "${CMAKE_CURRENT_SOURCE_DIR}/reduce.h"

gloo/transport/tcp/unbound_buffer.cc

@@ -203,6 +203,12 @@ void UnboundBuffer::throwIfException() {
     std::rethrow_exception(ex_);
   }
 }
+bool UnboundBuffer::testRecv() {
+  return recvCompletions_ != 0;
+}
+bool UnboundBuffer::testSend() {
+  return sendCompletions_ != 0;
+}
 
 } // namespace tcp
 } // namespace transport

gloo/transport/tcp/unbound_buffer.h

@@ -61,6 +61,9 @@ class UnboundBuffer : public ::gloo::transport::UnboundBuffer {
   void handleRecvCompletion(int rank);
   void handleSendCompletion(int rank);
 
+  bool testRecv() override;
+  bool testSend() override;
+
  protected:
   std::shared_ptr<Context> context_;
 

gloo/transport/unbound_buffer.cc

@@ -6,13 +6,20 @@
  * LICENSE file in the root directory of this source tree.
  */
 
+#include <cstdlib>
 #include "gloo/transport/unbound_buffer.h"
 
 namespace gloo {
 namespace transport {
 
 // Have to provide implementation for pure virtual destructor.
 UnboundBuffer::~UnboundBuffer() {}
+bool UnboundBuffer::testRecv()  {
+  abort();
+}
+bool UnboundBuffer::testSend() {
+  abort();
+}
 
 } // namespace transport
 } // namespace gloo

gloo/transport/unbound_buffer.h

@@ -117,6 +117,14 @@ class UnboundBuffer {
       uint64_t slot,
       size_t offset = 0,
       size_t nbytes = kUnspecifiedByteCount) = 0;
+
+  // Tests if a reception is possibly complete. Only valid if called after
+  // `recv` and before `waitRecv`
+  virtual bool testRecv();
+
+  // Tests if a sending is possibly complete. Only valid if called after
+  // `send` and before `waitSend`
+  virtual bool testSend();
 };
 
 } // namespace transport