including sendRecv implementation and test (#27)

Author: csegarragonz

include/faabric/scheduler/MpiWorld.h

@@ -86,6 +86,16 @@ class MpiWorld
 
     void awaitAsyncRequest(int requestId);
 
+    void sendRecv(uint8_t* sendBuffer,
+                  int sendcount,
+                  faabric_datatype_t* sendDataType,
+                  int recvRank,
+                  uint8_t* recvBuffer,
+                  int recvCount,
+                  faabric_datatype_t* recvDataType,
+                  int sendRank,
+                  MPI_Status* status);
+
     void scatter(int sendRank,
                  int recvRank,
                  const uint8_t* sendBuffer,

src/proto/faabric.proto

@@ -35,6 +35,7 @@ message MPIMessage {
         ALLREDUCE = 8;
         ALLTOALL = 9;
         RMA_WRITE = 10;
+        SENDRECV = 11;
     };
 
     MPIMessageType messageType = 1;

src/scheduler/MpiWorld.cpp

@@ -312,6 +312,43 @@ void MpiWorld::send(int sendRank,
     }
 }
 
+void MpiWorld::sendRecv(uint8_t* sendBuffer,
+                        int sendCount,
+                        faabric_datatype_t* sendDataType,
+                        int recvRank,
+                        uint8_t* recvBuffer,
+                        int recvCount,
+                        faabric_datatype_t* recvDataType,
+                        int sendRank,
+                        MPI_Status* status)
+{
+    auto logger = faabric::util::getLogger();
+    logger->trace("MPI - Sendrecv");
+
+    if (recvRank > this->size - 1) {
+        throw std::runtime_error(fmt::format(
+          "Receive rank {} bigger than world size {}", recvRank, this->size));
+    }
+    if (sendRank > this->size - 1) {
+        throw std::runtime_error(fmt::format(
+          "Send rank {} bigger than world size {}", sendRank, this->size));
+    }
+
+    // Post async recv
+    int recvId = irecv(recvRank, sendRank, recvBuffer, recvDataType, recvCount);
+    // Then send the message
+    // TODO change MPIMessage to MPIMessage::SENDRECV. This requires a change
+    // in the signature of doISendRecv.
+    send(sendRank,
+         recvRank,
+         sendBuffer,
+         sendDataType,
+         sendCount,
+         faabric::MPIMessage::NORMAL);
+    // And wait
+    awaitAsyncRequest(recvId);
+}
+
 void MpiWorld::broadcast(int sendRank,
                          const uint8_t* buffer,
                          faabric_datatype_t* dataType,

tests/test/scheduler/test_mpi_world.cpp

@@ -229,6 +229,65 @@ TEST_CASE("Test send and recv on same host", "[mpi]")
     }
 }
 
+TEST_CASE("Test sendrecv", "[mpi]")
+{
+    cleanFaabric();
+
+    auto msg = faabric::util::messageFactory(user, func);
+    scheduler::MpiWorld world;
+    world.create(msg, worldId, worldSize);
+
+    // Register two ranks
+    int rankA = 1;
+    int rankB = 2;
+    world.registerRank(rankA);
+    world.registerRank(rankB);
+
+    // Prepare data
+    MPI_Status status{};
+    std::vector<int> messageDataAB = { 0, 1, 2 };
+    std::vector<int> messageDataBA = { 3, 2, 1, 0 };
+
+    // sendRecv is blocking, so we run two threads.
+    // Run sendrecv from A
+    std::vector<std::thread> threads;
+    threads.emplace_back([&] {
+        std::vector<int> recvBufferA(messageDataBA.size(), 0);
+        world.sendRecv(BYTES(messageDataAB.data()),
+                       messageDataAB.size(),
+                       MPI_INT,
+                       rankB,
+                       BYTES(recvBufferA.data()),
+                       messageDataBA.size(),
+                       MPI_INT,
+                       rankA,
+                       &status);
+        // Test integrity of results
+        REQUIRE(recvBufferA == messageDataBA);
+    });
+    // Run sendrecv from B
+    threads.emplace_back([&] {
+        std::vector<int> recvBufferB(messageDataAB.size(), 0);
+        world.sendRecv(BYTES(messageDataBA.data()),
+                       messageDataBA.size(),
+                       MPI_INT,
+                       rankA,
+                       BYTES(recvBufferB.data()),
+                       messageDataAB.size(),
+                       MPI_INT,
+                       rankB,
+                       &status);
+        // Test integrity of results
+        REQUIRE(recvBufferB == messageDataAB);
+    });
+    // Wait for both to finish
+    for (auto& t : threads) {
+        if (t.joinable()) {
+            t.join();
+        }
+    }
+}
+
 TEST_CASE("Test async send and recv", "[mpi]")
 {
     cleanFaabric();