udptrr.cpp 41 KB

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