Phy.hpp 37 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208
  1. /*
  2. * Copyright (c)2013-2020 ZeroTier, Inc.
  3. *
  4. * Use of this software is governed by the Business Source License included
  5. * in the LICENSE.TXT file in the project's root directory.
  6. *
  7. * Change Date: 2026-01-01
  8. *
  9. * On the date above, in accordance with the Business Source License, use
  10. * of this software will be governed by version 2.0 of the Apache License.
  11. */
  12. /****/
  13. #ifndef ZT_PHY_HPP
  14. #define ZT_PHY_HPP
  15. #include <stdio.h>
  16. #include <stdlib.h>
  17. #include <string.h>
  18. #include <list>
  19. #include <stdexcept>
  20. #if defined(_WIN32) || defined(_WIN64)
  21. #include <winsock2.h>
  22. #include <ws2tcpip.h>
  23. #include <windows.h>
  24. #define ZT_PHY_SOCKFD_TYPE SOCKET
  25. #define ZT_PHY_SOCKFD_NULL (INVALID_SOCKET)
  26. #define ZT_PHY_SOCKFD_VALID(s) ((s) != INVALID_SOCKET)
  27. #define ZT_PHY_CLOSE_SOCKET(s) ::closesocket(s)
  28. #define ZT_PHY_MAX_SOCKETS (FD_SETSIZE)
  29. #define ZT_PHY_MAX_INTERCEPTS ZT_PHY_MAX_SOCKETS
  30. #define ZT_PHY_SOCKADDR_STORAGE_TYPE struct sockaddr_storage
  31. #else // not Windows
  32. #include <errno.h>
  33. #include <signal.h>
  34. #include <unistd.h>
  35. #include <fcntl.h>
  36. #include <sys/time.h>
  37. #include <sys/types.h>
  38. #include <sys/select.h>
  39. #include <sys/socket.h>
  40. #include <sys/un.h>
  41. #include <arpa/inet.h>
  42. #include <netinet/in.h>
  43. #include <netinet/tcp.h>
  44. #include "../node/Metrics.hpp"
  45. #if defined(__linux__) || defined(linux) || defined(__LINUX__) || defined(__linux)
  46. #ifndef IPV6_DONTFRAG
  47. #define IPV6_DONTFRAG 62
  48. #endif
  49. #endif
  50. #define ZT_PHY_SOCKFD_TYPE int
  51. #define ZT_PHY_SOCKFD_NULL (-1)
  52. #define ZT_PHY_SOCKFD_VALID(s) ((s) > -1)
  53. #define ZT_PHY_CLOSE_SOCKET(s) ::close(s)
  54. #define ZT_PHY_MAX_SOCKETS (FD_SETSIZE)
  55. #define ZT_PHY_MAX_INTERCEPTS ZT_PHY_MAX_SOCKETS
  56. #define ZT_PHY_SOCKADDR_STORAGE_TYPE struct sockaddr_storage
  57. #endif // Windows or not
  58. namespace ZeroTier {
  59. /**
  60. * Opaque socket type
  61. */
  62. typedef void PhySocket;
  63. /**
  64. * Simple templated non-blocking sockets implementation
  65. *
  66. * Yes there is boost::asio and libuv, but I like small binaries and I hate
  67. * build dependencies. Both drag in a whole bunch of pasta with them.
  68. *
  69. * This class is templated on a pointer to a handler class which must
  70. * implement the following functions:
  71. *
  72. * For all platforms:
  73. *
  74. * phyOnDatagram(PhySocket *sock,void **uptr,const struct sockaddr *localAddr,const struct sockaddr *from,void *data,unsigned long len)
  75. * phyOnTcpConnect(PhySocket *sock,void **uptr,bool success)
  76. * phyOnTcpAccept(PhySocket *sockL,PhySocket *sockN,void **uptrL,void **uptrN,const struct sockaddr *from)
  77. * phyOnTcpClose(PhySocket *sock,void **uptr)
  78. * phyOnTcpData(PhySocket *sock,void **uptr,void *data,unsigned long len)
  79. * phyOnTcpWritable(PhySocket *sock,void **uptr)
  80. * phyOnFileDescriptorActivity(PhySocket *sock,void **uptr,bool readable,bool writable)
  81. *
  82. * On Linux/OSX/Unix only (not required/used on Windows or elsewhere):
  83. *
  84. * phyOnUnixAccept(PhySocket *sockL,PhySocket *sockN,void **uptrL,void **uptrN)
  85. * phyOnUnixClose(PhySocket *sock,void **uptr)
  86. * phyOnUnixData(PhySocket *sock,void **uptr,void *data,unsigned long len)
  87. * phyOnUnixWritable(PhySocket *sock,void **uptr)
  88. *
  89. * These templates typically refer to function objects. Templates are used to
  90. * avoid the call overhead of indirection, which is surprisingly high for high
  91. * bandwidth applications pushing a lot of packets.
  92. *
  93. * The 'sock' pointer above is an opaque pointer to a socket. Each socket
  94. * has a 'uptr' user-settable/modifiable pointer associated with it, which
  95. * can be set on bind/connect calls and is passed as a void ** to permit
  96. * resetting at any time. The ACCEPT handler takes two sets of sock and
  97. * uptr: sockL and uptrL for the listen socket, and sockN and uptrN for
  98. * the new TCP connection socket that has just been created.
  99. *
  100. * Handlers are always called. On outgoing TCP connection, CONNECT is always
  101. * called on either success or failure followed by DATA and/or WRITABLE as
  102. * indicated. On socket close, handlers are called unless close() is told
  103. * explicitly not to call handlers. It is safe to close a socket within a
  104. * handler, and in that case close() can be told not to call handlers to
  105. * prevent recursion.
  106. *
  107. * This isn't thread-safe with the exception of whack(), which is safe to
  108. * call from another thread to abort poll().
  109. */
  110. template <typename HANDLER_PTR_TYPE>
  111. class Phy
  112. {
  113. private:
  114. HANDLER_PTR_TYPE _handler;
  115. enum PhySocketType
  116. {
  117. ZT_PHY_SOCKET_CLOSED = 0x00, // socket is closed, will be removed on next poll()
  118. ZT_PHY_SOCKET_TCP_OUT_PENDING = 0x01,
  119. ZT_PHY_SOCKET_TCP_OUT_CONNECTED = 0x02,
  120. ZT_PHY_SOCKET_TCP_IN = 0x03,
  121. ZT_PHY_SOCKET_TCP_LISTEN = 0x04,
  122. ZT_PHY_SOCKET_UDP = 0x05,
  123. ZT_PHY_SOCKET_FD = 0x06,
  124. ZT_PHY_SOCKET_UNIX_IN = 0x07,
  125. ZT_PHY_SOCKET_UNIX_LISTEN = 0x08
  126. };
  127. struct PhySocketImpl {
  128. PhySocketImpl() {}
  129. PhySocketType type;
  130. ZT_PHY_SOCKFD_TYPE sock;
  131. void *uptr; // user-settable pointer
  132. uint16_t localPort;
  133. ZT_PHY_SOCKADDR_STORAGE_TYPE saddr; // remote for TCP_OUT and TCP_IN, local for TCP_LISTEN, RAW, and UDP
  134. };
  135. std::list<PhySocketImpl> _socks;
  136. fd_set _readfds;
  137. fd_set _writefds;
  138. #if defined(_WIN32) || defined(_WIN64)
  139. fd_set _exceptfds;
  140. #endif
  141. long _nfds;
  142. ZT_PHY_SOCKFD_TYPE _whackReceiveSocket;
  143. ZT_PHY_SOCKFD_TYPE _whackSendSocket;
  144. bool _noDelay;
  145. bool _noCheck;
  146. public:
  147. /**
  148. * @param handler Pointer of type HANDLER_PTR_TYPE to handler
  149. * @param noDelay If true, disable TCP NAGLE algorithm on TCP sockets
  150. * @param noCheck If true, attempt to set UDP SO_NO_CHECK option to disable sending checksums
  151. */
  152. Phy(HANDLER_PTR_TYPE handler,bool noDelay,bool noCheck) :
  153. _handler(handler)
  154. {
  155. FD_ZERO(&_readfds);
  156. FD_ZERO(&_writefds);
  157. #if defined(_WIN32) || defined(_WIN64)
  158. FD_ZERO(&_exceptfds);
  159. SOCKET pipes[2];
  160. { // hack copied from StackOverflow, behaves a bit like pipe() on *nix systems
  161. struct sockaddr_in inaddr;
  162. struct sockaddr addr;
  163. SOCKET lst=::socket(AF_INET, SOCK_STREAM,IPPROTO_TCP);
  164. if (lst == INVALID_SOCKET)
  165. throw std::runtime_error("unable to create pipes for select() abort");
  166. memset(&inaddr, 0, sizeof(inaddr));
  167. memset(&addr, 0, sizeof(addr));
  168. inaddr.sin_family = AF_INET;
  169. inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
  170. inaddr.sin_port = 0;
  171. int yes=1;
  172. setsockopt(lst,SOL_SOCKET,SO_REUSEADDR,(char*)&yes,sizeof(yes));
  173. bind(lst,(struct sockaddr *)&inaddr,sizeof(inaddr));
  174. listen(lst,1);
  175. int len=sizeof(inaddr);
  176. getsockname(lst, &addr,&len);
  177. pipes[0]=::socket(AF_INET, SOCK_STREAM,0);
  178. if (pipes[0] == INVALID_SOCKET)
  179. throw std::runtime_error("unable to create pipes for select() abort");
  180. connect(pipes[0],&addr,len);
  181. pipes[1]=accept(lst,0,0);
  182. closesocket(lst);
  183. }
  184. #else // not Windows
  185. int pipes[2];
  186. if (::pipe(pipes))
  187. throw std::runtime_error("unable to create pipes for select() abort");
  188. #endif // Windows or not
  189. _nfds = (pipes[0] > pipes[1]) ? (long)pipes[0] : (long)pipes[1];
  190. _whackReceiveSocket = pipes[0];
  191. _whackSendSocket = pipes[1];
  192. _noDelay = noDelay;
  193. _noCheck = noCheck;
  194. }
  195. ~Phy()
  196. {
  197. for(typename std::list<PhySocketImpl>::const_iterator s(_socks.begin());s!=_socks.end();++s) {
  198. if (s->type != ZT_PHY_SOCKET_CLOSED)
  199. this->close((PhySocket *)&(*s),true);
  200. }
  201. ZT_PHY_CLOSE_SOCKET(_whackReceiveSocket);
  202. ZT_PHY_CLOSE_SOCKET(_whackSendSocket);
  203. }
  204. /**
  205. * @param s Socket object
  206. * @return Underlying OS-type (usually int or long) file descriptor associated with object
  207. */
  208. static inline ZT_PHY_SOCKFD_TYPE getDescriptor(PhySocket* s) throw()
  209. {
  210. return reinterpret_cast<PhySocketImpl*>(s)->sock;
  211. }
  212. /**
  213. * @param s Socket object
  214. * @return Pointer to user object
  215. */
  216. static inline void** getuptr(PhySocket* s) throw()
  217. {
  218. return &(reinterpret_cast<PhySocketImpl*>(s)->uptr);
  219. }
  220. /**
  221. * Return the local port corresponding to this PhySocket
  222. *
  223. * @param s Socket object
  224. *
  225. * @return Local port corresponding to this PhySocket
  226. */
  227. static inline uint16_t getLocalPort(PhySocket* s) throw()
  228. {
  229. return reinterpret_cast<PhySocketImpl*>(s)->localPort;
  230. }
  231. /**
  232. * Cause poll() to stop waiting immediately
  233. *
  234. * This can be used to reset the polling loop after changes that require
  235. * attention, or to shut down a background thread that is waiting, etc.
  236. */
  237. inline void whack()
  238. {
  239. #if defined(_WIN32) || defined(_WIN64)
  240. ::send(_whackSendSocket, (const char*)this, 1, 0);
  241. #else
  242. (void)(::write(_whackSendSocket, (PhySocket*)this, 1));
  243. #endif
  244. }
  245. /**
  246. * @return Number of open sockets
  247. */
  248. inline unsigned long count() const throw()
  249. {
  250. return _socks.size();
  251. }
  252. /**
  253. * @return Maximum number of sockets allowed
  254. */
  255. inline unsigned long maxCount() const throw()
  256. {
  257. return ZT_PHY_MAX_SOCKETS;
  258. }
  259. /**
  260. * Wrap a raw file descriptor in a PhySocket structure
  261. *
  262. * This can be used to select/poll on a raw file descriptor as part of this
  263. * class's I/O loop. By default the fd is set for read notification but
  264. * this can be controlled with setNotifyReadable(). When any detected
  265. * condition is present, the phyOnFileDescriptorActivity() callback is
  266. * called with one or both of its arguments 'true'.
  267. *
  268. * The Phy<>::close() method *must* be called when you're done with this
  269. * file descriptor to remove it from the select/poll set, but unlike other
  270. * types of sockets Phy<> does not actually close the underlying fd or
  271. * otherwise manage its life cycle. There is also no close notification
  272. * callback for this fd, since Phy<> doesn't actually perform reading or
  273. * writing or detect error conditions. This is only useful for adding a
  274. * file descriptor to Phy<> to select/poll on it.
  275. *
  276. * @param fd Raw file descriptor
  277. * @param uptr User pointer to supply to callbacks
  278. * @return PhySocket wrapping fd or NULL on failure (out of memory or too many sockets)
  279. */
  280. inline PhySocket *wrapSocket(ZT_PHY_SOCKFD_TYPE fd,void *uptr = (void *)0)
  281. {
  282. if (_socks.size() >= ZT_PHY_MAX_SOCKETS)
  283. return (PhySocket *)0;
  284. try {
  285. _socks.push_back(PhySocketImpl());
  286. } catch ( ... ) {
  287. return (PhySocket *)0;
  288. }
  289. PhySocketImpl &sws = _socks.back();
  290. if ((long)fd > _nfds)
  291. _nfds = (long)fd;
  292. FD_SET(fd,&_readfds);
  293. sws.type = ZT_PHY_SOCKET_UNIX_IN; /* TODO: Type was changed to allow for CBs with new RPC model */
  294. sws.sock = fd;
  295. sws.uptr = uptr;
  296. memset(&(sws.saddr),0,sizeof(struct sockaddr_storage));
  297. // no sockaddr for this socket type, leave saddr null
  298. return (PhySocket *)&sws;
  299. }
  300. /**
  301. * Bind a UDP socket
  302. *
  303. * @param localAddress Local endpoint address and port
  304. * @param uptr Initial value of user pointer associated with this socket (default: NULL)
  305. * @param bufferSize Desired socket receive/send buffer size -- will set as close to this as possible (default: 0, leave alone)
  306. * @return Socket or NULL on failure to bind
  307. */
  308. inline PhySocket *udpBind(const struct sockaddr *localAddress,void *uptr = (void *)0,int bufferSize = 0)
  309. {
  310. if (_socks.size() >= ZT_PHY_MAX_SOCKETS)
  311. return (PhySocket *)0;
  312. ZT_PHY_SOCKFD_TYPE s = ::socket(localAddress->sa_family,SOCK_DGRAM,0);
  313. if (!ZT_PHY_SOCKFD_VALID(s))
  314. return (PhySocket *)0;
  315. if (bufferSize > 0) {
  316. int bs = bufferSize;
  317. while (bs >= 65536) {
  318. int tmpbs = bs;
  319. if (setsockopt(s,SOL_SOCKET,SO_RCVBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
  320. break;
  321. bs -= 4096;
  322. }
  323. bs = bufferSize;
  324. while (bs >= 65536) {
  325. int tmpbs = bs;
  326. if (setsockopt(s,SOL_SOCKET,SO_SNDBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
  327. break;
  328. bs -= 4096;
  329. }
  330. }
  331. #if defined(_WIN32) || defined(_WIN64)
  332. {
  333. BOOL f;
  334. if (localAddress->sa_family == AF_INET6) {
  335. f = TRUE; setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(const char *)&f,sizeof(f));
  336. f = FALSE; setsockopt(s,IPPROTO_IPV6,IPV6_DONTFRAG,(const char *)&f,sizeof(f));
  337. }
  338. f = FALSE; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
  339. f = TRUE; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(const char *)&f,sizeof(f));
  340. }
  341. #else // not Windows
  342. {
  343. int f;
  344. if (localAddress->sa_family == AF_INET6) {
  345. f = 1; setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f));
  346. #ifdef IPV6_MTU_DISCOVER
  347. f = 0; setsockopt(s,IPPROTO_IPV6,IPV6_MTU_DISCOVER,&f,sizeof(f));
  348. #endif
  349. #ifdef IPV6_DONTFRAG
  350. f = 0; setsockopt(s,IPPROTO_IPV6,IPV6_DONTFRAG,&f,sizeof(f));
  351. #endif
  352. }
  353. f = 0; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
  354. f = 1; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(void *)&f,sizeof(f));
  355. #ifdef IP_DONTFRAG
  356. f = 0; setsockopt(s,IPPROTO_IP,IP_DONTFRAG,&f,sizeof(f));
  357. #endif
  358. #ifdef IP_MTU_DISCOVER
  359. f = 0; setsockopt(s,IPPROTO_IP,IP_MTU_DISCOVER,&f,sizeof(f));
  360. #endif
  361. #ifdef SO_NO_CHECK
  362. // For now at least we only set SO_NO_CHECK on IPv4 sockets since some
  363. // IPv6 stacks incorrectly discard zero checksum packets. May remove
  364. // this restriction later once broken stuff dies more.
  365. if ((localAddress->sa_family == AF_INET)&&(_noCheck)) {
  366. f = 1; setsockopt(s,SOL_SOCKET,SO_NO_CHECK,(void *)&f,sizeof(f));
  367. }
  368. #endif
  369. }
  370. #endif // Windows or not
  371. if (::bind(s,localAddress,(localAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in))) {
  372. ZT_PHY_CLOSE_SOCKET(s);
  373. return (PhySocket *)0;
  374. }
  375. #if defined(_WIN32) || defined(_WIN64)
  376. { u_long iMode=1; ioctlsocket(s,FIONBIO,&iMode); }
  377. #else
  378. fcntl(s,F_SETFL,O_NONBLOCK);
  379. #endif
  380. try {
  381. _socks.push_back(PhySocketImpl());
  382. } catch ( ... ) {
  383. ZT_PHY_CLOSE_SOCKET(s);
  384. return (PhySocket *)0;
  385. }
  386. PhySocketImpl &sws = _socks.back();
  387. if ((long)s > _nfds)
  388. _nfds = (long)s;
  389. FD_SET(s,&_readfds);
  390. sws.type = ZT_PHY_SOCKET_UDP;
  391. sws.sock = s;
  392. sws.uptr = uptr;
  393. #ifdef __UNIX_LIKE__
  394. struct sockaddr_in *sin = (struct sockaddr_in *)localAddress;
  395. sws.localPort = htons(sin->sin_port);
  396. #endif
  397. memset(&(sws.saddr),0,sizeof(struct sockaddr_storage));
  398. memcpy(&(sws.saddr),localAddress,(localAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in));
  399. return (PhySocket *)&sws;
  400. }
  401. /**
  402. * Set the IP TTL for the next outgoing packet (for IPv4 UDP sockets only)
  403. *
  404. * @param ttl New TTL (0 or >255 will set it to 255)
  405. * @return True on success
  406. */
  407. inline bool setIp4UdpTtl(PhySocket *sock,unsigned int ttl)
  408. {
  409. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  410. #if defined(_WIN32) || defined(_WIN64)
  411. DWORD tmp = ((ttl == 0)||(ttl > 255)) ? 255 : (DWORD)ttl;
  412. return (::setsockopt(sws.sock,IPPROTO_IP,IP_TTL,(const char *)&tmp,sizeof(tmp)) == 0);
  413. #else
  414. int tmp = ((ttl == 0)||(ttl > 255)) ? 255 : (int)ttl;
  415. return (::setsockopt(sws.sock,IPPROTO_IP,IP_TTL,(void *)&tmp,sizeof(tmp)) == 0);
  416. #endif
  417. }
  418. /**
  419. * Send a UDP packet
  420. *
  421. * @param sock UDP socket
  422. * @param remoteAddress Destination address (must be correct type for socket)
  423. * @param data Data to send
  424. * @param len Length of packet
  425. * @return True if packet appears to have been sent successfully
  426. */
  427. inline bool udpSend(PhySocket *sock,const struct sockaddr *remoteAddress,const void *data,unsigned long len)
  428. {
  429. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  430. bool sent = false;
  431. #if defined(_WIN32) || defined(_WIN64)
  432. sent = ((long)::sendto(
  433. sws.sock,
  434. reinterpret_cast<const char *>(data),
  435. len,
  436. 0,
  437. remoteAddress,
  438. (remoteAddress->sa_family == AF_INET6) ?
  439. sizeof(struct sockaddr_in6) :
  440. sizeof(struct sockaddr_in)) == (long)len);
  441. #else
  442. sent = ((long)::sendto(
  443. sws.sock,
  444. data,
  445. len,
  446. 0,
  447. remoteAddress,
  448. (remoteAddress->sa_family == AF_INET6) ?
  449. sizeof(struct sockaddr_in6) :
  450. sizeof(struct sockaddr_in)) == (long)len);
  451. #endif
  452. if (sent) {
  453. Metrics::udp_send += len;
  454. }
  455. return sent;
  456. }
  457. #ifdef __UNIX_LIKE__
  458. /**
  459. * Listen for connections on a Unix domain socket
  460. *
  461. * @param path Path to Unix domain socket
  462. * @param uptr Arbitrary pointer to associate
  463. * @return PhySocket or NULL if cannot bind
  464. */
  465. inline PhySocket *unixListen(const char *path,void *uptr = (void *)0)
  466. {
  467. struct sockaddr_un sun;
  468. if (_socks.size() >= ZT_PHY_MAX_SOCKETS)
  469. return (PhySocket *)0;
  470. memset(&sun,0,sizeof(sun));
  471. sun.sun_family = AF_UNIX;
  472. if (strlen(path) >= sizeof(sun.sun_path))
  473. return (PhySocket *)0;
  474. strcpy(sun.sun_path,path);
  475. ZT_PHY_SOCKFD_TYPE s = ::socket(PF_UNIX,SOCK_STREAM,0);
  476. if (!ZT_PHY_SOCKFD_VALID(s))
  477. return (PhySocket *)0;
  478. ::fcntl(s,F_SETFL,O_NONBLOCK);
  479. ::unlink(path);
  480. if (::bind(s,(struct sockaddr *)&sun,sizeof(struct sockaddr_un)) != 0) {
  481. ZT_PHY_CLOSE_SOCKET(s);
  482. return (PhySocket *)0;
  483. }
  484. if (::listen(s,128) != 0) {
  485. ZT_PHY_CLOSE_SOCKET(s);
  486. return (PhySocket *)0;
  487. }
  488. try {
  489. _socks.push_back(PhySocketImpl());
  490. } catch ( ... ) {
  491. ZT_PHY_CLOSE_SOCKET(s);
  492. return (PhySocket *)0;
  493. }
  494. PhySocketImpl &sws = _socks.back();
  495. if ((long)s > _nfds)
  496. _nfds = (long)s;
  497. FD_SET(s,&_readfds);
  498. sws.type = ZT_PHY_SOCKET_UNIX_LISTEN;
  499. sws.sock = s;
  500. sws.uptr = uptr;
  501. memset(&(sws.saddr),0,sizeof(struct sockaddr_storage));
  502. memcpy(&(sws.saddr),&sun,sizeof(struct sockaddr_un));
  503. return (PhySocket *)&sws;
  504. }
  505. #endif // __UNIX_LIKE__
  506. /**
  507. * Bind a local listen socket to listen for new TCP connections
  508. *
  509. * @param localAddress Local address and port
  510. * @param uptr Initial value of uptr for new socket (default: NULL)
  511. * @return Socket or NULL on failure to bind
  512. */
  513. inline PhySocket *tcpListen(const struct sockaddr *localAddress,void *uptr = (void *)0)
  514. {
  515. if (_socks.size() >= ZT_PHY_MAX_SOCKETS)
  516. return (PhySocket *)0;
  517. ZT_PHY_SOCKFD_TYPE s = ::socket(localAddress->sa_family,SOCK_STREAM,0);
  518. if (!ZT_PHY_SOCKFD_VALID(s))
  519. return (PhySocket *)0;
  520. #if defined(_WIN32) || defined(_WIN64)
  521. {
  522. BOOL f;
  523. f = TRUE; ::setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(const char *)&f,sizeof(f));
  524. f = TRUE; ::setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
  525. f = (_noDelay ? TRUE : FALSE); setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f));
  526. u_long iMode=1;
  527. ioctlsocket(s,FIONBIO,&iMode);
  528. }
  529. #else
  530. {
  531. int f;
  532. f = 1; ::setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f));
  533. f = 1; ::setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
  534. f = (_noDelay ? 1 : 0); setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f));
  535. fcntl(s,F_SETFL,O_NONBLOCK);
  536. }
  537. #endif
  538. if (::bind(s,localAddress,(localAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in))) {
  539. ZT_PHY_CLOSE_SOCKET(s);
  540. return (PhySocket *)0;
  541. }
  542. if (::listen(s,1024)) {
  543. ZT_PHY_CLOSE_SOCKET(s);
  544. return (PhySocket *)0;
  545. }
  546. try {
  547. _socks.push_back(PhySocketImpl());
  548. } catch ( ... ) {
  549. ZT_PHY_CLOSE_SOCKET(s);
  550. return (PhySocket *)0;
  551. }
  552. PhySocketImpl &sws = _socks.back();
  553. if ((long)s > _nfds)
  554. _nfds = (long)s;
  555. FD_SET(s,&_readfds);
  556. sws.type = ZT_PHY_SOCKET_TCP_LISTEN;
  557. sws.sock = s;
  558. sws.uptr = uptr;
  559. memset(&(sws.saddr),0,sizeof(struct sockaddr_storage));
  560. memcpy(&(sws.saddr),localAddress,(localAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in));
  561. return (PhySocket *)&sws;
  562. }
  563. /**
  564. * Start a non-blocking connect; CONNECT handler is called on success or failure
  565. *
  566. * A return value of NULL indicates a synchronous failure such as a
  567. * failure to open a socket. The TCP connection handler is not called
  568. * in this case.
  569. *
  570. * It is possible on some platforms for an "instant connect" to occur,
  571. * such as when connecting to a loopback address. In this case, the
  572. * 'connected' result parameter will be set to 'true' and if the
  573. * 'callConnectHandler' flag is true (the default) the TCP connect
  574. * handler will be called before the function returns.
  575. *
  576. * These semantics can be a bit confusing, but they're less so than
  577. * the underlying semantics of asynchronous TCP connect.
  578. *
  579. * @param remoteAddress Remote address
  580. * @param connected Result parameter: set to whether an "instant connect" has occurred (true if yes)
  581. * @param uptr Initial value of uptr for new socket (default: NULL)
  582. * @param callConnectHandler If true, call TCP connect handler even if result is known before function exit (default: true)
  583. * @return New socket or NULL on failure
  584. */
  585. inline PhySocket *tcpConnect(const struct sockaddr *remoteAddress,bool &connected,void *uptr = (void *)0,bool callConnectHandler = true)
  586. {
  587. if (_socks.size() >= ZT_PHY_MAX_SOCKETS)
  588. return (PhySocket *)0;
  589. ZT_PHY_SOCKFD_TYPE s = ::socket(remoteAddress->sa_family,SOCK_STREAM,0);
  590. if (!ZT_PHY_SOCKFD_VALID(s)) {
  591. connected = false;
  592. return (PhySocket *)0;
  593. }
  594. #if defined(_WIN32) || defined(_WIN64)
  595. {
  596. BOOL f;
  597. if (remoteAddress->sa_family == AF_INET6) { f = TRUE; ::setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(const char *)&f,sizeof(f)); }
  598. f = TRUE; ::setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
  599. f = (_noDelay ? TRUE : FALSE); setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f));
  600. u_long iMode=1;
  601. ioctlsocket(s,FIONBIO,&iMode);
  602. }
  603. #else
  604. {
  605. int f;
  606. if (remoteAddress->sa_family == AF_INET6) { f = 1; ::setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f)); }
  607. f = 1; ::setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
  608. f = (_noDelay ? 1 : 0); setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f));
  609. fcntl(s,F_SETFL,O_NONBLOCK);
  610. }
  611. #endif
  612. connected = true;
  613. if (::connect(s,remoteAddress,(remoteAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in))) {
  614. connected = false;
  615. #if defined(_WIN32) || defined(_WIN64)
  616. if (WSAGetLastError() != WSAEWOULDBLOCK) {
  617. #else
  618. if (errno != EINPROGRESS) {
  619. #endif
  620. ZT_PHY_CLOSE_SOCKET(s);
  621. return (PhySocket *)0;
  622. } // else connection is proceeding asynchronously...
  623. }
  624. try {
  625. _socks.push_back(PhySocketImpl());
  626. } catch ( ... ) {
  627. ZT_PHY_CLOSE_SOCKET(s);
  628. return (PhySocket *)0;
  629. }
  630. PhySocketImpl &sws = _socks.back();
  631. if ((long)s > _nfds)
  632. _nfds = (long)s;
  633. if (connected) {
  634. FD_SET(s,&_readfds);
  635. sws.type = ZT_PHY_SOCKET_TCP_OUT_CONNECTED;
  636. } else {
  637. FD_SET(s,&_writefds);
  638. #if defined(_WIN32) || defined(_WIN64)
  639. FD_SET(s,&_exceptfds);
  640. #endif
  641. sws.type = ZT_PHY_SOCKET_TCP_OUT_PENDING;
  642. }
  643. sws.sock = s;
  644. sws.uptr = uptr;
  645. memset(&(sws.saddr),0,sizeof(struct sockaddr_storage));
  646. memcpy(&(sws.saddr),remoteAddress,(remoteAddress->sa_family == AF_INET6) ? sizeof(struct sockaddr_in6) : sizeof(struct sockaddr_in));
  647. if ((callConnectHandler)&&(connected)) {
  648. try {
  649. _handler->phyOnTcpConnect((PhySocket *)&sws,&(sws.uptr),true);
  650. } catch ( ... ) {}
  651. }
  652. return (PhySocket *)&sws;
  653. }
  654. /**
  655. * Try to set buffer sizes as close to the given value as possible
  656. *
  657. * This will try the specified value and then lower values in 16K increments
  658. * until one works.
  659. *
  660. * @param sock Socket
  661. * @param receiveBufferSize Desired size of receive buffer
  662. * @param sendBufferSize Desired size of send buffer
  663. */
  664. inline void setBufferSizes(const PhySocket *sock,int receiveBufferSize,int sendBufferSize)
  665. {
  666. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  667. if (receiveBufferSize > 0) {
  668. while (receiveBufferSize > 0) {
  669. int tmpbs = receiveBufferSize;
  670. if (::setsockopt(sws.sock,SOL_SOCKET,SO_RCVBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
  671. break;
  672. receiveBufferSize -= 16384;
  673. }
  674. }
  675. if (sendBufferSize > 0) {
  676. while (sendBufferSize > 0) {
  677. int tmpbs = sendBufferSize;
  678. if (::setsockopt(sws.sock,SOL_SOCKET,SO_SNDBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
  679. break;
  680. sendBufferSize -= 16384;
  681. }
  682. }
  683. }
  684. /**
  685. * Attempt to send data to a stream socket (non-blocking)
  686. *
  687. * If -1 is returned, the socket should no longer be used as it is now
  688. * destroyed. If callCloseHandler is true, the close handler will be
  689. * called before the function returns.
  690. *
  691. * This can be used with TCP, Unix, or socket pair sockets.
  692. *
  693. * @param sock An open stream socket (other socket types will fail)
  694. * @param data Data to send
  695. * @param len Length of data
  696. * @param callCloseHandler If true, call close handler on socket closing failure condition (default: true)
  697. * @return Number of bytes actually sent or -1 on fatal error (socket closure)
  698. */
  699. inline long streamSend(PhySocket *sock,const void *data,unsigned long len,bool callCloseHandler = true)
  700. {
  701. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  702. #if defined(_WIN32) || defined(_WIN64)
  703. long n = (long)::send(sws.sock,reinterpret_cast<const char *>(data),len,0);
  704. if (n == SOCKET_ERROR) {
  705. switch(WSAGetLastError()) {
  706. case WSAEINTR:
  707. case WSAEWOULDBLOCK:
  708. return 0;
  709. default:
  710. this->close(sock,callCloseHandler);
  711. return -1;
  712. }
  713. }
  714. #else // not Windows
  715. long n = (long)::send(sws.sock,data,len,0);
  716. if (n < 0) {
  717. switch(errno) {
  718. #ifdef EAGAIN
  719. case EAGAIN:
  720. #endif
  721. #if defined(EWOULDBLOCK) && ( !defined(EAGAIN) || (EWOULDBLOCK != EAGAIN) )
  722. case EWOULDBLOCK:
  723. #endif
  724. #ifdef EINTR
  725. case EINTR:
  726. #endif
  727. return 0;
  728. default:
  729. this->close(sock,callCloseHandler);
  730. return -1;
  731. }
  732. }
  733. #endif // Windows or not
  734. return n;
  735. }
  736. #ifdef __UNIX_LIKE__
  737. /**
  738. * Attempt to send data to a Unix domain socket connection (non-blocking)
  739. *
  740. * If -1 is returned, the socket should no longer be used as it is now
  741. * destroyed. If callCloseHandler is true, the close handler will be
  742. * called before the function returns.
  743. *
  744. * @param sock An open Unix socket (other socket types will fail)
  745. * @param data Data to send
  746. * @param len Length of data
  747. * @param callCloseHandler If true, call close handler on socket closing failure condition (default: true)
  748. * @return Number of bytes actually sent or -1 on fatal error (socket closure)
  749. */
  750. inline long unixSend(PhySocket *sock,const void *data,unsigned long len,bool callCloseHandler = true)
  751. {
  752. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  753. long n = (long)::write(sws.sock,data,len);
  754. if (n < 0) {
  755. switch(errno) {
  756. #ifdef EAGAIN
  757. case EAGAIN:
  758. #endif
  759. #if defined(EWOULDBLOCK) && ( !defined(EAGAIN) || (EWOULDBLOCK != EAGAIN) )
  760. case EWOULDBLOCK:
  761. #endif
  762. #ifdef EINTR
  763. case EINTR:
  764. #endif
  765. return 0;
  766. default:
  767. this->close(sock,callCloseHandler);
  768. return -1;
  769. }
  770. }
  771. return n;
  772. }
  773. #endif // __UNIX_LIKE__
  774. /**
  775. * For streams, sets whether we want to be notified that the socket is writable
  776. *
  777. * This can be used with TCP, Unix, or socket pair sockets.
  778. *
  779. * Call whack() if this is being done from another thread and you want
  780. * it to take effect immediately. Otherwise it is only guaranteed to
  781. * take effect on the next poll().
  782. *
  783. * @param sock Stream connection socket
  784. * @param notifyWritable Want writable notifications?
  785. */
  786. inline void setNotifyWritable(PhySocket *sock,bool notifyWritable)
  787. {
  788. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  789. if (notifyWritable) {
  790. FD_SET(sws.sock,&_writefds);
  791. } else {
  792. FD_CLR(sws.sock,&_writefds);
  793. }
  794. }
  795. /**
  796. * Set whether we want to be notified that a socket is readable
  797. *
  798. * This is primarily for raw sockets added with wrapSocket(). It could be
  799. * used with others, but doing so would essentially lock them and prevent
  800. * data from being read from them until this is set to 'true' again.
  801. *
  802. * @param sock Socket to modify
  803. * @param notifyReadable True if socket should be monitored for readability
  804. */
  805. inline void setNotifyReadable(PhySocket *sock,bool notifyReadable)
  806. {
  807. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  808. if (notifyReadable) {
  809. FD_SET(sws.sock,&_readfds);
  810. } else {
  811. FD_CLR(sws.sock,&_readfds);
  812. }
  813. }
  814. /**
  815. * Wait for activity and handle one or more events
  816. *
  817. * Note that this is not guaranteed to wait up to 'timeout' even
  818. * if nothing happens, as whack() or other events such as signals
  819. * may cause premature termination.
  820. *
  821. * @param timeout Timeout in milliseconds or 0 for none (forever)
  822. */
  823. inline void poll(unsigned long timeout)
  824. {
  825. char buf[131072];
  826. struct sockaddr_storage ss;
  827. struct timeval tv;
  828. fd_set rfds,wfds,efds;
  829. memcpy(&rfds,&_readfds,sizeof(rfds));
  830. memcpy(&wfds,&_writefds,sizeof(wfds));
  831. #if defined(_WIN32) || defined(_WIN64)
  832. memcpy(&efds,&_exceptfds,sizeof(efds));
  833. #else
  834. FD_ZERO(&efds);
  835. #endif
  836. tv.tv_sec = (long)(timeout / 1000);
  837. tv.tv_usec = (long)((timeout % 1000) * 1000);
  838. if (::select((int)_nfds + 1,&rfds,&wfds,&efds,(timeout > 0) ? &tv : (struct timeval *)0) <= 0)
  839. return;
  840. if (FD_ISSET(_whackReceiveSocket,&rfds)) {
  841. char tmp[16];
  842. #if defined(_WIN32) || defined(_WIN64)
  843. ::recv(_whackReceiveSocket,tmp,16,0);
  844. #else
  845. ::read(_whackReceiveSocket,tmp,16);
  846. #endif
  847. }
  848. for(typename std::list<PhySocketImpl>::iterator s(_socks.begin());s!=_socks.end();) {
  849. switch (s->type) {
  850. case ZT_PHY_SOCKET_TCP_OUT_PENDING:
  851. #if defined(_WIN32) || defined(_WIN64)
  852. if (FD_ISSET(s->sock,&efds)) {
  853. this->close((PhySocket *)&(*s),true);
  854. } else // ... if
  855. #endif
  856. if (FD_ISSET(s->sock,&wfds)) {
  857. socklen_t slen = sizeof(ss);
  858. if (::getpeername(s->sock,(struct sockaddr *)&ss,&slen) != 0) {
  859. this->close((PhySocket *)&(*s),true);
  860. } else {
  861. s->type = ZT_PHY_SOCKET_TCP_OUT_CONNECTED;
  862. FD_SET(s->sock,&_readfds);
  863. FD_CLR(s->sock,&_writefds);
  864. #if defined(_WIN32) || defined(_WIN64)
  865. FD_CLR(s->sock,&_exceptfds);
  866. #endif
  867. try {
  868. _handler->phyOnTcpConnect((PhySocket *)&(*s),&(s->uptr),true);
  869. } catch ( ... ) {}
  870. }
  871. }
  872. break;
  873. case ZT_PHY_SOCKET_TCP_OUT_CONNECTED:
  874. case ZT_PHY_SOCKET_TCP_IN: {
  875. ZT_PHY_SOCKFD_TYPE sock = s->sock; // if closed, s->sock becomes invalid as s is no longer dereferencable
  876. if (FD_ISSET(sock,&rfds)) {
  877. long n = (long)::recv(sock,buf,sizeof(buf),0);
  878. if (n <= 0) {
  879. this->close((PhySocket *)&(*s),true);
  880. } else {
  881. try {
  882. _handler->phyOnTcpData((PhySocket *)&(*s),&(s->uptr),(void *)buf,(unsigned long)n);
  883. } catch ( ... ) {}
  884. }
  885. }
  886. if ((FD_ISSET(sock,&wfds))&&(FD_ISSET(sock,&_writefds))) {
  887. try {
  888. _handler->phyOnTcpWritable((PhySocket *)&(*s),&(s->uptr));
  889. } catch ( ... ) {}
  890. }
  891. } break;
  892. case ZT_PHY_SOCKET_TCP_LISTEN:
  893. if (FD_ISSET(s->sock,&rfds)) {
  894. memset(&ss,0,sizeof(ss));
  895. socklen_t slen = sizeof(ss);
  896. ZT_PHY_SOCKFD_TYPE newSock = ::accept(s->sock,(struct sockaddr *)&ss,&slen);
  897. if (ZT_PHY_SOCKFD_VALID(newSock)) {
  898. if (_socks.size() >= ZT_PHY_MAX_SOCKETS) {
  899. ZT_PHY_CLOSE_SOCKET(newSock);
  900. } else {
  901. #if defined(_WIN32) || defined(_WIN64)
  902. { BOOL f = (_noDelay ? TRUE : FALSE); setsockopt(newSock,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
  903. { u_long iMode=1; ioctlsocket(newSock,FIONBIO,&iMode); }
  904. #else
  905. { int f = (_noDelay ? 1 : 0); setsockopt(newSock,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
  906. fcntl(newSock,F_SETFL,O_NONBLOCK);
  907. #endif
  908. _socks.push_back(PhySocketImpl());
  909. PhySocketImpl &sws = _socks.back();
  910. FD_SET(newSock,&_readfds);
  911. if ((long)newSock > _nfds)
  912. _nfds = (long)newSock;
  913. sws.type = ZT_PHY_SOCKET_TCP_IN;
  914. sws.sock = newSock;
  915. sws.uptr = (void *)0;
  916. memcpy(&(sws.saddr),&ss,sizeof(struct sockaddr_storage));
  917. try {
  918. _handler->phyOnTcpAccept((PhySocket *)&(*s),(PhySocket *)&(_socks.back()),&(s->uptr),&(sws.uptr),(const struct sockaddr *)&(sws.saddr));
  919. } catch ( ... ) {}
  920. }
  921. }
  922. }
  923. break;
  924. case ZT_PHY_SOCKET_UDP:
  925. if (FD_ISSET(s->sock, &rfds)) {
  926. #if (defined(__linux__) || defined(linux) || defined(__linux)) && defined(MSG_WAITFORONE)
  927. #define RECVMMSG_WINDOW_SIZE 128
  928. #define RECVMMSG_BUF_SIZE 1500
  929. iovec iovs[RECVMMSG_WINDOW_SIZE];
  930. uint8_t bufs[RECVMMSG_WINDOW_SIZE][RECVMMSG_BUF_SIZE];
  931. sockaddr_storage addrs[RECVMMSG_WINDOW_SIZE];
  932. memset(addrs, 0, sizeof(addrs));
  933. mmsghdr mm[RECVMMSG_WINDOW_SIZE];
  934. memset(mm, 0, sizeof(mm));
  935. for (int i = 0; i < RECVMMSG_WINDOW_SIZE; ++i) {
  936. iovs[i].iov_base = (void*)bufs[i];
  937. iovs[i].iov_len = RECVMMSG_BUF_SIZE;
  938. mm[i].msg_hdr.msg_name = (void*)&(addrs[i]);
  939. mm[i].msg_hdr.msg_iov = &(iovs[i]);
  940. mm[i].msg_hdr.msg_iovlen = 1;
  941. }
  942. for (int k = 0; k < 1024; ++k) {
  943. for (int i = 0; i < RECVMMSG_WINDOW_SIZE; ++i) {
  944. mm[i].msg_hdr.msg_namelen = sizeof(sockaddr_storage);
  945. mm[i].msg_len = 0;
  946. }
  947. int received_count = recvmmsg(s->sock, mm, RECVMMSG_WINDOW_SIZE, MSG_WAITFORONE, nullptr);
  948. if (received_count > 0) {
  949. for (int i = 0; i < received_count; ++i) {
  950. long n = (long)mm[i].msg_len;
  951. if (n > 0) {
  952. try {
  953. _handler->phyOnDatagram((PhySocket*)&(*s), &(s->uptr), (const struct sockaddr*)&(s->saddr), (const struct sockaddr*)&(addrs[i]), bufs[i], (unsigned long)n);
  954. }
  955. catch (...) {
  956. }
  957. }
  958. }
  959. }
  960. else {
  961. break;
  962. }
  963. }
  964. #else
  965. for (int k = 0; k < 1024; ++k) {
  966. memset(&ss, 0, sizeof(ss));
  967. socklen_t slen = sizeof(ss);
  968. long n = (long)::recvfrom(s->sock, buf, sizeof(buf), 0, (struct sockaddr*)&ss, &slen);
  969. if (n > 0) {
  970. try {
  971. _handler->phyOnDatagram((PhySocket*)&(*s), &(s->uptr), (const struct sockaddr*)&(s->saddr), (const struct sockaddr*)&ss, (void*)buf, (unsigned long)n);
  972. }
  973. catch (...) {
  974. }
  975. }
  976. else if (n < 0)
  977. break;
  978. }
  979. #endif
  980. }
  981. break;
  982. case ZT_PHY_SOCKET_UNIX_IN: {
  983. #ifdef __UNIX_LIKE__
  984. ZT_PHY_SOCKFD_TYPE sock = s->sock; // if closed, s->sock becomes invalid as s is no longer dereferencable
  985. if ((FD_ISSET(sock,&wfds))&&(FD_ISSET(sock,&_writefds))) {
  986. try {
  987. _handler->phyOnUnixWritable((PhySocket *)&(*s),&(s->uptr));
  988. } catch ( ... ) {}
  989. }
  990. if (FD_ISSET(sock,&rfds)) {
  991. long n = (long)::read(sock,buf,sizeof(buf));
  992. if (n <= 0) {
  993. this->close((PhySocket *)&(*s),true);
  994. } else {
  995. try {
  996. _handler->phyOnUnixData((PhySocket *)&(*s),&(s->uptr),(void *)buf,(unsigned long)n);
  997. } catch ( ... ) {}
  998. }
  999. }
  1000. #endif // __UNIX_LIKE__
  1001. } break;
  1002. case ZT_PHY_SOCKET_UNIX_LISTEN:
  1003. #ifdef __UNIX_LIKE__
  1004. if (FD_ISSET(s->sock,&rfds)) {
  1005. memset(&ss,0,sizeof(ss));
  1006. socklen_t slen = sizeof(ss);
  1007. ZT_PHY_SOCKFD_TYPE newSock = ::accept(s->sock,(struct sockaddr *)&ss,&slen);
  1008. if (ZT_PHY_SOCKFD_VALID(newSock)) {
  1009. if (_socks.size() >= ZT_PHY_MAX_SOCKETS) {
  1010. ZT_PHY_CLOSE_SOCKET(newSock);
  1011. } else {
  1012. fcntl(newSock,F_SETFL,O_NONBLOCK);
  1013. _socks.push_back(PhySocketImpl());
  1014. PhySocketImpl &sws = _socks.back();
  1015. FD_SET(newSock,&_readfds);
  1016. if ((long)newSock > _nfds)
  1017. _nfds = (long)newSock;
  1018. sws.type = ZT_PHY_SOCKET_UNIX_IN;
  1019. sws.sock = newSock;
  1020. sws.uptr = (void *)0;
  1021. memcpy(&(sws.saddr),&ss,sizeof(struct sockaddr_storage));
  1022. try {
  1023. //_handler->phyOnUnixAccept((PhySocket *)&(*s),(PhySocket *)&(_socks.back()),&(s->uptr),&(sws.uptr));
  1024. } catch ( ... ) {}
  1025. }
  1026. }
  1027. }
  1028. #endif // __UNIX_LIKE__
  1029. break;
  1030. case ZT_PHY_SOCKET_FD: {
  1031. ZT_PHY_SOCKFD_TYPE sock = s->sock;
  1032. const bool readable = ((FD_ISSET(sock,&rfds))&&(FD_ISSET(sock,&_readfds)));
  1033. const bool writable = ((FD_ISSET(sock,&wfds))&&(FD_ISSET(sock,&_writefds)));
  1034. if ((readable)||(writable)) {
  1035. try {
  1036. //_handler->phyOnFileDescriptorActivity((PhySocket *)&(*s),&(s->uptr),readable,writable);
  1037. } catch ( ... ) {}
  1038. }
  1039. } break;
  1040. default:
  1041. break;
  1042. }
  1043. if (s->type == ZT_PHY_SOCKET_CLOSED)
  1044. _socks.erase(s++);
  1045. else ++s;
  1046. }
  1047. }
  1048. /**
  1049. * @param sock Socket to close
  1050. * @param callHandlers If true, call handlers for TCP connect (success: false) or close (default: true)
  1051. */
  1052. inline void close(PhySocket *sock,bool callHandlers = true)
  1053. {
  1054. if (!sock)
  1055. return;
  1056. PhySocketImpl &sws = *(reinterpret_cast<PhySocketImpl *>(sock));
  1057. if (sws.type == ZT_PHY_SOCKET_CLOSED)
  1058. return;
  1059. FD_CLR(sws.sock,&_readfds);
  1060. FD_CLR(sws.sock,&_writefds);
  1061. #if defined(_WIN32) || defined(_WIN64)
  1062. FD_CLR(sws.sock,&_exceptfds);
  1063. #endif
  1064. if (sws.type != ZT_PHY_SOCKET_FD)
  1065. ZT_PHY_CLOSE_SOCKET(sws.sock);
  1066. #ifdef __UNIX_LIKE__
  1067. if (sws.type == ZT_PHY_SOCKET_UNIX_LISTEN)
  1068. ::unlink(((struct sockaddr_un *)(&(sws.saddr)))->sun_path);
  1069. #endif // __UNIX_LIKE__
  1070. if (callHandlers) {
  1071. switch(sws.type) {
  1072. case ZT_PHY_SOCKET_TCP_OUT_PENDING:
  1073. try {
  1074. _handler->phyOnTcpConnect(sock,&(sws.uptr),false);
  1075. } catch ( ... ) {}
  1076. break;
  1077. case ZT_PHY_SOCKET_TCP_OUT_CONNECTED:
  1078. case ZT_PHY_SOCKET_TCP_IN:
  1079. try {
  1080. _handler->phyOnTcpClose(sock,&(sws.uptr));
  1081. } catch ( ... ) {}
  1082. break;
  1083. case ZT_PHY_SOCKET_UNIX_IN:
  1084. #ifdef __UNIX_LIKE__
  1085. try {
  1086. _handler->phyOnUnixClose(sock,&(sws.uptr));
  1087. } catch ( ... ) {}
  1088. #endif // __UNIX_LIKE__
  1089. break;
  1090. default:
  1091. break;
  1092. }
  1093. }
  1094. // Causes entry to be deleted from list in poll(), ignored elsewhere
  1095. sws.type = ZT_PHY_SOCKET_CLOSED;
  1096. if ((long)sws.sock >= (long)_nfds) {
  1097. long nfds = (long)_whackSendSocket;
  1098. if ((long)_whackReceiveSocket > nfds)
  1099. nfds = (long)_whackReceiveSocket;
  1100. for(typename std::list<PhySocketImpl>::iterator s(_socks.begin());s!=_socks.end();++s) {
  1101. if ((s->type != ZT_PHY_SOCKET_CLOSED)&&((long)s->sock > nfds))
  1102. nfds = (long)s->sock;
  1103. }
  1104. _nfds = nfds;
  1105. }
  1106. }
  1107. };
  1108. } // namespace ZeroTier
  1109. #endif