route.cpp 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214
  1. #include <pthread.h>
  2. #include "Enclave_t.h"
  3. #include "config.hpp"
  4. #include "utils.hpp"
  5. #include "sort.hpp"
  6. #include "route.hpp"
  7. #define PROFILE_ROUTING
  8. struct MsgBuffer {
  9. pthread_mutex_t mutex;
  10. uint8_t *buf;
  11. // The number of messages (not bytes) in, or on their way into, the
  12. // buffer
  13. uint32_t reserved;
  14. // The number of messages definitely in the buffer
  15. uint32_t inserted;
  16. MsgBuffer() : buf(NULL), reserved(0), inserted(0) {
  17. pthread_mutex_init(&mutex, NULL);
  18. }
  19. ~MsgBuffer() {
  20. delete[] buf;
  21. }
  22. // The number passed is messages, not bytes
  23. void alloc(uint32_t msgs) {
  24. delete[] buf;
  25. buf = NULL;
  26. reserved = 0;
  27. inserted = 0;
  28. // This may throw bad_alloc, but we'll catch it higher up
  29. buf = new uint8_t[size_t(msgs) * g_teems_config.msg_size];
  30. }
  31. // You can't copy a MsgBuffer
  32. MsgBuffer(const MsgBuffer&) = delete;
  33. MsgBuffer &operator=(const MsgBuffer&) = delete;
  34. };
  35. enum RouteStep {
  36. ROUTE_NOT_STARTED,
  37. ROUTE_ROUND_1,
  38. ROUTE_ROUND_2
  39. };
  40. // The round1 MsgBuffer stores messages we ingest while waiting for
  41. // round 1 to start, which will be sorted and sent out in round 1. The
  42. // round2 MsgBuffer stores messages we receive in round 1, which will be
  43. // padded, sorted, and sent out in round 2.
  44. static struct RouteState {
  45. MsgBuffer round1;
  46. MsgBuffer round2;
  47. RouteStep step;
  48. uint32_t tot_msg_per_ing;
  49. uint32_t max_msg_to_each_str;
  50. uint32_t max_round2_msgs;
  51. } route_state;
  52. // Computes ceil(x/y) where x and y are integers, x>=0, y>0.
  53. #define CEILDIV(x,y) (((x)+(y)-1)/(y))
  54. // Call this near the end of ecall_config_load, but before
  55. // comms_init_nodestate. Returns true on success, false on failure.
  56. bool route_init()
  57. {
  58. // Compute the maximum number of messages we could receive by direct
  59. // ingestion
  60. // Each ingestion node will have at most
  61. // ceil(user_count/num_ingestion_nodes) users, and each user will
  62. // send at most m_priv_out messages.
  63. uint32_t users_per_ing = CEILDIV(g_teems_config.user_count,
  64. g_teems_config.num_ingestion_nodes);
  65. uint32_t tot_msg_per_ing = users_per_ing * g_teems_config.m_priv_out;
  66. // Compute the maximum number of messages we could receive in round 1
  67. // Each ingestion node will send us an our_weight/tot_weight
  68. // fraction of the messages they hold
  69. uint32_t max_msg_from_each_ing = CEILDIV(tot_msg_per_ing,
  70. g_teems_config.tot_weight) * g_teems_config.my_weight;
  71. // And the maximum number we can receive in total is that times the
  72. // number of ingestion nodes
  73. uint32_t max_round1_msgs = max_msg_from_each_ing *
  74. g_teems_config.num_ingestion_nodes;
  75. // Compute the maximum number of messages we could send in round 2
  76. // Each storage node has at most this many users
  77. uint32_t users_per_str = CEILDIV(g_teems_config.user_count,
  78. g_teems_config.num_storage_nodes);
  79. // And so can receive at most this many messages
  80. uint32_t tot_msg_per_str = users_per_str *
  81. g_teems_config.m_priv_in;
  82. // Which will be at most this many from us
  83. uint32_t max_msg_to_each_str = CEILDIV(tot_msg_per_str,
  84. g_teems_config.tot_weight) * g_teems_config.my_weight;
  85. // But we can't send more messages to each storage server than we
  86. // could receive in total
  87. if (max_msg_to_each_str > max_round1_msgs) {
  88. max_msg_to_each_str = max_round1_msgs;
  89. }
  90. // And the max total number of outgoing messages in round 2 is then
  91. uint32_t max_round2_msgs = max_msg_to_each_str *
  92. g_teems_config.num_storage_nodes;
  93. // In case we have a weird configuration where users can send more
  94. // messages per epoch than they can receive, ensure the round 2
  95. // buffer is large enough to hold the incoming messages as well
  96. if (max_round2_msgs < max_round1_msgs) {
  97. max_round2_msgs = max_round1_msgs;
  98. }
  99. /*
  100. printf("round1_msgs = %u, round2_msgs = %u\n",
  101. max_round1_msgs, max_round2_msgs);
  102. */
  103. // Create the route state
  104. try {
  105. route_state.round1.alloc(tot_msg_per_ing);
  106. route_state.round2.alloc(max_round2_msgs);
  107. } catch (std::bad_alloc&) {
  108. printf("Memory allocation failed in route_init\n");
  109. return false;
  110. }
  111. route_state.step = ROUTE_NOT_STARTED;
  112. route_state.tot_msg_per_ing = tot_msg_per_ing;
  113. route_state.max_msg_to_each_str = max_msg_to_each_str;
  114. route_state.max_round2_msgs = max_round2_msgs;
  115. threadid_t nthreads = g_teems_config.nthreads;
  116. #ifdef PROFILE_ROUTING
  117. unsigned long start = printf_with_rtclock("begin precompute evalplans (%u,%hu) (%u,%hu)\n", tot_msg_per_ing, nthreads, max_round2_msgs, nthreads);
  118. #endif
  119. sort_precompute_evalplan(tot_msg_per_ing, nthreads);
  120. sort_precompute_evalplan(max_round2_msgs, nthreads);
  121. #ifdef PROFILE_ROUTING
  122. printf_with_rtclock_diff(start, "end precompute evalplans\n");
  123. #endif
  124. return true;
  125. }
  126. // Precompute the WaksmanNetworks needed for the sorts. If you pass -1,
  127. // it will return the number of different sizes it needs. If you pass
  128. // [0,sizes-1], it will compute one WaksmanNetwork with that size index
  129. // and return the number of available WaksmanNetworks of that size.
  130. size_t ecall_precompute_sort(int sizeidx)
  131. {
  132. size_t ret = 0;
  133. switch(sizeidx) {
  134. case 0:
  135. #ifdef PROFILE_ROUTING
  136. {unsigned long start = printf_with_rtclock("begin precompute WaksmanNetwork (%u)\n", route_state.tot_msg_per_ing);
  137. #endif
  138. ret = sort_precompute(route_state.tot_msg_per_ing);
  139. #ifdef PROFILE_ROUTING
  140. printf_with_rtclock_diff(start, "end precompute Waksman Network (%u)\n", route_state.tot_msg_per_ing);}
  141. #endif
  142. break;
  143. case 1:
  144. #ifdef PROFILE_ROUTING
  145. {unsigned long start = printf_with_rtclock("begin precompute WaksmanNetwork (%u)\n", route_state.max_round2_msgs);
  146. #endif
  147. ret = sort_precompute(route_state.max_round2_msgs);
  148. #ifdef PROFILE_ROUTING
  149. printf_with_rtclock_diff(start, "end precompute Waksman Network (%u)\n", route_state.max_round2_msgs);}
  150. #endif
  151. break;
  152. default:
  153. ret = 2;
  154. break;
  155. }
  156. return ret;
  157. }
  158. // Directly ingest a buffer of num_msgs messages into the round1 buffer.
  159. // Return true on success, false on failure.
  160. bool ecall_ingest_raw(uint8_t *msgs, uint32_t num_msgs)
  161. {
  162. uint16_t msg_size = g_teems_config.msg_size;
  163. MsgBuffer &round1 = route_state.round1;
  164. pthread_mutex_lock(&round1.mutex);
  165. uint32_t start = round1.reserved;
  166. if (start + num_msgs > route_state.tot_msg_per_ing) {
  167. pthread_mutex_unlock(&round1.mutex);
  168. printf("Max %u messages exceeded\n",
  169. route_state.tot_msg_per_ing);
  170. return false;
  171. }
  172. round1.reserved += num_msgs;
  173. pthread_mutex_unlock(&round1.mutex);
  174. memmove(round1.buf + start * msg_size,
  175. msgs, num_msgs * msg_size);
  176. pthread_mutex_lock(&round1.mutex);
  177. round1.inserted += num_msgs;
  178. pthread_mutex_unlock(&round1.mutex);
  179. return true;
  180. }