Ingestion processing uses the ingbuf in route_state now. Removed unnecessary memcpy's in ingestion processing. Clients maintain a SimID and a ClientID, and send messages to the correct ClientID format.

App/net.cpp

@@ -11,6 +11,7 @@
 #define COMMAND_CHUNK 0x02
 
 #define VERBOSE_NET
+// #define DEBUG_NET_CLIENTS
 NetIO *g_netio = NULL;
 
 NodeIO::NodeIO(tcp::socket &&socket, nodenum_t nodenum) :
@@ -237,7 +238,7 @@ void NetIO::authenticate_new_client(tcp::socket* csocket,
         return;
     }
 
-#ifdef VERBOSE_NET
+#ifdef DEBUG_NET_CLIENTS
         printf("Accept handler success\n");
 #endif
 
@@ -288,7 +289,7 @@ void NetIO::authenticate_new_client(tcp::socket* csocket,
 void NetIO::start_accept()
 {
     tcp::socket *csocket = new tcp::socket(io_context());
-#ifdef VERBOSE_NET
+#ifdef DEBUG_NET_CLIENTS
     std::cout << "Accepting on " << myconf.clistenhost << ":" << myconf.clistenport << "\n";
 #endif
     client_acceptor->async_accept(*csocket,

App/start.cpp

@@ -119,6 +119,68 @@ static void epoch(NetIO &netio, char **args) {
     ++epoch_num;
 }
 
+static void epoch_clients(NetIO &netio) {
+
+    static uint32_t epoch_num = 1;
+    Epoch epoch(netio.io_context(), epoch_num);
+    epoch.proceed();
+    epoch.wait();
+    // Launch threads to refill the precomputed Waksman networks we
+    // used, but just let them run in the background.
+    size_t num_sizes = ecall_precompute_sort(-1);
+    for (int i=0;i<int(num_sizes);++i) {
+        boost::thread t([i] {
+            ecall_precompute_sort(i);
+        });
+        t.detach();
+    }
+    ++epoch_num;
+}
+
+static void route_clients_test(NetIO &netio)
+{
+
+    // Precompute some WaksmanNetworks
+    const Config &config = netio.config();
+    size_t num_sizes = ecall_precompute_sort(-1);
+    for (int i=0;i<int(num_sizes);++i) {
+        std::vector<boost::thread> ts;
+        for (int j=0; j<config.nthreads; ++j) {
+            ts.emplace_back([i] {
+                ecall_precompute_sort(i);
+            });
+        }
+        for (auto& t: ts) {
+            t.join();
+        }
+    }
+
+    // The epoch interval, in microseconds
+    uint32_t epoch_interval_us = 1000000;
+    printf("Waiting on client's first messages\n");
+    sleep(3);
+
+    // Run epoch
+    for (int i=0; i<1; ++i) {
+        struct timespec tp;
+        clock_gettime(CLOCK_REALTIME_COARSE, &tp);
+        unsigned long start = tp.tv_sec * 1000000 + tp.tv_nsec/1000;
+
+        epoch_clients(netio);
+
+        clock_gettime(CLOCK_REALTIME_COARSE, &tp);
+        unsigned long end = tp.tv_sec * 1000000 + tp.tv_nsec/1000;
+        unsigned long diff = end - start;
+        printf("Epoch time: %lu.%06lu s\n", diff/1000000, diff%1000000);
+        // Sleep for the rest of the epoch interval
+        if (diff < epoch_interval_us) {
+            usleep(epoch_interval_us - (useconds_t)diff);
+        }
+    }
+    netio.close();
+}
+
+
 static void route_test(NetIO &netio, char **args)
 {
     // Count the number of arguments
@@ -187,4 +249,10 @@ void start(NetIO &netio, char **args)
         route_test(netio, args);
         return;
     }
+
+    if (*args && !strcmp(*args, "route_clients")) {
+        ++args;
+        route_clients_test(netio);
+        return;
+    }
 }

Client/clients.cpp

@@ -70,6 +70,7 @@ void displayMessage(unsigned char *msg, uint16_t msg_size) {
     sid = *((clientid_t*) ptr);
     ptr+=sizeof(sid);
     rid = *((clientid_t*) ptr);
+    ptr+=sizeof(sid);
     printf("Sender ID: %d, Receiver ID: %d, Token: N/A\n", sid, rid );
     printf("Message: ");
     for(int j = 0; j<msg_size - sizeof(sid)*2; j++) {
@@ -127,7 +128,8 @@ static inline uint32_t ptMsgBundleSize(uint16_t priv_out, uint16_t msg_size) {
 
 bool config_parse(Config &config, const std::string configstr,
     std::vector<NodeConfig> &ingestion_nodes,
-    std::vector<NodeConfig> &storage_nodes)
+    std::vector<NodeConfig> &storage_nodes,
+    std::vector<uint16_t> &storage_map)
 {
     bool found_params = false;
     bool ret = true;
@@ -136,6 +138,7 @@ bool config_parse(Config &config, const std::string configstr,
     boost::property_tree::ptree conftree;
 
     read_json(configstream, conftree);
+    uint16_t node_num = 0;
 
     for (auto & entry : conftree) {
         if (!entry.first.compare("params")) {
@@ -165,6 +168,7 @@ bool config_parse(Config &config, const std::string configstr,
             }
             found_params = true;
         } else if (!entry.first.compare("nodes")) {
+
             for (auto & node : entry.second) {
                 NodeConfig nc;
                 // All nodes need to be assigned their role in manifest.yaml
@@ -197,7 +201,9 @@ bool config_parse(Config &config, const std::string configstr,
                 }
                 if(nc.roles & ROLE_STORAGE) {
                     storage_nodes.push_back(std::move(nc));
+                    storage_map.push_back(node_num);
                 }
+                node_num++;
             }
         } else {
             std::cerr << "Unknown key in config: " <<
@@ -340,7 +346,7 @@ void Client::generateMessageBundle(uint8_t priv_out, uint32_t msg_size,
     unsigned char *pt_msgbundle)
 {
     unsigned char *ptr = pt_msgbundle;
-    uint64_t header = (id << 8) + CLIENT_MESSAGE_BUNDLE;
+    uint64_t header = (sim_id << 8) + CLIENT_MESSAGE_BUNDLE;
 
     // Setup header
     memcpy(ptr, (uint8_t*) &header, sizeof(header));
@@ -350,7 +356,8 @@ void Client::generateMessageBundle(uint8_t priv_out, uint32_t msg_size,
     for(uint32_t i = 0; i < priv_out; i++) {
         memcpy(ptr, &id, sizeof(id));
         ptr+=(sizeof(id));
-        memcpy(ptr, &id, sizeof(i));
+
+        memcpy(ptr, &id, sizeof(id));
         ptr+=(sizeof(id));
 
         uint32_t remaining_message_size = msg_size - (sizeof(id)*2);
@@ -401,7 +408,9 @@ void Client::sendMessageBundle(uint16_t priv_out, uint16_t msg_size,
 
     encryptMessageBundle(enc_bundle_size, pt_msgbundle, enc_msgbundle);
 
+#ifdef VERBOSE_CLIENT
     displayPtMessageBundle(pt_msgbundle, priv_out, msg_size);
+#endif
 
     //displayEncMessageBundle(enc_msgbundle, priv_out, msg_size);
 
@@ -414,7 +423,7 @@ int Client::sendAuthMessage()
     uint32_t auth_size = sizeof(uint64_t) + SGX_AESGCM_KEY_SIZE;
     unsigned char *auth_string = (unsigned char*) malloc(auth_size);
     unsigned char *as_ptr = auth_string;
-    uint64_t header = (id << 8) + CLIENT_AUTHENTICATE;
+    uint64_t header = (sim_id << 8) + CLIENT_AUTHENTICATE;
     memcpy(as_ptr, &header, sizeof(header));
     as_ptr+=sizeof(header);
 
@@ -434,11 +443,13 @@ int Client::sendAuthMessage()
     uint64_t *iv_ctr = (uint64_t*) iv;
     (*iv_ctr)+=1;
 
+#ifdef VERBOSE_CLIENT
     printf("Client %d auth_string: \n", id);
     for(int i=0; i<auth_size; i++) {
         printf("%x", auth_string[i]);
     }
     printf("\n");
+#endif
 
     boost::asio::write(*ingestion_sock,
         boost::asio::buffer(auth_string, auth_size));
@@ -449,10 +460,13 @@ int Client::sendAuthMessage()
 void generateClients(boost::asio::io_context &io_context,
     uint32_t cstart, uint32_t cstop, Client* &clients,
     aes_key &EMK, Config &config, std::vector<NodeConfig> &ingestion_nodes,
+    std::vector<NodeConfig> &storage_nodes, std::vector<uint16_t> &storage_map,
     uint32_t num_clients_total, uint32_t clients_per_ing,
     uint32_t ing_with_additional)
 {
     aes_key client_key;
+    uint16_t num_stg_nodes = storage_nodes.size();
+    uint16_t *stg_map = new uint16_t[num_stg_nodes];
 
     for(uint32_t i=cstart; i<cstop; i++) {
         uint16_t ing_node_this_client = i/clients_per_ing;
@@ -462,7 +476,7 @@ void generateClients(boost::asio::io_context &io_context,
         }
 
         int ret = generateClientEncryptionKey(i, EMK, client_key);
-        clients[i].initClient(i, client_key);
+        clients[i].initClient(i, client_key, num_stg_nodes, storage_map);
 
         clients[i].initializeSocket(io_context, ingestion_nodes[ing_node_this_client]);
         clients[i].sendAuthMessage();
@@ -516,6 +530,7 @@ int main(int argc, char **argv)
     uint16_t nthreads = 1;
     const char *progname = argv[0];
     std::vector<NodeConfig> ingestion_nodes, storage_nodes;
+    std::vector<uint16_t> storage_map;
     ++argv;
 
     // Parse options
@@ -538,12 +553,16 @@ int main(int argc, char **argv)
     std::string configstr;
     std::getline(std::cin, configstr);
 
+    // The epoch interval, in microseconds
+    uint32_t epoch_interval_us = 1000000;
+
     Config config;
     aes_key EMK, TMK;
     boost::asio::io_context io_context;
     boost::asio::ip::tcp::resolver resolver(io_context);
 
-    if (!config_parse(config, configstr, ingestion_nodes, storage_nodes)) {
+    if (!config_parse(config, configstr, ingestion_nodes,
+        storage_nodes, storage_map)) {
        exit(1);
     }
 
@@ -570,7 +589,8 @@ int main(int argc, char **argv)
         printf("Thread %d, cstart = %d, cstop = %d\n", i, cstart, cstop);
         threads[i] = std::thread(generateClients, std::ref(io_context),
             cstart, cstop, std::ref(clients), std::ref(EMK), std::ref(config),
-            std::ref(ingestion_nodes), num_clients_total,
+            std::ref(ingestion_nodes), std::ref(storage_nodes),
+            std::ref(storage_map), num_clients_total,
             clients_per_ing, ing_with_additional);
     }
 
@@ -581,6 +601,10 @@ int main(int argc, char **argv)
     // Multithreaded client message bundle generation and send
     uint32_t epoch = 0;
     while(epoch < 1) {
+        struct timespec tp;
+        clock_gettime(CLOCK_REALTIME_COARSE, &tp);
+        unsigned long start = tp.tv_sec * 1000000 + tp.tv_nsec/1000;
+
         for(int i=0; i<nthreads; i++) {
             uint32_t cstart, cstop;
             cstart = i * clients_per_thread;
@@ -591,7 +615,19 @@ int main(int argc, char **argv)
         for(int i=0; i<nthreads; i++) {
             threads[i].join();
         }
+
+        clock_gettime(CLOCK_REALTIME_COARSE, &tp);
+        unsigned long end = tp.tv_sec * 1000000 + tp.tv_nsec/1000;
+        unsigned long diff = end - start;
+        // Sleep for the rest of the epoch interval
+        printf("Done with submissions for 1 epoch \n");
+        if (diff < epoch_interval_us) {
+            printf("diff = %ld\n", diff);
+            usleep(epoch_interval_us - (useconds_t)diff);
+        }
         epoch++;
     }
+
+    sleep(10000);
     delete [] clients;
 }

Enclave/ingest.cpp

@@ -66,9 +66,8 @@ void Ingestion::initialize(uint32_t cnum, uint32_t cstart, sgx_aes_gcm_128bit_ke
     clients.keys = new sgx_aes_gcm_128bit_key_t[cnum];
     generateClientKeys(ESK);
 
-    // Initialize the MsgBuffer to correct size
     max_buffer_size = g_teems_config.m_priv_out * cnum;
-    buffer.alloc(max_buffer_size);
+    buffer = &(route_state.ingbuf);
 }
 
 bool Ingestion::authenticate(clientid_t cid, unsigned char *auth_string)
@@ -76,8 +75,7 @@ bool Ingestion::authenticate(clientid_t cid, unsigned char *auth_string)
     int auth_success = 0;
     // Fetch corresponding client key
     clientid_t lcid = cid - g_ing.clients.start;
-    sgx_aes_gcm_128bit_key_t ckey;
-    memcpy(ckey, (g_ing.clients).keys[lcid], SGX_AESGCM_KEY_SIZE);
+    sgx_aes_gcm_128bit_key_t &ckey = (g_ing.clients).keys[lcid];
 
     unsigned char computed_auth[SGX_AESGCM_KEY_SIZE];
     unsigned char zeroes[SGX_AESGCM_KEY_SIZE];
@@ -96,7 +94,6 @@ bool Ingestion::authenticate(clientid_t cid, unsigned char *auth_string)
     auth_success = memcmp(auth_string, computed_auth, SGX_AESGCM_KEY_SIZE);
 
     if(auth_success == 0) {
-        printf("authentication SUCCESS\n");
         return true;
     } else {
         printf("authentication FAIL\n");
@@ -109,33 +106,32 @@ bool Ingestion::processMsgBundle(clientid_t cid, unsigned char *msgbundle,
 
     // Fetch corresponding client key
     clientid_t lcid = cid - g_ing.clients.start;
-    sgx_aes_gcm_128bit_key_t ckey;
-    memcpy(ckey, (g_ing.clients).keys[lcid], SGX_AESGCM_KEY_SIZE);
+    sgx_aes_gcm_128bit_key_t &ckey = (g_ing.clients).keys[lcid];
     unsigned char *iv = msgbundle;
     msgbundle += SGX_AESGCM_IV_SIZE;
 
     uint16_t msg_size = g_teems_config.msg_size;
     uint32_t msgbundle_size = num_msgs * msg_size;
     unsigned char *dec_msgbundle = (unsigned char *) malloc (msgbundle_size);
-    sgx_aes_gcm_128bit_tag_t tag;
-    memcpy(tag, msgbundle + msgbundle_size, SGX_AESGCM_MAC_SIZE);
-    //sgx_aes_gcm_128bit_tag_t *tag = (sgx_aes_gcm_128bit_tag_t*) (msgbundle + msgbundle_size);
+    //sgx_aes_gcm_128bit_tag_t tag;
+    //memcpy(tag, msgbundle + msgbundle_size, SGX_AESGCM_MAC_SIZE);
+    sgx_aes_gcm_128bit_tag_t *tag = (sgx_aes_gcm_128bit_tag_t*) (msgbundle + msgbundle_size);
 
     sgx_status_t ret = sgx_rijndael128GCM_decrypt(&ckey, msgbundle, msgbundle_size,
-        dec_msgbundle, iv, SGX_AESGCM_IV_SIZE, NULL, 0, &tag);
+        dec_msgbundle, iv, SGX_AESGCM_IV_SIZE, NULL, 0, tag);
     if(ret!=SGX_SUCCESS) {
         printf("Ingestion::processMsgBundle FAIL\n");
         printf("Error code: %d", (uint32_t) ret);
     }
 
     // Append msgbundle to g_ing.buffer;
-    MsgBuffer &msg_queue = g_ing.buffer;
+    MsgBuffer &msg_queue = *(g_ing.buffer);
 
     pthread_mutex_lock(&msg_queue.mutex);
     uint32_t head = msg_queue.reserved;
     if (head + num_msgs > g_ing.max_buffer_size) {
         pthread_mutex_unlock(&msg_queue.mutex);
-        printf("Max %u messages exceeded\n",
+        printf("Ingestions: Max %u messages exceeded\n",
             g_ing.max_buffer_size);
         return false;
     }

Enclave/ingest.hpp

@@ -1,6 +1,8 @@
 #ifndef __INGEST_HPP__
 #define __INGEST_HPP__
 
+//#define DEBUG_INGESTION
+
 struct ClientList {
     uint32_t num;
     uint32_t start;
@@ -12,7 +14,7 @@ class Ingestion{
 private:
 
     ClientList clients;
-    MsgBuffer buffer;
+    MsgBuffer *buffer;
     uint32_t max_buffer_size;
 
 public:

Enclave/route.cpp

@@ -9,29 +9,7 @@
 
 #define PROFILE_ROUTING
 
-enum RouteStep {
-    ROUTE_NOT_STARTED,
-    ROUTE_ROUND_1,
-    ROUTE_ROUND_2
-};
-
-// The ingbuf MsgBuffer stores messages an ingestion node ingests while
-// waiting for round 1 to start, which will be sorted and sent out in
-// round 1.  The round1 MsgBuffer stores messages a routing node
-// receives in round 1, which will be padded, sorted, and sent out in
-// round 2.  The round2 MsgBuffer stores messages a storage node
-// receives in round 2.
-
-static struct RouteState {
-    MsgBuffer ingbuf;
-    MsgBuffer round1;
-    MsgBuffer round2;
-    RouteStep step;
-    uint32_t tot_msg_per_ing;
-    uint32_t max_msg_to_each_stg;
-    uint32_t max_round2_msgs;
-    void *cbpointer;
-} route_state;
+RouteState route_state;
 
 // Computes ceil(x/y) where x and y are integers, x>=0, y>0.
 

Enclave/route.hpp

@@ -55,6 +55,33 @@ struct MsgBuffer {
     MsgBuffer &operator=(const MsgBuffer&) = delete;
 };
 
+
+enum RouteStep {
+    ROUTE_NOT_STARTED,
+    ROUTE_ROUND_1,
+    ROUTE_ROUND_2
+};
+
+// The ingbuf MsgBuffer stores messages an ingestion node ingests while
+// waiting for round 1 to start, which will be sorted and sent out in
+// round 1.  The round1 MsgBuffer stores messages a routing node
+// receives in round 1, which will be padded, sorted, and sent out in
+// round 2.  The round2 MsgBuffer stores messages a storage node
+// receives in round 2.
+
+struct RouteState {
+    MsgBuffer ingbuf;
+    MsgBuffer round1;
+    MsgBuffer round2;
+    RouteStep step;
+    uint32_t tot_msg_per_ing;
+    uint32_t max_msg_to_each_stg;
+    uint32_t max_round2_msgs;
+    void *cbpointer;
+};
+
+extern RouteState route_state;
+
 // Call this near the end of ecall_config_load, but before
 // comms_init_nodestate. Returns true on success, false on failure.
 bool route_init();