route.cpp 61 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438
  1. #include "Enclave_t.h"
  2. #include "config.hpp"
  3. #include "utils.hpp"
  4. #include "sort.hpp"
  5. #include "comms.hpp"
  6. #include "obliv.hpp"
  7. #include "storage.hpp"
  8. #include "route.hpp"
  9. #define PROFILE_ROUTING
  10. #define TRACE_ROUTING
  11. RouteState route_state;
  12. // Computes ceil(x/y) where x and y are integers, x>=0, y>0.
  13. #define CEILDIV(x,y) (((x)+(y)-1)/(y))
  14. #ifdef TRACE_ROUTING
  15. // Show (the first 300 of, if there are more) the headers and the first
  16. // few bytes of the body of each message in the buffer
  17. static void show_messages(const char *label, const unsigned char *buffer,
  18. size_t num)
  19. {
  20. if (label) {
  21. printf("%s\n", label);
  22. }
  23. for (size_t i=0; i<num && i<300; ++i) {
  24. const uint32_t *ibuf = (const uint32_t *)buffer;
  25. if (g_teems_config.private_routing) {
  26. printf("%3d R:%08x S:%08x [%08x]\n", i, ibuf[0], ibuf[1],
  27. ibuf[2]);
  28. } else {
  29. printf("%3d R:%08x P:%08x S:%08x [%08x]\n", i, ibuf[0], ibuf[1],
  30. ibuf[2], ibuf[3]);
  31. }
  32. buffer += g_teems_config.msg_size;
  33. }
  34. printf("\n");
  35. }
  36. #endif
  37. // Call this near the end of ecall_config_load, but before
  38. // comms_init_nodestate. Returns true on success, false on failure.
  39. bool route_init()
  40. {
  41. // Compute the maximum number of messages we could receive by direct
  42. // ingestion
  43. // Each ingestion node will have at most
  44. // ceil(user_count/num_ingestion_nodes) users, and each user will
  45. // send at most m_priv_out messages.
  46. uint32_t users_per_ing = CEILDIV(g_teems_config.user_count,
  47. g_teems_config.num_ingestion_nodes);
  48. uint32_t tot_msg_per_ing;
  49. if (g_teems_config.private_routing) {
  50. tot_msg_per_ing = users_per_ing * g_teems_config.m_priv_out;
  51. } else {
  52. tot_msg_per_ing = users_per_ing * g_teems_config.m_pub_out;
  53. }
  54. // Compute the maximum number of messages we could receive in round 1
  55. // In private routing, each ingestion node will send us an
  56. // our_weight/tot_weight fraction of the messages they hold
  57. uint32_t max_msg_from_each_ing;
  58. max_msg_from_each_ing = CEILDIV(tot_msg_per_ing, g_teems_config.tot_weight) *
  59. g_teems_config.my_weight;
  60. // And the maximum number we can receive in total is that times the
  61. // number of ingestion nodes
  62. uint32_t max_round1_msgs = max_msg_from_each_ing *
  63. g_teems_config.num_ingestion_nodes;
  64. // Compute the maximum number of messages we could send in round 2
  65. // Each storage node has at most this many users
  66. uint32_t users_per_stg = CEILDIV(g_teems_config.user_count,
  67. g_teems_config.num_storage_nodes);
  68. // And so can receive at most this many messages
  69. uint32_t tot_msg_per_stg;
  70. if (g_teems_config.private_routing) {
  71. tot_msg_per_stg = users_per_stg * g_teems_config.m_priv_in;
  72. } else {
  73. tot_msg_per_stg = users_per_stg * g_teems_config.m_pub_in;
  74. }
  75. // Which will be at most this many from us
  76. uint32_t max_msg_to_each_stg;
  77. max_msg_to_each_stg = CEILDIV(tot_msg_per_stg, g_teems_config.tot_weight) *
  78. g_teems_config.my_weight;
  79. // But we can't send more messages to each storage server than we
  80. // could receive in total
  81. if (max_msg_to_each_stg > max_round1_msgs) {
  82. max_msg_to_each_stg = max_round1_msgs;
  83. }
  84. // And the max total number of outgoing messages in round 2 is then
  85. uint32_t max_round2_msgs = max_msg_to_each_stg *
  86. g_teems_config.num_storage_nodes;
  87. // In case we have a weird configuration where users can send more
  88. // messages per epoch than they can receive, ensure the round 2
  89. // buffer is large enough to hold the incoming messages as well
  90. if (max_round2_msgs < max_round1_msgs) {
  91. max_round2_msgs = max_round1_msgs;
  92. }
  93. // The max number of messages that can arrive at a storage server
  94. uint32_t max_stg_msgs;
  95. max_stg_msgs = tot_msg_per_stg + g_teems_config.tot_weight;
  96. // Calculating public-routing buffer sizes
  97. // Weights are not used in public routing
  98. uint32_t max_round1b_msgs_to_adj_rtr =
  99. (g_teems_config.num_routing_nodes-1)*(g_teems_config.num_routing_nodes-1);
  100. // Ensure columnroute constraint that column height is >= 2*(num_routing_nodes-1)^2
  101. uint32_t max_round1a_msgs = std::max(max_round1_msgs, 2*max_round1b_msgs_to_adj_rtr);
  102. uint32_t max_round1c_msgs = max_round1a_msgs;
  103. /*
  104. printf("users_per_ing=%u, tot_msg_per_ing=%u, max_msg_from_each_ing=%u, max_round1_msgs=%u, users_per_stg=%u, tot_msg_per_stg=%u, max_msg_to_each_stg=%u, max_round2_msgs=%u, max_stg_msgs=%u\n", users_per_ing, tot_msg_per_ing, max_msg_from_each_ing, max_round1_msgs, users_per_stg, tot_msg_per_stg, max_msg_to_each_stg, max_round2_msgs, max_stg_msgs);
  105. */
  106. #ifdef TRACK_HEAP_USAGE
  107. printf("route_init H1 heap %u\n", g_peak_heap_used);
  108. #endif
  109. // Create the route state
  110. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  111. try {
  112. if (my_roles & ROLE_INGESTION) {
  113. route_state.ingbuf.alloc(tot_msg_per_ing);
  114. }
  115. #ifdef TRACK_HEAP_USAGE
  116. printf("route_init alloc %u msgs\n", tot_msg_per_ing);
  117. printf("route_init H2 heap %u\n", g_peak_heap_used);
  118. #endif
  119. if (my_roles & ROLE_ROUTING) {
  120. route_state.round1.alloc(max_round2_msgs);
  121. #ifdef TRACK_HEAP_USAGE
  122. printf("route_init alloc %u msgs\n", max_round2_msgs);
  123. printf("route_init H3 heap %u\n", g_peak_heap_used);
  124. #endif
  125. if (!g_teems_config.private_routing) {
  126. route_state.round1a.alloc(max_round1a_msgs);
  127. route_state.round1a_sorted.alloc(max_round1a_msgs);
  128. // double round 1b buffers to sort with some round 1a messages
  129. route_state.round1b_prev.alloc(2*max_round1b_msgs_to_adj_rtr);
  130. route_state.round1b_next.alloc(2*max_round1b_msgs_to_adj_rtr);
  131. route_state.round1c.alloc(max_round1c_msgs);
  132. #ifdef TRACK_HEAP_USAGE
  133. printf("route_init alloc %u msgs\n", max_round1a_msgs);
  134. printf("route_init alloc %u msgs\n", max_round1a_msgs);
  135. printf("route_init alloc %u msgs\n", 2*max_round1b_msgs_to_adj_rtr);
  136. printf("route_init alloc %u msgs\n", 2*max_round1b_msgs_to_adj_rtr);
  137. printf("route_init alloc %u msgs\n", max_round1c_msgs);
  138. printf("route_init H4 heap %u\n", g_peak_heap_used);
  139. #endif
  140. }
  141. }
  142. if (my_roles & ROLE_STORAGE) {
  143. route_state.round2.alloc(max_stg_msgs);
  144. #ifdef TRACK_HEAP_USAGE
  145. printf("route_init alloc %u msgs\n", max_stg_msgs);
  146. printf("route_init H5 heap %u\n", g_peak_heap_used);
  147. #endif
  148. if (!storage_init(users_per_stg, max_stg_msgs)) {
  149. return false;
  150. }
  151. }
  152. } catch (std::bad_alloc&) {
  153. printf("Memory allocation failed in route_init\n");
  154. return false;
  155. }
  156. route_state.step = ROUTE_NOT_STARTED;
  157. route_state.tot_msg_per_ing = tot_msg_per_ing;
  158. route_state.max_round1_msgs = max_round1_msgs;
  159. route_state.max_round1a_msgs = max_round1a_msgs;
  160. route_state.max_round1b_msgs_to_adj_rtr = max_round1b_msgs_to_adj_rtr;
  161. route_state.max_round1c_msgs = max_round1c_msgs;
  162. route_state.max_msg_to_each_stg = max_msg_to_each_stg;
  163. route_state.max_round2_msgs = max_round2_msgs;
  164. route_state.max_stg_msgs = max_stg_msgs;
  165. route_state.cbpointer = NULL;
  166. threadid_t nthreads = g_teems_config.nthreads;
  167. #ifdef PROFILE_ROUTING
  168. unsigned long start = printf_with_rtclock("begin precompute evalplans (%u,%hu) (%u,%hu)\n", tot_msg_per_ing, nthreads, max_round2_msgs, nthreads);
  169. #endif
  170. if (my_roles & ROLE_INGESTION) {
  171. sort_precompute_evalplan(tot_msg_per_ing, nthreads);
  172. }
  173. if (my_roles & ROLE_ROUTING) {
  174. sort_precompute_evalplan(max_round2_msgs, nthreads);
  175. if(!g_teems_config.private_routing) {
  176. sort_precompute_evalplan(max_round1a_msgs, nthreads);
  177. sort_precompute_evalplan(2*max_round1b_msgs_to_adj_rtr, nthreads);
  178. }
  179. }
  180. if (my_roles & ROLE_STORAGE) {
  181. sort_precompute_evalplan(max_stg_msgs, nthreads);
  182. }
  183. #ifdef PROFILE_ROUTING
  184. printf_with_rtclock_diff(start, "end precompute evalplans\n");
  185. #endif
  186. #ifdef TRACK_HEAP_USAGE
  187. printf("route_init end heap %u\n", g_peak_heap_used);
  188. #endif
  189. return true;
  190. }
  191. // Call when shutting system down to deallocate routing state
  192. void route_close() {
  193. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  194. if (my_roles & ROLE_STORAGE) {
  195. storage_close();
  196. }
  197. }
  198. // Precompute the WaksmanNetworks needed for the sorts. If you pass -1,
  199. // it will return the number of different sizes it needs to regenerate.
  200. // If you pass [0,sizes-1], it will compute one WaksmanNetwork with that
  201. // size index and return the number of available WaksmanNetworks of that
  202. // size. If you pass anything else, it will return the number of
  203. // different sizes it needs at all.
  204. // The list of sizes that need refilling, updated when you pass -1
  205. static std::vector<uint32_t> used_sizes;
  206. size_t ecall_precompute_sort(int sizeidx)
  207. {
  208. size_t ret = 0;
  209. if (sizeidx == -1) {
  210. used_sizes = sort_get_used();
  211. ret = used_sizes.size();
  212. } else if (sizeidx >= 0 && sizeidx < used_sizes.size()) {
  213. uint32_t size = used_sizes[sizeidx];
  214. #ifdef PROFILE_ROUTING
  215. unsigned long start = printf_with_rtclock("begin precompute WaksmanNetwork (%u)\n", size);
  216. #endif
  217. ret = sort_precompute(size);
  218. #ifdef PROFILE_ROUTING
  219. printf_with_rtclock_diff(start, "end precompute Waksman Network (%u)\n", size);
  220. #endif
  221. } else {
  222. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  223. if (my_roles & ROLE_INGESTION) {
  224. used_sizes.push_back(route_state.tot_msg_per_ing);
  225. }
  226. if (my_roles & ROLE_ROUTING) {
  227. used_sizes.push_back(route_state.max_round2_msgs);
  228. if(!g_teems_config.private_routing) {
  229. used_sizes.push_back(route_state.max_round1a_msgs);
  230. used_sizes.push_back(2*route_state.max_round1b_msgs_to_adj_rtr);
  231. used_sizes.push_back(2*route_state.max_round1b_msgs_to_adj_rtr);
  232. used_sizes.push_back(route_state.max_round1c_msgs);
  233. }
  234. }
  235. if (my_roles & ROLE_STORAGE) {
  236. used_sizes.push_back(route_state.max_stg_msgs);
  237. if(!g_teems_config.private_routing) {
  238. used_sizes.push_back(route_state.max_stg_msgs);
  239. }
  240. }
  241. ret = used_sizes.size();
  242. }
  243. return ret;
  244. }
  245. static uint8_t* msgbuffer_get_buf(MsgBuffer &msgbuf,
  246. NodeCommState &, uint32_t tot_enc_chunk_size)
  247. {
  248. uint16_t msg_size = g_teems_config.msg_size;
  249. // Chunks will be encrypted and have a MAC tag attached which will
  250. // not correspond to plaintext bytes, so we can trim them.
  251. // The minimum number of chunks needed to transmit this message
  252. uint32_t min_num_chunks =
  253. (tot_enc_chunk_size + (FRAME_SIZE-1)) / FRAME_SIZE;
  254. // The number of plaintext bytes this message could contain
  255. uint32_t plaintext_bytes = tot_enc_chunk_size -
  256. SGX_AESGCM_MAC_SIZE * min_num_chunks;
  257. assert ((plaintext_bytes % uint32_t(msg_size)) == 0);
  258. uint32_t num_msgs = plaintext_bytes/uint32_t(msg_size);
  259. pthread_mutex_lock(&msgbuf.mutex);
  260. uint32_t start = msgbuf.reserved;
  261. if (start + num_msgs > msgbuf.bufsize) {
  262. pthread_mutex_unlock(&msgbuf.mutex);
  263. printf("Max %u messages exceeded\n", msgbuf.bufsize);
  264. return NULL;
  265. }
  266. msgbuf.reserved += num_msgs;
  267. pthread_mutex_unlock(&msgbuf.mutex);
  268. return msgbuf.buf + start * msg_size;
  269. }
  270. static void round1a_received(NodeCommState &nodest,
  271. uint8_t *data, uint32_t plaintext_len, uint32_t);
  272. static void round1b_prev_received(NodeCommState &nodest,
  273. uint8_t *data, uint32_t plaintext_len, uint32_t);
  274. static void round1b_next_received(NodeCommState &nodest,
  275. uint8_t *data, uint32_t plaintext_len, uint32_t);
  276. static void round1c_received(NodeCommState &nodest, uint8_t *data,
  277. uint32_t plaintext_len, uint32_t);
  278. static void round2_received(NodeCommState &nodest,
  279. uint8_t *data, uint32_t plaintext_len, uint32_t);
  280. // A round 1 message was received by a routing node from an ingestion
  281. // node; we put it into the round 2 buffer for processing in round 2
  282. static void round1_received(NodeCommState &nodest,
  283. uint8_t *data, uint32_t plaintext_len, uint32_t)
  284. {
  285. uint16_t msg_size = g_teems_config.msg_size;
  286. assert((plaintext_len % uint32_t(msg_size)) == 0);
  287. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  288. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  289. uint8_t their_roles = g_teems_config.roles[nodest.node_num];
  290. pthread_mutex_lock(&route_state.round1.mutex);
  291. route_state.round1.inserted += num_msgs;
  292. route_state.round1.nodes_received += 1;
  293. nodenum_t nodes_received = route_state.round1.nodes_received;
  294. bool completed_prev_round = route_state.round1.completed_prev_round;
  295. pthread_mutex_unlock(&route_state.round1.mutex);
  296. // What is the next message we expect from this node?
  297. if (g_teems_config.private_routing) {
  298. if ((our_roles & ROLE_STORAGE) && (their_roles & ROLE_ROUTING)) {
  299. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  300. uint32_t tot_enc_chunk_size) {
  301. return msgbuffer_get_buf(route_state.round2, commst,
  302. tot_enc_chunk_size);
  303. };
  304. nodest.in_msg_received = round2_received;
  305. }
  306. // Otherwise, it's just the next round 1 message, so don't change
  307. // the handlers.
  308. } else {
  309. if (their_roles & ROLE_ROUTING) {
  310. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  311. uint32_t tot_enc_chunk_size) {
  312. return msgbuffer_get_buf(route_state.round1a, commst,
  313. tot_enc_chunk_size);
  314. };
  315. nodest.in_msg_received = round1a_received;
  316. }
  317. // Otherwise, it's just the next round 1 message, so don't change
  318. // the handlers.
  319. }
  320. if (nodes_received == g_teems_config.num_ingestion_nodes &&
  321. completed_prev_round) {
  322. route_state.step = ROUTE_ROUND_1;
  323. void *cbpointer = route_state.cbpointer;
  324. route_state.cbpointer = NULL;
  325. ocall_routing_round_complete(cbpointer, 1);
  326. }
  327. }
  328. // A round 1a message was received by a routing node
  329. static void round1a_received(NodeCommState &nodest,
  330. uint8_t *data, uint32_t plaintext_len, uint32_t)
  331. {
  332. uint16_t msg_size = g_teems_config.msg_size;
  333. assert((plaintext_len % uint32_t(msg_size)) == 0);
  334. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  335. pthread_mutex_lock(&route_state.round1a.mutex);
  336. route_state.round1a.inserted += num_msgs;
  337. route_state.round1a.nodes_received += 1;
  338. nodenum_t nodes_received = route_state.round1a.nodes_received;
  339. bool completed_prev_round = route_state.round1a.completed_prev_round;
  340. pthread_mutex_unlock(&route_state.round1a.mutex);
  341. // Both are routing nodes
  342. // We only expect a message from the previous and next nodes (if they exist)
  343. nodenum_t my_node_num = g_teems_config.my_node_num;
  344. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  345. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  346. if ((prev_nodes > 0) &&
  347. (nodest.node_num == g_teems_config.routing_nodes[num_routing_nodes-1])) {
  348. // Node is previous routing node
  349. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  350. uint32_t tot_enc_chunk_size) {
  351. return msgbuffer_get_buf(route_state.round1b_prev, commst,
  352. tot_enc_chunk_size);
  353. };
  354. nodest.in_msg_received = round1b_prev_received;
  355. } else if ((prev_nodes < num_routing_nodes-1) &&
  356. (nodest.node_num == g_teems_config.routing_nodes[1])) {
  357. // Node is next routing node
  358. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  359. uint32_t tot_enc_chunk_size) {
  360. return msgbuffer_get_buf(route_state.round1b_next, commst,
  361. tot_enc_chunk_size);
  362. };
  363. nodest.in_msg_received = round1b_next_received;
  364. } else {
  365. // other routing nodes will not send to this node until round 1c
  366. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  367. uint32_t tot_enc_chunk_size) {
  368. return msgbuffer_get_buf(route_state.round1c, commst,
  369. tot_enc_chunk_size);
  370. };
  371. nodest.in_msg_received = round1c_received;
  372. }
  373. if (nodes_received == g_teems_config.num_routing_nodes &&
  374. completed_prev_round) {
  375. route_state.step = ROUTE_ROUND_1A;
  376. void *cbpointer = route_state.cbpointer;
  377. route_state.cbpointer = NULL;
  378. ocall_routing_round_complete(cbpointer, ROUND_1A);
  379. }
  380. }
  381. // A round 1b message was received from the previous routing node
  382. static void round1b_prev_received(NodeCommState &nodest,
  383. uint8_t *data, uint32_t plaintext_len, uint32_t)
  384. {
  385. uint16_t msg_size = g_teems_config.msg_size;
  386. assert((plaintext_len % uint32_t(msg_size)) == 0);
  387. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  388. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  389. uint8_t their_roles = g_teems_config.roles[nodest.node_num];
  390. pthread_mutex_lock(&route_state.round1b_prev.mutex);
  391. route_state.round1b_prev.inserted += num_msgs;
  392. route_state.round1b_prev.nodes_received += 1;
  393. nodenum_t nodes_received = route_state.round1b_prev.nodes_received;
  394. bool completed_prev_round = route_state.round1b_prev.completed_prev_round;
  395. pthread_mutex_unlock(&route_state.round1b_prev.mutex);
  396. pthread_mutex_lock(&route_state.round1b_next.mutex);
  397. nodes_received += route_state.round1b_next.nodes_received;
  398. pthread_mutex_unlock(&route_state.round1b_next.mutex);
  399. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  400. uint32_t tot_enc_chunk_size) {
  401. return msgbuffer_get_buf(route_state.round1c, commst,
  402. tot_enc_chunk_size);
  403. };
  404. nodest.in_msg_received = round1c_received;
  405. nodenum_t my_node_num = g_teems_config.my_node_num;
  406. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  407. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  408. nodenum_t adjacent_nodes;
  409. if (num_routing_nodes == 1) {
  410. adjacent_nodes = 0;
  411. } else if ((prev_nodes == 0) || (prev_nodes == num_routing_nodes-1)) {
  412. adjacent_nodes = 1;
  413. } else {
  414. adjacent_nodes = 2;
  415. }
  416. if (nodes_received == adjacent_nodes && completed_prev_round) {
  417. route_state.step = ROUTE_ROUND_1B;
  418. void *cbpointer = route_state.cbpointer;
  419. route_state.cbpointer = NULL;
  420. ocall_routing_round_complete(cbpointer, ROUND_1B);
  421. }
  422. }
  423. // A round 1b message was received from the next routing node
  424. static void round1b_next_received(NodeCommState &nodest,
  425. uint8_t *data, uint32_t plaintext_len, uint32_t)
  426. {
  427. uint16_t msg_size = g_teems_config.msg_size;
  428. assert((plaintext_len % uint32_t(msg_size)) == 0);
  429. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  430. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  431. uint8_t their_roles = g_teems_config.roles[nodest.node_num];
  432. pthread_mutex_lock(&route_state.round1b_next.mutex);
  433. route_state.round1b_next.inserted += num_msgs;
  434. route_state.round1b_next.nodes_received += 1;
  435. nodenum_t nodes_received = route_state.round1b_next.nodes_received;
  436. bool completed_prev_round = route_state.round1b_next.completed_prev_round;
  437. pthread_mutex_unlock(&route_state.round1b_next.mutex);
  438. pthread_mutex_lock(&route_state.round1b_prev.mutex);
  439. nodes_received += route_state.round1b_prev.nodes_received;
  440. pthread_mutex_unlock(&route_state.round1b_prev.mutex);
  441. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  442. uint32_t tot_enc_chunk_size) {
  443. return msgbuffer_get_buf(route_state.round1c, commst,
  444. tot_enc_chunk_size);
  445. };
  446. nodest.in_msg_received = round1c_received;
  447. nodenum_t my_node_num = g_teems_config.my_node_num;
  448. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  449. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  450. nodenum_t adjacent_nodes;
  451. if (num_routing_nodes == 1) {
  452. adjacent_nodes = 0;
  453. } else if ((prev_nodes == 0) || (prev_nodes == num_routing_nodes-1)) {
  454. adjacent_nodes = 1;
  455. } else {
  456. adjacent_nodes = 2;
  457. }
  458. if (nodes_received == adjacent_nodes && completed_prev_round) {
  459. route_state.step = ROUTE_ROUND_1B;
  460. void *cbpointer = route_state.cbpointer;
  461. route_state.cbpointer = NULL;
  462. ocall_routing_round_complete(cbpointer, ROUND_1B);
  463. }
  464. }
  465. // Message received in round 1c
  466. static void round1c_received(NodeCommState &nodest, uint8_t *data,
  467. uint32_t plaintext_len, uint32_t)
  468. {
  469. uint16_t msg_size = g_teems_config.msg_size;
  470. assert((plaintext_len % uint32_t(msg_size)) == 0);
  471. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  472. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  473. uint8_t their_roles = g_teems_config.roles[nodest.node_num];
  474. pthread_mutex_lock(&route_state.round1c.mutex);
  475. route_state.round1c.inserted += num_msgs;
  476. route_state.round1c.nodes_received += 1;
  477. nodenum_t nodes_received = route_state.round1c.nodes_received;
  478. bool completed_prev_round = route_state.round1c.completed_prev_round;
  479. pthread_mutex_unlock(&route_state.round1c.mutex);
  480. // What is the next message we expect from this node?
  481. if (our_roles & ROLE_STORAGE) {
  482. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  483. uint32_t tot_enc_chunk_size) {
  484. return msgbuffer_get_buf(route_state.round2, commst,
  485. tot_enc_chunk_size);
  486. };
  487. nodest.in_msg_received = round2_received;
  488. } else if (their_roles & ROLE_INGESTION) {
  489. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  490. uint32_t tot_enc_chunk_size) {
  491. return msgbuffer_get_buf(route_state.round1, commst,
  492. tot_enc_chunk_size);
  493. };
  494. nodest.in_msg_received = round1_received;
  495. } else {
  496. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  497. uint32_t tot_enc_chunk_size) {
  498. return msgbuffer_get_buf(route_state.round1a, commst,
  499. tot_enc_chunk_size);
  500. };
  501. nodest.in_msg_received = round1a_received;
  502. }
  503. if (nodes_received == g_teems_config.num_routing_nodes &&
  504. completed_prev_round) {
  505. route_state.step = ROUTE_ROUND_1C;
  506. void *cbpointer = route_state.cbpointer;
  507. route_state.cbpointer = NULL;
  508. ocall_routing_round_complete(cbpointer, ROUND_1C);
  509. }
  510. }
  511. // A round 2 message was received by a storage node from a routing node
  512. static void round2_received(NodeCommState &nodest,
  513. uint8_t *data, uint32_t plaintext_len, uint32_t)
  514. {
  515. uint16_t msg_size = g_teems_config.msg_size;
  516. assert((plaintext_len % uint32_t(msg_size)) == 0);
  517. uint32_t num_msgs = plaintext_len / uint32_t(msg_size);
  518. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  519. uint8_t their_roles = g_teems_config.roles[nodest.node_num];
  520. pthread_mutex_lock(&route_state.round2.mutex);
  521. route_state.round2.inserted += num_msgs;
  522. route_state.round2.nodes_received += 1;
  523. nodenum_t nodes_received = route_state.round2.nodes_received;
  524. bool completed_prev_round = route_state.round2.completed_prev_round;
  525. pthread_mutex_unlock(&route_state.round2.mutex);
  526. // What is the next message we expect from this node?
  527. if ((our_roles & ROLE_ROUTING) && (their_roles & ROLE_INGESTION)) {
  528. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  529. uint32_t tot_enc_chunk_size) {
  530. return msgbuffer_get_buf(route_state.round1, commst,
  531. tot_enc_chunk_size);
  532. };
  533. nodest.in_msg_received = round1_received;
  534. }
  535. // Otherwise, it's just the next round 2 message, so don't change
  536. // the handlers.
  537. if (nodes_received == g_teems_config.num_routing_nodes &&
  538. completed_prev_round) {
  539. route_state.step = ROUTE_ROUND_2;
  540. void *cbpointer = route_state.cbpointer;
  541. route_state.cbpointer = NULL;
  542. ocall_routing_round_complete(cbpointer, 2);
  543. }
  544. }
  545. // For a given other node, set the received message handler to the first
  546. // message we would expect from them, given their roles and our roles.
  547. void route_init_msg_handler(nodenum_t node_num)
  548. {
  549. // Our roles and their roles
  550. uint8_t our_roles = g_teems_config.roles[g_teems_config.my_node_num];
  551. uint8_t their_roles = g_teems_config.roles[node_num];
  552. // The node communication state
  553. NodeCommState &nodest = g_commstates[node_num];
  554. // If we are a routing node (possibly among other roles) and they
  555. // are an ingestion node (possibly among other roles), a round 1
  556. // routing message is the first thing we expect from them. We put
  557. // these messages into the round1 buffer for processing.
  558. if ((our_roles & ROLE_ROUTING) && (their_roles & ROLE_INGESTION)) {
  559. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  560. uint32_t tot_enc_chunk_size) {
  561. return msgbuffer_get_buf(route_state.round1, commst,
  562. tot_enc_chunk_size);
  563. };
  564. nodest.in_msg_received = round1_received;
  565. }
  566. // Otherwise, if we are a storage node (possibly among other roles)
  567. // and they are a routing node (possibly among other roles), a round
  568. // 2 routing message is the first thing we expect from them
  569. else if ((our_roles & ROLE_STORAGE) && (their_roles & ROLE_ROUTING)) {
  570. nodest.in_msg_get_buf = [&](NodeCommState &commst,
  571. uint32_t tot_enc_chunk_size) {
  572. return msgbuffer_get_buf(route_state.round2, commst,
  573. tot_enc_chunk_size);
  574. };
  575. nodest.in_msg_received = round2_received;
  576. }
  577. // Otherwise, we don't expect a message from this node. Set the
  578. // unknown message handler.
  579. else {
  580. nodest.in_msg_get_buf = default_in_msg_get_buf;
  581. nodest.in_msg_received = unknown_in_msg_received;
  582. }
  583. }
  584. // Directly ingest a buffer of num_msgs messages into the ingbuf buffer.
  585. // Return true on success, false on failure.
  586. bool ecall_ingest_raw(uint8_t *msgs, uint32_t num_msgs)
  587. {
  588. uint16_t msg_size = g_teems_config.msg_size;
  589. MsgBuffer &ingbuf = route_state.ingbuf;
  590. pthread_mutex_lock(&ingbuf.mutex);
  591. uint32_t start = ingbuf.reserved;
  592. if (start + num_msgs > route_state.tot_msg_per_ing) {
  593. pthread_mutex_unlock(&ingbuf.mutex);
  594. printf("Max %u messages exceeded\n",
  595. route_state.tot_msg_per_ing);
  596. return false;
  597. }
  598. ingbuf.reserved += num_msgs;
  599. pthread_mutex_unlock(&ingbuf.mutex);
  600. memmove(ingbuf.buf + start * msg_size,
  601. msgs, num_msgs * msg_size);
  602. pthread_mutex_lock(&ingbuf.mutex);
  603. ingbuf.inserted += num_msgs;
  604. pthread_mutex_unlock(&ingbuf.mutex);
  605. return true;
  606. }
  607. // Send messages round-robin, used in rounds 1 and 1c. Note that N here is not private.
  608. template<typename T>
  609. static void send_round_robin_msgs(MsgBuffer &round, const uint8_t *msgs, const T *indices,
  610. uint32_t N)
  611. {
  612. uint16_t msg_size = g_teems_config.msg_size;
  613. uint16_t tot_weight;
  614. tot_weight = g_teems_config.tot_weight;
  615. nodenum_t my_node_num = g_teems_config.my_node_num;
  616. uint32_t full_rows;
  617. uint32_t last_row;
  618. full_rows = N / uint32_t(tot_weight);
  619. last_row = N % uint32_t(tot_weight);
  620. for (auto &routing_node: g_teems_config.routing_nodes) {
  621. uint8_t weight = g_teems_config.weights[routing_node].weight;
  622. if (weight == 0) {
  623. // This shouldn't happen, but just in case
  624. continue;
  625. }
  626. uint16_t start_weight =
  627. g_teems_config.weights[routing_node].startweight;
  628. // The number of messages headed for this routing node from the
  629. // full rows
  630. uint32_t num_msgs_full_rows = full_rows * uint32_t(weight);
  631. // The number of messages headed for this routing node from the
  632. // incomplete last row is:
  633. // 0 if last_row < start_weight
  634. // last_row-start_weight if start_weight <= last_row < start_weight + weight
  635. // weight if start_weight + weight <= last_row
  636. uint32_t num_msgs_last_row = 0;
  637. if (start_weight <= last_row && last_row < start_weight + weight) {
  638. num_msgs_last_row = last_row-start_weight;
  639. } else if (start_weight + weight <= last_row) {
  640. num_msgs_last_row = weight;
  641. }
  642. // The total number of messages headed for this routing node
  643. uint32_t num_msgs = num_msgs_full_rows + num_msgs_last_row;
  644. if (routing_node == my_node_num) {
  645. // Special case: we're sending to ourselves; just put the
  646. // messages in our own buffer
  647. pthread_mutex_lock(&round.mutex);
  648. uint32_t start = round.reserved;
  649. if (start + num_msgs > round.bufsize) {
  650. pthread_mutex_unlock(&round.mutex);
  651. printf("Max %u messages exceeded\n", round.bufsize);
  652. return;
  653. }
  654. round.reserved += num_msgs;
  655. pthread_mutex_unlock(&round.mutex);
  656. uint8_t *buf = round.buf + start * msg_size;
  657. for (uint32_t i=0; i<full_rows; ++i) {
  658. const T *idxp = indices + i*tot_weight + start_weight;
  659. for (uint32_t j=0; j<weight; ++j) {
  660. memmove(buf, msgs + idxp[j].index()*msg_size, msg_size);
  661. buf += msg_size;
  662. }
  663. }
  664. const T *idxp = indices + full_rows*tot_weight + start_weight;
  665. for (uint32_t j=0; j<num_msgs_last_row; ++j) {
  666. memmove(buf, msgs + idxp[j].index()*msg_size, msg_size);
  667. buf += msg_size;
  668. }
  669. pthread_mutex_lock(&round.mutex);
  670. round.inserted += num_msgs;
  671. round.nodes_received += 1;
  672. pthread_mutex_unlock(&round.mutex);
  673. } else {
  674. NodeCommState &nodecom = g_commstates[routing_node];
  675. nodecom.message_start(num_msgs * msg_size);
  676. for (uint32_t i=0; i<full_rows; ++i) {
  677. const T *idxp = indices + i*tot_weight + start_weight;
  678. for (uint32_t j=0; j<weight; ++j) {
  679. nodecom.message_data(msgs + idxp[j].index()*msg_size, msg_size);
  680. }
  681. }
  682. const T *idxp = indices + full_rows*tot_weight + start_weight;
  683. for (uint32_t j=0; j<num_msgs_last_row; ++j) {
  684. nodecom.message_data(msgs + idxp[j].index()*msg_size, msg_size);
  685. }
  686. }
  687. }
  688. }
  689. // Send the round 1a messages from the round 1 buffer, which only occurs in public-channel routing.
  690. // msgs points to the message buffer, indices points to the the sorted indices, and N is the number
  691. // of non-padding items.
  692. static void send_round1a_msgs(const uint8_t *msgs, const UidPriorityKey *indices, uint32_t N) {
  693. uint16_t msg_size = g_teems_config.msg_size;
  694. nodenum_t my_node_num = g_teems_config.my_node_num;
  695. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  696. uint32_t min_msgs_per_node = route_state.max_round1a_msgs / num_routing_nodes;
  697. uint32_t extra_msgs = route_state.max_round1a_msgs % num_routing_nodes;
  698. for (auto &routing_node: g_teems_config.routing_nodes) {
  699. // In this unweighted setting, start_weight represents the position among routing nodes
  700. uint16_t prev_nodes = g_teems_config.weights[routing_node].startweight;
  701. uint32_t start_msg, num_msgs;
  702. if (prev_nodes >= extra_msgs) {
  703. start_msg = min_msgs_per_node * prev_nodes + extra_msgs;
  704. num_msgs = min_msgs_per_node;
  705. } else {
  706. start_msg = min_msgs_per_node * prev_nodes + prev_nodes;
  707. num_msgs = min_msgs_per_node + 1;
  708. }
  709. // take number of messages into account
  710. if (start_msg >= N) {
  711. num_msgs = 0;
  712. } else if (start_msg + num_msgs > N) {
  713. num_msgs = N - start_msg;
  714. }
  715. if (routing_node == my_node_num) {
  716. // Special case: we're sending to ourselves; just put the
  717. // messages in our own buffer
  718. MsgBuffer &round1a = route_state.round1a;
  719. pthread_mutex_lock(&round1a.mutex);
  720. uint32_t start = round1a.reserved;
  721. if (start + num_msgs > round1a.bufsize) {
  722. pthread_mutex_unlock(&round1a.mutex);
  723. printf("Max %u messages exceeded in round 1a\n", round1a.bufsize);
  724. return;
  725. }
  726. round1a.reserved += num_msgs;
  727. pthread_mutex_unlock(&round1a.mutex);
  728. uint8_t *buf = round1a.buf + start * msg_size;
  729. for (uint32_t i=0; i<num_msgs; ++i) {
  730. const UidPriorityKey *idxp = indices + start_msg + i;
  731. memmove(buf, msgs + idxp->index()*msg_size, msg_size);
  732. buf += msg_size;
  733. }
  734. pthread_mutex_lock(&round1a.mutex);
  735. round1a.inserted += num_msgs;
  736. round1a.nodes_received += 1;
  737. pthread_mutex_unlock(&round1a.mutex);
  738. } else {
  739. NodeCommState &nodecom = g_commstates[routing_node];
  740. nodecom.message_start(num_msgs * msg_size);
  741. for (uint32_t i=0; i<num_msgs; ++i) {
  742. const UidPriorityKey *idxp = indices + start_msg + i;
  743. nodecom.message_data(msgs + idxp->index()*msg_size, msg_size);
  744. }
  745. }
  746. }
  747. }
  748. // Send the round 1b messages from the round 1a buffer, which only occurs in public-channel routing.
  749. // msgs points to the message buffer, and N is the number of non-padding items.
  750. static void send_round1b_msgs(const uint8_t *msgs, uint32_t N) {
  751. uint16_t msg_size = g_teems_config.msg_size;
  752. nodenum_t my_node_num = g_teems_config.my_node_num;
  753. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  754. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  755. // send to previous node
  756. if (prev_nodes > 0) {
  757. nodenum_t prev_node = g_teems_config.routing_nodes[num_routing_nodes-1];
  758. NodeCommState &nodecom = g_commstates[prev_node];
  759. uint32_t num_msgs = min(N, route_state.max_round1b_msgs_to_adj_rtr);
  760. nodecom.message_start(num_msgs * msg_size);
  761. for (uint32_t i=0; i<num_msgs; ++i) {
  762. nodecom.message_data(msgs + i*msg_size, msg_size);
  763. }
  764. }
  765. // send to next node
  766. if (prev_nodes < num_routing_nodes-1) {
  767. nodenum_t next_node = g_teems_config.routing_nodes[1];
  768. NodeCommState &nodecom = g_commstates[next_node];
  769. if (N <= route_state.max_round1a_msgs - route_state.max_round1b_msgs_to_adj_rtr) {
  770. // No messages to exchange with next node
  771. nodecom.message_start(0);
  772. // No need to call message_data()
  773. } else {
  774. uint32_t start_msg =
  775. route_state.max_round1a_msgs - route_state.max_round1b_msgs_to_adj_rtr;
  776. uint32_t num_msgs = N - start_msg;
  777. nodecom.message_start(num_msgs * msg_size);
  778. for (uint32_t i=0; i<num_msgs; ++i) {
  779. nodecom.message_data(msgs + i*msg_size, msg_size);
  780. }
  781. }
  782. }
  783. }
  784. // Send the round 2 messages from the previous-round buffer, which are already
  785. // padded and shuffled, so this can be done non-obliviously. tot_msgs
  786. // is the total number of messages in the input buffer, which may
  787. // include padding messages added by the shuffle. Those messages are
  788. // not sent anywhere. There are num_msgs_per_stg messages for each
  789. // storage node labelled for that node.
  790. static void send_round2_msgs(uint32_t tot_msgs, uint32_t num_msgs_per_stg, MsgBuffer &prevround)
  791. {
  792. uint16_t msg_size = g_teems_config.msg_size;
  793. const uint8_t* buf = prevround.buf;
  794. nodenum_t num_storage_nodes = g_teems_config.num_storage_nodes;
  795. nodenum_t my_node_num = g_teems_config.my_node_num;
  796. uint8_t *myself_buf = NULL;
  797. for (nodenum_t i=0; i<num_storage_nodes; ++i) {
  798. nodenum_t node = g_teems_config.storage_nodes[i];
  799. if (node != my_node_num) {
  800. g_commstates[node].message_start(msg_size * num_msgs_per_stg);
  801. } else {
  802. MsgBuffer &round2 = route_state.round2;
  803. pthread_mutex_lock(&round2.mutex);
  804. uint32_t start = round2.reserved;
  805. if (start + num_msgs_per_stg > round2.bufsize) {
  806. pthread_mutex_unlock(&round2.mutex);
  807. printf("Max %u messages exceeded\n", round2.bufsize);
  808. return;
  809. }
  810. round2.reserved += num_msgs_per_stg;
  811. pthread_mutex_unlock(&round2.mutex);
  812. myself_buf = round2.buf + start * msg_size;
  813. }
  814. }
  815. while (tot_msgs) {
  816. nodenum_t storage_node_id =
  817. nodenum_t((*(const uint32_t *)buf)>>DEST_UID_BITS);
  818. if (storage_node_id < num_storage_nodes) {
  819. nodenum_t node = g_teems_config.storage_map[storage_node_id];
  820. if (node == my_node_num) {
  821. memmove(myself_buf, buf, msg_size);
  822. myself_buf += msg_size;
  823. } else {
  824. g_commstates[node].message_data(buf, msg_size);
  825. }
  826. }
  827. buf += msg_size;
  828. --tot_msgs;
  829. }
  830. if (myself_buf) {
  831. MsgBuffer &round2 = route_state.round2;
  832. pthread_mutex_lock(&round2.mutex);
  833. round2.inserted += num_msgs_per_stg;
  834. round2.nodes_received += 1;
  835. pthread_mutex_unlock(&round2.mutex);
  836. }
  837. }
  838. static void round1a_processing(void *cbpointer) {
  839. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  840. MsgBuffer &round1 = route_state.round1;
  841. if (my_roles & ROLE_ROUTING) {
  842. route_state.cbpointer = cbpointer;
  843. pthread_mutex_lock(&round1.mutex);
  844. // Ensure there are no pending messages currently being inserted
  845. // into the buffer
  846. while (round1.reserved != round1.inserted) {
  847. pthread_mutex_unlock(&round1.mutex);
  848. pthread_mutex_lock(&round1.mutex);
  849. }
  850. #ifdef TRACE_ROUTING
  851. show_messages("Start of round 1a", round1.buf, round1.inserted);
  852. #endif
  853. #ifdef PROFILE_ROUTING
  854. uint32_t inserted = round1.inserted;
  855. unsigned long start_round1a = printf_with_rtclock("begin round1a processing (%u)\n", inserted);
  856. // Sort the messages we've received
  857. unsigned long start_sort = printf_with_rtclock("begin oblivious sort (%u,%u)\n", inserted, route_state.max_round1_msgs);
  858. #endif
  859. // Sort received messages by increasing user ID and
  860. // priority. Smaller priority number indicates higher priority.
  861. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, round1.buf,
  862. g_teems_config.msg_size, round1.inserted, route_state.max_round1_msgs,
  863. send_round1a_msgs);
  864. #ifdef PROFILE_ROUTING
  865. printf_with_rtclock_diff(start_sort, "end oblivious sort (%u,%u)\n", inserted, route_state.max_round1_msgs);
  866. printf_with_rtclock_diff(start_round1a, "end round1a processing (%u)\n", inserted);
  867. #endif
  868. round1.reset();
  869. pthread_mutex_unlock(&round1.mutex);
  870. MsgBuffer &round1a = route_state.round1a;
  871. pthread_mutex_lock(&round1a.mutex);
  872. round1a.completed_prev_round = true;
  873. nodenum_t nodes_received = round1a.nodes_received;
  874. pthread_mutex_unlock(&round1a.mutex);
  875. if (nodes_received == g_teems_config.num_routing_nodes) {
  876. route_state.step = ROUTE_ROUND_1A;
  877. route_state.cbpointer = NULL;
  878. ocall_routing_round_complete(cbpointer, ROUND_1A);
  879. }
  880. } else {
  881. route_state.step = ROUTE_ROUND_1A;
  882. route_state.round1a.completed_prev_round = true;
  883. ocall_routing_round_complete(cbpointer, ROUND_1A);
  884. }
  885. }
  886. static void round1b_processing(void *cbpointer) {
  887. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  888. nodenum_t my_node_num = g_teems_config.my_node_num;
  889. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  890. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  891. MsgBuffer &round1a = route_state.round1a;
  892. MsgBuffer &round1a_sorted = route_state.round1a_sorted;
  893. if (my_roles & ROLE_ROUTING) {
  894. route_state.cbpointer = cbpointer;
  895. pthread_mutex_lock(&round1a.mutex);
  896. // Ensure there are no pending messages currently being inserted
  897. // into the buffer
  898. while (round1a.reserved != round1a.inserted) {
  899. pthread_mutex_unlock(&round1a.mutex);
  900. pthread_mutex_lock(&round1a.mutex);
  901. }
  902. pthread_mutex_lock(&round1a_sorted.mutex);
  903. #ifdef TRACE_ROUTING
  904. show_messages("Start of round 1b", round1a.buf, round1a.inserted);
  905. #endif
  906. #ifdef PROFILE_ROUTING
  907. uint32_t inserted = round1a.inserted;
  908. unsigned long start_round1b = printf_with_rtclock("begin round1b processing (%u)\n", inserted);
  909. // Sort the messages we've received
  910. unsigned long start_sort = printf_with_rtclock("begin oblivious sort (%u,%u)\n", inserted, route_state.max_round1a_msgs);
  911. #endif
  912. // Sort received messages by increasing user ID and
  913. // priority. Smaller priority number indicates higher priority.
  914. if (inserted > 0) {
  915. // copy items in sorted order into round1a_sorted
  916. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, round1a.buf,
  917. g_teems_config.msg_size, round1a.inserted, route_state.max_round1a_msgs,
  918. route_state.round1a_sorted.buf);
  919. send_round1b_msgs(round1a_sorted.buf, round1a.inserted);
  920. } else {
  921. send_round1b_msgs(NULL, 0);
  922. }
  923. #ifdef PROFILE_ROUTING
  924. printf_with_rtclock_diff(start_sort, "end oblivious sort (%u,%u)\n", inserted, route_state.max_round1a_msgs);
  925. printf_with_rtclock_diff(start_round1b, "end round1b processing (%u)\n", inserted);
  926. #endif
  927. pthread_mutex_unlock(&round1a_sorted.mutex);
  928. pthread_mutex_unlock(&round1a.mutex);
  929. MsgBuffer &round1b_prev = route_state.round1b_prev;
  930. pthread_mutex_lock(&round1b_prev.mutex);
  931. round1b_prev.completed_prev_round = true;
  932. nodenum_t nodes_received = round1b_prev.nodes_received;
  933. pthread_mutex_unlock(&round1b_prev.mutex);
  934. MsgBuffer &round1b_next = route_state.round1b_next;
  935. pthread_mutex_lock(&round1b_next.mutex);
  936. round1b_next.completed_prev_round = true;
  937. nodes_received += round1b_next.nodes_received;
  938. pthread_mutex_unlock(&round1b_next.mutex);
  939. nodenum_t adjacent_nodes;
  940. if (num_routing_nodes == 1) {
  941. adjacent_nodes = 0;
  942. } else if ((prev_nodes == 0) || (prev_nodes == num_routing_nodes-1)) {
  943. adjacent_nodes = 1;
  944. } else {
  945. adjacent_nodes = 2;
  946. }
  947. if (nodes_received == adjacent_nodes) {
  948. route_state.step = ROUTE_ROUND_1B;
  949. route_state.cbpointer = NULL;
  950. ocall_routing_round_complete(cbpointer, ROUND_1B);
  951. }
  952. } else {
  953. route_state.step = ROUTE_ROUND_1B;
  954. route_state.round1b_prev.completed_prev_round = true;
  955. route_state.round1b_next.completed_prev_round = true;
  956. ocall_routing_round_complete(cbpointer, ROUND_1B);
  957. }
  958. }
  959. static void copy_msgs(uint8_t *dst, uint32_t start_msg, uint32_t num_copy, const uint8_t *src,
  960. const UidPriorityKey *indices)
  961. {
  962. uint16_t msg_size = g_teems_config.msg_size;
  963. const UidPriorityKey *idxp = indices + start_msg;
  964. uint8_t *buf = dst;
  965. for (uint32_t i=0; i<num_copy; i++) {
  966. memmove(buf, src + idxp[i].index()*msg_size, msg_size);
  967. buf += msg_size;
  968. }
  969. }
  970. static void round1c_processing(void *cbpointer) {
  971. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  972. nodenum_t my_node_num = g_teems_config.my_node_num;
  973. nodenum_t num_routing_nodes = g_teems_config.num_routing_nodes;
  974. uint16_t prev_nodes = g_teems_config.weights[my_node_num].startweight;
  975. uint16_t msg_size = g_teems_config.msg_size;
  976. uint32_t max_round1b_msgs_to_adj_rtr = route_state.max_round1b_msgs_to_adj_rtr;
  977. uint32_t max_round1a_msgs = route_state.max_round1a_msgs;
  978. MsgBuffer &round1a = route_state.round1a;
  979. MsgBuffer &round1a_sorted = route_state.round1a_sorted;
  980. MsgBuffer &round1b_prev = route_state.round1b_prev;
  981. MsgBuffer &round1b_next = route_state.round1b_next;
  982. if (my_roles & ROLE_ROUTING) {
  983. route_state.cbpointer = cbpointer;
  984. pthread_mutex_lock(&round1b_prev.mutex);
  985. pthread_mutex_lock(&round1b_next.mutex);
  986. // Ensure there are no pending messages currently being inserted
  987. // into the round 1b buffers
  988. while (round1b_prev.reserved != round1b_prev.inserted) {
  989. pthread_mutex_unlock(&round1b_prev.mutex);
  990. pthread_mutex_lock(&round1b_prev.mutex);
  991. }
  992. while (round1b_next.reserved != round1b_next.inserted) {
  993. pthread_mutex_unlock(&round1b_next.mutex);
  994. pthread_mutex_lock(&round1b_next.mutex);
  995. }
  996. pthread_mutex_lock(&round1a.mutex);
  997. pthread_mutex_lock(&round1a_sorted.mutex);
  998. #ifdef TRACE_ROUTING
  999. show_messages("Start of round 1c", round1a.buf, round1a.inserted);
  1000. #endif
  1001. #ifdef PROFILE_ROUTING
  1002. unsigned long start_round1c = printf_with_rtclock("begin round1c processing (%u)\n", round1a.inserted);
  1003. #endif
  1004. // sort round1b_prev msgs with initial msgs in round1a_sorted
  1005. if (prev_nodes > 0) {
  1006. // Copy initial msgs in round1a_sorted to round1b_prev buffer for sorting
  1007. // Note that all inserted values and buffer sizes are non-secret
  1008. uint32_t num_init_round1a = min(round1a.inserted,
  1009. max_round1b_msgs_to_adj_rtr);
  1010. uint32_t num_round1b_prev = round1b_prev.inserted;
  1011. if (num_round1b_prev + num_init_round1a <= max_round1b_msgs_to_adj_rtr) {
  1012. // all our round 1a messages "belong" to previous router and can be removed here
  1013. round1a.inserted = 0;
  1014. } else {
  1015. // copy initial round1a msgs after round1b_prev msgs
  1016. memmove(round1b_prev.buf+num_round1b_prev*msg_size, round1a_sorted.buf,
  1017. num_init_round1a*msg_size);
  1018. // sort and take final msgs as initial round1a msgs
  1019. #ifdef PROFILE_ROUTING
  1020. unsigned long start_sort = printf_with_rtclock("begin round1b_prev oblivious sort (%u,%u)\n", num_round1b_prev + num_init_round1a, 2*max_round1b_msgs_to_adj_rtr);
  1021. #endif
  1022. uint32_t num_copy = num_round1b_prev+num_init_round1a-max_round1b_msgs_to_adj_rtr;
  1023. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, round1b_prev.buf,
  1024. msg_size, num_round1b_prev + num_init_round1a,
  1025. 2*max_round1b_msgs_to_adj_rtr,
  1026. [&](const uint8_t *src, const UidPriorityKey *indices, uint32_t Nr) {
  1027. return copy_msgs(round1a_sorted.buf, max_round1b_msgs_to_adj_rtr,
  1028. num_copy, src, indices);
  1029. }
  1030. );
  1031. round1a.inserted -= (max_round1b_msgs_to_adj_rtr-num_round1b_prev);
  1032. #ifdef PROFILE_ROUTING
  1033. printf_with_rtclock_diff(start_sort, "end round1b_prev oblivious sort (%u,%u)\n", num_round1b_prev + num_init_round1a, 2*max_round1b_msgs_to_adj_rtr);
  1034. #endif
  1035. }
  1036. }
  1037. // sort round1b_next msgs with final msgs in round1a_sorted
  1038. if ((prev_nodes < num_routing_nodes-1) && (round1b_next.inserted > 0)) {
  1039. // Copy final msgs in round1a_sorted to round1b_next buffer for sorting
  1040. // Note that all inserted values and buffer sizes are non-secret
  1041. // round1b_next.inserted>0, so round1a >= max_round1a_msgs-max_round1b_msgs_to_adj_rtr
  1042. uint32_t round1a_msg_start = max_round1a_msgs-max_round1b_msgs_to_adj_rtr;
  1043. uint32_t num_final_round1a = round1a.inserted - round1a_msg_start;
  1044. uint32_t num_round1b_next = round1b_next.inserted;
  1045. memmove(round1b_next.buf+num_round1b_next*msg_size,
  1046. round1a_sorted.buf + round1a_msg_start*msg_size,
  1047. num_final_round1a*msg_size);
  1048. // sort and take initial msgs as final round1a msgs
  1049. #ifdef PROFILE_ROUTING
  1050. unsigned long start_sort = printf_with_rtclock("begin round1b_next oblivious sort (%u,%u)\n", num_round1b_next + num_final_round1a, 2*max_round1b_msgs_to_adj_rtr);
  1051. #endif
  1052. uint32_t num_copy = min(num_final_round1a+num_round1b_next,
  1053. max_round1b_msgs_to_adj_rtr);
  1054. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, round1b_next.buf,
  1055. msg_size, num_round1b_next + num_final_round1a, 2*max_round1b_msgs_to_adj_rtr,
  1056. [&](const uint8_t *src, const UidPriorityKey *indices, uint32_t Nr) {
  1057. return copy_msgs(round1a_sorted.buf + round1a_msg_start*msg_size, 0,
  1058. num_copy, src, indices);
  1059. }
  1060. );
  1061. round1a.inserted += (num_copy - num_final_round1a);
  1062. #ifdef PROFILE_ROUTING
  1063. printf_with_rtclock_diff(start_sort, "end round1b_next oblivious sort (%u,%u)\n", num_round1b_next + num_final_round1a, 2*max_round1b_msgs_to_adj_rtr);
  1064. #endif
  1065. }
  1066. #ifdef TRACE_ROUTING
  1067. printf("round1a_sorted.inserted = %lu. round1a.inserted = %lu\n",
  1068. round1a_sorted.inserted, round1a.inserted);
  1069. show_messages("In round 1c", round1a_sorted.buf,
  1070. round1a.inserted);
  1071. #endif
  1072. #ifdef PROFILE_ROUTING
  1073. unsigned long start_sort = printf_with_rtclock("begin full oblivious sort (%u,%u)\n", round1a.inserted, route_state.max_round1a_msgs);
  1074. #endif
  1075. // Sort received messages by increasing user ID and
  1076. // priority. Smaller priority number indicates higher priority.
  1077. if (round1a.inserted > 0) {
  1078. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, round1a_sorted.buf,
  1079. msg_size, round1a.inserted, route_state.max_round1a_msgs,
  1080. [&](const uint8_t *msgs, const UidPriorityKey *indices, uint32_t N) {
  1081. send_round_robin_msgs<UidPriorityKey>(route_state.round1c, msgs, indices, N);
  1082. });
  1083. } else {
  1084. send_round_robin_msgs<UidPriorityKey>(route_state.round1c, NULL, NULL, 0);
  1085. }
  1086. #ifdef PROFILE_ROUTING
  1087. printf_with_rtclock_diff(start_sort, "end full oblivious sort (%u,%u)\n", round1a.inserted, route_state.max_round1a_msgs);
  1088. printf_with_rtclock_diff(start_round1c, "end round1c processing (%u)\n", round1a.inserted);
  1089. #endif
  1090. round1a.reset();
  1091. round1a_sorted.reset();
  1092. round1b_prev.reset();
  1093. round1b_next.reset();
  1094. pthread_mutex_unlock(&round1a_sorted.mutex);
  1095. pthread_mutex_unlock(&round1a.mutex);
  1096. pthread_mutex_unlock(&round1b_next.mutex);
  1097. pthread_mutex_unlock(&round1b_prev.mutex);
  1098. MsgBuffer &round1c = route_state.round1c;
  1099. pthread_mutex_lock(&round1c.mutex);
  1100. round1c.completed_prev_round = true;
  1101. nodenum_t nodes_received = round1c.nodes_received;
  1102. pthread_mutex_unlock(&round1c.mutex);
  1103. if (nodes_received == num_routing_nodes) {
  1104. route_state.step = ROUTE_ROUND_1C;
  1105. route_state.cbpointer = NULL;
  1106. ocall_routing_round_complete(cbpointer, ROUND_1C);
  1107. }
  1108. } else {
  1109. route_state.step = ROUTE_ROUND_1C;
  1110. route_state.round1c.completed_prev_round = true;
  1111. ocall_routing_round_complete(cbpointer, ROUND_1C);
  1112. }
  1113. }
  1114. // Process messages in round 2
  1115. static void round2_processing(uint8_t my_roles, void *cbpointer, MsgBuffer &prevround) {
  1116. if (my_roles & ROLE_ROUTING) {
  1117. route_state.cbpointer = cbpointer;
  1118. pthread_mutex_lock(&prevround.mutex);
  1119. // Ensure there are no pending messages currently being inserted
  1120. // into the buffer
  1121. while (prevround.reserved != prevround.inserted) {
  1122. pthread_mutex_unlock(&prevround.mutex);
  1123. pthread_mutex_lock(&prevround.mutex);
  1124. }
  1125. // If the _total_ number of messages we received in round 1
  1126. // is less than the max number of messages we could send to
  1127. // _each_ storage node, then cap the number of messages we
  1128. // will send to each storage node to that number.
  1129. uint32_t msgs_per_stg = route_state.max_msg_to_each_stg;
  1130. if (prevround.inserted < msgs_per_stg) {
  1131. msgs_per_stg = prevround.inserted;
  1132. }
  1133. // Note: at this point, it is required that each message in
  1134. // the prevround buffer have a _valid_ storage node id field.
  1135. // Obliviously tally the number of messages we received in
  1136. // the previous round destined for each storage node
  1137. #ifdef TRACE_ROUTING
  1138. show_messages("Start of round 2", prevround.buf, prevround.inserted);
  1139. #endif
  1140. #ifdef PROFILE_ROUTING
  1141. unsigned long start_round2 = printf_with_rtclock("begin round2 processing (%u,%u)\n", prevround.inserted, prevround.bufsize);
  1142. unsigned long start_tally = printf_with_rtclock("begin tally (%u)\n", prevround.inserted);
  1143. #endif
  1144. uint16_t msg_size = g_teems_config.msg_size;
  1145. nodenum_t num_storage_nodes = g_teems_config.num_storage_nodes;
  1146. std::vector<uint32_t> tally = obliv_tally_stg(
  1147. prevround.buf, msg_size, prevround.inserted, num_storage_nodes);
  1148. #ifdef PROFILE_ROUTING
  1149. printf_with_rtclock_diff(start_tally, "end tally (%u)\n", prevround.inserted);
  1150. #endif
  1151. // Note: tally contains private values! It's OK to
  1152. // non-obliviously check for an error condition, though.
  1153. // While we're at it, obliviously change the tally of
  1154. // messages received to a tally of padding messages
  1155. // required.
  1156. uint32_t tot_padding = 0;
  1157. for (nodenum_t i=0; i<num_storage_nodes; ++i) {
  1158. if (tally[i] > msgs_per_stg) {
  1159. printf("Received too many messages for storage node %u\n", i);
  1160. assert(tally[i] <= msgs_per_stg);
  1161. }
  1162. tally[i] = msgs_per_stg - tally[i];
  1163. tot_padding += tally[i];
  1164. }
  1165. prevround.reserved += tot_padding;
  1166. assert(prevround.reserved <= prevround.bufsize);
  1167. // Obliviously add padding for each storage node according
  1168. // to the (private) padding tally.
  1169. #ifdef PROFILE_ROUTING
  1170. unsigned long start_pad = printf_with_rtclock("begin pad (%u)\n", tot_padding);
  1171. #endif
  1172. obliv_pad_stg(prevround.buf + prevround.inserted * msg_size,
  1173. msg_size, tally, tot_padding);
  1174. #ifdef PROFILE_ROUTING
  1175. printf_with_rtclock_diff(start_pad, "end pad (%u)\n", tot_padding);
  1176. #endif
  1177. prevround.inserted += tot_padding;
  1178. // Obliviously shuffle the messages
  1179. #ifdef PROFILE_ROUTING
  1180. unsigned long start_shuffle = printf_with_rtclock("begin shuffle (%u,%u)\n", prevround.inserted, prevround.bufsize);
  1181. #endif
  1182. uint32_t num_shuffled = shuffle_mtobliv(g_teems_config.nthreads,
  1183. prevround.buf, msg_size, prevround.inserted, prevround.bufsize);
  1184. #ifdef PROFILE_ROUTING
  1185. printf_with_rtclock_diff(start_shuffle, "end shuffle (%u,%u)\n", prevround.inserted, prevround.bufsize);
  1186. printf_with_rtclock_diff(start_round2, "end round2 processing (%u,%u)\n", prevround.inserted, prevround.bufsize);
  1187. #endif
  1188. // Now we can handle the messages non-obliviously, since we
  1189. // know there will be exactly msgs_per_stg messages to each
  1190. // storage node, and the oblivious shuffle broke the
  1191. // connection between where each message came from and where
  1192. // it's going.
  1193. send_round2_msgs(num_shuffled, msgs_per_stg, prevround);
  1194. prevround.reset();
  1195. pthread_mutex_unlock(&prevround.mutex);
  1196. }
  1197. if (my_roles & ROLE_STORAGE) {
  1198. route_state.cbpointer = cbpointer;
  1199. MsgBuffer &round2 = route_state.round2;
  1200. pthread_mutex_lock(&round2.mutex);
  1201. round2.completed_prev_round = true;
  1202. nodenum_t nodes_received = round2.nodes_received;
  1203. pthread_mutex_unlock(&round2.mutex);
  1204. if (nodes_received == g_teems_config.num_routing_nodes) {
  1205. route_state.step = ROUTE_ROUND_2;
  1206. route_state.cbpointer = NULL;
  1207. ocall_routing_round_complete(cbpointer, 2);
  1208. }
  1209. } else {
  1210. route_state.step = ROUTE_ROUND_2;
  1211. route_state.round2.completed_prev_round = true;
  1212. ocall_routing_round_complete(cbpointer, 2);
  1213. }
  1214. }
  1215. // Perform the next round of routing. The callback pointer will be
  1216. // passed to ocall_routing_round_complete when the round is complete.
  1217. void ecall_routing_proceed(void *cbpointer)
  1218. {
  1219. uint8_t my_roles = g_teems_config.roles[g_teems_config.my_node_num];
  1220. if (route_state.step == ROUTE_NOT_STARTED) {
  1221. if (my_roles & ROLE_INGESTION) {
  1222. ingestion_epoch++;
  1223. route_state.cbpointer = cbpointer;
  1224. MsgBuffer &ingbuf = route_state.ingbuf;
  1225. pthread_mutex_lock(&ingbuf.mutex);
  1226. // Ensure there are no pending messages currently being inserted
  1227. // into the buffer
  1228. while (ingbuf.reserved != ingbuf.inserted) {
  1229. pthread_mutex_unlock(&ingbuf.mutex);
  1230. pthread_mutex_lock(&ingbuf.mutex);
  1231. }
  1232. // Sort the messages we've received
  1233. #ifdef TRACE_ROUTING
  1234. show_messages("Start of round 1", ingbuf.buf, ingbuf.inserted);
  1235. #endif
  1236. #ifdef PROFILE_ROUTING
  1237. uint32_t inserted = ingbuf.inserted;
  1238. unsigned long start_round1 = printf_with_rtclock("begin round1 processing (%u)\n", inserted);
  1239. unsigned long start_sort = printf_with_rtclock("begin oblivious sort (%u,%u)\n", inserted, route_state.tot_msg_per_ing);
  1240. #endif
  1241. if (g_teems_config.private_routing) {
  1242. sort_mtobliv<UidKey>(g_teems_config.nthreads, ingbuf.buf,
  1243. g_teems_config.msg_size, ingbuf.inserted,
  1244. route_state.tot_msg_per_ing,
  1245. [&](const uint8_t *msgs, const UidKey *indices, uint32_t N) {
  1246. send_round_robin_msgs<UidKey>(route_state.round1, msgs, indices, N);
  1247. });
  1248. } else {
  1249. // Sort received messages by increasing user ID and
  1250. // priority. Smaller priority number indicates higher priority.
  1251. sort_mtobliv<UidPriorityKey>(g_teems_config.nthreads, ingbuf.buf,
  1252. g_teems_config.msg_size, ingbuf.inserted, route_state.tot_msg_per_ing,
  1253. [&](const uint8_t *msgs, const UidPriorityKey *indices, uint32_t N) {
  1254. send_round_robin_msgs<UidPriorityKey>(route_state.round1, msgs, indices, N);
  1255. });
  1256. }
  1257. #ifdef PROFILE_ROUTING
  1258. printf_with_rtclock_diff(start_sort, "end oblivious sort (%u,%u)\n", inserted, route_state.tot_msg_per_ing);
  1259. printf_with_rtclock_diff(start_round1, "end round1 processing (%u)\n", inserted);
  1260. #endif
  1261. ingbuf.reset();
  1262. pthread_mutex_unlock(&ingbuf.mutex);
  1263. }
  1264. if (my_roles & ROLE_ROUTING) {
  1265. MsgBuffer &round1 = route_state.round1;
  1266. pthread_mutex_lock(&round1.mutex);
  1267. round1.completed_prev_round = true;
  1268. nodenum_t nodes_received = round1.nodes_received;
  1269. pthread_mutex_unlock(&round1.mutex);
  1270. if (nodes_received == g_teems_config.num_ingestion_nodes) {
  1271. route_state.step = ROUTE_ROUND_1;
  1272. route_state.cbpointer = NULL;
  1273. ocall_routing_round_complete(cbpointer, 1);
  1274. }
  1275. } else {
  1276. route_state.step = ROUTE_ROUND_1;
  1277. route_state.round1.completed_prev_round = true;
  1278. ocall_routing_round_complete(cbpointer, 1);
  1279. }
  1280. } else if (route_state.step == ROUTE_ROUND_1) {
  1281. if (g_teems_config.private_routing) { // private routing next round
  1282. round2_processing(my_roles, cbpointer, route_state.round1);
  1283. } else { // public routing next round
  1284. round1a_processing(cbpointer);
  1285. }
  1286. } else if (route_state.step == ROUTE_ROUND_1A) {
  1287. round1b_processing(cbpointer);
  1288. } else if (route_state.step == ROUTE_ROUND_1B) {
  1289. round1c_processing(cbpointer);
  1290. } else if (route_state.step == ROUTE_ROUND_1C) {
  1291. round2_processing(my_roles, cbpointer, route_state.round1c);
  1292. } else if (route_state.step == ROUTE_ROUND_2) {
  1293. if (my_roles & ROLE_STORAGE) {
  1294. MsgBuffer &round2 = route_state.round2;
  1295. pthread_mutex_lock(&round2.mutex);
  1296. // Ensure there are no pending messages currently being inserted
  1297. // into the buffer
  1298. while (round2.reserved != round2.inserted) {
  1299. pthread_mutex_unlock(&round2.mutex);
  1300. pthread_mutex_lock(&round2.mutex);
  1301. }
  1302. unsigned long start = printf_with_rtclock("begin storage processing (%u)\n", round2.inserted);
  1303. storage_received(round2);
  1304. printf_with_rtclock_diff(start, "end storage processing (%u)\n", round2.inserted);
  1305. // We're done
  1306. route_state.step = ROUTE_NOT_STARTED;
  1307. ocall_routing_round_complete(cbpointer, 0);
  1308. } else {
  1309. // We're done
  1310. route_state.step = ROUTE_NOT_STARTED;
  1311. ocall_routing_round_complete(cbpointer, 0);
  1312. }
  1313. }
  1314. }