networklayer_tcp.c 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369
  1. /*
  2. * This work is licensed under a Creative Commons CCZero 1.0 Universal License.
  3. * See http://creativecommons.org/publicdomain/zero/1.0/ for more information.
  4. */
  5. #include <stdlib.h> // malloc, free
  6. #ifdef _WIN32
  7. #include <malloc.h>
  8. #include <winsock2.h>
  9. #include <sys/types.h>
  10. #include <Windows.h>
  11. #include <ws2tcpip.h>
  12. #define CLOSESOCKET(S) closesocket(S)
  13. #else
  14. #include <sys/select.h>
  15. #include <netinet/in.h>
  16. #include <sys/socketvar.h>
  17. #include <sys/ioctl.h>
  18. #include <unistd.h> // read, write, close
  19. #define CLOSESOCKET(S) close(S)
  20. #endif
  21. #include <arpa/inet.h>
  22. #include <stdio.h>
  23. #include <errno.h> // errno, EINTR
  24. #include <fcntl.h> // fcntl
  25. #include "networklayer_tcp.h" // UA_MULTITHREADING is defined in here
  26. #ifdef UA_MULTITHREADING
  27. #include <urcu/uatomic.h>
  28. #endif
  29. #define MAXBACKLOG 100
  30. struct Networklayer_TCP;
  31. /* Forwarded to the server as a (UA_Connection) and used for callbacks back into
  32. the networklayer */
  33. typedef struct {
  34. UA_Connection connection;
  35. UA_Int32 sockfd;
  36. struct NetworkLayerTCP *layer;
  37. } TCPConnection;
  38. /* Internal mapping of sockets to connections */
  39. typedef struct {
  40. TCPConnection *connection;
  41. UA_Int32 sockfd;
  42. } ConnectionLink;
  43. typedef struct NetworkLayerTCP {
  44. UA_ConnectionConfig conf;
  45. fd_set fdset;
  46. UA_Int32 serversockfd;
  47. UA_Int32 highestfd;
  48. UA_UInt16 conLinksSize;
  49. ConnectionLink *conLinks;
  50. UA_UInt32 port;
  51. /* We remove the connection links only in the main thread. Attach
  52. to-be-deleted links with atomic operations */
  53. struct deleteLink {
  54. UA_Int32 sockfd;
  55. struct deleteLink *next;
  56. } *deleteLinkList;
  57. } NetworkLayerTCP;
  58. static UA_StatusCode setNonBlocking(int sockid) {
  59. #ifdef _WIN32
  60. u_long iMode = 1;
  61. if(ioctlsocket(sockid, FIONBIO, &iMode) != NO_ERROR)
  62. return UA_STATUSCODE_BADINTERNALERROR;
  63. #else
  64. int opts = fcntl(sockid,F_GETFL);
  65. if(opts < 0 || fcntl(sockid,F_SETFL,opts|O_NONBLOCK) < 0)
  66. return UA_STATUSCODE_BADINTERNALERROR;
  67. #endif
  68. return UA_STATUSCODE_GOOD;
  69. }
  70. static void freeConnectionCallback(UA_Server *server, TCPConnection *connection) {
  71. free(connection);
  72. }
  73. // after every select, reset the set of sockets we want to listen on
  74. static void setFDSet(NetworkLayerTCP *layer) {
  75. FD_ZERO(&layer->fdset);
  76. FD_SET(layer->serversockfd, &layer->fdset);
  77. layer->highestfd = layer->serversockfd;
  78. for(UA_Int32 i=0;i<layer->conLinksSize;i++) {
  79. FD_SET(layer->conLinks[i].sockfd, &layer->fdset);
  80. if(layer->conLinks[i].sockfd > layer->highestfd)
  81. layer->highestfd = layer->conLinks[i].sockfd;
  82. }
  83. }
  84. // the callbacks are thread-safe if UA_MULTITHREADING is defined
  85. void closeConnection(TCPConnection *handle);
  86. void writeCallback(TCPConnection *handle, UA_ByteStringArray gather_buf);
  87. static UA_StatusCode NetworkLayerTCP_add(NetworkLayerTCP *layer, UA_Int32 newsockfd) {
  88. setNonBlocking(newsockfd);
  89. TCPConnection *c = malloc(sizeof(TCPConnection));
  90. c->sockfd = newsockfd;
  91. c->layer = layer;
  92. c->connection.state = UA_CONNECTION_OPENING;
  93. c->connection.localConf = layer->conf;
  94. c->connection.channel = UA_NULL;
  95. c->connection.close = (void (*)(void*))closeConnection;
  96. c->connection.write = (void (*)(void*, UA_ByteStringArray))writeCallback;
  97. layer->conLinks = realloc(layer->conLinks, sizeof(ConnectionLink)*(layer->conLinksSize+1));
  98. layer->conLinks[layer->conLinksSize].connection = c;
  99. layer->conLinks[layer->conLinksSize].sockfd = newsockfd;
  100. layer->conLinksSize++;
  101. return UA_STATUSCODE_GOOD;
  102. }
  103. // Takes the linked list of closed connections and returns the work for the server loop
  104. static UA_UInt32 batchDeleteLinks(NetworkLayerTCP *layer, UA_WorkItem **returnWork) {
  105. UA_WorkItem *work = malloc(sizeof(UA_WorkItem)*layer->conLinksSize);
  106. #ifdef UA_MULTITHREADING
  107. struct deleteLink *d = uatomic_xchg(&layer->deleteLinkList, UA_NULL);
  108. #else
  109. struct deleteLink *d = layer->deleteLinkList;
  110. layer->deleteLinkList = UA_NULL;
  111. #endif
  112. UA_UInt32 count = 0;
  113. while(d) {
  114. UA_Int32 i;
  115. for(i = 0;i<layer->conLinksSize;i++) {
  116. if(layer->conLinks[i].sockfd == d->sockfd)
  117. break;
  118. }
  119. if(i < layer->conLinksSize) {
  120. TCPConnection *c = layer->conLinks[i].connection;
  121. layer->conLinksSize--;
  122. layer->conLinks[i] = layer->conLinks[layer->conLinksSize];
  123. work[count] = (UA_WorkItem)
  124. {.type = UA_WORKITEMTYPE_DELAYEDMETHODCALL,
  125. .work.methodCall = {.data = c,
  126. .method = (void (*)(UA_Server*,void*))freeConnectionCallback} };
  127. }
  128. struct deleteLink *oldd = d;
  129. d = d->next;
  130. free(oldd);
  131. count++;
  132. }
  133. *returnWork = work;
  134. return count;
  135. }
  136. #ifdef UA_MULTITHREADING
  137. void closeConnection(TCPConnection *handle) {
  138. if(uatomic_xchg(&handle->connection.state, UA_CONNECTION_CLOSING) == UA_CONNECTION_CLOSING)
  139. return;
  140. UA_Connection_detachSecureChannel(&handle->connection);
  141. shutdown(handle->sockfd,2);
  142. CLOSESOCKET(handle->sockfd);
  143. // Remove the link later in the main thread
  144. struct deleteLink *d = malloc(sizeof(struct deleteLink));
  145. d->sockfd = handle->sockfd;
  146. while(1) {
  147. d->next = handle->layer->deleteLinkList;
  148. if(uatomic_cmpxchg(&handle->layer->deleteLinkList, d->next, d) == d->next)
  149. break;
  150. }
  151. }
  152. #else
  153. void closeConnection(TCPConnection *handle) {
  154. if(handle->connection.state == UA_CONNECTION_CLOSING)
  155. return;
  156. handle->connection.state = UA_CONNECTION_CLOSING;
  157. UA_Connection_detachSecureChannel(&handle->connection);
  158. shutdown(handle->sockfd,2);
  159. CLOSESOCKET(handle->sockfd);
  160. // Remove the link later in the main thread
  161. struct deleteLink *d = malloc(sizeof(struct deleteLink));
  162. d->sockfd = handle->sockfd;
  163. d->next = handle->layer->deleteLinkList;
  164. handle->layer->deleteLinkList = d;
  165. }
  166. #endif
  167. /** Accesses only the sockfd in the handle. Can be run from parallel threads. */
  168. void writeCallback(TCPConnection *handle, UA_ByteStringArray gather_buf) {
  169. UA_UInt32 total_len = 0, nWritten = 0;
  170. #ifdef _WIN32
  171. LPWSABUF buf = _alloca(gather_buf.stringsSize * sizeof(WSABUF));
  172. int result = 0;
  173. for(UA_UInt32 i = 0; i<gather_buf.stringsSize; i++) {
  174. buf[i].buf = gather_buf.strings[i].data;
  175. buf[i].len = gather_buf.strings[i].length;
  176. total_len += gather_buf.strings[i].length;
  177. }
  178. while(nWritten < total_len) {
  179. UA_UInt32 n = 0;
  180. do {
  181. result = WSASend(handle->sockfd, buf, gather_buf.stringsSize ,
  182. (LPDWORD)&n, 0, NULL, NULL);
  183. if(result != 0)
  184. printf("Error WSASend, code: %d \n", WSAGetLastError());
  185. } while(errno == EINTR);
  186. nWritten += n;
  187. }
  188. #else
  189. struct iovec iov[gather_buf.stringsSize];
  190. for(UA_UInt32 i=0;i<gather_buf.stringsSize;i++) {
  191. iov[i] = (struct iovec) {.iov_base = gather_buf.strings[i].data,
  192. .iov_len = gather_buf.strings[i].length};
  193. total_len += gather_buf.strings[i].length;
  194. }
  195. struct msghdr message = {.msg_name = NULL, .msg_namelen = 0, .msg_iov = iov,
  196. .msg_iovlen = gather_buf.stringsSize, .msg_control = NULL,
  197. .msg_controllen = 0, .msg_flags = 0};
  198. while (nWritten < total_len) {
  199. UA_Int32 n = 0;
  200. do {
  201. n = sendmsg(handle->sockfd, &message, 0);
  202. } while (n == -1L && errno == EINTR);
  203. nWritten += n;
  204. }
  205. #endif
  206. }
  207. UA_StatusCode NetworkLayerTCP_start(NetworkLayerTCP *layer) {
  208. #ifdef _WIN32
  209. WORD wVersionRequested;
  210. WSADATA wsaData;
  211. wVersionRequested = MAKEWORD(2, 2);
  212. WSAStartup(wVersionRequested, &wsaData);
  213. if((layer->serversockfd = socket(PF_INET, SOCK_STREAM,0)) == INVALID_SOCKET) {
  214. printf("ERROR opening socket, code: %d\n", WSAGetLastError());
  215. return UA_STATUSCODE_BADINTERNALERROR;
  216. }
  217. #else
  218. if((layer->serversockfd = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
  219. perror("ERROR opening socket");
  220. return UA_STATUSCODE_BADINTERNALERROR;
  221. }
  222. #endif
  223. const struct sockaddr_in serv_addr = {
  224. .sin_family = AF_INET, .sin_addr.s_addr = INADDR_ANY,
  225. .sin_port = htons(layer->port), .sin_zero = {0}};
  226. int optval = 1;
  227. if(setsockopt(layer->serversockfd, SOL_SOCKET,
  228. SO_REUSEADDR, (const char *)&optval,
  229. sizeof(optval)) == -1) {
  230. perror("setsockopt");
  231. CLOSESOCKET(layer->serversockfd);
  232. return UA_STATUSCODE_BADINTERNALERROR;
  233. }
  234. if(bind(layer->serversockfd, (const struct sockaddr *)&serv_addr,
  235. sizeof(serv_addr)) < 0) {
  236. perror("binding");
  237. CLOSESOCKET(layer->serversockfd);
  238. return UA_STATUSCODE_BADINTERNALERROR;
  239. }
  240. setNonBlocking(layer->serversockfd);
  241. listen(layer->serversockfd, MAXBACKLOG);
  242. printf("Listening for TCP connections on %s:%d\n",
  243. inet_ntoa(serv_addr.sin_addr),
  244. ntohs(serv_addr.sin_port));
  245. return UA_STATUSCODE_GOOD;
  246. }
  247. UA_Int32 NetworkLayerTCP_getWork(NetworkLayerTCP *layer, UA_WorkItem **workItems,
  248. UA_UInt16 timeout) {
  249. UA_WorkItem *items = UA_NULL;
  250. UA_Int32 itemsCount = batchDeleteLinks(layer, &items);
  251. setFDSet(layer);
  252. struct timeval tmptv = {0, timeout};
  253. UA_Int32 resultsize = select(layer->highestfd+1, &layer->fdset, NULL, NULL, &tmptv);
  254. if(resultsize < 0) {
  255. *workItems = items;
  256. return itemsCount;
  257. }
  258. // accept new connections (can only be a single one)
  259. if(FD_ISSET(layer->serversockfd,&layer->fdset)) {
  260. resultsize--;
  261. struct sockaddr_in cli_addr;
  262. socklen_t cli_len = sizeof(cli_addr);
  263. int newsockfd = accept(layer->serversockfd, (struct sockaddr *) &cli_addr, &cli_len);
  264. if (newsockfd >= 0)
  265. NetworkLayerTCP_add(layer, newsockfd);
  266. }
  267. items = realloc(items, sizeof(UA_WorkItem)*(itemsCount+resultsize));
  268. // read from established sockets
  269. UA_Int32 j = itemsCount;
  270. UA_ByteString buf = {-1, NULL};
  271. for(UA_Int32 i=0;i<layer->conLinksSize && j<itemsCount+resultsize;i++) {
  272. if(!(FD_ISSET(layer->conLinks[i].sockfd, &layer->fdset)))
  273. continue;
  274. if(buf.data == NULL)
  275. buf.data = malloc(layer->conf.recvBufferSize);
  276. #ifdef _WIN32
  277. buf.length = recv(layer->conLinks[i].connection.sockfd, (char *)buf.data,
  278. layer->conf.recvBufferSize, 0);
  279. #else
  280. buf.length = read(layer->conLinks[i].sockfd, buf.data, layer->conf.recvBufferSize);
  281. #endif
  282. if (errno != 0 || buf.length == 0) {
  283. closeConnection(layer->conLinks[i].connection); // work is returned in the next iteration
  284. } else {
  285. items[j].type = UA_WORKITEMTYPE_BINARYNETWORKMESSAGE;
  286. items[j].work.binaryNetworkMessage.message = buf;
  287. items[j].work.binaryNetworkMessage.connection = &layer->conLinks[i].connection->connection;
  288. buf.data = NULL;
  289. j++;
  290. }
  291. }
  292. if(buf.data != NULL)
  293. free(buf.data);
  294. if(j == 0) {
  295. free(items);
  296. *workItems = NULL;
  297. } else
  298. *workItems = items;
  299. return j;
  300. }
  301. UA_Int32 NetworkLayerTCP_stop(NetworkLayerTCP * layer, UA_WorkItem **workItems) {
  302. for(UA_Int32 index = 0;index < layer->conLinksSize;index++)
  303. closeConnection(layer->conLinks[index].connection);
  304. #ifdef _WIN32
  305. WSACleanup();
  306. #endif
  307. return batchDeleteLinks(layer, workItems);
  308. }
  309. void NetworkLayerTCP_delete(NetworkLayerTCP *layer) {
  310. free(layer->conLinks);
  311. free(layer);
  312. }
  313. UA_NetworkLayer NetworkLayerTCP_new(UA_ConnectionConfig conf, UA_UInt32 port) {
  314. NetworkLayerTCP *tcplayer = malloc(sizeof(NetworkLayerTCP));
  315. tcplayer->conf = conf;
  316. tcplayer->conLinksSize = 0;
  317. tcplayer->conLinks = NULL;
  318. tcplayer->port = port;
  319. tcplayer->deleteLinkList = UA_NULL;
  320. UA_NetworkLayer nl;
  321. nl.nlHandle = tcplayer;
  322. nl.start = (UA_StatusCode (*)(void*))NetworkLayerTCP_start;
  323. nl.getWork = (UA_Int32 (*)(void*, UA_WorkItem**, UA_UInt16)) NetworkLayerTCP_getWork;
  324. nl.stop = (UA_Int32 (*)(void*, UA_WorkItem**)) NetworkLayerTCP_stop;
  325. nl.delete = (void (*)(void*))NetworkLayerTCP_delete;
  326. return nl;
  327. }