| /******************************************************************************* |
| * Copyright (c) 2009, 2020 IBM Corp. and others |
| * |
| * All rights reserved. This program and the accompanying materials |
| * are made available under the terms of the Eclipse Public License v2.0 |
| * and Eclipse Distribution License v1.0 which accompany this distribution. |
| * |
| * The Eclipse Public License is available at |
| * https://www.eclipse.org/legal/epl-2.0/ |
| * and the Eclipse Distribution License is available at |
| * http://www.eclipse.org/org/documents/edl-v10.php. |
| * |
| * Contributors: |
| * Ian Craggs - initial implementation and documentation |
| * Ian Craggs - async client updates |
| * Ian Craggs - fix for bug 484496 |
| * Juergen Kosel, Ian Craggs - fix for issue #135 |
| * Ian Craggs - issue #217 |
| * Ian Craggs - fix for issue #186 |
| * Ian Craggs - remove StackTrace print debugging calls |
| *******************************************************************************/ |
| |
| /** |
| * @file |
| * \brief Socket related functions |
| * |
| * Some other related functions are in the SocketBuffer module |
| */ |
| |
| |
| #include "Socket.h" |
| #include "Log.h" |
| #include "SocketBuffer.h" |
| #include "Messages.h" |
| #include "StackTrace.h" |
| #if defined(OPENSSL) |
| #include "SSLSocket.h" |
| #endif |
| |
| #include <stdlib.h> |
| #include <string.h> |
| #include <signal.h> |
| #include <ctype.h> |
| |
| #include "Heap.h" |
| |
| int Socket_setnonblocking(int sock); |
| int Socket_error(char* aString, int sock); |
| int Socket_addSocket(int newSd); |
| int isReady(int socket, fd_set* read_set, fd_set* write_set); |
| int Socket_writev(int socket, iobuf* iovecs, int count, unsigned long* bytes); |
| int Socket_close_only(int socket); |
| int Socket_continueWrite(int socket); |
| int Socket_continueWrites(fd_set* pwset); |
| char* Socket_getaddrname(struct sockaddr* sa, int sock); |
| int Socket_abortWrite(int socket); |
| |
| #if defined(_WIN32) || defined(_WIN64) |
| #define iov_len len |
| #define iov_base buf |
| #endif |
| |
| /** |
| * Structure to hold all socket data for this module |
| */ |
| Sockets mod_s; |
| static fd_set wset; |
| |
| /** |
| * Set a socket non-blocking, OS independently |
| * @param sock the socket to set non-blocking |
| * @return TCP call error code |
| */ |
| int Socket_setnonblocking(int sock) |
| { |
| int rc; |
| #if defined(_WIN32) || defined(_WIN64) |
| u_long flag = 1L; |
| |
| FUNC_ENTRY; |
| rc = ioctl(sock, FIONBIO, &flag); |
| #else |
| int flags; |
| |
| FUNC_ENTRY; |
| if ((flags = fcntl(sock, F_GETFL, 0))) |
| flags = 0; |
| rc = fcntl(sock, F_SETFL, flags | O_NONBLOCK); |
| #endif |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Gets the specific error corresponding to SOCKET_ERROR |
| * @param aString the function that was being used when the error occurred |
| * @param sock the socket on which the error occurred |
| * @return the specific TCP error code |
| */ |
| int Socket_error(char* aString, int sock) |
| { |
| int err; |
| |
| #if defined(_WIN32) || defined(_WIN64) |
| err = WSAGetLastError(); |
| #else |
| err = errno; |
| #endif |
| if (err != EINTR && err != EAGAIN && err != EINPROGRESS && err != EWOULDBLOCK) |
| { |
| if (strcmp(aString, "shutdown") != 0 || (err != ENOTCONN && err != ECONNRESET)) |
| Log(TRACE_MINIMUM, -1, "Socket error %s(%d) in %s for socket %d", strerror(err), err, aString, sock); |
| } |
| return err; |
| } |
| |
| |
| /** |
| * Initialize the socket module |
| */ |
| void Socket_outInitialize(void) |
| { |
| #if defined(_WIN32) || defined(_WIN64) |
| WORD winsockVer = 0x0202; |
| WSADATA wsd; |
| |
| FUNC_ENTRY; |
| WSAStartup(winsockVer, &wsd); |
| #else |
| FUNC_ENTRY; |
| signal(SIGPIPE, SIG_IGN); |
| #endif |
| |
| SocketBuffer_initialize(); |
| mod_s.clientsds = ListInitialize(); |
| mod_s.connect_pending = ListInitialize(); |
| mod_s.write_pending = ListInitialize(); |
| mod_s.cur_clientsds = NULL; |
| FD_ZERO(&(mod_s.rset)); /* Initialize the descriptor set */ |
| FD_ZERO(&(mod_s.pending_wset)); |
| mod_s.maxfdp1 = 0; |
| memcpy((void*)&(mod_s.rset_saved), (void*)&(mod_s.rset), sizeof(mod_s.rset_saved)); |
| FUNC_EXIT; |
| } |
| |
| |
| /** |
| * Terminate the socket module |
| */ |
| void Socket_outTerminate(void) |
| { |
| FUNC_ENTRY; |
| ListFree(mod_s.connect_pending); |
| ListFree(mod_s.write_pending); |
| ListFree(mod_s.clientsds); |
| SocketBuffer_terminate(); |
| #if defined(_WIN32) || defined(_WIN64) |
| WSACleanup(); |
| #endif |
| FUNC_EXIT; |
| } |
| |
| |
| /** |
| * Add a socket to the list of socket to check with select |
| * @param newSd the new socket to add |
| */ |
| int Socket_addSocket(int newSd) |
| { |
| int rc = 0; |
| |
| FUNC_ENTRY; |
| if (ListFindItem(mod_s.clientsds, &newSd, intcompare) == NULL) /* make sure we don't add the same socket twice */ |
| { |
| if (mod_s.clientsds->count >= FD_SETSIZE) |
| { |
| Log(LOG_ERROR, -1, "addSocket: exceeded FD_SETSIZE %d", FD_SETSIZE); |
| rc = SOCKET_ERROR; |
| } |
| else |
| { |
| int* pnewSd = (int*)malloc(sizeof(newSd)); |
| |
| if (!pnewSd) |
| { |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| *pnewSd = newSd; |
| if (!ListAppend(mod_s.clientsds, pnewSd, sizeof(newSd))) |
| { |
| free(pnewSd); |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| FD_SET(newSd, &(mod_s.rset_saved)); |
| mod_s.maxfdp1 = max(mod_s.maxfdp1, newSd + 1); |
| rc = Socket_setnonblocking(newSd); |
| if (rc == SOCKET_ERROR) |
| Log(LOG_ERROR, -1, "addSocket: setnonblocking"); |
| } |
| } |
| else |
| Log(LOG_ERROR, -1, "addSocket: socket %d already in the list", newSd); |
| |
| exit: |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Don't accept work from a client unless it is accepting work back, i.e. its socket is writeable |
| * this seems like a reasonable form of flow control, and practically, seems to work. |
| * @param socket the socket to check |
| * @param read_set the socket read set (see select doc) |
| * @param write_set the socket write set (see select doc) |
| * @return boolean - is the socket ready to go? |
| */ |
| int isReady(int socket, fd_set* read_set, fd_set* write_set) |
| { |
| int rc = 1; |
| |
| FUNC_ENTRY; |
| if (ListFindItem(mod_s.connect_pending, &socket, intcompare) && FD_ISSET(socket, write_set)) |
| ListRemoveItem(mod_s.connect_pending, &socket, intcompare); |
| else |
| rc = FD_ISSET(socket, read_set) && FD_ISSET(socket, write_set) && Socket_noPendingWrites(socket); |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Returns the next socket ready for communications as indicated by select |
| * @param more_work flag to indicate more work is waiting, and thus a timeout value of 0 should |
| * be used for the select |
| * @param tp the timeout to be used for the select, unless overridden |
| * @return the socket next ready, or 0 if none is ready |
| */ |
| int Socket_getReadySocket(int more_work, struct timeval *tp, mutex_type mutex) |
| { |
| int rc = 0; |
| static struct timeval zero = {0L, 0L}; /* 0 seconds */ |
| static struct timeval one = {1L, 0L}; /* 1 second */ |
| struct timeval timeout = one; |
| |
| FUNC_ENTRY; |
| Thread_lock_mutex(mutex); |
| if (mod_s.clientsds->count == 0) |
| goto exit; |
| |
| if (more_work) |
| timeout = zero; |
| else if (tp) |
| timeout = *tp; |
| |
| while (mod_s.cur_clientsds != NULL) |
| { |
| if (isReady(*((int*)(mod_s.cur_clientsds->content)), &(mod_s.rset), &wset)) |
| break; |
| ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds); |
| } |
| |
| if (mod_s.cur_clientsds == NULL) |
| { |
| int rc1; |
| fd_set pwset; |
| |
| memcpy((void*)&(mod_s.rset), (void*)&(mod_s.rset_saved), sizeof(mod_s.rset)); |
| memcpy((void*)&(pwset), (void*)&(mod_s.pending_wset), sizeof(pwset)); |
| /* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */ |
| Thread_unlock_mutex(mutex); |
| rc = select(mod_s.maxfdp1, &(mod_s.rset), &pwset, NULL, &timeout); |
| Thread_lock_mutex(mutex); |
| if (rc == SOCKET_ERROR) |
| { |
| Socket_error("read select", 0); |
| goto exit; |
| } |
| Log(TRACE_MAX, -1, "Return code %d from read select", rc); |
| |
| if (Socket_continueWrites(&pwset) == SOCKET_ERROR) |
| { |
| rc = 0; |
| goto exit; |
| } |
| |
| memcpy((void*)&wset, (void*)&(mod_s.rset_saved), sizeof(wset)); |
| if ((rc1 = select(mod_s.maxfdp1, NULL, &(wset), NULL, &zero)) == SOCKET_ERROR) |
| { |
| Socket_error("write select", 0); |
| rc = rc1; |
| goto exit; |
| } |
| Log(TRACE_MAX, -1, "Return code %d from write select", rc1); |
| |
| if (rc == 0 && rc1 == 0) |
| goto exit; /* no work to do */ |
| |
| mod_s.cur_clientsds = mod_s.clientsds->first; |
| while (mod_s.cur_clientsds != NULL) |
| { |
| int cursock = *((int*)(mod_s.cur_clientsds->content)); |
| if (isReady(cursock, &(mod_s.rset), &wset)) |
| break; |
| ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds); |
| } |
| } |
| |
| if (mod_s.cur_clientsds == NULL) |
| rc = 0; |
| else |
| { |
| rc = *((int*)(mod_s.cur_clientsds->content)); |
| ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds); |
| } |
| exit: |
| Thread_unlock_mutex(mutex); |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } /* end getReadySocket */ |
| |
| |
| /** |
| * Reads one byte from a socket |
| * @param socket the socket to read from |
| * @param c the character read, returned |
| * @return completion code |
| */ |
| int Socket_getch(int socket, char* c) |
| { |
| int rc = SOCKET_ERROR; |
| |
| FUNC_ENTRY; |
| if ((rc = SocketBuffer_getQueuedChar(socket, c)) != SOCKETBUFFER_INTERRUPTED) |
| goto exit; |
| |
| if ((rc = recv(socket, c, (size_t)1, 0)) == SOCKET_ERROR) |
| { |
| int err = Socket_error("recv - getch", socket); |
| if (err == EWOULDBLOCK || err == EAGAIN) |
| { |
| rc = TCPSOCKET_INTERRUPTED; |
| SocketBuffer_interrupted(socket, 0); |
| } |
| } |
| else if (rc == 0) |
| rc = SOCKET_ERROR; /* The return value from recv is 0 when the peer has performed an orderly shutdown. */ |
| else if (rc == 1) |
| { |
| SocketBuffer_queueChar(socket, *c); |
| rc = TCPSOCKET_COMPLETE; |
| } |
| exit: |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Attempts to read a number of bytes from a socket, non-blocking. If a previous read did not |
| * finish, then retrieve that data. |
| * @param socket the socket to read from |
| * @param bytes the number of bytes to read |
| * @param actual_len the actual number of bytes read |
| * @return completion code |
| */ |
| char *Socket_getdata(int socket, size_t bytes, size_t* actual_len, int *rc) |
| { |
| char* buf; |
| |
| FUNC_ENTRY; |
| if (bytes == 0) |
| { |
| buf = SocketBuffer_complete(socket); |
| goto exit; |
| } |
| |
| buf = SocketBuffer_getQueuedData(socket, bytes, actual_len); |
| |
| if ((*rc = recv(socket, buf + (*actual_len), (int)(bytes - (*actual_len)), 0)) == SOCKET_ERROR) |
| { |
| *rc = Socket_error("recv - getdata", socket); |
| if (*rc != EAGAIN && *rc != EWOULDBLOCK) |
| { |
| buf = NULL; |
| goto exit; |
| } |
| } |
| else if (*rc == 0) /* rc 0 means the other end closed the socket, albeit "gracefully" */ |
| { |
| buf = NULL; |
| goto exit; |
| } |
| else |
| *actual_len += *rc; |
| |
| if (*actual_len == bytes) |
| SocketBuffer_complete(socket); |
| else /* we didn't read the whole packet */ |
| { |
| SocketBuffer_interrupted(socket, *actual_len); |
| Log(TRACE_MAX, -1, "%d bytes expected but %d bytes now received", (int)bytes, (int)*actual_len); |
| } |
| exit: |
| FUNC_EXIT; |
| return buf; |
| } |
| |
| |
| /** |
| * Indicate whether any data is pending outbound for a socket. |
| * @return boolean - true == data pending. |
| */ |
| int Socket_noPendingWrites(int socket) |
| { |
| int cursock = socket; |
| return ListFindItem(mod_s.write_pending, &cursock, intcompare) == NULL; |
| } |
| |
| |
| /** |
| * Attempts to write a series of iovec buffers to a socket in *one* system call so that |
| * they are sent as one packet. |
| * @param socket the socket to write to |
| * @param iovecs an array of buffers to write |
| * @param count number of buffers in iovecs |
| * @param bytes number of bytes actually written returned |
| * @return completion code, especially TCPSOCKET_INTERRUPTED |
| */ |
| int Socket_writev(int socket, iobuf* iovecs, int count, unsigned long* bytes) |
| { |
| int rc; |
| |
| FUNC_ENTRY; |
| *bytes = 0L; |
| #if defined(_WIN32) || defined(_WIN64) |
| rc = WSASend(socket, iovecs, count, (LPDWORD)bytes, 0, NULL, NULL); |
| if (rc == SOCKET_ERROR) |
| { |
| int err = Socket_error("WSASend - putdatas", socket); |
| if (err == EWOULDBLOCK || err == EAGAIN) |
| rc = TCPSOCKET_INTERRUPTED; |
| } |
| #else |
| /*#define TCPSOCKET_INTERRUPTED_TESTING |
| This section forces the occasional return of TCPSOCKET_INTERRUPTED, |
| for testing purposes only! |
| */ |
| #if defined(TCPSOCKET_INTERRUPTED_TESTING) |
| static int i = 0; |
| if (++i >= 10 && i < 21) |
| { |
| if (1) |
| { |
| printf("Deliberately simulating TCPSOCKET_INTERRUPTED\n"); |
| rc = TCPSOCKET_INTERRUPTED; /* simulate a network wait */ |
| } |
| else |
| { |
| printf("Deliberately simulating SOCKET_ERROR\n"); |
| rc = SOCKET_ERROR; |
| } |
| /* should *bytes always be 0? */ |
| if (i == 20) |
| { |
| printf("Shutdown socket\n"); |
| shutdown(socket, SHUT_WR); |
| } |
| } |
| else |
| { |
| #endif |
| rc = writev(socket, iovecs, count); |
| if (rc == SOCKET_ERROR) |
| { |
| int err = Socket_error("writev - putdatas", socket); |
| if (err == EWOULDBLOCK || err == EAGAIN) |
| rc = TCPSOCKET_INTERRUPTED; |
| } |
| else |
| *bytes = rc; |
| #if defined(TCPSOCKET_INTERRUPTED_TESTING) |
| } |
| #endif |
| #endif |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Attempts to write a series of buffers to a socket in *one* system call so that they are |
| * sent as one packet. |
| * @param socket the socket to write to |
| * @param buf0 the first buffer |
| * @param buf0len the length of data in the first buffer |
| * @param count number of buffers |
| * @param buffers an array of buffers to write |
| * @param buflens an array of corresponding buffer lengths |
| * @return completion code, especially TCPSOCKET_INTERRUPTED |
| */ |
| int Socket_putdatas(int socket, char* buf0, size_t buf0len, PacketBuffers bufs) |
| { |
| unsigned long bytes = 0L; |
| iobuf iovecs[5]; |
| int frees1[5]; |
| int rc = TCPSOCKET_INTERRUPTED, i; |
| size_t total = buf0len; |
| |
| FUNC_ENTRY; |
| if (!Socket_noPendingWrites(socket)) |
| { |
| Log(LOG_SEVERE, -1, "Trying to write to socket %d for which there is already pending output", socket); |
| rc = SOCKET_ERROR; |
| goto exit; |
| } |
| |
| for (i = 0; i < bufs.count; i++) |
| total += bufs.buflens[i]; |
| |
| iovecs[0].iov_base = buf0; |
| iovecs[0].iov_len = (ULONG)buf0len; |
| frees1[0] = 1; /* this buffer should be freed by SocketBuffer if the write is interrupted */ |
| for (i = 0; i < bufs.count; i++) |
| { |
| iovecs[i+1].iov_base = bufs.buffers[i]; |
| iovecs[i+1].iov_len = (ULONG)bufs.buflens[i]; |
| frees1[i+1] = bufs.frees[i]; |
| } |
| |
| if ((rc = Socket_writev(socket, iovecs, bufs.count+1, &bytes)) != SOCKET_ERROR) |
| { |
| if (bytes == total) |
| rc = TCPSOCKET_COMPLETE; |
| else |
| { |
| int* sockmem = (int*)malloc(sizeof(int)); |
| |
| if (!sockmem) |
| { |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| Log(TRACE_MIN, -1, "Partial write: %lu bytes of %lu actually written on socket %d", |
| bytes, total, socket); |
| #if defined(OPENSSL) |
| SocketBuffer_pendingWrite(socket, NULL, bufs.count+1, iovecs, frees1, total, bytes); |
| #else |
| SocketBuffer_pendingWrite(socket, bufs.count+1, iovecs, frees1, total, bytes); |
| #endif |
| *sockmem = socket; |
| if (!ListAppend(mod_s.write_pending, sockmem, sizeof(int))) |
| { |
| free(sockmem); |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| FD_SET(socket, &(mod_s.pending_wset)); |
| rc = TCPSOCKET_INTERRUPTED; |
| } |
| } |
| exit: |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Add a socket to the pending write list, so that it is checked for writing in select. This is used |
| * in connect processing when the TCP connect is incomplete, as we need to check the socket for both |
| * ready to read and write states. |
| * @param socket the socket to add |
| */ |
| void Socket_addPendingWrite(int socket) |
| { |
| FD_SET(socket, &(mod_s.pending_wset)); |
| } |
| |
| |
| /** |
| * Clear a socket from the pending write list - if one was added with Socket_addPendingWrite |
| * @param socket the socket to remove |
| */ |
| void Socket_clearPendingWrite(int socket) |
| { |
| if (FD_ISSET(socket, &(mod_s.pending_wset))) |
| FD_CLR(socket, &(mod_s.pending_wset)); |
| } |
| |
| |
| /** |
| * Close a socket without removing it from the select list. |
| * @param socket the socket to close |
| * @return completion code |
| */ |
| int Socket_close_only(int socket) |
| { |
| int rc; |
| |
| FUNC_ENTRY; |
| #if defined(_WIN32) || defined(_WIN64) |
| if (shutdown(socket, SD_BOTH) == SOCKET_ERROR) |
| Socket_error("shutdown", socket); |
| if ((rc = closesocket(socket)) == SOCKET_ERROR) |
| Socket_error("close", socket); |
| #else |
| if (shutdown(socket, SHUT_WR) == SOCKET_ERROR) |
| Socket_error("shutdown", socket); |
| if ((rc = recv(socket, NULL, (size_t)0, 0)) == SOCKET_ERROR) |
| Socket_error("shutdown", socket); |
| if ((rc = close(socket)) == SOCKET_ERROR) |
| Socket_error("close", socket); |
| #endif |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Close a socket and remove it from the select list. |
| * @param socket the socket to close |
| * @return completion code |
| */ |
| void Socket_close(int socket) |
| { |
| FUNC_ENTRY; |
| Socket_close_only(socket); |
| FD_CLR(socket, &(mod_s.rset_saved)); |
| if (FD_ISSET(socket, &(mod_s.pending_wset))) |
| FD_CLR(socket, &(mod_s.pending_wset)); |
| if (mod_s.cur_clientsds != NULL && *(int*)(mod_s.cur_clientsds->content) == socket) |
| mod_s.cur_clientsds = mod_s.cur_clientsds->next; |
| Socket_abortWrite(socket); |
| SocketBuffer_cleanup(socket); |
| ListRemoveItem(mod_s.connect_pending, &socket, intcompare); |
| ListRemoveItem(mod_s.write_pending, &socket, intcompare); |
| |
| if (ListRemoveItem(mod_s.clientsds, &socket, intcompare)) |
| Log(TRACE_MIN, -1, "Removed socket %d", socket); |
| else |
| Log(LOG_ERROR, -1, "Failed to remove socket %d", socket); |
| if (socket + 1 >= mod_s.maxfdp1) |
| { |
| /* now we have to reset mod_s.maxfdp1 */ |
| ListElement* cur_clientsds = NULL; |
| |
| mod_s.maxfdp1 = 0; |
| while (ListNextElement(mod_s.clientsds, &cur_clientsds)) |
| mod_s.maxfdp1 = max(*((int*)(cur_clientsds->content)), mod_s.maxfdp1); |
| ++(mod_s.maxfdp1); |
| Log(TRACE_MAX, -1, "Reset max fdp1 to %d", mod_s.maxfdp1); |
| } |
| FUNC_EXIT; |
| } |
| |
| |
| /** |
| * Create a new socket and TCP connect to an address/port |
| * @param addr the address string |
| * @param port the TCP port |
| * @param sock returns the new socket |
| * @param timeout the timeout in milliseconds |
| * @return completion code |
| */ |
| #if defined(__GNUC__) && defined(__linux__) |
| int Socket_new(const char* addr, size_t addr_len, int port, int* sock, long timeout) |
| #else |
| int Socket_new(const char* addr, size_t addr_len, int port, int* sock) |
| #endif |
| { |
| int type = SOCK_STREAM; |
| char *addr_mem; |
| struct sockaddr_in address; |
| #if defined(AF_INET6) |
| struct sockaddr_in6 address6; |
| #endif |
| int rc = SOCKET_ERROR; |
| #if defined(_WIN32) || defined(_WIN64) |
| short family; |
| #else |
| sa_family_t family = AF_INET; |
| #endif |
| struct addrinfo *result = NULL; |
| struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL}; |
| |
| FUNC_ENTRY; |
| *sock = -1; |
| memset(&address6, '\0', sizeof(address6)); |
| |
| if (addr[0] == '[') |
| { |
| ++addr; |
| --addr_len; |
| } |
| |
| if ((addr_mem = malloc( addr_len + 1u )) == NULL) |
| { |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| memcpy( addr_mem, addr, addr_len ); |
| addr_mem[addr_len] = '\0'; |
| |
| #if 0 /*defined(__GNUC__) && defined(__linux__)*/ |
| /* Commented out because the CI tests get intermittent ECONNABORTED return values |
| * and I don't know why yet. |
| */ |
| /* set getaddrinfo timeout if available */ |
| struct gaicb ar = {addr_mem, NULL, &hints, NULL}; |
| struct gaicb *reqs[] = {&ar}; |
| |
| unsigned long int seconds = timeout / 1000L; |
| unsigned long int nanos = (timeout - (seconds * 1000L)) * 1000000L; |
| struct timespec timeoutspec = {seconds, nanos}; |
| |
| rc = getaddrinfo_a(GAI_NOWAIT, reqs, 1, NULL); |
| if (rc == 0) |
| rc = gai_suspend((const struct gaicb* const *) reqs, 1, &timeoutspec); |
| |
| if (rc == 0) |
| { |
| rc = gai_error(reqs[0]); |
| result = ar.ar_result; |
| } |
| #else |
| rc = getaddrinfo(addr_mem, NULL, &hints, &result); |
| #endif |
| |
| if (rc == 0) |
| { |
| struct addrinfo* res = result; |
| |
| while (res) |
| { /* prefer ip4 addresses */ |
| if (res->ai_family == AF_INET || res->ai_next == NULL) |
| break; |
| res = res->ai_next; |
| } |
| |
| if (res == NULL) |
| rc = -1; |
| else |
| #if defined(AF_INET6) |
| if (res->ai_family == AF_INET6) |
| { |
| address6.sin6_port = htons(port); |
| address6.sin6_family = family = AF_INET6; |
| memcpy(&address6.sin6_addr, &((struct sockaddr_in6*)(res->ai_addr))->sin6_addr, sizeof(address6.sin6_addr)); |
| } |
| else |
| #endif |
| if (res->ai_family == AF_INET) |
| { |
| memset(&address.sin_zero, 0, sizeof(address.sin_zero)); |
| address.sin_port = htons(port); |
| address.sin_family = family = AF_INET; |
| address.sin_addr = ((struct sockaddr_in*)(res->ai_addr))->sin_addr; |
| } |
| else |
| rc = -1; |
| |
| freeaddrinfo(result); |
| } |
| else |
| Log(LOG_ERROR, -1, "getaddrinfo failed for addr %s with rc %d", addr_mem, rc); |
| |
| if (rc != 0) |
| Log(LOG_ERROR, -1, "%s is not a valid IP address", addr_mem); |
| else |
| { |
| *sock = (int)socket(family, type, 0); |
| if (*sock == INVALID_SOCKET) |
| rc = Socket_error("socket", *sock); |
| else |
| { |
| #if defined(NOSIGPIPE) |
| int opt = 1; |
| |
| if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0) |
| Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock); |
| #endif |
| /*#define SMALL_TCP_BUFFER_TESTING |
| This section sets the TCP send buffer to a small amount to provoke TCPSOCKET_INTERRUPTED |
| return codes from send, for testing only! |
| */ |
| #if defined(SMALL_TCP_BUFFER_TESTING) |
| if (1) |
| { |
| int optsend = 100; //2 * 1440; |
| printf("Setting optsend to %d\n", optsend); |
| if (setsockopt(*sock, SOL_SOCKET, SO_SNDBUF, (void*)&optsend, sizeof(optsend)) != 0) |
| Log(LOG_ERROR, -1, "Could not set SO_SNDBUF for socket %d", *sock); |
| } |
| #endif |
| Log(TRACE_MIN, -1, "New socket %d for %s, port %d", *sock, addr, port); |
| if (Socket_addSocket(*sock) == SOCKET_ERROR) |
| rc = Socket_error("addSocket", *sock); |
| else |
| { |
| /* this could complete immmediately, even though we are non-blocking */ |
| if (family == AF_INET) |
| rc = connect(*sock, (struct sockaddr*)&address, sizeof(address)); |
| #if defined(AF_INET6) |
| else |
| rc = connect(*sock, (struct sockaddr*)&address6, sizeof(address6)); |
| #endif |
| if (rc == SOCKET_ERROR) |
| rc = Socket_error("connect", *sock); |
| if (rc == EINPROGRESS || rc == EWOULDBLOCK) |
| { |
| int* pnewSd = (int*)malloc(sizeof(int)); |
| |
| if (!pnewSd) |
| { |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| *pnewSd = *sock; |
| if (!ListAppend(mod_s.connect_pending, pnewSd, sizeof(int))) |
| { |
| free(pnewSd); |
| rc = PAHO_MEMORY_ERROR; |
| goto exit; |
| } |
| Log(TRACE_MIN, 15, "Connect pending"); |
| } |
| } |
| /* Prevent socket leak by closing unusable sockets, |
| as reported in https://github.com/eclipse/paho.mqtt.c/issues/135 */ |
| if (rc != 0 && (rc != EINPROGRESS) && (rc != EWOULDBLOCK)) |
| { |
| Socket_close(*sock); /* close socket and remove from our list of sockets */ |
| *sock = -1; /* as initialized before */ |
| } |
| } |
| } |
| |
| exit: |
| if (addr_mem) |
| free(addr_mem); |
| |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| static Socket_writeComplete* writecomplete = NULL; |
| |
| void Socket_setWriteCompleteCallback(Socket_writeComplete* mywritecomplete) |
| { |
| writecomplete = mywritecomplete; |
| } |
| |
| |
| |
| /** |
| * Continue an outstanding write for a particular socket |
| * @param socket that socket |
| * @return completion code: 0=incomplete, 1=complete, -1=socket error |
| */ |
| int Socket_continueWrite(int socket) |
| { |
| int rc = 0; |
| pending_writes* pw; |
| unsigned long curbuflen = 0L, /* cumulative total of buffer lengths */ |
| bytes = 0L; |
| int curbuf = -1, i; |
| iobuf iovecs1[5]; |
| |
| FUNC_ENTRY; |
| pw = SocketBuffer_getWrite(socket); |
| |
| #if defined(OPENSSL) |
| if (pw->ssl) |
| { |
| rc = SSLSocket_continueWrite(pw); |
| goto exit; |
| } |
| #endif |
| |
| for (i = 0; i < pw->count; ++i) |
| { |
| if (pw->bytes <= curbuflen) |
| { /* if previously written length is less than the buffer we are currently looking at, |
| add the whole buffer */ |
| iovecs1[++curbuf].iov_len = pw->iovecs[i].iov_len; |
| iovecs1[curbuf].iov_base = pw->iovecs[i].iov_base; |
| } |
| else if (pw->bytes < curbuflen + pw->iovecs[i].iov_len) |
| { /* if previously written length is in the middle of the buffer we are currently looking at, |
| add some of the buffer */ |
| size_t offset = pw->bytes - curbuflen; |
| iovecs1[++curbuf].iov_len = pw->iovecs[i].iov_len - (ULONG)offset; |
| iovecs1[curbuf].iov_base = (char*)pw->iovecs[i].iov_base + offset; |
| break; |
| } |
| curbuflen += pw->iovecs[i].iov_len; |
| } |
| |
| if ((rc = Socket_writev(socket, iovecs1, curbuf+1, &bytes)) != SOCKET_ERROR) |
| { |
| pw->bytes += bytes; |
| if ((rc = (pw->bytes == pw->total))) |
| { /* topic and payload buffers are freed elsewhere, when all references to them have been removed */ |
| for (i = 0; i < pw->count; i++) |
| { |
| if (pw->frees[i]) |
| { |
| free(pw->iovecs[i].iov_base); |
| pw->iovecs[i].iov_base = NULL; |
| } |
| } |
| rc = 1; /* signal complete */ |
| Log(TRACE_MIN, -1, "ContinueWrite: partial write now complete for socket %d", socket); |
| } |
| else |
| { |
| rc = 0; /* signal not complete */ |
| Log(TRACE_MIN, -1, "ContinueWrite wrote +%lu bytes on socket %d", bytes, socket); |
| } |
| } |
| else /* if we got SOCKET_ERROR we need to clean up anyway - a partial write is no good anymore */ |
| { |
| for (i = 0; i < pw->count; i++) |
| { |
| if (pw->frees[i]) |
| { |
| free(pw->iovecs[i].iov_base); |
| pw->iovecs[i].iov_base = NULL; |
| } |
| } |
| } |
| #if defined(OPENSSL) |
| exit: |
| #endif |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| |
| /** |
| * Continue an outstanding write for a particular socket |
| * @param socket that socket |
| * @return completion code: 0=incomplete, 1=complete, -1=socket error |
| */ |
| int Socket_abortWrite(int socket) |
| { |
| int i = -1, rc = 0; |
| pending_writes* pw; |
| |
| FUNC_ENTRY; |
| if ((pw = SocketBuffer_getWrite(socket)) == NULL) |
| goto exit; |
| |
| #if defined(OPENSSL) |
| if (pw->ssl) |
| goto exit; |
| #endif |
| |
| for (i = 0; i < pw->count; i++) |
| { |
| if (pw->frees[i]) |
| { |
| Log(TRACE_MIN, -1, "Cleaning in abortWrite for socket %d", socket); |
| free(pw->iovecs[i].iov_base); |
| } |
| } |
| exit: |
| FUNC_EXIT_RC(rc); |
| return rc; |
| } |
| |
| |
| /** |
| * Continue any outstanding writes for a socket set |
| * @param pwset the set of sockets |
| * @return completion code |
| */ |
| int Socket_continueWrites(fd_set* pwset) |
| { |
| int rc1 = 0; |
| ListElement* curpending = mod_s.write_pending->first; |
| |
| FUNC_ENTRY; |
| while (curpending && curpending->content) |
| { |
| int socket = *(int*)(curpending->content); |
| int rc = 0; |
| |
| if (FD_ISSET(socket, pwset) && ((rc = Socket_continueWrite(socket)) != 0)) |
| { |
| if (!SocketBuffer_writeComplete(socket)) |
| Log(LOG_SEVERE, -1, "Failed to remove pending write from socket buffer list"); |
| FD_CLR(socket, &(mod_s.pending_wset)); |
| if (!ListRemove(mod_s.write_pending, curpending->content)) |
| { |
| Log(LOG_SEVERE, -1, "Failed to remove pending write from list"); |
| ListNextElement(mod_s.write_pending, &curpending); |
| } |
| curpending = mod_s.write_pending->current; |
| |
| if (writecomplete) |
| (*writecomplete)(socket, rc); |
| } |
| else |
| ListNextElement(mod_s.write_pending, &curpending); |
| } |
| FUNC_EXIT_RC(rc1); |
| return rc1; |
| } |
| |
| |
| /** |
| * Convert a numeric address to character string |
| * @param sa socket numerical address |
| * @param sock socket |
| * @return the peer information |
| */ |
| char* Socket_getaddrname(struct sockaddr* sa, int sock) |
| { |
| /** |
| * maximum length of the address string |
| */ |
| #define ADDRLEN INET6_ADDRSTRLEN+1 |
| /** |
| * maximum length of the port string |
| */ |
| #define PORTLEN 10 |
| static char addr_string[ADDRLEN + PORTLEN]; |
| |
| #if defined(_WIN32) || defined(_WIN64) |
| int buflen = ADDRLEN*2; |
| wchar_t buf[ADDRLEN*2]; |
| if (WSAAddressToStringW(sa, sizeof(struct sockaddr_in6), NULL, buf, (LPDWORD)&buflen) == SOCKET_ERROR) |
| Socket_error("WSAAddressToString", sock); |
| else |
| wcstombs(addr_string, buf, sizeof(addr_string)); |
| /* TODO: append the port information - format: [00:00:00::]:port */ |
| /* strcpy(&addr_string[strlen(addr_string)], "what?"); */ |
| #else |
| struct sockaddr_in *sin = (struct sockaddr_in *)sa; |
| inet_ntop(sin->sin_family, &sin->sin_addr, addr_string, ADDRLEN); |
| sprintf(&addr_string[strlen(addr_string)], ":%d", ntohs(sin->sin_port)); |
| #endif |
| return addr_string; |
| } |
| |
| |
| /** |
| * Get information about the other end connected to a socket |
| * @param sock the socket to inquire on |
| * @return the peer information |
| */ |
| char* Socket_getpeer(int sock) |
| { |
| struct sockaddr_in6 sa; |
| socklen_t sal = sizeof(sa); |
| |
| if (getpeername(sock, (struct sockaddr*)&sa, &sal) == SOCKET_ERROR) |
| { |
| Socket_error("getpeername", sock); |
| return "unknown"; |
| } |
| |
| return Socket_getaddrname((struct sockaddr*)&sa, sock); |
| } |
| |
| |
| #if defined(Socket_TEST) |
| |
| int main(int argc, char *argv[]) |
| { |
| Socket_connect("127.0.0.1", 1883); |
| Socket_connect("localhost", 1883); |
| Socket_connect("loadsadsacalhost", 1883); |
| } |
| |
| #endif |