mpi: use boost's lockfree ringbuffer for local messaging

include/faabric/mpi/MpiWorld.h

@@ -28,7 +28,7 @@ namespace faabric::mpi {
 // as the broker already has mocking capabilities
 std::vector<std::shared_ptr<MPIMessage>> getMpiMockedMessages(int sendRank);
 
-typedef faabric::util::FixedCapacityQueue<std::shared_ptr<MPIMessage>>
+typedef faabric::util::SpinLockQueue<std::shared_ptr<MPIMessage>>
   InMemoryMpiQueue;
 
 class MpiWorld

include/faabric/util/queue.h

@@ -4,6 +4,7 @@
 #include <faabric/util/locks.h>
 #include <faabric/util/logging.h>
 
+#include <boost/lockfree/spsc_queue.hpp>
 #include <condition_variable>
 #include <queue>
 #include <readerwriterqueue/readerwritercircularbuffer.h>
@@ -215,6 +216,38 @@ class FixedCapacityQueue
     moodycamel::BlockingReaderWriterCircularBuffer<T> mq;
 };
 
+// High-performance, spin-lock single-producer, single-consumer queue. This
+// queue spin-locks, so use at your own risk!
+template<typename T>
+class SpinLockQueue
+{
+  public:
+    void enqueue(T& value, long timeoutMs = DEFAULT_QUEUE_TIMEOUT_MS) {
+        while (!mq.push(value)) { ; };
+    }
+
+    T dequeue(long timeoutMs = DEFAULT_QUEUE_TIMEOUT_MS) {
+        T value;
+
+        while (!mq.pop(value)) { ; }
+
+        return value;
+    }
+
+    long size() {
+        throw std::runtime_error("Size for fast queue unimplemented!");
+    }
+
+    void drain() {
+        while (mq.pop()) { ; }
+    }
+
+    void reset() { ; }
+
+  private:
+    boost::lockfree::spsc_queue<T, boost::lockfree::capacity<1024>> mq;
+};
+
 class TokenPool
 {
   public:

src/mpi/MpiWorld.cpp

@@ -530,7 +530,7 @@ void MpiWorld::send(int sendRank,
     if (isLocal) {
         SPDLOG_TRACE(
           "MPI - send {} -> {} ({})", sendRank, recvRank, messageType);
-        getLocalQueue(sendRank, recvRank)->enqueue(std::move(m));
+        getLocalQueue(sendRank, recvRank)->enqueue(m);
     } else {
         SPDLOG_TRACE(
           "MPI - send remote {} -> {} ({})", sendRank, recvRank, messageType);
@@ -1395,15 +1395,17 @@ void MpiWorld::allToAll(int rank,
 // queues.
 void MpiWorld::probe(int sendRank, int recvRank, MPI_Status* status)
 {
+    throw std::runtime_error("MPI_Probe not supported!");
+    /*
     const std::shared_ptr<InMemoryMpiQueue>& queue =
       getLocalQueue(sendRank, recvRank);
-    // 30/12/21 - Peek will throw a runtime error
     std::shared_ptr<MPIMessage> m = *(queue->peek());
 
     faabric_datatype_t* datatype = getFaabricDatatypeFromId(m->type());
     status->bytesSize = m->count() * datatype->size;
     status->MPI_ERROR = 0;
     status->MPI_SOURCE = m->sender();
+    */
 }
 
 void MpiWorld::barrier(int thisRank)