networklayer_udp.c 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286
  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 <strings.h> //bzero
  15. #include <sys/select.h>
  16. #include <netinet/in.h>
  17. #include <sys/socketvar.h>
  18. #include <sys/ioctl.h>
  19. #include <unistd.h> // read, write, close
  20. #include <arpa/inet.h>
  21. #define CLOSESOCKET(S) close(S)
  22. #endif
  23. #include <stdio.h>
  24. #include <errno.h> // errno, EINTR
  25. #include <fcntl.h> // fcntl
  26. #include "networklayer_udp.h" // UA_MULTITHREADING is defined in here
  27. #ifdef UA_MULTITHREADING
  28. #include <urcu/uatomic.h>
  29. #endif
  30. #define MAXBACKLOG 100
  31. struct Networklayer_UDP;
  32. /* Forwarded to the server as a (UA_Connection) and used for callbacks back into
  33. the networklayer */
  34. typedef struct {
  35. UA_Connection connection;
  36. struct sockaddr from;
  37. socklen_t fromlen;
  38. struct NetworkLayerUDP *layer;
  39. } UDPConnection;
  40. typedef struct NetworkLayerUDP {
  41. UA_ConnectionConfig conf;
  42. fd_set fdset;
  43. UA_Int32 serversockfd;
  44. UA_UInt32 port;
  45. /* We remove the connection links only in the main thread. Attach
  46. to-be-deleted links with atomic operations */
  47. struct deleteLink {
  48. UA_Int32 sockfd;
  49. struct deleteLink *next;
  50. } *deleteLinkList;
  51. } NetworkLayerUDP;
  52. static UA_StatusCode setNonBlocking(int sockid) {
  53. #ifdef _WIN32
  54. u_long iMode = 1;
  55. if(ioctlsocket(sockid, FIONBIO, &iMode) != NO_ERROR)
  56. return UA_STATUSCODE_BADINTERNALERROR;
  57. #else
  58. int opts = fcntl(sockid,F_GETFL);
  59. if(opts < 0 || fcntl(sockid,F_SETFL,opts|O_NONBLOCK) < 0)
  60. return UA_STATUSCODE_BADINTERNALERROR;
  61. #endif
  62. return UA_STATUSCODE_GOOD;
  63. }
  64. // after every select, reset the set of sockets we want to listen on
  65. static void setFDSet(NetworkLayerUDP *layer) {
  66. FD_ZERO(&layer->fdset);
  67. FD_SET(layer->serversockfd, &layer->fdset);
  68. }
  69. // the callbacks are thread-safe if UA_MULTITHREADING is defined
  70. void closeConnectionUDP(UDPConnection *write){
  71. }
  72. void writeCallbackUDP(UDPConnection *handle, UA_ByteStringArray gather_buf);
  73. /** Accesses only the sockfd in the handle. Can be run from parallel threads. */
  74. void writeCallbackUDP(UDPConnection *handle, UA_ByteStringArray gather_buf) {
  75. UA_UInt32 total_len = 0, nWritten = 0;
  76. #ifdef _WIN32
  77. LPWSABUF buf = _alloca(gather_buf.stringsSize * sizeof(WSABUF));
  78. int result = 0;
  79. for(UA_UInt32 i = 0; i<gather_buf.stringsSize; i++) {
  80. buf[i].buf = gather_buf.strings[i].data;
  81. buf[i].len = gather_buf.strings[i].length;
  82. total_len += gather_buf.strings[i].length;
  83. }
  84. while(nWritten < total_len) {
  85. UA_UInt32 n = 0;
  86. do {
  87. result = WSASendto(handle->sockfd, buf, gather_buf.stringsSize ,
  88. (LPDWORD)&n, 0, NULL, NULL);
  89. //FIXME:
  90. if(result != 0)
  91. printf("Error WSASend, code: %d \n", WSAGetLastError());
  92. } while(errno == EINTR);
  93. nWritten += n;
  94. }
  95. #else
  96. struct iovec iov[gather_buf.stringsSize];
  97. for(UA_UInt32 i=0;i<gather_buf.stringsSize;i++) {
  98. iov[i] = (struct iovec) {.iov_base = gather_buf.strings[i].data,
  99. .iov_len = gather_buf.strings[i].length};
  100. total_len += gather_buf.strings[i].length;
  101. }
  102. struct sockaddr_in *sin = UA_NULL;
  103. if (handle->from.sa_family == AF_INET)
  104. {
  105. sin = (struct sockaddr_in *) &(handle->from);
  106. }else{
  107. //FIXME:
  108. return;
  109. }
  110. struct msghdr message = {.msg_name = sin, .msg_namelen = handle->fromlen, .msg_iov = iov,
  111. .msg_iovlen = gather_buf.stringsSize, .msg_control = NULL,
  112. .msg_controllen = 0, .msg_flags = 0};
  113. while (nWritten < total_len) {
  114. UA_Int32 n = 0;
  115. do {
  116. n = sendmsg(handle->layer->serversockfd, &message, 0);
  117. if(n==-1L){
  118. printf("ERROR:%i\n", errno);
  119. }
  120. } while (n == -1L && errno == EINTR);
  121. nWritten += n;
  122. }
  123. #endif
  124. //remove the session
  125. //TODO: without this line we have a memleak
  126. //with this line we have some illegal memory flagged by valgrind
  127. free(handle);
  128. }
  129. UA_StatusCode NetworkLayerUDP_start(NetworkLayerUDP *layer) {
  130. #ifdef _WIN32
  131. WORD wVersionRequested;
  132. WSADATA wsaData;
  133. wVersionRequested = MAKEWORD(2, 2);
  134. WSAStartup(wVersionRequested, &wsaData);
  135. if((layer->serversockfd = socket(PF_INET, SOCK_DGRAM,0)) == INVALID_SOCKET) {
  136. printf("ERROR opening socket, code: %d\n", WSAGetLastError());
  137. return UA_STATUSCODE_BADINTERNALERROR;
  138. }
  139. #else
  140. if((layer->serversockfd = socket(PF_INET, SOCK_DGRAM, 0)) < 0) {
  141. perror("ERROR opening socket");
  142. return UA_STATUSCODE_BADINTERNALERROR;
  143. }
  144. #endif
  145. const struct sockaddr_in serv_addr = {
  146. .sin_family = AF_INET, .sin_addr.s_addr = INADDR_ANY,
  147. .sin_port = htons(layer->port), .sin_zero = {0}};
  148. int optval = 1;
  149. if(setsockopt(layer->serversockfd, SOL_SOCKET,
  150. SO_REUSEADDR, (const char *)&optval,
  151. sizeof(optval)) == -1) {
  152. perror("setsockopt");
  153. CLOSESOCKET(layer->serversockfd);
  154. return UA_STATUSCODE_BADINTERNALERROR;
  155. }
  156. if(bind(layer->serversockfd, (const struct sockaddr *)&serv_addr,
  157. sizeof(serv_addr)) < 0) {
  158. perror("binding");
  159. CLOSESOCKET(layer->serversockfd);
  160. return UA_STATUSCODE_BADINTERNALERROR;
  161. }
  162. setNonBlocking(layer->serversockfd);
  163. printf("Listening for UDP connections on %s:%d\n",
  164. inet_ntoa(serv_addr.sin_addr),
  165. ntohs(serv_addr.sin_port));
  166. return UA_STATUSCODE_GOOD;
  167. }
  168. UA_Int32 NetworkLayerUDP_getWork(NetworkLayerUDP *layer, UA_WorkItem **workItems,
  169. UA_UInt16 timeout) {
  170. UA_WorkItem *items = UA_NULL;
  171. UA_Int32 itemsCount = 0;
  172. setFDSet(layer);
  173. struct timeval tmptv = {0, timeout};
  174. UA_Int32 resultsize = select(layer->serversockfd+1, &layer->fdset, NULL, NULL, &tmptv);
  175. if(resultsize < 0) {
  176. *workItems = items;
  177. return itemsCount;
  178. }
  179. items = realloc(items, sizeof(UA_WorkItem)*(itemsCount+resultsize));
  180. // read from established sockets
  181. UA_Int32 j = itemsCount;
  182. UA_ByteString buf = { -1, NULL};
  183. if(!buf.data) {
  184. buf.data = malloc(sizeof(UA_Byte) * layer->conf.recvBufferSize);
  185. if(!buf.data){
  186. //TODO:
  187. }
  188. }
  189. struct sockaddr sender;
  190. socklen_t sendsize = sizeof(sender);
  191. bzero(&sender, sizeof(sender));
  192. #ifdef _WIN32
  193. buf.length = recvfrom(layer->conLinks[i].sockfd, (char *)buf.data,
  194. layer->conf.recvBufferSize, 0);
  195. //todo: fixme
  196. #else
  197. buf.length = recvfrom(layer->serversockfd, buf.data, layer->conf.recvBufferSize, 0, &sender, &sendsize);
  198. #endif
  199. if (buf.length <= 0) {
  200. } else {
  201. UDPConnection *c = malloc(sizeof(UDPConnection));
  202. if(!c)
  203. return UA_STATUSCODE_BADINTERNALERROR;
  204. c->layer = layer;
  205. c->from = sender;
  206. c->fromlen = sendsize;
  207. c->connection.state = UA_CONNECTION_OPENING;
  208. c->connection.localConf = layer->conf;
  209. c->connection.channel = UA_NULL;
  210. c->connection.close = (void (*)(void*))closeConnectionUDP;
  211. c->connection.write = (void (*)(void*, UA_ByteStringArray))writeCallbackUDP;
  212. items[j].type = UA_WORKITEMTYPE_BINARYNETWORKMESSAGE;
  213. items[j].work.binaryNetworkMessage.message = buf;
  214. items[j].work.binaryNetworkMessage.connection = (UA_Connection*)c;
  215. buf.data = NULL;
  216. j++;
  217. }
  218. if(buf.data)
  219. free(buf.data);
  220. if(j == 0) {
  221. free(items);
  222. *workItems = NULL;
  223. } else
  224. *workItems = items;
  225. return j;
  226. }
  227. UA_Int32 NetworkLayerUDP_stop(NetworkLayerUDP * layer, UA_WorkItem **workItems) {
  228. CLOSESOCKET(layer->serversockfd);
  229. return 0;
  230. }
  231. void NetworkLayerUDP_delete(NetworkLayerUDP *layer) {
  232. free(layer);
  233. }
  234. UA_NetworkLayer NetworkLayerUDP_new(UA_ConnectionConfig conf, UA_UInt32 port) {
  235. NetworkLayerUDP *udplayer = malloc(sizeof(NetworkLayerUDP));
  236. udplayer->conf = conf;
  237. udplayer->port = port;
  238. udplayer->deleteLinkList = UA_NULL;
  239. UA_NetworkLayer nl;
  240. nl.nlHandle = udplayer;
  241. nl.start = (UA_StatusCode (*)(void*))NetworkLayerUDP_start;
  242. nl.getWork = (UA_Int32 (*)(void*, UA_WorkItem**, UA_UInt16)) NetworkLayerUDP_getWork;
  243. nl.stop = (UA_Int32 (*)(void*, UA_WorkItem**)) NetworkLayerUDP_stop;
  244. nl.delete = (void (*)(void*))NetworkLayerUDP_delete;
  245. return nl;
  246. }