udpsha.cpp 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. #include "udplib.hpp"
  14. #include "udpsha.hpp"
  15. #include "jsocket.hpp"
  16. #include "jlog.hpp"
  17. #include "roxie.hpp"
  18. #include "roxiemem.hpp"
  19. #ifdef _WIN32
  20. #include <winsock2.h>
  21. #else
  22. #include <sys/socket.h>
  23. #endif
  24. #if defined(_DEBUG) && defined(_WIN32) && !defined(USING_MPATROL)
  25. #define new new(_NORMAL_BLOCK, __FILE__, __LINE__)
  26. #endif
  27. using roxiemem::DataBuffer;
  28. using roxiemem::IDataBufferManager;
  29. IDataBufferManager *bufferManager;
  30. unsigned udpTraceLevel = 0;
  31. unsigned udpTraceCategories = (unsigned) -1;
  32. unsigned udpFlowSocketsSize = 131072;
  33. unsigned udpLocalWriteSocketSize = 1024000;
  34. unsigned udpSnifferReadThreadPriority = 3;
  35. unsigned udpSnifferSendThreadPriority = 3;
  36. unsigned multicastTTL = 1;
  37. MODULE_INIT(INIT_PRIORITY_STANDARD)
  38. {
  39. bufferManager = roxiemem::createDataBufferManager(roxiemem::DATA_ALIGNMENT_SIZE);
  40. return true;
  41. }
  42. MODULE_EXIT()
  43. {
  44. bufferManager->Release();
  45. }
  46. // Maintaining a table so each node in the system has a unique index
  47. static IpAddressArray nodeTable;
  48. extern UDPLIB_API const IpAddress &getNodeAddress(unsigned index)
  49. {
  50. return nodeTable.item(index);
  51. }
  52. extern UDPLIB_API unsigned addRoxieNode(const char *ipString)
  53. {
  54. IpAddress ip(ipString);
  55. if (ip.isNull())
  56. throw MakeStringException(ROXIE_UDP_ERROR, "Could not resolve address %s", ipString);
  57. ForEachItemIn(idx, nodeTable)
  58. {
  59. if (ip.ipequals(nodeTable.item(idx)))
  60. return idx;
  61. }
  62. nodeTable.append(ip);
  63. return nodeTable.ordinality()-1;
  64. }
  65. extern UDPLIB_API unsigned getNumNodes()
  66. {
  67. assertex(nodeTable.ordinality());
  68. return nodeTable.ordinality();
  69. }
  70. //---------------------------------------------------------------------------------------------
  71. void queue_t::set_queue_size(unsigned int queue_s)
  72. {
  73. queue_size = queue_s;
  74. element_count = queue_size;
  75. elements = new queue_element[queue_size];
  76. free_space.signal(queue_size);
  77. active_buffers = 0;
  78. first = 0;
  79. last = 0;
  80. }
  81. queue_t::queue_t(unsigned int queue_s)
  82. {
  83. set_queue_size(queue_s);
  84. signal_free_sl = 0;
  85. }
  86. queue_t::queue_t()
  87. {
  88. signal_free_sl = 0;
  89. }
  90. queue_t::~queue_t()
  91. {
  92. delete [] elements;
  93. }
  94. bool queue_t::empty()
  95. {
  96. c_region.enter();
  97. bool res = (active_buffers == 0);
  98. c_region.leave();
  99. return res;
  100. }
  101. int queue_t::free_slots()
  102. {
  103. int res=0;
  104. while (!res)
  105. {
  106. c_region.enter();
  107. res = queue_size - active_buffers;
  108. if (!res)
  109. signal_free_sl++;
  110. c_region.leave();
  111. if (!res)
  112. {
  113. while (!free_sl.wait(3000))
  114. {
  115. if (udpTraceLevel >= 1)
  116. DBGLOG("queue_t::free_slots blocked for 3 seconds waiting for free_sl semaphore");
  117. }
  118. }
  119. }
  120. return res;
  121. }
  122. void queue_t::interrupt()
  123. {
  124. data_avail.interrupt();
  125. }
  126. void queue_t::pushOwn(DataBuffer *buf)
  127. {
  128. while (!free_space.wait(3000))
  129. {
  130. if (udpTraceLevel >= 1)
  131. DBGLOG("queue_t::pushOwn blocked for 3 seconds waiting for free_space semaphore, activeBuffers == %d", active_buffers);
  132. }
  133. c_region.enter();
  134. int next = (last + 1) % element_count;
  135. elements[last].data = buf;
  136. last = next;
  137. active_buffers++;
  138. c_region.leave();
  139. data_avail.signal();
  140. }
  141. DataBuffer *queue_t::pop()
  142. {
  143. data_avail.wait();
  144. DataBuffer *ret = NULL;
  145. bool must_signal;
  146. {
  147. CriticalBlock b(c_region);
  148. if (!active_buffers)
  149. return NULL;
  150. ret = elements[first].data;
  151. first = (first + 1) % element_count;
  152. active_buffers--;
  153. must_signal = signal_free_sl>0;
  154. if (must_signal)
  155. signal_free_sl--;
  156. }
  157. free_space.signal();
  158. if (must_signal)
  159. free_sl.signal();
  160. return ret;
  161. }
  162. bool queue_t::removeData(void *key, PKT_CMP_FUN pkCmpFn)
  163. {
  164. bool ret = false;
  165. CriticalBlock b(c_region);
  166. if (active_buffers)
  167. {
  168. unsigned ix = first;
  169. loop
  170. {
  171. if (elements[ix].data &&
  172. ((key == NULL) || (pkCmpFn == NULL) || pkCmpFn((void*) elements[ix].data, key)))
  173. {
  174. ::Release(elements[ix].data);
  175. elements[ix].data = NULL; // safer than trying to remove it and close up queue - race conditions with code elsewhere
  176. ret = true;
  177. }
  178. ix++;
  179. if (ix==element_count)
  180. ix = 0;
  181. if (ix == last)
  182. break;
  183. }
  184. }
  185. return ret;
  186. }
  187. bool queue_t::dataQueued(void *key, PKT_CMP_FUN pkCmpFn)
  188. {
  189. bool ret = false;
  190. CriticalBlock b(c_region);
  191. if (active_buffers)
  192. {
  193. unsigned ix = first;
  194. loop
  195. {
  196. if (elements[ix].data && pkCmpFn((void*) elements[ix].data, key))
  197. {
  198. ret = true;
  199. break;
  200. }
  201. ix++;
  202. if (ix==element_count)
  203. ix = 0;
  204. if (ix==last)
  205. break;
  206. }
  207. }
  208. return ret;
  209. }
  210. #ifndef _WIN32
  211. #define HOSTENT hostent
  212. #include <netdb.h>
  213. #endif
  214. int check_set(const char *path, int value)
  215. {
  216. #ifdef __linux__
  217. FILE *f = fopen(path,"r");
  218. char res[32];
  219. char *r = 0;
  220. int si = 0;
  221. if (f) {
  222. r = fgets(res, sizeof(res), f);
  223. fclose(f);
  224. }
  225. if (r)
  226. si = atoi(r);
  227. if (!si)
  228. {
  229. DBGLOG("WARNING: Failed to read value for %s", path);
  230. return 0;
  231. }
  232. else if (si<value)
  233. return -1;
  234. #endif
  235. return 0;
  236. }
  237. int check_max_socket_read_buffer(int size) {
  238. return check_set("/proc/sys/net/core/rmem_max", size);
  239. }
  240. int check_max_socket_write_buffer(int size) {
  241. return check_set("/proc/sys/net/core/wmem_max", size);
  242. }
  243. #ifdef __linux__
  244. void setLinuxThreadPriority(int level)
  245. {
  246. pthread_t self = pthread_self();
  247. int policy;
  248. sched_param param;
  249. int rc;
  250. if (( rc = pthread_getschedparam(self, &policy, &param)) != 0)
  251. DBGLOG("pthread_getschedparam error: %d", rc);
  252. if (level < 0)
  253. UNIMPLEMENTED;
  254. else if (!level)
  255. {
  256. param.sched_priority = 0;
  257. policy = SCHED_OTHER;
  258. }
  259. else
  260. {
  261. policy = SCHED_RR;
  262. param.sched_priority = level;
  263. }
  264. if(( rc = pthread_setschedparam(self, policy, &param)) != 0)
  265. DBGLOG("pthread_setschedparam error: %d policy=%i pr=%i id=%" I64F "i TID=%i", rc, policy, param.sched_priority, (unsigned __int64) self, threadLogID());
  266. else
  267. DBGLOG("priority set id=%" I64F "i policy=%i pri=%i TID=%i", (unsigned __int64) self, policy, param.sched_priority, threadLogID());
  268. }
  269. #endif
  270. extern UDPLIB_API void queryMemoryPoolStats(StringBuffer &memStats)
  271. {
  272. if (bufferManager)
  273. bufferManager->poolStats(memStats);
  274. }
  275. /*
  276. Crazy thoughts on network-wide flow control
  277. Avoid sending data that clashes with other outbound or inbound data
  278. is outbound really an issue?
  279. if only inbound, should be easier
  280. can have each inbound node police its own, for a start
  281. udplib already tries to do this
  282. when sending permission to send, best to pick someone that is not sending to anyone else
  283. udplib already tries to do this
  284. but it can still lead to idleness - id node 1 sending to node 2, and node2 to node 1, node3 can't find anyone idle.
  285. If you do need global:
  286. Every bit of data getting sent (perhaps over a certain size threshold?) gets permission from central traffic cop
  287. Outbound packet says source node, target node size
  288. Reply says source,target,size
  289. Cop allows immediately if nothing inflight between those pairs
  290. Cop assumes completion
  291. Cop redundancy
  292. - a backup cop is listening in?
  293. - use multicast for requests and replies?
  294. - no reply implies what?
  295. - backup cop just needs heartbeat from active cop
  296. - permission expires
  297. - multiple cops for blocks of targets?
  298. - but I want global view of who is sending
  299. */