Phy.hpp 36 KB

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