Fixed the bug with distibuting clients over their ingestion servers. Clients are now distributed over ingestion servers with the same modulo calculation like clients to storage server mapping

Client/clients.cpp

@@ -339,13 +339,11 @@ int generateClientKeys(clientid_t client_number, aes_key &ESK,
     }
 
     /*
-    if(client_number % 10 == 0) {
-        printf("Client %d Storage Key: (after Gen) ", client_number);
-        for(int i=0;i<SGX_AESGCM_KEY_SIZE;i++) {
-            printf("%x", client_stg_key[i]);
-        }
-        printf("\n");
+    printf("Client %d Ingestion: (after Gen) ", client_number);
+    for(int i=0;i<SGX_AESGCM_KEY_SIZE;i++) {
+        printf("%x", client_ing_key[i]);
     }
+    printf("\n");
     */
 
     return 1;
@@ -632,16 +630,10 @@ void generateClients(boost::asio::io_context &io_context,
     uint32_t num_clients_total = config.user_count;
     uint16_t num_stg_nodes = storage_nodes.size();
     uint16_t num_ing_nodes = ingestion_nodes.size();
-    uint32_t clients_per_ing = CEILDIV(num_clients_total, num_ing_nodes);
-    uint16_t ing_with_additional = num_clients_total % num_ing_nodes;
 
     for(uint32_t i=cstart; i<cstop; i++) {
-        uint16_t ing_no = i/clients_per_ing;
-        if(ing_no > ing_with_additional && ing_with_additional!=0) {
-            uint16_t leftover = num_clients_total - (ing_with_additional * clients_per_ing);
-            ing_no = ing_with_additional + (leftover / (clients_per_ing-1));
-        }
 
+        uint16_t ing_no = i % num_ing_nodes;
         uint16_t stg_no = i % num_stg_nodes;
         uint16_t stg_node_id = storage_map[stg_no];
         uint16_t ing_node_id = ingestion_map[ing_no];
@@ -779,7 +771,7 @@ void initializeClients(boost::asio::io_context &io_context, uint16_t nthreads)
     for(int i=0; i<nthreads; i++) {
         uint32_t cstart, cstop;
         cstart = i * clients_per_thread;
-        cstop = (i==nthreads-1)? num_clients_total: (i+1) * clients_per_thread;
+        cstop = (i==(nthreads-1))? num_clients_total: (i+1) * clients_per_thread;
 
 #ifdef VERBOSE_CLIENT
         printf("Thread %d, cstart = %d, cstop = %d\n", i, cstart, cstop);

Enclave/config.cpp

@@ -155,18 +155,8 @@ bool ecall_config_load(threadid_t nthreads, bool private_routing,
         if(my_role & ROLE_INGESTION) {
             uint32_t num_ing_nodes = g_teems_config.num_ingestion_nodes;
             uint32_t clients_per_server = CEILDIV(num_clients_total, num_ing_nodes);
-            uint32_t ing_with_additional = num_clients_total % num_ing_nodes;
-            uint32_t num_smaller_ing = (uint32_t) ing_smaller.size();
             uint32_t num_clients_this_ing = clients_per_server;
-            num_clients_this_ing += (num_smaller_ing < ing_with_additional)? 1: 0;
-            uint32_t client_start = num_smaller_ing * clients_per_server;
-            if (ing_with_additional > 0) {
-                if(ing_with_additional > num_smaller_ing) {
-                    client_start+= num_smaller_ing;
-                } else {
-                    client_start+= ing_with_additional;
-                }
-            }
+            uint32_t client_start = ing_smaller.size();
             g_ing.initialize(num_clients_this_ing, client_start, g_teems_config.ESK);
         }
     }

Enclave/ingest.cpp

@@ -73,7 +73,8 @@ void Ingestion::initialize(uint32_t num, uint32_t start, sgx_aes_gcm_128bit_key_
 
 bool Ingestion::authenticate(clientid_t cid, unsigned char *auth_message)
 {
-    uint32_t lcid = cid-cstart;
+    uint32_t num_ing_nodes = g_teems_config.num_ingestion_nodes;
+    uint32_t lcid = cid / num_ing_nodes;
     const sgx_aes_gcm_128bit_key_t *ckey = &(clients[lcid].key);
     return(authenticateClient(auth_message, ckey));
 }
@@ -82,7 +83,8 @@ bool Ingestion::processMsgBundle(clientid_t cid, unsigned char *msgbundle,
     uint32_t num_msgs) {
 
     // Fetch corresponding client key
-    clientid_t lcid = cid - g_ing.cstart;
+    uint32_t num_ing_nodes = g_teems_config.num_ingestion_nodes;
+    clientid_t lcid = cid / num_ing_nodes;
     sgx_aes_gcm_128bit_key_t *ckey = &(clients[lcid].key);
     unsigned char *iv = msgbundle;
     msgbundle += SGX_AESGCM_IV_SIZE;
@@ -120,13 +122,6 @@ bool Ingestion::processMsgBundle(clientid_t cid, unsigned char *msgbundle,
         msg_queue.reserved += num_msgs;
         pthread_mutex_unlock(&msg_queue.mutex);
 
-        /*
-        if(lcid==0) {
-            printf("\n\nIngestion: Message for lcid 0, S, R = %ld, %ld\n\n\n", *((uint32_t*) dec_msgbundle),
-                *((uint32_t*) (dec_msgbundle + 4)));
-        }
-        */
-
         memmove(msg_queue.buf + head * msg_size,
             dec_msgbundle, num_msgs * msg_size);
 
@@ -143,8 +138,10 @@ bool Ingestion::processMsgBundle(clientid_t cid, unsigned char *msgbundle,
 
 void Ingestion::generateClientKeys(sgx_aes_gcm_128bit_key_t &ESK)
 {
-    printf("In Ingestion::genCK, num_clients = %d, client_start = %d, client_end = %d\n",
-        cnum, cstart, cnum + cstart);
+    //printf("In Ingestion::genCK, num_clients = %d, client_start = %d, client_end = %d\n",
+    //    cnum, cstart, cnum + cstart);
+    uint32_t num_ing_nodes = g_teems_config.num_ingestion_nodes;
+
     for(uint32_t i=0; i<cnum; i++)
     {
         unsigned char zeroes[SGX_AESGCM_KEY_SIZE];
@@ -153,7 +150,7 @@ void Ingestion::generateClientKeys(sgx_aes_gcm_128bit_key_t &ESK)
         memset(zeroes, 0, SGX_AESGCM_KEY_SIZE);
         memset(iv, 0, SGX_AESGCM_IV_SIZE);
 
-        uint32_t client_num = cstart + i;
+        uint32_t client_num = cstart + (i * num_ing_nodes);
         memcpy(iv, (uint8_t*) (&client_num), sizeof(client_num));
 
         sgx_status_t ret = SGX_SUCCESS;
@@ -164,6 +161,14 @@ void Ingestion::generateClientKeys(sgx_aes_gcm_128bit_key_t &ESK)
             printf("Ingestion::GCK FAIL\n");
         }
 
+        /*
+        printf("Client %d Ingestion: (after Gen) ", client_num);
+        for(int j=0;j<SGX_AESGCM_KEY_SIZE;j++) {
+            printf("%x", (clients[i].key)[j]);
+        }
+        printf("\n");
+        */
+
     }
 }