udptrr.cpp 41 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977
  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 <string>
  14. #include <map>
  15. #include <queue>
  16. #include "jthread.hpp"
  17. #include "jlog.hpp"
  18. #include "jisem.hpp"
  19. #include "jsocket.hpp"
  20. #include "udplib.hpp"
  21. #include "udptrr.hpp"
  22. #include "udptrs.hpp"
  23. #include "udpipmap.hpp"
  24. #include "udpmsgpk.hpp"
  25. #include "roxiemem.hpp"
  26. #include "roxie.hpp"
  27. #ifdef _WIN32
  28. #include <io.h>
  29. #include <winsock2.h>
  30. #else
  31. #include <sys/socket.h>
  32. #include <sys/time.h>
  33. #include <sys/resource.h>
  34. #endif
  35. #include <thread>
  36. using roxiemem::DataBuffer;
  37. using roxiemem::IRowManager;
  38. unsigned udpMaxPendingPermits;
  39. RelaxedAtomic<unsigned> flowPermitsSent = {0};
  40. RelaxedAtomic<unsigned> flowRequestsReceived = {0};
  41. RelaxedAtomic<unsigned> dataPacketsReceived = {0};
  42. static unsigned lastFlowPermitsSent = 0;
  43. static unsigned lastFlowRequestsReceived = 0;
  44. static unsigned lastDataPacketsReceived = 0;
  45. class CReceiveManager : implements IReceiveManager, public CInterface
  46. {
  47. /*
  48. * The ReceiveManager has several threads:
  49. * 1. receive_receive_flow (priority 3)
  50. * - waits for packets on flow port
  51. * - maintains list of nodes that have pending requests
  52. * - sends ok_to_send to one sender (or more) at a time
  53. * 2. receive_data (priority 4)
  54. * - reads data packets off data socket
  55. * - runs at v. high priority
  56. * - used to have an option to perform collation on this thread but a bad idea:
  57. * - can block (ends up in memory manager via attachDataBuffer).
  58. * - Does not apply back pressure
  59. * - Just enqueues them. We don't give permission to send more than the queue can hold, but it's a soft limit
  60. * 3. PacketCollator (standard priority)
  61. * - dequeues packets
  62. * - collates packets
  63. *
  64. */
  65. /*
  66. * Handling lost packets
  67. *
  68. * We try to make lost packets unlikely by telling agents when to send (and making sure they don't send unless
  69. * there's a good chance that socket buffer will have room). But we can't legislate for network issues.
  70. *
  71. * What packets can be lost?
  72. * 1. Data packets - handled via sliding window of resendable packets (or by retrying whole query after a timeout, of resend logic disabled)
  73. * 2. RequestToSend - the sender's resend thread checks periodically. There's a short initial timeout for getting a reply (either "request_received"
  74. * or "okToSend"), then a longer timeout for actually sending.
  75. * 3. OkToSend - there is a timeout after which the permission is considered invalid (based on how long it SHOULD take to send them).
  76. * The requestToSend retry mechanism would then make sure retried.
  77. * MORE - if I don't get a response from OkToSend I should assume lost and requeue it.
  78. * 4. complete - covered by same timeout as okToSend. A lost complete will mean incoming data to that node stalls for the duration of this timeout,
  79. *
  80. */
  81. class UdpSenderEntry // one per node in the system
  82. {
  83. // This is created the first time a message from a previously unseen IP arrives, and remains alive indefinitely
  84. // Note that the various members are accessed by different threads, but no member is accessed from more than one thread
  85. // (except where noted) so protection is not required
  86. // Note that UDP ordering rules mean we can't guarantee that we don't see a "request_to_send" for the next transfer before
  87. // we see the "complete" for the current one. Even if we were sure network stack would not reorder, these come from different
  88. // threads on the sender side and the order is not 100% guaranteed, so we need to cope with it.
  89. // We also need to recover gracefully (and preferably quickly) if any flow or data messages go missing. Currently the sender
  90. // will resend the rts if no ok_to_send within timeout, but there may be a better way?
  91. public:
  92. // Used only by receive_flow thread
  93. IpAddress dest;
  94. ISocket *flowSocket = nullptr;
  95. UdpSenderEntry *prevSender = nullptr; // Used to form list of all senders that have outstanding requests
  96. UdpSenderEntry *nextSender = nullptr; // Used to form list of all senders that have outstanding requests
  97. flowType::flowCmd state = flowType::send_completed; // Meaning I'm not on any queue
  98. sequence_t flowSeq = 0; // the sender's most recent flow sequence number
  99. sequence_t sendSeq = 0; // the sender's most recent sequence number from request-to-send, representing sequence number of next packet it will send
  100. unsigned timeouts = 0; // How many consecutive timeouts have happened on the current request
  101. unsigned requestTime = 0; // When we received the active requestToSend
  102. unsigned timeStamp = 0; // When we last sent okToSend
  103. private:
  104. // Updated by receive_data thread, read atomically by receive_flow
  105. mutable CriticalSection psCrit;
  106. PacketTracker packetsSeen;
  107. public:
  108. UdpSenderEntry(const IpAddress &_dest, unsigned port) : dest(_dest)
  109. {
  110. SocketEndpoint ep(port, dest);
  111. flowSocket = ISocket::udp_connect(ep);
  112. }
  113. ~UdpSenderEntry()
  114. {
  115. if (flowSocket)
  116. {
  117. flowSocket->close();
  118. flowSocket->Release();
  119. }
  120. }
  121. bool noteSeen(UdpPacketHeader &hdr)
  122. {
  123. if (udpResendLostPackets)
  124. {
  125. CriticalBlock b(psCrit);
  126. return packetsSeen.noteSeen(hdr);
  127. }
  128. else
  129. return false;
  130. }
  131. bool canSendAny() const
  132. {
  133. // We can send some if (a) the first available new packet is less than TRACKER_BITS above the first unreceived packet or
  134. // (b) we are assuming arrival in order, and there are some marked seen that are > first unseen OR
  135. // (c) the oldest in-flight packet has expired
  136. if (!udpResendLostPackets)
  137. return true;
  138. {
  139. CriticalBlock b(psCrit);
  140. if (packetsSeen.canRecord(sendSeq))
  141. return true;
  142. if (udpAssumeSequential && packetsSeen.hasGaps())
  143. return true;
  144. }
  145. if (msTick()-requestTime > udpResendTimeout)
  146. return true;
  147. return false;
  148. }
  149. void acknowledgeRequest(const IpAddress &returnAddress, sequence_t _flowSeq, sequence_t _sendSeq)
  150. {
  151. if (flowSeq==_flowSeq)
  152. {
  153. // It's a duplicate request-to-send - ignore it? Or assume it means they lost our ok-to-send ? MORE - probably depends on state
  154. if (udpTraceLevel || udpTraceFlow)
  155. {
  156. StringBuffer s;
  157. DBGLOG("UdpFlow: ignoring duplicate requestToSend %" SEQF "u from node %s", _flowSeq, dest.getIpText(s).str());
  158. }
  159. return;
  160. }
  161. flowSeq = _flowSeq;
  162. sendSeq = _sendSeq;
  163. requestTime = msTick();
  164. timeouts = 0;
  165. try
  166. {
  167. UdpPermitToSendMsg msg;
  168. msg.cmd = flowType::request_received;
  169. msg.flowSeq = _flowSeq;
  170. msg.destNode = returnAddress;
  171. msg.max_data = 0;
  172. if (udpResendLostPackets)
  173. {
  174. CriticalBlock b(psCrit);
  175. msg.seen = packetsSeen.copy();
  176. }
  177. if (udpTraceLevel > 3 || udpTraceFlow)
  178. {
  179. StringBuffer ipStr;
  180. DBGLOG("UdpReceiver: sending request_received msg seq %" SEQF "u to node=%s", _flowSeq, dest.getIpText(ipStr).str());
  181. }
  182. flowSocket->write(&msg, udpResendLostPackets ? sizeof(UdpPermitToSendMsg) : offsetof(UdpPermitToSendMsg, seen));
  183. flowPermitsSent++;
  184. }
  185. catch(IException *e)
  186. {
  187. StringBuffer d, s;
  188. DBGLOG("UdpReceiver: acknowledgeRequest failed node=%s %s", dest.getIpText(d).str(), e->errorMessage(s).str());
  189. e->Release();
  190. }
  191. }
  192. void requestToSend(unsigned maxTransfer, const IpAddress &returnAddress)
  193. {
  194. try
  195. {
  196. UdpPermitToSendMsg msg;
  197. msg.cmd = maxTransfer ? flowType::ok_to_send : flowType::request_received;
  198. msg.flowSeq = flowSeq;
  199. msg.destNode = returnAddress;
  200. msg.max_data = maxTransfer;
  201. if (udpResendLostPackets)
  202. {
  203. CriticalBlock b(psCrit);
  204. msg.seen = packetsSeen.copy();
  205. }
  206. if (udpTraceLevel > 3 || udpTraceFlow)
  207. {
  208. StringBuffer ipStr;
  209. DBGLOG("UdpReceiver: sending ok_to_send %u msg seq %" SEQF "u to node=%s", maxTransfer, flowSeq, dest.getIpText(ipStr).str());
  210. }
  211. flowSocket->write(&msg, udpResendLostPackets ? sizeof(UdpPermitToSendMsg) : offsetof(UdpPermitToSendMsg, seen));
  212. flowPermitsSent++;
  213. }
  214. catch(IException *e)
  215. {
  216. StringBuffer d, s;
  217. DBGLOG("UdpReceiver: requestToSend failed node=%s %s", dest.getIpText(d).str(), e->errorMessage(s).str());
  218. e->Release();
  219. }
  220. }
  221. };
  222. class SenderList
  223. {
  224. UdpSenderEntry *head = nullptr;
  225. UdpSenderEntry *tail = nullptr;
  226. unsigned numEntries = 0;
  227. void checkListIsValid(UdpSenderEntry *lookfor)
  228. {
  229. #ifdef _DEBUG
  230. UdpSenderEntry *prev = nullptr;
  231. UdpSenderEntry *finger = head;
  232. unsigned length = 0;
  233. while (finger)
  234. {
  235. if (finger==lookfor)
  236. lookfor = nullptr;
  237. prev = finger;
  238. finger = finger->nextSender;
  239. length++;
  240. }
  241. assert(prev == tail);
  242. assert(lookfor==nullptr);
  243. assert(numEntries==length);
  244. #endif
  245. }
  246. public:
  247. unsigned length() const { return numEntries; }
  248. operator UdpSenderEntry *() const
  249. {
  250. return head;
  251. }
  252. void append(UdpSenderEntry *sender)
  253. {
  254. if (tail)
  255. {
  256. tail->nextSender = sender;
  257. sender->prevSender = tail;
  258. tail = sender;
  259. }
  260. else
  261. {
  262. head = tail = sender;
  263. }
  264. numEntries++;
  265. checkListIsValid(sender);
  266. }
  267. void remove(UdpSenderEntry *sender)
  268. {
  269. if (sender->prevSender)
  270. sender->prevSender->nextSender = sender->nextSender;
  271. else
  272. head = sender->nextSender;
  273. if (sender->nextSender)
  274. sender->nextSender->prevSender = sender->prevSender;
  275. else
  276. tail = sender->prevSender;
  277. sender->prevSender = nullptr;
  278. sender->nextSender = nullptr;
  279. numEntries--;
  280. checkListIsValid(nullptr);
  281. }
  282. };
  283. IpMapOf<UdpSenderEntry> sendersTable;
  284. class receive_receive_flow : public Thread
  285. {
  286. CReceiveManager &parent;
  287. Owned<ISocket> flow_socket;
  288. const unsigned flow_port;
  289. const unsigned maxSlotsPerSender;
  290. std::atomic<bool> running = { false };
  291. SenderList pendingRequests; // List of people wanting permission to send
  292. SenderList pendingPermits; // List of people given permission to send
  293. void enqueueRequest(UdpSenderEntry *requester, sequence_t flowSeq, sequence_t sendSeq)
  294. {
  295. switch (requester->state)
  296. {
  297. case flowType::ok_to_send:
  298. pendingPermits.remove(requester);
  299. // Fall through
  300. case flowType::send_completed:
  301. pendingRequests.append(requester);
  302. requester->state = flowType::request_to_send;
  303. break;
  304. case flowType::request_to_send:
  305. // Perhaps the sender never saw our permission? Already on queue...
  306. break;
  307. default:
  308. // Unexpected state, should never happen!
  309. DBGLOG("ERROR: Unexpected state %s in enqueueRequest", flowType::name(requester->state));
  310. throwUnexpected();
  311. break;
  312. }
  313. requester->acknowledgeRequest(myNode.getIpAddress(), flowSeq, sendSeq); // Acknowledge receipt of the request
  314. }
  315. void okToSend(UdpSenderEntry *requester, unsigned slots)
  316. {
  317. switch (requester->state)
  318. {
  319. case flowType::request_to_send:
  320. pendingRequests.remove(requester);
  321. // Fall through
  322. case flowType::send_completed:
  323. pendingPermits.append(requester);
  324. requester->state = flowType::ok_to_send;
  325. break;
  326. case flowType::ok_to_send:
  327. // Perhaps the sender never saw our permission? Already on queue...
  328. break;
  329. default:
  330. // Unexpected state, should never happen!
  331. DBGLOG("ERROR: Unexpected state %s in okToSend", flowType::name(requester->state));
  332. throwUnexpected();
  333. break;
  334. }
  335. requester->timeStamp = msTick();
  336. requester->requestToSend(slots, myNode.getIpAddress());
  337. }
  338. void noteDone(UdpSenderEntry *requester, UdpRequestToSendMsg &msg)
  339. {
  340. switch (requester->state)
  341. {
  342. case flowType::request_to_send:
  343. // A bit unexpected but will happen if our previous permission timed out and we pushed to back of the requests queue
  344. pendingRequests.remove(requester);
  345. break;
  346. case flowType::ok_to_send:
  347. pendingPermits.remove(requester);
  348. break;
  349. case flowType::send_completed:
  350. DBGLOG("Duplicate completed message received: msg %s flowSeq %" SEQF "u sendSeq %" SEQF "u. Ignoring", flowType::name(msg.cmd), msg.flowSeq, msg.sendSeq);
  351. break;
  352. default:
  353. // Unexpected state, should never happen! Ignore.
  354. DBGLOG("ERROR: Unexpected state %s in noteDone", flowType::name(requester->state));
  355. break;
  356. }
  357. requester->state = flowType::send_completed;
  358. }
  359. public:
  360. receive_receive_flow(CReceiveManager &_parent, unsigned flow_p, unsigned _maxSlotsPerSender)
  361. : Thread("UdpLib::receive_receive_flow"), parent(_parent), flow_port(flow_p), maxSlotsPerSender(_maxSlotsPerSender)
  362. {
  363. if (check_max_socket_read_buffer(udpFlowSocketsSize) < 0)
  364. throw MakeStringException(ROXIE_UDP_ERROR, "System Socket max read buffer is less than %i", udpFlowSocketsSize);
  365. flow_socket.setown(ISocket::udp_create(flow_port));
  366. flow_socket->set_receive_buffer_size(udpFlowSocketsSize);
  367. size32_t actualSize = flow_socket->get_receive_buffer_size();
  368. DBGLOG("UdpReceiver: receive_receive_flow created port=%d sockbuffsize=%d actual %d", flow_port, udpFlowSocketsSize, actualSize);
  369. }
  370. ~receive_receive_flow()
  371. {
  372. running = false;
  373. if (flow_socket)
  374. flow_socket->close();
  375. join();
  376. }
  377. virtual void start()
  378. {
  379. running = true;
  380. Thread::start();
  381. }
  382. virtual int run() override
  383. {
  384. DBGLOG("UdpReceiver: receive_receive_flow started");
  385. #ifdef __linux__
  386. setLinuxThreadPriority(3);
  387. #else
  388. adjustPriority(1);
  389. #endif
  390. UdpRequestToSendMsg msg;
  391. unsigned timeout = 5000;
  392. while (running)
  393. {
  394. try
  395. {
  396. if (udpTraceLevel > 5 || udpTraceFlow)
  397. {
  398. DBGLOG("UdpReceiver: wait_read(%u)", timeout);
  399. }
  400. bool dataAvail = flow_socket->wait_read(timeout);
  401. if (dataAvail)
  402. {
  403. const unsigned l = sizeof(msg);
  404. unsigned int res ;
  405. flow_socket->readtms(&msg, l, l, res, 0);
  406. flowRequestsReceived++;
  407. assert(res==l);
  408. if (udpTraceLevel > 5 || udpTraceFlow)
  409. {
  410. StringBuffer ipStr;
  411. DBGLOG("UdpReceiver: received %s msg flowSeq %" SEQF "u sendSeq %" SEQF "u from node=%s", flowType::name(msg.cmd), msg.flowSeq, msg.sendSeq, msg.sourceNode.getTraceText(ipStr).str());
  412. }
  413. UdpSenderEntry *sender = &parent.sendersTable[msg.sourceNode];
  414. switch (msg.cmd)
  415. {
  416. case flowType::request_to_send:
  417. enqueueRequest(sender, msg.flowSeq, msg.sendSeq);
  418. break;
  419. case flowType::send_completed:
  420. noteDone(sender, msg);
  421. break;
  422. case flowType::request_to_send_more:
  423. noteDone(sender, msg);
  424. enqueueRequest(sender, msg.flowSeq+1, msg.sendSeq);
  425. break;
  426. default:
  427. DBGLOG("UdpReceiver: received unrecognized flow control message cmd=%i", msg.cmd);
  428. }
  429. }
  430. timeout = 5000; // The default timeout is 5 seconds if nothing is waiting for response...
  431. if (pendingPermits)
  432. {
  433. unsigned now = msTick();
  434. for (UdpSenderEntry *finger = pendingPermits; finger != nullptr; )
  435. {
  436. if (now - finger->timeStamp >= udpRequestToSendAckTimeout)
  437. {
  438. if (udpTraceLevel || udpTraceFlow || udpTraceTimeouts)
  439. {
  440. StringBuffer s;
  441. DBGLOG("permit to send %" SEQF "u to node %s timed out after %u ms, rescheduling", finger->flowSeq, finger->dest.getIpText(s).str(), udpRequestToSendAckTimeout);
  442. }
  443. UdpSenderEntry *next = finger->nextSender;
  444. pendingPermits.remove(finger);
  445. if (++finger->timeouts > udpMaxRetryTimedoutReqs && udpMaxRetryTimedoutReqs != 0)
  446. {
  447. if (udpTraceLevel || udpTraceFlow || udpTraceTimeouts)
  448. {
  449. StringBuffer s;
  450. DBGLOG("permit to send %" SEQF "u to node %s timed out %u times - abandoning", finger->flowSeq, finger->dest.getIpText(s).str(), finger->timeouts);
  451. }
  452. }
  453. else
  454. {
  455. // Put it back on the queue (at the back)
  456. finger->timeStamp = now;
  457. pendingRequests.append(finger);
  458. finger->state = flowType::request_to_send;
  459. }
  460. finger = next;
  461. }
  462. else
  463. {
  464. timeout = finger->timeStamp + udpRequestToSendAckTimeout - now;
  465. break;
  466. }
  467. }
  468. }
  469. unsigned slots = parent.input_queue->available();
  470. bool anyCanSend = false;
  471. for (UdpSenderEntry *finger = pendingRequests; finger != nullptr; finger = finger->nextSender)
  472. {
  473. if (pendingPermits.length()>=udpMaxPendingPermits)
  474. break;
  475. if (!slots) // || slots<minSlotsPerSender)
  476. {
  477. timeout = 1; // Slots should free up very soon!
  478. break;
  479. }
  480. // If requester would not be able to send me any (because of the ones in flight) then wait
  481. if (finger->canSendAny())
  482. {
  483. unsigned requestSlots = slots;
  484. if (requestSlots>maxSlotsPerSender)
  485. requestSlots = maxSlotsPerSender;
  486. okToSend(finger, requestSlots);
  487. slots -= requestSlots;
  488. if (timeout > udpRequestToSendAckTimeout)
  489. timeout = udpRequestToSendAckTimeout;
  490. anyCanSend = true;
  491. }
  492. else
  493. {
  494. if (udpTraceFlow)
  495. {
  496. StringBuffer s;
  497. DBGLOG("Sender %s can't be given permission to send yet as resend buffer full", finger->dest.getIpText(s).str());
  498. }
  499. }
  500. }
  501. if (slots && pendingRequests.length() && pendingPermits.length()<udpMaxPendingPermits && !anyCanSend)
  502. {
  503. if (udpTraceFlow)
  504. {
  505. StringBuffer s;
  506. DBGLOG("All senders blocked by resend buffers");
  507. }
  508. timeout = 1; // Hopefully one of the senders should unblock soon
  509. }
  510. }
  511. catch (IException *e)
  512. {
  513. if (running)
  514. {
  515. StringBuffer s;
  516. DBGLOG("UdpReceiver: failed %i %s", flow_port, e->errorMessage(s).str());
  517. }
  518. e->Release();
  519. }
  520. catch (...)
  521. {
  522. DBGLOG("UdpReceiver: receive_receive_flow::run unknown exception");
  523. }
  524. }
  525. return 0;
  526. }
  527. };
  528. class receive_data : public Thread
  529. {
  530. CReceiveManager &parent;
  531. ISocket *receive_socket;
  532. std::atomic<bool> running = { false };
  533. Semaphore started;
  534. public:
  535. receive_data(CReceiveManager &_parent) : Thread("UdpLib::receive_data"), parent(_parent)
  536. {
  537. unsigned ip_buffer = parent.input_queue_size*DATA_PAYLOAD*2;
  538. if (ip_buffer < udpFlowSocketsSize) ip_buffer = udpFlowSocketsSize;
  539. if (check_max_socket_read_buffer(ip_buffer) < 0)
  540. throw MakeStringException(ROXIE_UDP_ERROR, "System socket max read buffer is less than %u", ip_buffer);
  541. receive_socket = ISocket::udp_create(parent.data_port);
  542. receive_socket->set_receive_buffer_size(ip_buffer);
  543. size32_t actualSize = receive_socket->get_receive_buffer_size();
  544. DBGLOG("UdpReceiver: rcv_data_socket created port=%d requested sockbuffsize=%d actual sockbuffsize=%d", parent.data_port, ip_buffer, actualSize);
  545. running = false;
  546. }
  547. virtual void start()
  548. {
  549. running = true;
  550. Thread::start();
  551. started.wait();
  552. }
  553. ~receive_data()
  554. {
  555. running = false;
  556. if (receive_socket)
  557. receive_socket->close();
  558. join();
  559. ::Release(receive_socket);
  560. }
  561. virtual int run()
  562. {
  563. DBGLOG("UdpReceiver: receive_data started");
  564. #ifdef __linux__
  565. setLinuxThreadPriority(4);
  566. #else
  567. adjustPriority(2);
  568. #endif
  569. DataBuffer *b = NULL;
  570. started.signal();
  571. unsigned lastOOOReport = 0;
  572. unsigned lastPacketsOOO = 0;
  573. while (running)
  574. {
  575. try
  576. {
  577. unsigned int res;
  578. b = bufferManager->allocate();
  579. receive_socket->read(b->data, 1, DATA_PAYLOAD, res, 5);
  580. dataPacketsReceived++;
  581. UdpPacketHeader &hdr = *(UdpPacketHeader *) b->data;
  582. assert(hdr.length == res && hdr.length > sizeof(hdr));
  583. UdpSenderEntry *sender = &parent.sendersTable[hdr.node];
  584. if (sender->noteSeen(hdr))
  585. {
  586. if (udpTraceLevel > 5) // don't want to interrupt this thread if we can help it
  587. {
  588. StringBuffer s;
  589. DBGLOG("UdpReceiver: discarding unwanted resent packet %" SEQF "u %x from %s", hdr.sendSeq, hdr.pktSeq, hdr.node.getTraceText(s).str());
  590. }
  591. parent.noteDuplicate(b);
  592. ::Release(b);
  593. }
  594. else
  595. {
  596. if (udpTraceLevel > 5) // don't want to interrupt this thread if we can help it
  597. {
  598. StringBuffer s;
  599. DBGLOG("UdpReceiver: %u bytes received packet %" SEQF "u %x from %s", res, hdr.sendSeq, hdr.pktSeq, hdr.node.getTraceText(s).str());
  600. }
  601. parent.input_queue->pushOwn(b);
  602. }
  603. b = NULL;
  604. }
  605. catch (IException *e)
  606. {
  607. ::Release(b);
  608. b = NULL;
  609. if (running && e->errorCode() != JSOCKERR_timeout_expired)
  610. {
  611. StringBuffer s;
  612. DBGLOG("UdpReceiver: receive_data::run read failed port=%u - Exp: %s", parent.data_port, e->errorMessage(s).str());
  613. MilliSleep(1000); // Give a chance for mem free
  614. }
  615. e->Release();
  616. }
  617. catch (...)
  618. {
  619. ::Release(b);
  620. b = NULL;
  621. DBGLOG("UdpReceiver: receive_data::run unknown exception port %u", parent.data_port);
  622. MilliSleep(1000);
  623. }
  624. if (udpStatsReportInterval)
  625. {
  626. unsigned now = msTick();
  627. if (now-lastOOOReport > udpStatsReportInterval)
  628. {
  629. lastOOOReport = now;
  630. if (packetsOOO > lastPacketsOOO)
  631. {
  632. DBGLOG("%u more packets received out-of-order by this server (%u total)", packetsOOO-lastPacketsOOO, packetsOOO-0);
  633. lastPacketsOOO = packetsOOO;
  634. }
  635. if (flowRequestsReceived > lastFlowRequestsReceived)
  636. {
  637. DBGLOG("%u more flow requests received by this server (%u total)", flowRequestsReceived-lastFlowRequestsReceived, flowRequestsReceived-0);
  638. lastFlowRequestsReceived = flowRequestsReceived;
  639. }
  640. if (flowPermitsSent > lastFlowPermitsSent)
  641. {
  642. DBGLOG("%u more flow permits sent by this server (%u total)", flowPermitsSent-lastFlowPermitsSent, flowPermitsSent-0);
  643. lastFlowPermitsSent = flowPermitsSent;
  644. }
  645. if (dataPacketsReceived > lastDataPacketsReceived)
  646. {
  647. DBGLOG("%u more data packets received by this server (%u total)", dataPacketsReceived-lastDataPacketsReceived, dataPacketsReceived-0);
  648. lastDataPacketsReceived = dataPacketsReceived;
  649. }
  650. }
  651. }
  652. }
  653. ::Release(b);
  654. return 0;
  655. }
  656. };
  657. class CPacketCollator : public Thread
  658. {
  659. CReceiveManager &parent;
  660. public:
  661. CPacketCollator(CReceiveManager &_parent) : Thread("CPacketCollator"), parent(_parent) {}
  662. virtual int run()
  663. {
  664. DBGLOG("UdpReceiver: CPacketCollator::run");
  665. parent.collatePackets();
  666. return 0;
  667. }
  668. } collatorThread;
  669. friend class receive_receive_flow;
  670. friend class receive_send_flow;
  671. friend class receive_data;
  672. friend class ReceiveFlowManager;
  673. queue_t *input_queue;
  674. int input_queue_size;
  675. receive_receive_flow *receive_flow;
  676. receive_data *data;
  677. int receive_flow_port;
  678. int data_port;
  679. std::atomic<bool> running = { false };
  680. bool encrypted = false;
  681. typedef std::map<ruid_t, CMessageCollator*> uid_map;
  682. uid_map collators;
  683. SpinLock collatorsLock; // protects access to collators map
  684. public:
  685. IMPLEMENT_IINTERFACE;
  686. CReceiveManager(int server_flow_port, int d_port, int client_flow_port, int queue_size, int m_slot_pr_client, bool _encrypted)
  687. : collatorThread(*this), encrypted(_encrypted), sendersTable([client_flow_port](const ServerIdentifier ip) { return new UdpSenderEntry(ip.getIpAddress(), client_flow_port);})
  688. {
  689. #ifndef _WIN32
  690. setpriority(PRIO_PROCESS, 0, -15);
  691. #endif
  692. receive_flow_port = server_flow_port;
  693. data_port = d_port;
  694. input_queue_size = queue_size;
  695. input_queue = new queue_t(queue_size);
  696. data = new receive_data(*this);
  697. receive_flow = new receive_receive_flow(*this, server_flow_port, m_slot_pr_client);
  698. running = true;
  699. collatorThread.start();
  700. data->start();
  701. receive_flow->start();
  702. MilliSleep(15);
  703. }
  704. ~CReceiveManager()
  705. {
  706. running = false;
  707. input_queue->interrupt();
  708. collatorThread.join();
  709. delete data;
  710. delete receive_flow;
  711. delete input_queue;
  712. }
  713. virtual void detachCollator(const IMessageCollator *msgColl)
  714. {
  715. ruid_t ruid = msgColl->queryRUID();
  716. if (udpTraceLevel >= 2) DBGLOG("UdpReceiver: detach %p %u", msgColl, ruid);
  717. {
  718. SpinBlock b(collatorsLock);
  719. collators.erase(ruid);
  720. }
  721. msgColl->Release();
  722. }
  723. void collatePackets()
  724. {
  725. while(running)
  726. {
  727. DataBuffer *dataBuff = input_queue->pop(true);
  728. collatePacket(dataBuff);
  729. }
  730. }
  731. void noteDuplicate(DataBuffer *dataBuff)
  732. {
  733. const UdpPacketHeader *pktHdr = (UdpPacketHeader*) dataBuff->data;
  734. Linked <CMessageCollator> msgColl;
  735. SpinBlock b(collatorsLock);
  736. try
  737. {
  738. msgColl.set(collators[pktHdr->ruid]);
  739. }
  740. catch (IException *E)
  741. {
  742. EXCLOG(E);
  743. E->Release();
  744. }
  745. catch (...)
  746. {
  747. IException *E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unexpected exception caught in CPacketCollator::run");
  748. EXCLOG(E);
  749. E->Release();
  750. }
  751. if (msgColl)
  752. msgColl->noteDuplicate((pktHdr->pktSeq & UDP_PACKET_RESENT) != 0);
  753. }
  754. void collatePacket(DataBuffer *dataBuff)
  755. {
  756. const UdpPacketHeader *pktHdr = (UdpPacketHeader*) dataBuff->data;
  757. if (udpTraceLevel >= 4)
  758. {
  759. StringBuffer s;
  760. DBGLOG("UdpReceiver: CPacketCollator - unQed packet - ruid=" RUIDF " id=0x%.8X mseq=%u pkseq=0x%.8X len=%d node=%s",
  761. pktHdr->ruid, pktHdr->msgId, pktHdr->msgSeq, pktHdr->pktSeq, pktHdr->length, pktHdr->node.getTraceText(s).str());
  762. }
  763. Linked <CMessageCollator> msgColl;
  764. bool isDefault = false;
  765. {
  766. SpinBlock b(collatorsLock);
  767. try
  768. {
  769. msgColl.set(collators[pktHdr->ruid]);
  770. if (!msgColl)
  771. {
  772. msgColl.set(collators[RUID_DISCARD]);
  773. isDefault = true;
  774. unwantedDiscarded++;
  775. }
  776. }
  777. catch (IException *E)
  778. {
  779. EXCLOG(E);
  780. E->Release();
  781. }
  782. catch (...)
  783. {
  784. IException *E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unexpected exception caught in CPacketCollator::run");
  785. EXCLOG(E);
  786. E->Release();
  787. }
  788. }
  789. if (udpTraceLevel && isDefault)
  790. {
  791. StringBuffer s;
  792. DBGLOG("UdpReceiver: CPacketCollator NO msg collator found - using default - ruid=" RUIDF " id=0x%.8X mseq=%u pkseq=0x%.8X node=%s", pktHdr->ruid, pktHdr->msgId, pktHdr->msgSeq, pktHdr->pktSeq, pktHdr->node.getTraceText(s).str());
  793. }
  794. if (msgColl && msgColl->attach_databuffer(dataBuff))
  795. dataBuff = nullptr;
  796. else
  797. dataBuff->Release();
  798. }
  799. virtual IMessageCollator *createMessageCollator(IRowManager *rowManager, ruid_t ruid)
  800. {
  801. CMessageCollator *msgColl = new CMessageCollator(rowManager, ruid, encrypted);
  802. if (udpTraceLevel > 2)
  803. DBGLOG("UdpReceiver: createMessageCollator %p %u", msgColl, ruid);
  804. {
  805. SpinBlock b(collatorsLock);
  806. collators[ruid] = msgColl;
  807. }
  808. msgColl->Link();
  809. return msgColl;
  810. }
  811. };
  812. IReceiveManager *createReceiveManager(int server_flow_port, int data_port, int client_flow_port,
  813. int udpQueueSize, unsigned maxSlotsPerSender,
  814. bool encrypted)
  815. {
  816. assertex (maxSlotsPerSender <= (unsigned) udpQueueSize);
  817. assertex (maxSlotsPerSender <= (unsigned) TRACKER_BITS);
  818. return new CReceiveManager(server_flow_port, data_port, client_flow_port, udpQueueSize, maxSlotsPerSender, encrypted);
  819. }
  820. /*
  821. Thoughts on flow control / streaming:
  822. 1. The "continuation packet" mechanism does have some advantages
  823. - easy recovery from agent failures
  824. - agent recovers easily from Roxie server failures
  825. - flow control is simple (but is it effective?)
  826. 2. Abandoning continuation packet in favour of streaming would give us the following issues:
  827. - would need some flow control to stop getting ahead of a Roxie server that consumed slowly
  828. - flow control is non trivial if you want to avoid tying up a agent thread and want agent to be able to recover from Roxie server failure
  829. - Need to work out how to do GSS - the nextGE info needs to be passed back in the flow control?
  830. - can't easily recover from agent failures if you already started processing
  831. - unless you assume that the results from agent are always deterministic and can retry and skip N
  832. - potentially ties up a agent thread for a while
  833. - do we need to have a larger thread pool but limit how many actually active?
  834. 3. Order of work
  835. - Just adding streaming while ignoring flow control and continuation stuff (i.e. we still stop for permission to continue periodically)
  836. - Shouldn't make anything any _worse_ ...
  837. - except that won't be able to recover from a agent dying mid-stream (at least not without some considerable effort)
  838. - what will happen then?
  839. - May also break server-side caching (that no-one has used AFAIK). Maybe restrict to nohits as we change....
  840. - Add some flow control
  841. - would prevent agent getting too far ahead in cases that are inadequately flow-controlled today
  842. - shouldn't make anything any worse...
  843. - Think about removing continuation mechanism from some cases
  844. Per Gavin, streaming would definitely help for the lowest frequency term. It may help for the others as well if it avoided any significant start up costs - e.g., opening the indexes,
  845. creating the segment monitors, creating the various cursors, and serialising the context (especially because there are likely to be multiple cursors).
  846. To add streaming:
  847. - Need to check for meta availability other than when first received
  848. - when ?
  849. - Need to cope with a getNext() blocking without it causing issues
  850. - perhaps should recode getNext() of variable-size rows first?
  851. More questions:
  852. - Can we afford the memory for the resend info?
  853. - Save maxPacketsPerSender per sender ?
  854. - are we really handling restart and sequence wraparound correctly?
  855. - what about server-side caching? Makes it hard
  856. - but maybe we should only cache tiny replies anyway....
  857. Problems found while testing implemetnation:
  858. - the unpacker cursor read code is crap
  859. - there is a potential to deadlock when need to make a callback agent->server during a streamed result (indexread5 illustrates)
  860. - resolution callback code doesn't really need to be query specific - could go to the default handler
  861. - but other callbacks - ALIVE, EXCEPTION, and debugger are not so clear
  862. - It's not at all clear where to move the code for processing metadata
  863. - callback paradigm would solve both - but it has to be on a client thread (e.g. from within call to next()).
  864. The following are used in "pseudo callback" mode:
  865. #define ROXIE_DEBUGREQUEST 0x3ffffff7u
  866. #define ROXIE_DEBUGCALLBACK 0x3ffffff8u
  867. #define ROXIE_PING 0x3ffffff9u
  868. - goes to own handler anyway
  869. #define ROXIE_TRACEINFO 0x3ffffffau
  870. - could go in meta? Not time critical. Could all go to single handler? (a bit hard since we want to intercept for caller...)
  871. #define ROXIE_FILECALLBACK 0x3ffffffbu
  872. - could go to single handler
  873. #define ROXIE_ALIVE 0x3ffffffcu
  874. - currently getting delayed a bit too much potentially if downstream processing is slow? Do I even need it if streaming?
  875. #define ROXIE_KEYEDLIMIT_EXCEEDED 0x3ffffffdu
  876. - could go in metadata of standard response
  877. #define ROXIE_LIMIT_EXCEEDED 0x3ffffffeu
  878. - ditto
  879. #define ROXIE_EXCEPTION 0x3fffffffu
  880. - ditto
  881. And the continuation metadata.
  882. What if EVERYTHING was a callback? - here's an exception... here's some more rows... here's some tracing... here's some continuation metadata
  883. Somewhere sometime I need to marshall from one thread to another though (maybe more than once unless I can guarantee callback is always very fast)
  884. OR (is it the same) everything is metadata ? Metadata can contain any of the above information (apart from rows - or maybe they are just another type)
  885. If I can't deal quickly with a packet of information, I queue it up? Spanning complicates things though. I need to be able to spot complete portions of metadata
  886. (and in kind-of the same way I need to be able to spot complete rows of data even when they span multiple packets.) I think data is really a bit different from the rest -
  887. you expect it to be continuous and you want the others to interrupt the flow.
  888. If continuation info was restricted to a "yes/no" (i.e. had to be continued on same node as started on) could have simple "Is there any continuation" bit. Others are sent in their
  889. own packets so are a little different. Does that make it harder to recover? Not sure that it does really (just means that the window at which a failure causes a problem starts earlier).
  890. However it may be an issue tying up agent thread for a while (and do we know when to untie it if the Roxie server abandons/restarts?)
  891. Perhaps it makes sense to pause at this point (with streaming disabled and with retry mechanism optional)
  892. */