Socket.c 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503
  1. /*******************************************************************************
  2. * Copyright (c) 2009, 2022 IBM Corp., Ian Craggs and others
  3. *
  4. * All rights reserved. This program and the accompanying materials
  5. * are made available under the terms of the Eclipse Public License v2.0
  6. * and Eclipse Distribution License v1.0 which accompany this distribution.
  7. *
  8. * The Eclipse Public License is available at
  9. * https://www.eclipse.org/legal/epl-2.0/
  10. * and the Eclipse Distribution License is available at
  11. * http://www.eclipse.org/org/documents/edl-v10.php.
  12. *
  13. * Contributors:
  14. * Ian Craggs - initial implementation and documentation
  15. * Ian Craggs - async client updates
  16. * Ian Craggs - fix for bug 484496
  17. * Juergen Kosel, Ian Craggs - fix for issue #135
  18. * Ian Craggs - issue #217
  19. * Ian Craggs - fix for issue #186
  20. * Ian Craggs - remove StackTrace print debugging calls
  21. *******************************************************************************/
  22. /**
  23. * @file
  24. * \brief Socket related functions
  25. *
  26. * Some other related functions are in the SocketBuffer module
  27. */
  28. #include "Socket.h"
  29. #include "Log.h"
  30. #include "SocketBuffer.h"
  31. #include "Messages.h"
  32. #include "StackTrace.h"
  33. #if defined(OPENSSL)
  34. #include "SSLSocket.h"
  35. #endif
  36. #include <stdlib.h>
  37. #include <string.h>
  38. #include <signal.h>
  39. #include <ctype.h>
  40. #include "Heap.h"
  41. #if defined(USE_SELECT)
  42. int isReady(int socket, fd_set* read_set, fd_set* write_set);
  43. int Socket_continueWrites(fd_set* pwset, SOCKET* socket, mutex_type mutex);
  44. #else
  45. int isReady(int index);
  46. int Socket_continueWrites(SOCKET* socket, mutex_type mutex);
  47. #endif
  48. int Socket_setnonblocking(SOCKET sock);
  49. int Socket_error(char* aString, SOCKET sock);
  50. int Socket_addSocket(SOCKET newSd);
  51. int Socket_writev(SOCKET socket, iobuf* iovecs, int count, unsigned long* bytes);
  52. int Socket_close_only(SOCKET socket);
  53. int Socket_continueWrite(SOCKET socket);
  54. char* Socket_getaddrname(struct sockaddr* sa, SOCKET sock);
  55. int Socket_abortWrite(SOCKET socket);
  56. #if defined(_WIN32) || defined(_WIN64)
  57. #define iov_len len
  58. #define iov_base buf
  59. #define snprintf _snprintf
  60. #endif
  61. /**
  62. * Structure to hold all socket data for this module
  63. */
  64. Sockets mod_s;
  65. #if defined(USE_SELECT)
  66. static fd_set wset;
  67. #endif
  68. extern mutex_type socket_mutex;
  69. /**
  70. * Set a socket non-blocking, OS independently
  71. * @param sock the socket to set non-blocking
  72. * @return TCP call error code
  73. */
  74. int Socket_setnonblocking(SOCKET sock)
  75. {
  76. int rc;
  77. #if defined(_WIN32) || defined(_WIN64)
  78. u_long flag = 1L;
  79. FUNC_ENTRY;
  80. rc = ioctl(sock, FIONBIO, &flag);
  81. #else
  82. int flags;
  83. FUNC_ENTRY;
  84. if ((flags = fcntl(sock, F_GETFL, 0)))
  85. flags = 0;
  86. rc = fcntl(sock, F_SETFL, flags | O_NONBLOCK);
  87. #endif
  88. FUNC_EXIT_RC(rc);
  89. return rc;
  90. }
  91. /**
  92. * Gets the specific error corresponding to SOCKET_ERROR
  93. * @param aString the function that was being used when the error occurred
  94. * @param sock the socket on which the error occurred
  95. * @return the specific TCP error code
  96. */
  97. int Socket_error(char* aString, SOCKET sock)
  98. {
  99. int err;
  100. #if defined(_WIN32) || defined(_WIN64)
  101. err = WSAGetLastError();
  102. #else
  103. err = errno;
  104. #endif
  105. if (err != EINTR && err != EAGAIN && err != EINPROGRESS && err != EWOULDBLOCK)
  106. {
  107. if (strcmp(aString, "shutdown") != 0 || (err != ENOTCONN && err != ECONNRESET))
  108. Log(TRACE_MINIMUM, -1, "Socket error %s(%d) in %s for socket %d", strerror(err), err, aString, sock);
  109. }
  110. return err;
  111. }
  112. /**
  113. * Initialize the socket module
  114. */
  115. void Socket_outInitialize(void)
  116. {
  117. #if defined(_WIN32) || defined(_WIN64)
  118. WORD winsockVer = 0x0202;
  119. WSADATA wsd;
  120. FUNC_ENTRY;
  121. WSAStartup(winsockVer, &wsd);
  122. #else
  123. FUNC_ENTRY;
  124. signal(SIGPIPE, SIG_IGN);
  125. #endif
  126. SocketBuffer_initialize();
  127. mod_s.connect_pending = ListInitialize();
  128. mod_s.write_pending = ListInitialize();
  129. #if defined(USE_SELECT)
  130. mod_s.clientsds = ListInitialize();
  131. mod_s.cur_clientsds = NULL;
  132. FD_ZERO(&(mod_s.rset)); /* Initialize the descriptor set */
  133. FD_ZERO(&(mod_s.pending_wset));
  134. mod_s.maxfdp1 = 0;
  135. memcpy((void*)&(mod_s.rset_saved), (void*)&(mod_s.rset), sizeof(mod_s.rset_saved));
  136. #else
  137. mod_s.nfds = 0;
  138. mod_s.fds_read = NULL;
  139. mod_s.fds_write = NULL;
  140. mod_s.saved.cur_fd = -1;
  141. mod_s.saved.fds_write = NULL;
  142. mod_s.saved.fds_read = NULL;
  143. mod_s.saved.nfds = 0;
  144. #endif
  145. FUNC_EXIT;
  146. }
  147. /**
  148. * Terminate the socket module
  149. */
  150. void Socket_outTerminate(void)
  151. {
  152. FUNC_ENTRY;
  153. ListFree(mod_s.connect_pending);
  154. ListFree(mod_s.write_pending);
  155. #if defined(USE_SELECT)
  156. ListFree(mod_s.clientsds);
  157. #else
  158. if (mod_s.fds_read)
  159. free(mod_s.fds_read);
  160. if (mod_s.fds_write)
  161. free(mod_s.fds_write);
  162. if (mod_s.saved.fds_write)
  163. free(mod_s.saved.fds_write);
  164. if (mod_s.saved.fds_read)
  165. free(mod_s.saved.fds_read);
  166. #endif
  167. SocketBuffer_terminate();
  168. #if defined(_WIN32) || defined(_WIN64)
  169. WSACleanup();
  170. #endif
  171. FUNC_EXIT;
  172. }
  173. #if defined(USE_SELECT)
  174. /**
  175. * Add a socket to the list of socket to check with select
  176. * @param newSd the new socket to add
  177. */
  178. int Socket_addSocket(SOCKET newSd)
  179. {
  180. int rc = 0;
  181. FUNC_ENTRY;
  182. if (ListFindItem(mod_s.clientsds, &newSd, intcompare) == NULL) /* make sure we don't add the same socket twice */
  183. {
  184. if (mod_s.clientsds->count >= FD_SETSIZE)
  185. {
  186. Log(LOG_ERROR, -1, "addSocket: exceeded FD_SETSIZE %d", FD_SETSIZE);
  187. rc = SOCKET_ERROR;
  188. }
  189. else
  190. {
  191. SOCKET* pnewSd = (SOCKET*)malloc(sizeof(newSd));
  192. if (!pnewSd)
  193. {
  194. rc = PAHO_MEMORY_ERROR;
  195. goto exit;
  196. }
  197. *pnewSd = newSd;
  198. if (!ListAppend(mod_s.clientsds, pnewSd, sizeof(newSd)))
  199. {
  200. free(pnewSd);
  201. rc = PAHO_MEMORY_ERROR;
  202. goto exit;
  203. }
  204. FD_SET(newSd, &(mod_s.rset_saved));
  205. mod_s.maxfdp1 = max(mod_s.maxfdp1, (int)newSd + 1);
  206. rc = Socket_setnonblocking(newSd);
  207. if (rc == SOCKET_ERROR)
  208. Log(LOG_ERROR, -1, "addSocket: setnonblocking");
  209. }
  210. }
  211. else
  212. Log(LOG_ERROR, -1, "addSocket: socket %d already in the list", newSd);
  213. exit:
  214. FUNC_EXIT_RC(rc);
  215. return rc;
  216. }
  217. #else
  218. static int cmpfds(const void *p1, const void *p2)
  219. {
  220. SOCKET key1 = ((struct pollfd*)p1)->fd;
  221. SOCKET key2 = ((struct pollfd*)p2)->fd;
  222. return (key1 == key2) ? 0 : ((key1 < key2) ? -1 : 1);
  223. }
  224. static int cmpsockfds(const void *p1, const void *p2)
  225. {
  226. int key1 = *(int*)p1;
  227. SOCKET key2 = ((struct pollfd*)p2)->fd;
  228. return (key1 == key2) ? 0 : ((key1 < key2) ? -1 : 1);
  229. }
  230. /**
  231. * Add a socket to the list of socket to check with select
  232. * @param newSd the new socket to add
  233. */
  234. int Socket_addSocket(SOCKET newSd)
  235. {
  236. int rc = 0;
  237. FUNC_ENTRY;
  238. Thread_lock_mutex(socket_mutex);
  239. mod_s.nfds++;
  240. if (mod_s.fds_read)
  241. mod_s.fds_read = realloc(mod_s.fds_read, mod_s.nfds * sizeof(mod_s.fds_read[0]));
  242. else
  243. mod_s.fds_read = malloc(mod_s.nfds * sizeof(mod_s.fds_read[0]));
  244. if (!mod_s.fds_read)
  245. {
  246. rc = PAHO_MEMORY_ERROR;
  247. goto exit;
  248. }
  249. if (mod_s.fds_write)
  250. mod_s.fds_write = realloc(mod_s.fds_write, mod_s.nfds * sizeof(mod_s.fds_write[0]));
  251. else
  252. mod_s.fds_write = malloc(mod_s.nfds * sizeof(mod_s.fds_write[0]));
  253. if (!mod_s.fds_read)
  254. {
  255. rc = PAHO_MEMORY_ERROR;
  256. goto exit;
  257. }
  258. mod_s.fds_read[mod_s.nfds - 1].fd = newSd;
  259. mod_s.fds_write[mod_s.nfds - 1].fd = newSd;
  260. #if defined(_WIN32) || defined(_WIN64)
  261. mod_s.fds_read[mod_s.nfds - 1].events = POLLIN;
  262. mod_s.fds_write[mod_s.nfds - 1].events = POLLOUT;
  263. #else
  264. mod_s.fds_read[mod_s.nfds - 1].events = POLLIN | POLLNVAL;
  265. mod_s.fds_write[mod_s.nfds - 1].events = POLLOUT;
  266. #endif
  267. /* sort the poll fds array by socket number */
  268. qsort(mod_s.fds_read, (size_t)mod_s.nfds, sizeof(mod_s.fds_read[0]), cmpfds);
  269. qsort(mod_s.fds_write, (size_t)mod_s.nfds, sizeof(mod_s.fds_write[0]), cmpfds);
  270. rc = Socket_setnonblocking(newSd);
  271. if (rc == SOCKET_ERROR)
  272. Log(LOG_ERROR, -1, "addSocket: setnonblocking");
  273. exit:
  274. Thread_unlock_mutex(socket_mutex);
  275. FUNC_EXIT_RC(rc);
  276. return rc;
  277. }
  278. #endif
  279. #if defined(USE_SELECT)
  280. /**
  281. * Don't accept work from a client unless it is accepting work back, i.e. its socket is writeable
  282. * this seems like a reasonable form of flow control, and practically, seems to work.
  283. * @param socket the socket to check
  284. * @param read_set the socket read set (see select doc)
  285. * @param write_set the socket write set (see select doc)
  286. * @return boolean - is the socket ready to go?
  287. */
  288. int isReady(int socket, fd_set* read_set, fd_set* write_set)
  289. {
  290. int rc = 1;
  291. FUNC_ENTRY;
  292. if (ListFindItem(mod_s.connect_pending, &socket, intcompare) && FD_ISSET(socket, write_set))
  293. ListRemoveItem(mod_s.connect_pending, &socket, intcompare);
  294. else
  295. rc = FD_ISSET(socket, read_set) && FD_ISSET(socket, write_set) && Socket_noPendingWrites(socket);
  296. FUNC_EXIT_RC(rc);
  297. return rc;
  298. }
  299. #else
  300. /**
  301. * Don't accept work from a client unless it is accepting work back, i.e. its socket is writeable
  302. * this seems like a reasonable form of flow control, and practically, seems to work.
  303. * @param index the socket index to check
  304. * @return boolean - is the socket ready to go?
  305. */
  306. int isReady(int index)
  307. {
  308. int rc = 1;
  309. SOCKET* socket = &mod_s.saved.fds_write[index].fd;
  310. FUNC_ENTRY;
  311. if ((mod_s.saved.fds_read[index].revents & POLLHUP) || (mod_s.saved.fds_read[index].revents & POLLNVAL))
  312. ; /* signal work to be done if there is an error on the socket */
  313. else if (ListFindItem(mod_s.connect_pending, socket, intcompare) &&
  314. (mod_s.saved.fds_write[index].revents & POLLOUT))
  315. ListRemoveItem(mod_s.connect_pending, socket, intcompare);
  316. else
  317. rc = (mod_s.saved.fds_read[index].revents & POLLIN) &&
  318. (mod_s.saved.fds_write[index].revents & POLLOUT) &&
  319. Socket_noPendingWrites(*socket);
  320. FUNC_EXIT_RC(rc);
  321. return rc;
  322. }
  323. #endif
  324. #if defined(USE_SELECT)
  325. /**
  326. * Returns the next socket ready for communications as indicated by select
  327. * @param more_work flag to indicate more work is waiting, and thus a timeout value of 0 should
  328. * be used for the select
  329. * @param timeout the timeout to be used for the select, unless overridden
  330. * @param rc a value other than 0 indicates an error of the returned socket
  331. * @return the socket next ready, or 0 if none is ready
  332. */
  333. SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc)
  334. {
  335. SOCKET sock = 0;
  336. *rc = 0;
  337. int timeout_ms = 1000;
  338. FUNC_ENTRY;
  339. Thread_lock_mutex(mutex);
  340. if (mod_s.clientsds->count == 0)
  341. goto exit;
  342. if (more_work)
  343. timeout_ms = 0;
  344. else if (timeout >= 0)
  345. timeout_ms = timeout;
  346. while (mod_s.cur_clientsds != NULL)
  347. {
  348. if (isReady(*((int*)(mod_s.cur_clientsds->content)), &(mod_s.rset), &wset))
  349. break;
  350. ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds);
  351. }
  352. if (mod_s.cur_clientsds == NULL)
  353. {
  354. static struct timeval zero = {0L, 0L}; /* 0 seconds */
  355. int rc1, maxfdp1_saved;
  356. fd_set pwset;
  357. struct timeval timeout_tv = {0L, 0L};
  358. if (timeout_ms > 0L)
  359. {
  360. timeout_tv.tv_sec = timeout_ms / 1000;
  361. timeout_tv.tv_usec = (timeout_ms % 1000) * 1000; /* this field is microseconds! */
  362. }
  363. memcpy((void*)&(mod_s.rset), (void*)&(mod_s.rset_saved), sizeof(mod_s.rset));
  364. memcpy((void*)&(pwset), (void*)&(mod_s.pending_wset), sizeof(pwset));
  365. maxfdp1_saved = mod_s.maxfdp1;
  366. if (maxfdp1_saved == 0)
  367. {
  368. sock = 0;
  369. goto exit; /* no work to do */
  370. }
  371. /* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
  372. Thread_unlock_mutex(mutex);
  373. *rc = select(maxfdp1_saved, &(mod_s.rset), &pwset, NULL, &timeout_tv);
  374. Thread_lock_mutex(mutex);
  375. if (*rc == SOCKET_ERROR)
  376. {
  377. Socket_error("read select", 0);
  378. goto exit;
  379. }
  380. Log(TRACE_MAX, -1, "Return code %d from read select", *rc);
  381. if (Socket_continueWrites(&pwset, &sock, mutex) == SOCKET_ERROR)
  382. {
  383. *rc = SOCKET_ERROR;
  384. goto exit;
  385. }
  386. memcpy((void*)&wset, (void*)&(mod_s.rset_saved), sizeof(wset));
  387. if ((rc1 = select(mod_s.maxfdp1, NULL, &(wset), NULL, &zero)) == SOCKET_ERROR)
  388. {
  389. Socket_error("write select", 0);
  390. *rc = rc1;
  391. goto exit;
  392. }
  393. Log(TRACE_MAX, -1, "Return code %d from write select", rc1);
  394. if (*rc == 0 && rc1 == 0)
  395. {
  396. sock = 0;
  397. goto exit; /* no work to do */
  398. }
  399. mod_s.cur_clientsds = mod_s.clientsds->first;
  400. while (mod_s.cur_clientsds != NULL)
  401. {
  402. int cursock = *((int*)(mod_s.cur_clientsds->content));
  403. if (isReady(cursock, &(mod_s.rset), &wset))
  404. break;
  405. ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds);
  406. }
  407. }
  408. *rc = 0;
  409. if (mod_s.cur_clientsds == NULL)
  410. sock = 0;
  411. else
  412. {
  413. sock = *((int*)(mod_s.cur_clientsds->content));
  414. ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds);
  415. }
  416. exit:
  417. Thread_unlock_mutex(mutex);
  418. FUNC_EXIT_RC(sock);
  419. return sock;
  420. } /* end getReadySocket */
  421. #else
  422. /**
  423. * Returns the next socket ready for communications as indicated by select
  424. * @param more_work flag to indicate more work is waiting, and thus a timeout value of 0 should
  425. * be used for the select
  426. * @param timeout the timeout to be used in ms
  427. * @param rc a value other than 0 indicates an error of the returned socket
  428. * @return the socket next ready, or 0 if none is ready
  429. */
  430. SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc)
  431. {
  432. SOCKET sock = 0;
  433. *rc = 0;
  434. int timeout_ms = 1000;
  435. FUNC_ENTRY;
  436. Thread_lock_mutex(mutex);
  437. if (mod_s.nfds == 0 && mod_s.saved.nfds == 0)
  438. goto exit;
  439. if (more_work)
  440. timeout_ms = 0;
  441. else if (timeout >= 0)
  442. timeout_ms = timeout;
  443. while (mod_s.saved.cur_fd != -1)
  444. {
  445. if (isReady(mod_s.saved.cur_fd))
  446. break;
  447. mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
  448. }
  449. if (mod_s.saved.cur_fd == -1)
  450. {
  451. int rc1 = 0;
  452. if (mod_s.nfds != mod_s.saved.nfds)
  453. {
  454. mod_s.saved.nfds = mod_s.nfds;
  455. if (mod_s.saved.fds_read)
  456. mod_s.saved.fds_read = realloc(mod_s.saved.fds_read, mod_s.nfds * sizeof(struct pollfd));
  457. else
  458. mod_s.saved.fds_read = malloc(mod_s.nfds * sizeof(struct pollfd));
  459. if (mod_s.saved.fds_write)
  460. mod_s.saved.fds_write = realloc(mod_s.saved.fds_write, mod_s.nfds * sizeof(struct pollfd));
  461. else
  462. mod_s.saved.fds_write = malloc(mod_s.nfds * sizeof(struct pollfd));
  463. }
  464. memcpy(mod_s.saved.fds_read, mod_s.fds_read, mod_s.nfds * sizeof(struct pollfd));
  465. memcpy(mod_s.saved.fds_write, mod_s.fds_write, mod_s.nfds * sizeof(struct pollfd));
  466. if (mod_s.saved.nfds == 0)
  467. {
  468. sock = 0;
  469. goto exit; /* no work to do */
  470. }
  471. /* Check pending write set for writeable sockets */
  472. rc1 = poll(mod_s.saved.fds_write, mod_s.saved.nfds, 0);
  473. if (rc1 > 0 && Socket_continueWrites(&sock, mutex) == SOCKET_ERROR)
  474. {
  475. *rc = SOCKET_ERROR;
  476. goto exit;
  477. }
  478. /* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
  479. Thread_unlock_mutex(mutex);
  480. *rc = poll(mod_s.saved.fds_read, mod_s.saved.nfds, timeout_ms);
  481. Thread_lock_mutex(mutex);
  482. if (*rc == SOCKET_ERROR)
  483. {
  484. Socket_error("poll", 0);
  485. goto exit;
  486. }
  487. Log(TRACE_MAX, -1, "Return code %d from poll", *rc);
  488. if (rc1 == 0 && *rc == 0)
  489. {
  490. sock = 0;
  491. goto exit; /* no work to do */
  492. }
  493. mod_s.saved.cur_fd = 0;
  494. while (mod_s.saved.cur_fd != -1)
  495. {
  496. if (isReady(mod_s.saved.cur_fd))
  497. break;
  498. mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
  499. }
  500. }
  501. *rc = 0;
  502. if (mod_s.saved.cur_fd == -1)
  503. sock = 0;
  504. else
  505. {
  506. sock = mod_s.saved.fds_read[mod_s.saved.cur_fd].fd;
  507. mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
  508. }
  509. exit:
  510. Thread_unlock_mutex(mutex);
  511. FUNC_EXIT_RC(sock);
  512. return sock;
  513. } /* end getReadySocket */
  514. #endif
  515. /**
  516. * Reads one byte from a socket
  517. * @param socket the socket to read from
  518. * @param c the character read, returned
  519. * @return completion code
  520. */
  521. int Socket_getch(SOCKET socket, char* c)
  522. {
  523. int rc = SOCKET_ERROR;
  524. FUNC_ENTRY;
  525. if ((rc = SocketBuffer_getQueuedChar(socket, c)) != SOCKETBUFFER_INTERRUPTED)
  526. goto exit;
  527. if ((rc = recv(socket, c, (size_t)1, 0)) == SOCKET_ERROR)
  528. {
  529. int err = Socket_error("recv - getch", socket);
  530. if (err == EWOULDBLOCK || err == EAGAIN)
  531. {
  532. rc = TCPSOCKET_INTERRUPTED;
  533. SocketBuffer_interrupted(socket, 0);
  534. }
  535. }
  536. else if (rc == 0)
  537. rc = SOCKET_ERROR; /* The return value from recv is 0 when the peer has performed an orderly shutdown. */
  538. else if (rc == 1)
  539. {
  540. SocketBuffer_queueChar(socket, *c);
  541. rc = TCPSOCKET_COMPLETE;
  542. }
  543. exit:
  544. FUNC_EXIT_RC(rc);
  545. return rc;
  546. }
  547. /**
  548. * Attempts to read a number of bytes from a socket, non-blocking. If a previous read did not
  549. * finish, then retrieve that data.
  550. * @param socket the socket to read from
  551. * @param bytes the number of bytes to read
  552. * @param actual_len the actual number of bytes read
  553. * @return completion code
  554. */
  555. char *Socket_getdata(SOCKET socket, size_t bytes, size_t* actual_len, int *rc)
  556. {
  557. char* buf;
  558. FUNC_ENTRY;
  559. if (bytes == 0)
  560. {
  561. buf = SocketBuffer_complete(socket);
  562. goto exit;
  563. }
  564. buf = SocketBuffer_getQueuedData(socket, bytes, actual_len);
  565. if ((*rc = recv(socket, buf + (*actual_len), (int)(bytes - (*actual_len)), 0)) == SOCKET_ERROR)
  566. {
  567. *rc = Socket_error("recv - getdata", socket);
  568. if (*rc != EAGAIN && *rc != EWOULDBLOCK)
  569. {
  570. buf = NULL;
  571. goto exit;
  572. }
  573. }
  574. else if (*rc == 0) /* rc 0 means the other end closed the socket, albeit "gracefully" */
  575. {
  576. buf = NULL;
  577. goto exit;
  578. }
  579. else
  580. *actual_len += *rc;
  581. if (*actual_len == bytes)
  582. SocketBuffer_complete(socket);
  583. else /* we didn't read the whole packet */
  584. {
  585. SocketBuffer_interrupted(socket, *actual_len);
  586. Log(TRACE_MAX, -1, "%d bytes expected but %d bytes now received", (int)bytes, (int)*actual_len);
  587. }
  588. exit:
  589. FUNC_EXIT;
  590. return buf;
  591. }
  592. /**
  593. * Indicate whether any data is pending outbound for a socket.
  594. * @return boolean - true == no pending data.
  595. */
  596. int Socket_noPendingWrites(SOCKET socket)
  597. {
  598. SOCKET cursock = socket;
  599. return ListFindItem(mod_s.write_pending, &cursock, intcompare) == NULL;
  600. }
  601. /**
  602. * Attempts to write a series of iovec buffers to a socket in *one* system call so that
  603. * they are sent as one packet.
  604. * @param socket the socket to write to
  605. * @param iovecs an array of buffers to write
  606. * @param count number of buffers in iovecs
  607. * @param bytes number of bytes actually written returned
  608. * @return completion code, especially TCPSOCKET_INTERRUPTED
  609. */
  610. int Socket_writev(SOCKET socket, iobuf* iovecs, int count, unsigned long* bytes)
  611. {
  612. int rc;
  613. FUNC_ENTRY;
  614. *bytes = 0L;
  615. #if defined(_WIN32) || defined(_WIN64)
  616. rc = WSASend(socket, iovecs, count, (LPDWORD)bytes, 0, NULL, NULL);
  617. if (rc == SOCKET_ERROR)
  618. {
  619. int err = Socket_error("WSASend - putdatas", socket);
  620. if (err == EWOULDBLOCK || err == EAGAIN)
  621. rc = TCPSOCKET_INTERRUPTED;
  622. }
  623. #else
  624. /*#define TCPSOCKET_INTERRUPTED_TESTING
  625. This section forces the occasional return of TCPSOCKET_INTERRUPTED,
  626. for testing purposes only!
  627. */
  628. #if defined(TCPSOCKET_INTERRUPTED_TESTING)
  629. static int i = 0;
  630. if (++i >= 10 && i < 21)
  631. {
  632. if (1)
  633. {
  634. printf("Deliberately simulating TCPSOCKET_INTERRUPTED\n");
  635. rc = TCPSOCKET_INTERRUPTED; /* simulate a network wait */
  636. }
  637. else
  638. {
  639. printf("Deliberately simulating SOCKET_ERROR\n");
  640. rc = SOCKET_ERROR;
  641. }
  642. /* should *bytes always be 0? */
  643. if (i == 20)
  644. {
  645. printf("Shutdown socket\n");
  646. shutdown(socket, SHUT_WR);
  647. }
  648. }
  649. else
  650. {
  651. #endif
  652. rc = writev(socket, iovecs, count);
  653. if (rc == SOCKET_ERROR)
  654. {
  655. int err = Socket_error("writev - putdatas", socket);
  656. if (err == EWOULDBLOCK || err == EAGAIN)
  657. rc = TCPSOCKET_INTERRUPTED;
  658. }
  659. else
  660. *bytes = rc;
  661. #if defined(TCPSOCKET_INTERRUPTED_TESTING)
  662. }
  663. #endif
  664. #endif
  665. FUNC_EXIT_RC(rc);
  666. return rc;
  667. }
  668. /**
  669. * Attempts to write a series of buffers to a socket in *one* system call so that they are
  670. * sent as one packet.
  671. * @param socket the socket to write to
  672. * @param buf0 the first buffer
  673. * @param buf0len the length of data in the first buffer
  674. * @param count number of buffers
  675. * @param buffers an array of buffers to write
  676. * @param buflens an array of corresponding buffer lengths
  677. * @return completion code, especially TCPSOCKET_INTERRUPTED
  678. */
  679. int Socket_putdatas(SOCKET socket, char* buf0, size_t buf0len, PacketBuffers bufs)
  680. {
  681. unsigned long bytes = 0L;
  682. iobuf iovecs[5];
  683. int frees1[5];
  684. int rc = TCPSOCKET_INTERRUPTED, i;
  685. size_t total = buf0len;
  686. FUNC_ENTRY;
  687. if (!Socket_noPendingWrites(socket))
  688. {
  689. Log(LOG_SEVERE, -1, "Trying to write to socket %d for which there is already pending output", socket);
  690. rc = SOCKET_ERROR;
  691. goto exit;
  692. }
  693. for (i = 0; i < bufs.count; i++)
  694. total += bufs.buflens[i];
  695. iovecs[0].iov_base = buf0;
  696. iovecs[0].iov_len = (ULONG)buf0len;
  697. frees1[0] = 1; /* this buffer should be freed by SocketBuffer if the write is interrupted */
  698. for (i = 0; i < bufs.count; i++)
  699. {
  700. iovecs[i+1].iov_base = bufs.buffers[i];
  701. iovecs[i+1].iov_len = (ULONG)bufs.buflens[i];
  702. frees1[i+1] = bufs.frees[i];
  703. }
  704. if ((rc = Socket_writev(socket, iovecs, bufs.count+1, &bytes)) != SOCKET_ERROR)
  705. {
  706. if (bytes == total)
  707. rc = TCPSOCKET_COMPLETE;
  708. else
  709. {
  710. SOCKET* sockmem = (SOCKET*)malloc(sizeof(SOCKET));
  711. if (!sockmem)
  712. {
  713. rc = PAHO_MEMORY_ERROR;
  714. goto exit;
  715. }
  716. Log(TRACE_MIN, -1, "Partial write: %lu bytes of %lu actually written on socket %d",
  717. bytes, total, socket);
  718. #if defined(OPENSSL)
  719. SocketBuffer_pendingWrite(socket, NULL, bufs.count+1, iovecs, frees1, total, bytes);
  720. #else
  721. SocketBuffer_pendingWrite(socket, bufs.count+1, iovecs, frees1, total, bytes);
  722. #endif
  723. *sockmem = socket;
  724. if (!ListAppend(mod_s.write_pending, sockmem, sizeof(int)))
  725. {
  726. free(sockmem);
  727. rc = PAHO_MEMORY_ERROR;
  728. goto exit;
  729. }
  730. #if defined(USE_SELECT)
  731. FD_SET(socket, &(mod_s.pending_wset));
  732. #endif
  733. rc = TCPSOCKET_INTERRUPTED;
  734. }
  735. }
  736. exit:
  737. FUNC_EXIT_RC(rc);
  738. return rc;
  739. }
  740. /**
  741. * Add a socket to the pending write list, so that it is checked for writing in select. This is used
  742. * in connect processing when the TCP connect is incomplete, as we need to check the socket for both
  743. * ready to read and write states.
  744. * @param socket the socket to add
  745. */
  746. void Socket_addPendingWrite(SOCKET socket)
  747. {
  748. #if defined(USE_SELECT)
  749. FD_SET(socket, &(mod_s.pending_wset));
  750. #endif
  751. }
  752. /**
  753. * Clear a socket from the pending write list - if one was added with Socket_addPendingWrite
  754. * @param socket the socket to remove
  755. */
  756. void Socket_clearPendingWrite(SOCKET socket)
  757. {
  758. #if defined(USE_SELECT)
  759. if (FD_ISSET(socket, &(mod_s.pending_wset)))
  760. FD_CLR(socket, &(mod_s.pending_wset));
  761. #endif
  762. }
  763. /**
  764. * Close a socket without removing it from the select list.
  765. * @param socket the socket to close
  766. * @return completion code
  767. */
  768. int Socket_close_only(SOCKET socket)
  769. {
  770. int rc;
  771. FUNC_ENTRY;
  772. #if defined(_WIN32) || defined(_WIN64)
  773. if (shutdown(socket, SD_BOTH) == SOCKET_ERROR)
  774. Socket_error("shutdown", socket);
  775. if ((rc = closesocket(socket)) == SOCKET_ERROR)
  776. Socket_error("close", socket);
  777. #else
  778. if (shutdown(socket, SHUT_WR) == SOCKET_ERROR)
  779. Socket_error("shutdown", socket);
  780. if ((rc = recv(socket, NULL, (size_t)0, 0)) == SOCKET_ERROR)
  781. Socket_error("shutdown", socket);
  782. if ((rc = close(socket)) == SOCKET_ERROR)
  783. Socket_error("close", socket);
  784. #endif
  785. FUNC_EXIT_RC(rc);
  786. return rc;
  787. }
  788. #if defined(USE_SELECT)
  789. /**
  790. * Close a socket and remove it from the select list.
  791. * @param socket the socket to close
  792. * @return completion code
  793. */
  794. int Socket_close(SOCKET socket)
  795. {
  796. int rc = 0;
  797. FUNC_ENTRY;
  798. Socket_close_only(socket);
  799. FD_CLR(socket, &(mod_s.rset_saved));
  800. if (FD_ISSET(socket, &(mod_s.pending_wset)))
  801. FD_CLR(socket, &(mod_s.pending_wset));
  802. if (mod_s.cur_clientsds != NULL && *(int*)(mod_s.cur_clientsds->content) == socket)
  803. mod_s.cur_clientsds = mod_s.cur_clientsds->next;
  804. Socket_abortWrite(socket);
  805. SocketBuffer_cleanup(socket);
  806. ListRemoveItem(mod_s.connect_pending, &socket, intcompare);
  807. ListRemoveItem(mod_s.write_pending, &socket, intcompare);
  808. if (ListRemoveItem(mod_s.clientsds, &socket, intcompare))
  809. Log(TRACE_MIN, -1, "Removed socket %d", socket);
  810. else
  811. {
  812. Log(LOG_ERROR, -1, "Failed to remove socket %d", socket);
  813. rc = SOCKET_ERROR;
  814. goto exit;
  815. }
  816. if (socket + 1 >= mod_s.maxfdp1)
  817. {
  818. /* now we have to reset mod_s.maxfdp1 */
  819. ListElement* cur_clientsds = NULL;
  820. mod_s.maxfdp1 = 0;
  821. while (ListNextElement(mod_s.clientsds, &cur_clientsds))
  822. mod_s.maxfdp1 = max(*((int*)(cur_clientsds->content)), mod_s.maxfdp1);
  823. ++(mod_s.maxfdp1);
  824. Log(TRACE_MAX, -1, "Reset max fdp1 to %d", mod_s.maxfdp1);
  825. }
  826. exit:
  827. FUNC_EXIT_RC(rc);
  828. return rc;
  829. }
  830. #else
  831. /**
  832. * Close a socket and remove it from the select list.
  833. * @param socket the socket to close
  834. * @return completion code
  835. */
  836. int Socket_close(SOCKET socket)
  837. {
  838. struct pollfd* fd;
  839. int rc = 0;
  840. FUNC_ENTRY;
  841. Socket_close_only(socket);
  842. Socket_abortWrite(socket);
  843. SocketBuffer_cleanup(socket);
  844. ListRemoveItem(mod_s.connect_pending, &socket, intcompare);
  845. ListRemoveItem(mod_s.write_pending, &socket, intcompare);
  846. if (mod_s.nfds == 0)
  847. goto exit;
  848. fd = bsearch(&socket, mod_s.fds_read, (size_t)mod_s.nfds, sizeof(mod_s.fds_read[0]), cmpsockfds);
  849. if (fd)
  850. {
  851. struct pollfd* last_fd = &mod_s.fds_read[mod_s.nfds - 1];
  852. if (--mod_s.nfds == 0)
  853. {
  854. free(mod_s.fds_read);
  855. mod_s.fds_read = NULL;
  856. }
  857. else
  858. {
  859. if (fd != last_fd)
  860. {
  861. /* shift array to remove the socket in question */
  862. memmove(fd, fd + 1, (mod_s.nfds - (fd - mod_s.fds_read)) * sizeof(mod_s.fds_read[0]));
  863. }
  864. mod_s.fds_read = realloc(mod_s.fds_read, sizeof(mod_s.fds_read[0]) * mod_s.nfds);
  865. if (mod_s.fds_read == NULL)
  866. {
  867. rc = PAHO_MEMORY_ERROR;
  868. goto exit;
  869. }
  870. }
  871. Log(TRACE_MIN, -1, "Removed socket %d", socket);
  872. }
  873. else
  874. Log(LOG_ERROR, -1, "Failed to remove socket %d", socket);
  875. fd = bsearch(&socket, mod_s.fds_write, (size_t)(mod_s.nfds+1), sizeof(mod_s.fds_write[0]), cmpsockfds);
  876. if (fd)
  877. {
  878. struct pollfd* last_fd = &mod_s.fds_write[mod_s.nfds];
  879. if (mod_s.nfds == 0)
  880. {
  881. free(mod_s.fds_write);
  882. mod_s.fds_write = NULL;
  883. }
  884. else
  885. {
  886. if (fd != last_fd)
  887. {
  888. /* shift array to remove the socket in question */
  889. memmove(fd, fd + 1, (mod_s.nfds - (fd - mod_s.fds_write)) * sizeof(mod_s.fds_write[0]));
  890. }
  891. mod_s.fds_write = realloc(mod_s.fds_write, sizeof(mod_s.fds_write[0]) * mod_s.nfds);
  892. if (mod_s.fds_write == NULL)
  893. {
  894. rc = PAHO_MEMORY_ERROR;
  895. goto exit;
  896. }
  897. }
  898. Log(TRACE_MIN, -1, "Removed socket %d", socket);
  899. }
  900. else
  901. Log(LOG_ERROR, -1, "Failed to remove socket %d", socket);
  902. exit:
  903. FUNC_EXIT_RC(rc);
  904. return rc;
  905. }
  906. #endif
  907. /**
  908. * Create a new socket and TCP connect to an address/port
  909. * @param addr the address string
  910. * @param port the TCP port
  911. * @param sock returns the new socket
  912. * @param timeout the timeout in milliseconds
  913. * @return completion code 0=good, SOCKET_ERROR=fail
  914. */
  915. #if defined(__GNUC__) && defined(__linux__)
  916. int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock, long timeout)
  917. #else
  918. int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock)
  919. #endif
  920. {
  921. int type = SOCK_STREAM;
  922. char *addr_mem;
  923. struct sockaddr_in address;
  924. #if defined(AF_INET6)
  925. struct sockaddr_in6 address6;
  926. #endif
  927. int rc = SOCKET_ERROR;
  928. #if defined(_WIN32) || defined(_WIN64)
  929. short family;
  930. #else
  931. sa_family_t family = AF_INET;
  932. #endif
  933. struct addrinfo *result = NULL;
  934. struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL};
  935. FUNC_ENTRY;
  936. *sock = SOCKET_ERROR;
  937. memset(&address6, '\0', sizeof(address6));
  938. if (addr[0] == '[')
  939. {
  940. ++addr;
  941. --addr_len;
  942. }
  943. if ((addr_mem = malloc( addr_len + 1u )) == NULL)
  944. {
  945. rc = PAHO_MEMORY_ERROR;
  946. goto exit;
  947. }
  948. memcpy( addr_mem, addr, addr_len );
  949. addr_mem[addr_len] = '\0';
  950. #if 0 /*defined(__GNUC__) && defined(__linux__)*/
  951. /* Commented out because the CI tests get intermittent ECONNABORTED return values
  952. * and I don't know why yet.
  953. */
  954. /* set getaddrinfo timeout if available */
  955. struct gaicb ar = {addr_mem, NULL, &hints, NULL};
  956. struct gaicb *reqs[] = {&ar};
  957. unsigned long int seconds = timeout / 1000L;
  958. unsigned long int nanos = (timeout - (seconds * 1000L)) * 1000000L;
  959. struct timespec timeoutspec = {seconds, nanos};
  960. rc = getaddrinfo_a(GAI_NOWAIT, reqs, 1, NULL);
  961. if (rc == 0)
  962. rc = gai_suspend((const struct gaicb* const *) reqs, 1, &timeoutspec);
  963. if (rc == 0)
  964. {
  965. rc = gai_error(reqs[0]);
  966. result = ar.ar_result;
  967. }
  968. #else
  969. rc = getaddrinfo(addr_mem, NULL, &hints, &result);
  970. #endif
  971. if (rc == 0)
  972. {
  973. struct addrinfo* res = result;
  974. while (res)
  975. { /* prefer ip4 addresses */
  976. if (res->ai_family == AF_INET || res->ai_next == NULL)
  977. break;
  978. res = res->ai_next;
  979. }
  980. if (res == NULL)
  981. rc = SOCKET_ERROR;
  982. else
  983. #if defined(AF_INET6)
  984. if (res->ai_family == AF_INET6)
  985. {
  986. address6.sin6_port = htons(port);
  987. address6.sin6_family = family = AF_INET6;
  988. memcpy(&address6.sin6_addr, &((struct sockaddr_in6*)(res->ai_addr))->sin6_addr, sizeof(address6.sin6_addr));
  989. }
  990. else
  991. #endif
  992. if (res->ai_family == AF_INET)
  993. {
  994. memset(&address.sin_zero, 0, sizeof(address.sin_zero));
  995. address.sin_port = htons(port);
  996. address.sin_family = family = AF_INET;
  997. address.sin_addr = ((struct sockaddr_in*)(res->ai_addr))->sin_addr;
  998. }
  999. else
  1000. rc = SOCKET_ERROR;
  1001. freeaddrinfo(result);
  1002. }
  1003. else
  1004. {
  1005. Log(LOG_ERROR, -1, "getaddrinfo failed for addr %s with rc %d", addr_mem, rc);
  1006. rc = SOCKET_ERROR;
  1007. }
  1008. if (rc != 0)
  1009. Log(LOG_ERROR, -1, "%s is not a valid IP address", addr_mem);
  1010. else
  1011. {
  1012. *sock = socket(family, type, 0);
  1013. if (*sock == INVALID_SOCKET)
  1014. rc = Socket_error("socket", *sock);
  1015. else
  1016. {
  1017. #if defined(NOSIGPIPE)
  1018. int opt = 1;
  1019. if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0)
  1020. Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock);
  1021. #endif
  1022. /*#define SMALL_TCP_BUFFER_TESTING
  1023. This section sets the TCP send buffer to a small amount to provoke TCPSOCKET_INTERRUPTED
  1024. return codes from send, for testing only!
  1025. */
  1026. #if defined(SMALL_TCP_BUFFER_TESTING)
  1027. if (1)
  1028. {
  1029. int optsend = 100; //2 * 1440;
  1030. printf("Setting optsend to %d\n", optsend);
  1031. if (setsockopt(*sock, SOL_SOCKET, SO_SNDBUF, (void*)&optsend, sizeof(optsend)) != 0)
  1032. Log(LOG_ERROR, -1, "Could not set SO_SNDBUF for socket %d", *sock);
  1033. }
  1034. #endif
  1035. Log(TRACE_MIN, -1, "New socket %d for %s, port %d", *sock, addr, port);
  1036. if (Socket_addSocket(*sock) == SOCKET_ERROR)
  1037. rc = Socket_error("addSocket", *sock);
  1038. else
  1039. {
  1040. /* this could complete immediately, even though we are non-blocking */
  1041. if (family == AF_INET)
  1042. rc = connect(*sock, (struct sockaddr*)&address, sizeof(address));
  1043. #if defined(AF_INET6)
  1044. else
  1045. rc = connect(*sock, (struct sockaddr*)&address6, sizeof(address6));
  1046. #endif
  1047. if (rc == SOCKET_ERROR)
  1048. rc = Socket_error("connect", *sock);
  1049. if (rc == EINPROGRESS || rc == EWOULDBLOCK)
  1050. {
  1051. SOCKET* pnewSd = (SOCKET*)malloc(sizeof(SOCKET));
  1052. ListElement* result = NULL;
  1053. if (!pnewSd)
  1054. {
  1055. rc = PAHO_MEMORY_ERROR;
  1056. goto exit;
  1057. }
  1058. *pnewSd = *sock;
  1059. Thread_lock_mutex(socket_mutex);
  1060. result = ListAppend(mod_s.connect_pending, pnewSd, sizeof(SOCKET));
  1061. Thread_unlock_mutex(socket_mutex);
  1062. if (!result)
  1063. {
  1064. free(pnewSd);
  1065. rc = PAHO_MEMORY_ERROR;
  1066. goto exit;
  1067. }
  1068. Log(TRACE_MIN, 15, "Connect pending");
  1069. }
  1070. }
  1071. /* Prevent socket leak by closing unusable sockets,
  1072. as reported in https://github.com/eclipse/paho.mqtt.c/issues/135 */
  1073. if (rc != 0 && (rc != EINPROGRESS) && (rc != EWOULDBLOCK))
  1074. {
  1075. Thread_lock_mutex(socket_mutex);
  1076. Socket_close(*sock); /* close socket and remove from our list of sockets */
  1077. Thread_unlock_mutex(socket_mutex);
  1078. *sock = SOCKET_ERROR; /* as initialized before */
  1079. }
  1080. }
  1081. }
  1082. exit:
  1083. if (addr_mem)
  1084. free(addr_mem);
  1085. FUNC_EXIT_RC(rc);
  1086. return rc;
  1087. }
  1088. static Socket_writeContinue* writecontinue = NULL;
  1089. void Socket_setWriteContinueCallback(Socket_writeContinue* mywritecontinue)
  1090. {
  1091. writecontinue = mywritecontinue;
  1092. }
  1093. static Socket_writeComplete* writecomplete = NULL;
  1094. void Socket_setWriteCompleteCallback(Socket_writeComplete* mywritecomplete)
  1095. {
  1096. writecomplete = mywritecomplete;
  1097. }
  1098. static Socket_writeAvailable* writeAvailable = NULL;
  1099. void Socket_setWriteAvailableCallback(Socket_writeAvailable* mywriteavailable)
  1100. {
  1101. writeAvailable = mywriteavailable;
  1102. }
  1103. /**
  1104. * Continue an outstanding write for a particular socket
  1105. * @param socket that socket
  1106. * @return completion code: 0=incomplete, 1=complete, -1=socket error
  1107. */
  1108. int Socket_continueWrite(SOCKET socket)
  1109. {
  1110. int rc = 0;
  1111. pending_writes* pw;
  1112. unsigned long curbuflen = 0L, /* cumulative total of buffer lengths */
  1113. bytes = 0L;
  1114. int curbuf = -1, i;
  1115. iobuf iovecs1[5];
  1116. FUNC_ENTRY;
  1117. pw = SocketBuffer_getWrite(socket);
  1118. #if defined(OPENSSL)
  1119. if (pw->ssl)
  1120. {
  1121. rc = SSLSocket_continueWrite(pw);
  1122. goto exit;
  1123. }
  1124. #endif
  1125. for (i = 0; i < pw->count; ++i)
  1126. {
  1127. if (pw->bytes <= curbuflen)
  1128. { /* if previously written length is less than the buffer we are currently looking at,
  1129. add the whole buffer */
  1130. iovecs1[++curbuf].iov_len = pw->iovecs[i].iov_len;
  1131. iovecs1[curbuf].iov_base = pw->iovecs[i].iov_base;
  1132. }
  1133. else if (pw->bytes < curbuflen + pw->iovecs[i].iov_len)
  1134. { /* if previously written length is in the middle of the buffer we are currently looking at,
  1135. add some of the buffer */
  1136. size_t offset = pw->bytes - curbuflen;
  1137. iovecs1[++curbuf].iov_len = pw->iovecs[i].iov_len - (ULONG)offset;
  1138. iovecs1[curbuf].iov_base = (char*)pw->iovecs[i].iov_base + offset;
  1139. }
  1140. curbuflen += pw->iovecs[i].iov_len;
  1141. }
  1142. if ((rc = Socket_writev(socket, iovecs1, curbuf+1, &bytes)) != SOCKET_ERROR)
  1143. {
  1144. pw->bytes += bytes;
  1145. if ((rc = (pw->bytes == pw->total)))
  1146. { /* topic and payload buffers are freed elsewhere, when all references to them have been removed */
  1147. for (i = 0; i < pw->count; i++)
  1148. {
  1149. if (pw->frees[i])
  1150. {
  1151. free(pw->iovecs[i].iov_base);
  1152. pw->iovecs[i].iov_base = NULL;
  1153. }
  1154. }
  1155. rc = 1; /* signal complete */
  1156. Log(TRACE_MIN, -1, "ContinueWrite: partial write now complete for socket %d", socket);
  1157. }
  1158. else
  1159. {
  1160. rc = 0; /* signal not complete */
  1161. Log(TRACE_MIN, -1, "ContinueWrite wrote +%lu bytes on socket %d", bytes, socket);
  1162. }
  1163. }
  1164. else /* if we got SOCKET_ERROR we need to clean up anyway - a partial write is no good anymore */
  1165. {
  1166. for (i = 0; i < pw->count; i++)
  1167. {
  1168. if (pw->frees[i])
  1169. {
  1170. free(pw->iovecs[i].iov_base);
  1171. pw->iovecs[i].iov_base = NULL;
  1172. }
  1173. }
  1174. }
  1175. #if defined(OPENSSL)
  1176. exit:
  1177. #endif
  1178. FUNC_EXIT_RC(rc);
  1179. return rc;
  1180. }
  1181. /**
  1182. * Continue an outstanding write for a particular socket
  1183. * @param socket that socket
  1184. * @return completion code: 0=incomplete, 1=complete, -1=socket error
  1185. */
  1186. int Socket_abortWrite(SOCKET socket)
  1187. {
  1188. int i = -1, rc = 0;
  1189. pending_writes* pw;
  1190. FUNC_ENTRY;
  1191. if ((pw = SocketBuffer_getWrite(socket)) == NULL)
  1192. goto exit;
  1193. #if defined(OPENSSL)
  1194. if (pw->ssl)
  1195. {
  1196. rc = SSLSocket_abortWrite(pw);
  1197. goto exit;
  1198. }
  1199. #endif
  1200. for (i = 0; i < pw->count; i++)
  1201. {
  1202. if (pw->frees[i])
  1203. {
  1204. Log(TRACE_MIN, -1, "Cleaning in abortWrite for socket %d", socket);
  1205. free(pw->iovecs[i].iov_base);
  1206. }
  1207. }
  1208. exit:
  1209. FUNC_EXIT_RC(rc);
  1210. return rc;
  1211. }
  1212. #if defined(USE_SELECT)
  1213. /**
  1214. * Continue any outstanding writes for a socket set
  1215. * @param pwset the set of sockets
  1216. * @param sock in case of a socket error contains the affected socket
  1217. * @return completion code, 0 or SOCKET_ERROR
  1218. */
  1219. int Socket_continueWrites(fd_set* pwset, SOCKET* sock, mutex_type mutex)
  1220. #else
  1221. /**
  1222. * Continue any outstanding socket writes
  1223. * @param sock in case of a socket error contains the affected socket
  1224. * @return completion code, 0 or SOCKET_ERROR
  1225. */
  1226. int Socket_continueWrites(SOCKET* sock, mutex_type mutex)
  1227. #endif
  1228. {
  1229. int rc1 = 0;
  1230. ListElement* curpending = mod_s.write_pending->first;
  1231. FUNC_ENTRY;
  1232. while (curpending && curpending->content)
  1233. {
  1234. int socket = *(int*)(curpending->content);
  1235. int rc = 0;
  1236. #if defined(USE_SELECT)
  1237. if (FD_ISSET(socket, pwset) && ((rc = Socket_continueWrite(socket)) != 0))
  1238. #else
  1239. struct pollfd* fd;
  1240. /* find the socket in the fds structure */
  1241. fd = bsearch(&socket, mod_s.saved.fds_write, (size_t)mod_s.saved.nfds, sizeof(mod_s.saved.fds_write[0]), cmpsockfds);
  1242. if ((fd->revents & POLLOUT) && ((rc = Socket_continueWrite(socket)) != 0))
  1243. #endif
  1244. {
  1245. if (!SocketBuffer_writeComplete(socket))
  1246. Log(LOG_SEVERE, -1, "Failed to remove pending write from socket buffer list");
  1247. #if defined(USE_SELECT)
  1248. FD_CLR(socket, &(mod_s.pending_wset));
  1249. #endif
  1250. if (!ListRemove(mod_s.write_pending, curpending->content))
  1251. {
  1252. Log(LOG_SEVERE, -1, "Failed to remove pending write from list");
  1253. ListNextElement(mod_s.write_pending, &curpending);
  1254. }
  1255. curpending = mod_s.write_pending->current;
  1256. if (writeAvailable && rc > 0)
  1257. (*writeAvailable)(socket);
  1258. if (writecomplete)
  1259. {
  1260. Thread_unlock_mutex(mutex);
  1261. (*writecomplete)(socket, rc);
  1262. Thread_lock_mutex(mutex);
  1263. }
  1264. }
  1265. else
  1266. ListNextElement(mod_s.write_pending, &curpending);
  1267. if (writecontinue && rc == 0)
  1268. (*writecontinue)(socket);
  1269. if (rc == SOCKET_ERROR)
  1270. {
  1271. *sock = socket;
  1272. rc1 = SOCKET_ERROR;
  1273. }
  1274. }
  1275. FUNC_EXIT_RC(rc1);
  1276. return rc1;
  1277. }
  1278. /**
  1279. * Convert a numeric address to character string
  1280. * @param sa socket numerical address
  1281. * @param sock socket
  1282. * @return the peer information
  1283. */
  1284. char* Socket_getaddrname(struct sockaddr* sa, SOCKET sock)
  1285. {
  1286. /**
  1287. * maximum length of the address string
  1288. */
  1289. #define ADDRLEN INET6_ADDRSTRLEN+1
  1290. /**
  1291. * maximum length of the port string
  1292. */
  1293. #define PORTLEN 10
  1294. static char addr_string[ADDRLEN + PORTLEN];
  1295. #if defined(_WIN32) || defined(_WIN64)
  1296. int buflen = ADDRLEN*2;
  1297. wchar_t buf[ADDRLEN*2];
  1298. if (WSAAddressToStringW(sa, sizeof(struct sockaddr_in6), NULL, buf, (LPDWORD)&buflen) == SOCKET_ERROR)
  1299. Socket_error("WSAAddressToString", sock);
  1300. else
  1301. wcstombs(addr_string, buf, sizeof(addr_string));
  1302. /* TODO: append the port information - format: [00:00:00::]:port */
  1303. /* strcpy(&addr_string[strlen(addr_string)], "what?"); */
  1304. #else
  1305. struct sockaddr_in *sin = (struct sockaddr_in *)sa;
  1306. size_t buflen = sizeof(addr_string) - strlen(addr_string);
  1307. inet_ntop(sin->sin_family, &sin->sin_addr, addr_string, ADDRLEN);
  1308. if (snprintf(&addr_string[strlen(addr_string)], buflen, ":%d", ntohs(sin->sin_port)) >= buflen)
  1309. addr_string[sizeof(addr_string)-1] = '\0'; /* just in case of snprintf buffer filling */
  1310. #endif
  1311. return addr_string;
  1312. }
  1313. /**
  1314. * Get information about the other end connected to a socket
  1315. * @param sock the socket to inquire on
  1316. * @return the peer information
  1317. */
  1318. char* Socket_getpeer(SOCKET sock)
  1319. {
  1320. struct sockaddr_in6 sa;
  1321. socklen_t sal = sizeof(sa);
  1322. if (getpeername(sock, (struct sockaddr*)&sa, &sal) == SOCKET_ERROR)
  1323. {
  1324. Socket_error("getpeername", sock);
  1325. return "unknown";
  1326. }
  1327. return Socket_getaddrname((struct sockaddr*)&sa, sock);
  1328. }
  1329. #if defined(Socket_TEST)
  1330. int main(int argc, char *argv[])
  1331. {
  1332. Socket_connect("127.0.0.1", 1883);
  1333. Socket_connect("localhost", 1883);
  1334. Socket_connect("loadsadsacalhost", 1883);
  1335. }
  1336. #endif