Allocate a backing buffer in storage.cpp

Enclave/route.cpp

@@ -110,6 +110,10 @@ bool route_init()
         if (my_roles & ROLE_STORAGE) {
             route_state.round2.alloc(tot_msg_per_stg +
                 g_teems_config.tot_weight);
+            if (!storage_init(tot_msg_per_stg +
+                g_teems_config.tot_weight)) {
+                return false;
+            }
         }
     } catch (std::bad_alloc&) {
         printf("Memory allocation failed in route_init\n");
@@ -679,14 +683,11 @@ void ecall_routing_proceed(void *cbpointer)
 #ifdef PROFILE_ROUTING
             unsigned long start = printf_with_rtclock("begin storage processing (%u)\n", round2.inserted);
 #endif
-            storage_received(round2.buf, round2.inserted);
+            storage_received(round2);
 #ifdef PROFILE_ROUTING
             printf_with_rtclock_diff(start, "end storage processing (%u)\n", round2.inserted);
 #endif
 
-            round2.reset();
-            pthread_mutex_unlock(&round2.mutex);
-
             // We're done
             route_state.step = ROUTE_NOT_STARTED;
             ocall_routing_round_complete(cbpointer, 0);

Enclave/storage.cpp

@@ -3,13 +3,33 @@
 #include "storage.hpp"
 #include "ORExpand.hpp"
 
-// Handle the messages received by a storage node
-void storage_received(const uint8_t *msgs, uint32_t num_msgs)
+static struct {
+    // A local storage buffer, used when we need to do non-in-place
+    // sorts of the messages that have arrived
+    MsgBuffer stg_buf;
+} storage_state;
+
+// route_init will call this function; no one else should call it
+// explicitly.  The parameter is the number of messages that can fit in
+// the storage-side MsgBuffer.  Returns true on success, false on
+// failure.
+bool storage_init(uint32_t msg_buf_size)
+{
+    storage_state.stg_buf.alloc(msg_buf_size);
+    return true;
+}
+
+// Handle the messages received by a storage node.  Pass a _locked_
+// MsgBuffer.  This function will itself reset and unlock it when it's
+// done with it.
+void storage_received(MsgBuffer &storage_buf)
 {
     // A dummy function for now that just counts how many real and
     // padding messages arrived
     uint16_t msg_size = g_teems_config.msg_size;
     nodenum_t my_node_num = g_teems_config.my_node_num;
+    const uint8_t *msgs = storage_buf.buf;
+    uint32_t num_msgs = storage_buf.inserted;
     uint32_t real = 0, padding = 0;
     uint32_t uid_mask = (1 << DEST_UID_BITS) - 1;
 
@@ -32,4 +52,7 @@ void storage_received(const uint8_t *msgs, uint32_t num_msgs)
         msgs += msg_size;
     }
     printf("%u real, %u padding\n", real, padding);
+
+    storage_buf.reset();
+    pthread_mutex_unlock(&storage_buf.mutex);
 }

Enclave/storage.hpp

@@ -3,7 +3,17 @@
 
 #include <cstdint>
 
-// Handle the messages received by a storage node
-void storage_received(const uint8_t *msgs, uint32_t num_msgs);
+#include "route.hpp"
+
+// route_init will call this function; no one else should call it
+// explicitly.  The parameter is the number of messages that can fit in
+// the storage-side MsgBuffer.  Returns true on success, false on
+// failure.
+bool storage_init(uint32_t msg_buf_size);
+
+// Handle the messages received by a storage node.  Pass a _locked_
+// MsgBuffer.  This function will itself reset and unlock it when it's
+// done with it.
+void storage_received(MsgBuffer &storage_buf);
 
 #endif