ExternalVendorCode/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2024-12-24 15:16:40 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
9a5f6f020f
ZeroTierOne/osnet/NativeSocketManager.cpp
Adam Ierymenko 2436e22f46 More work on abstracting socket manager.
2014-10-21 15:18:50 -07:00

990 lines
27 KiB
C++
Raw Blame History

/*
* ZeroTier One - Global Peer to Peer Ethernet
* Copyright (C) 2011-2014 ZeroTier Networks LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* --
*
* ZeroTier may be used and distributed under the terms of the GPLv3, which
* are available at: http://www.gnu.org/licenses/gpl-3.0.html
*
* If you would like to embed ZeroTier into a commercial application or
* redistribute it in a modified binary form, please contact ZeroTier Networks
* LLC. Start here: http://www.zerotier.com/
*/
/* Native SocketManager for Windows and Unix */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <time.h>
#include <sys/types.h>
#include <algorithm>
#include "../node/Constants.hpp"
#include "NativeSocketManager.hpp"
#ifndef __WINDOWS__
#include <errno.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <signal.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#endif // !__WINDOWS__
// Uncomment to turn off TCP Nagle
//#define ZT_TCP_NODELAY
// Allow us to use the same value on Windows and *nix
#ifndef INVALID_SOCKET
#define INVALID_SOCKET (-1)
#endif
#ifdef __WINDOWS__
#define CLOSE_SOCKET(s) ::closesocket(s)
#else
#define CLOSE_SOCKET(s) ::close(s)
#endif
namespace ZeroTier {
//////////////////////////////////////////////////////////////////////////////
// Socket implementations
//////////////////////////////////////////////////////////////////////////////
class NativeSocket : public Socket
{
public:
#ifdef __WINDOWS__
NativeSocket(const Type &t,SOCKET s) : Socket(t),_sock(s) {}
SOCKET _sock;
#else
NativeSocket(const Type &t,int s) : Socket(t),_sock(s) {}
int _sock;
#endif
};
/**
* Native UDP socket
*/
class NativeUdpSocket : public NativeSocket
{
public:
#ifdef __WINDOWS__
NativeUdpSocket(Type t,SOCKET s) : NativeSocket(t,s) {}
#else
NativeUdpSocket(Type t,int s) : NativeSocket(t,s) {}
#endif
virtual ~NativeUdpSocket()
{
#ifdef __WINDOWS__
::closesocket(_sock);
#else
::close(_sock);
#endif
}
virtual bool send(const InetAddress &to,const void *msg,unsigned int msglen)
{
if (to.isV6()) {
#ifdef __WINDOWS__
return ((int)sendto(_sock,(const char *)msg,msglen,0,to.saddr(),to.saddrLen()) == (int)msglen);
#else
return ((int)sendto(_sock,msg,msglen,0,to.saddr(),to.saddrLen()) == (int)msglen);
#endif
} else {
#ifdef __WINDOWS__
return ((int)sendto(_sock,(const char *)msg,msglen,0,to.saddr(),to.saddrLen()) == (int)msglen);
#else
return ((int)sendto(_sock,msg,msglen,0,to.saddr(),to.saddrLen()) == (int)msglen);
#endif
}
}
inline bool notifyAvailableForRead(const SharedPtr<Socket> &self,NativeSocketManager *sm,void (*handler)(const SharedPtr<Socket> &,void *,const InetAddress &,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> &),void *arg)
{
Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> buf;
InetAddress from;
socklen_t salen = from.saddrSpaceLen();
int n = (int)recvfrom(_sock,(char *)(buf.data()),ZT_SOCKET_MAX_MESSAGE_LEN,0,from.saddr(),&salen);
if (n > 0) {
buf.setSize((unsigned int)n);
try {
handler(self,arg,from,buf);
} catch ( ... ) {} // handlers should not throw
}
return true;
}
inline bool notifyAvailableForWrite(const SharedPtr<Socket> &self,NativeSocketManager *sm)
{
return true;
}
};
/**
* A TCP socket encapsulating ZeroTier packets over a TCP stream connection
*
* This implements a simple packet encapsulation that is designed to look like
* a TLS connection. It's not a TLS connection, but it sends TLS format record
* headers. It could be extended in the future to implement a fake TLS
* handshake.
*
* At the moment, each packet is just made to look like TLS application data:
* <[1] TLS content type> - currently 0x17 for "application data"
* <[1] TLS major version> - currently 0x03 for TLS 1.2
* <[1] TLS minor version> - currently 0x03 for TLS 1.2
* <[2] payload length> - 16-bit length of payload in bytes
* <[...] payload> - Message payload
*
* The primary purpose of TCP sockets is to work over ports like HTTPS(443),
* allowing users behind particularly fascist firewalls to at least reach
* ZeroTier's supernodes. UDP is the preferred method of communication as
* encapsulating L2 and L3 protocols over TCP is inherently inefficient
* due to double-ACKs. So TCP is only used as a fallback.
*/
class NativeTcpSocket : public NativeSocket
{
public:
#ifdef __WINDOWS__
NativeTcpSocket(NativeSocketManager *sm,SOCKET s,Socket::Type t,bool c,const InetAddress &r) :
#else
NativeTcpSocket(NativeSocketManager *sm,int s,Socket::Type t,bool c,const InetAddress &r) :
#endif
NativeSocket(t,s),
_lastActivity(Utils::now()),
_sm(sm),
_inptr(0),
_outptr(0),
_connecting(c),
_remote(r) {}
virtual ~NativeTcpSocket()
{
#ifdef __WINDOWS__
::closesocket(_sock);
#else
::close(_sock);
#endif
}
virtual bool send(const InetAddress &to,const void *msg,unsigned int msglen)
{
if (msglen > ZT_SOCKET_MAX_MESSAGE_LEN)
return false; // message too big
if (!msglen)
return true; // sanity check
Mutex::Lock _l(_writeLock);
bool writeInProgress = ((_outptr != 0)||(_connecting));
if ((_outptr + 5 + msglen) > (unsigned int)sizeof(_outbuf))
return false;
_outbuf[_outptr++] = 0x17; // look like TLS data
_outbuf[_outptr++] = 0x03;
_outbuf[_outptr++] = 0x03; // look like TLS 1.2
_outbuf[_outptr++] = (unsigned char)((msglen >> 8) & 0xff);
_outbuf[_outptr++] = (unsigned char)(msglen & 0xff);
for(unsigned int i=0;i<msglen;++i)
_outbuf[_outptr++] = ((const unsigned char *)msg)[i];
if (!writeInProgress) {
// If no output was enqueued before this, try to send() it and then
// start a queued write if any remains after that.
int n = (int)::send(_sock,(const char *)_outbuf,_outptr,0);
if (n > 0)
memmove(_outbuf,_outbuf + (unsigned int)n,_outptr -= (unsigned int)n);
if (_outptr) {
_sm->_startNotifyWrite(this);
_sm->whack();
}
} // else just leave in _outbuf[] to get written when stream is available for write
return true;
}
inline bool notifyAvailableForRead(const SharedPtr<Socket> &self,NativeSocketManager *sm,void (*handler)(const SharedPtr<Socket> &,void *,const InetAddress &,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> &),void *arg)
{
unsigned char buf[65536];
int n = (int)::recv(_sock,(char *)buf,sizeof(buf),0);
if (n <= 0)
return false; // read error, stream probably closed
unsigned int p = _inptr,pl = 0;
for(int k=0;k<n;++k) {
_inbuf[p++] = buf[k];
if (p >= (int)sizeof(_inbuf))
return false; // read overrun, packet too large or invalid
if ((!pl)&&(p >= 5)) {
if (_inbuf[0] == 0x17) {
// fake TLS data frame, next two bytes are TLS version and are ignored
pl = (((unsigned int)_inbuf[3] << 8) | (unsigned int)_inbuf[4]) + 5;
} else return false; // in the future we may support fake TLS handshakes
}
if ((pl)&&(p >= pl)) {
Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> data(_inbuf + 5,pl - 5);
memmove(_inbuf,_inbuf + pl,p -= pl);
try {
handler(self,arg,_remote,data);
} catch ( ... ) {} // handlers should not throw
pl = 0;
}
}
_inptr = p;
return true;
}
inline bool notifyAvailableForWrite(const SharedPtr<Socket> &self,NativeSocketManager *sm)
{
Mutex::Lock _l(_writeLock);
if (_connecting)
_connecting = false;
if (_outptr) {
int n = (int)::send(_sock,(const char *)_outbuf,_outptr,0);
#ifdef __WINDOWS__
if (n == SOCKET_ERROR) {
switch(WSAGetLastError()) {
case WSAEINTR:
case WSAEWOULDBLOCK:
break;
default:
return false;
}
#else
if (n <= 0) {
switch(errno) {
#ifdef EAGAIN
case EAGAIN:
#endif
#if defined(EWOULDBLOCK) && ( !defined(EAGAIN) || (EWOULDBLOCK != EAGAIN) )
case EWOULDBLOCK:
#endif
#ifdef EINTR
case EINTR:
#endif
break;
default:
return false;
}
#endif
} else memmove(_outbuf,_outbuf + (unsigned int)n,_outptr -= (unsigned int)n);
}
if (!_outptr)
sm->_stopNotifyWrite(this);
return true;
}
unsigned char _inbuf[ZT_SOCKET_MAX_MESSAGE_LEN];
unsigned char _outbuf[ZT_SOCKET_MAX_MESSAGE_LEN * 4];
uint64_t _lastActivity; // updated whenever data is received, checked directly by SocketManager for stale TCP cleanup
NativeSocketManager *_sm;
unsigned int _inptr;
unsigned int _outptr;
bool _connecting; // manipulated directly by SocketManager, true if connect() is in progress
InetAddress _remote;
Mutex _writeLock;
};
//////////////////////////////////////////////////////////////////////////////
#ifdef __WINDOWS__
// hack copied from StackOverflow, behaves a bit like pipe() on *nix systems
static inline void winPipeHack(SOCKET fds[2])
{
struct sockaddr_in inaddr;
struct sockaddr addr;
SOCKET lst=::socket(AF_INET, SOCK_STREAM,IPPROTO_TCP);
memset(&inaddr, 0, sizeof(inaddr));
memset(&addr, 0, sizeof(addr));
inaddr.sin_family = AF_INET;
inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
inaddr.sin_port = 0;
int yes=1;
setsockopt(lst,SOL_SOCKET,SO_REUSEADDR,(char*)&yes,sizeof(yes));
bind(lst,(struct sockaddr *)&inaddr,sizeof(inaddr));
listen(lst,1);
int len=sizeof(inaddr);
getsockname(lst, &addr,&len);
fds[0]=::socket(AF_INET, SOCK_STREAM,0);
connect(fds[0],&addr,len);
fds[1]=accept(lst,0,0);
closesocket(lst);
}
#endif
NativeSocketManager::NativeSocketManager(int localUdpPort,int localTcpPort) :
SocketManager(),
_whackSendPipe(INVALID_SOCKET),
_whackReceivePipe(INVALID_SOCKET),
_tcpV4ListenSocket(INVALID_SOCKET),
_tcpV6ListenSocket(INVALID_SOCKET),
_nfds(0)
{
FD_ZERO(&_readfds);
FD_ZERO(&_writefds);
// Create a pipe or socket pair that can be used to interrupt select()
#ifdef __WINDOWS__
{
SOCKET tmps[2] = { INVALID_SOCKET,INVALID_SOCKET };
winPipeHack(tmps);
_whackSendPipe = tmps[0];
_whackReceivePipe = tmps[1];
u_long iMode=1;
ioctlsocket(tmps[1],FIONBIO,&iMode);
}
#else
{
int tmpfds[2];
if (::pipe(tmpfds))
throw std::runtime_error("pipe() failed");
_whackSendPipe = tmpfds[1];
_whackReceivePipe = tmpfds[0];
fcntl(_whackReceivePipe,F_SETFL,O_NONBLOCK);
}
#endif
FD_SET(_whackReceivePipe,&_readfds);
if (localTcpPort > 0) {
if (localTcpPort > 0xffff) {
_closeSockets();
throw std::runtime_error("invalid local TCP port number");
}
{ // bind TCP IPv6
_tcpV6ListenSocket = ::socket(AF_INET6,SOCK_STREAM,0);
#ifdef __WINDOWS__
if (_tcpV6ListenSocket != INVALID_SOCKET) {
{
BOOL f;
f = TRUE; ::setsockopt(_tcpV6ListenSocket,IPPROTO_IPV6,IPV6_V6ONLY,(const char *)&f,sizeof(f));
f = TRUE; ::setsockopt(_tcpV6ListenSocket,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
u_long iMode=1;
ioctlsocket(_tcpV6ListenSocket,FIONBIO,&iMode);
}
#else
if (_tcpV6ListenSocket > 0) {
{
int f;
f = 1; ::setsockopt(_tcpV6ListenSocket,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f));
f = 1; ::setsockopt(_tcpV6ListenSocket,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
fcntl(_tcpV6ListenSocket,F_SETFL,O_NONBLOCK);
}
#endif // __WINDOWS__ / not __WINDOWS__
struct sockaddr_in6 sin6;
memset(&sin6,0,sizeof(sin6));
sin6.sin6_family = AF_INET6;
sin6.sin6_port = htons(localTcpPort);
memcpy(&(sin6.sin6_addr),&in6addr_any,sizeof(struct in6_addr));
if (::bind(_tcpV6ListenSocket,(const struct sockaddr *)&sin6,sizeof(sin6))) {
_closeSockets();
throw std::runtime_error("unable to bind to local TCP port");
}
if (::listen(_tcpV6ListenSocket,16)) {
_closeSockets();
throw std::runtime_error("listen() failed");
}
FD_SET(_tcpV6ListenSocket,&_readfds);
}
}
{ // bind TCP IPv4
_tcpV4ListenSocket = ::socket(AF_INET,SOCK_STREAM,0);
#ifdef __WINDOWS__
if (_tcpV4ListenSocket == INVALID_SOCKET) {
#else
if (_tcpV4ListenSocket <= 0) {
#endif
_closeSockets();
throw std::runtime_error("unable to create IPv4 SOCK_STREAM socket");
}
#ifdef __WINDOWS__
{
BOOL f = TRUE; ::setsockopt(_tcpV4ListenSocket,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
u_long iMode=1;
ioctlsocket(_tcpV4ListenSocket,FIONBIO,&iMode);
}
#else
{
int f = 1; ::setsockopt(_tcpV4ListenSocket,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
fcntl(_tcpV4ListenSocket,F_SETFL,O_NONBLOCK);
}
#endif
struct sockaddr_in sin4;
memset(&sin4,0,sizeof(sin4));
sin4.sin_family = AF_INET;
sin4.sin_port = htons(localTcpPort);
sin4.sin_addr.s_addr = INADDR_ANY;
if (::bind(_tcpV4ListenSocket,(const struct sockaddr *)&sin4,sizeof(sin4))) {
_closeSockets();
throw std::runtime_error("unable to bind to local TCP port");
}
if (::listen(_tcpV4ListenSocket,16)) {
_closeSockets();
throw std::runtime_error("listen() failed");
}
FD_SET(_tcpV4ListenSocket,&_readfds);
}
}
if (localUdpPort > 0) {
if (localUdpPort > 0xffff) {
_closeSockets();
throw std::runtime_error("invalid local UDP port number");
}
{ // bind UDP IPv6
#ifdef __WINDOWS__
SOCKET s = ::socket(AF_INET6,SOCK_DGRAM,0);
if (s != INVALID_SOCKET) {
#else
int s = ::socket(AF_INET6,SOCK_DGRAM,0);
if (s > 0) {
#endif
{
int bs = 1048576;
while (bs >= 65536) {
int tmpbs = bs;
if (setsockopt(s,SOL_SOCKET,SO_RCVBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
break;
bs -= 16384;
}
bs = 1048576;
while (bs >= 65536) {
int tmpbs = bs;
if (setsockopt(s,SOL_SOCKET,SO_SNDBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
break;
bs -= 16384;
}
#ifdef __WINDOWS__
BOOL f;
f = TRUE; setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(const char *)&f,sizeof(f));
f = FALSE; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
f = FALSE; setsockopt(s,IPPROTO_IPV6,IPV6_DONTFRAG,(const char *)&f,sizeof(f));
f = TRUE; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(const char *)&f,sizeof(f));
#else
int f;
f = 1; setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,(void *)&f,sizeof(f));
f = 0; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
f = 1; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(void *)&f,sizeof(f));
#ifdef IP_DONTFRAG
f = 0; setsockopt(s,IPPROTO_IP,IP_DONTFRAG,&f,sizeof(f));
#endif
#ifdef IP_MTU_DISCOVER
f = 0; setsockopt(s,IPPROTO_IP,IP_MTU_DISCOVER,&f,sizeof(f));
#endif
#ifdef IPV6_MTU_DISCOVER
f = 0; setsockopt(s,IPPROTO_IPV6,IPV6_MTU_DISCOVER,&f,sizeof(f));
#endif
#endif
}
struct sockaddr_in6 sin6;
memset(&sin6,0,sizeof(sin6));
sin6.sin6_family = AF_INET6;
sin6.sin6_port = htons(localUdpPort);
memcpy(&(sin6.sin6_addr),&in6addr_any,sizeof(struct in6_addr));
if (::bind(s,(const struct sockaddr *)&sin6,sizeof(sin6))) {
CLOSE_SOCKET(s);
_closeSockets();
throw std::runtime_error("unable to bind to port");
}
_udpV6Socket = SharedPtr<Socket>(new NativeUdpSocket(Socket::ZT_SOCKET_TYPE_UDP_V6,s));
#ifdef __WINDOWS__
u_long iMode=1;
ioctlsocket(s,FIONBIO,&iMode);
#else
fcntl(s,F_SETFL,O_NONBLOCK);
#endif
FD_SET(s,&_readfds);
}
}
{ // bind UDP IPv4
#ifdef __WINDOWS__
SOCKET s = ::socket(AF_INET,SOCK_DGRAM,0);
if (s == INVALID_SOCKET) {
_closeSockets();
throw std::runtime_error("unable to create IPv4 SOCK_DGRAM socket");
}
#else
int s = ::socket(AF_INET,SOCK_DGRAM,0);
if (s <= 0) {
_closeSockets();
throw std::runtime_error("unable to create IPv4 SOCK_DGRAM socket");
}
#endif
{
int bs = 1048576;
while (bs >= 65536) {
int tmpbs = bs;
if (setsockopt(s,SOL_SOCKET,SO_RCVBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
break;
bs -= 16384;
}
bs = 1048576;
while (bs >= 65536) {
int tmpbs = bs;
if (setsockopt(s,SOL_SOCKET,SO_SNDBUF,(const char *)&tmpbs,sizeof(tmpbs)) == 0)
break;
bs -= 16384;
}
#ifdef __WINDOWS__
BOOL f;
f = FALSE; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(const char *)&f,sizeof(f));
f = FALSE; setsockopt(s,IPPROTO_IP,IP_DONTFRAGMENT,(const char *)&f,sizeof(f));
f = TRUE; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(const char *)&f,sizeof(f));
#else
int f;
f = 0; setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(void *)&f,sizeof(f));
f = 1; setsockopt(s,SOL_SOCKET,SO_BROADCAST,(void *)&f,sizeof(f));
#ifdef IP_DONTFRAG
f = 0; setsockopt(s,IPPROTO_IP,IP_DONTFRAG,&f,sizeof(f));
#endif
#ifdef IP_MTU_DISCOVER
f = 0; setsockopt(s,IPPROTO_IP,IP_MTU_DISCOVER,&f,sizeof(f));
#endif
#endif
}
struct sockaddr_in sin4;
memset(&sin4,0,sizeof(sin4));
sin4.sin_family = AF_INET;
sin4.sin_port = htons(localUdpPort);
sin4.sin_addr.s_addr = INADDR_ANY;
if (::bind(s,(const struct sockaddr *)&sin4,sizeof(sin4))) {
CLOSE_SOCKET(s);
_closeSockets();
throw std::runtime_error("unable to bind to port");
}
_udpV4Socket = SharedPtr<Socket>(new NativeUdpSocket(Socket::ZT_SOCKET_TYPE_UDP_V4,s));
#ifdef __WINDOWS__
u_long iMode=1;
ioctlsocket(s,FIONBIO,&iMode);
#else
fcntl(s,F_SETFL,O_NONBLOCK);
#endif
FD_SET(s,&_readfds);
}
}
_updateNfds();
}
NativeSocketManager::~NativeSocketManager()
{
Mutex::Lock _l(_pollLock);
_closeSockets();
}
bool NativeSocketManager::send(const InetAddress &to,bool tcp,bool autoConnectTcp,const void *msg,unsigned int msglen)
{
if (tcp) {
SharedPtr<Socket> ts;
{
Mutex::Lock _l(_tcpSockets_m);
std::map< InetAddress,SharedPtr<Socket> >::iterator opents(_tcpSockets.find(to));
if (opents != _tcpSockets.end())
ts = opents->second;
}
if (ts)
return ts->send(to,msg,msglen);
if (!autoConnectTcp)
return false;
#ifdef __WINDOWS__
SOCKET s = ::socket(to.isV4() ? AF_INET : AF_INET6,SOCK_STREAM,0);
if (s == INVALID_SOCKET)
return false;
{ u_long iMode=1; ioctlsocket(s,FIONBIO,&iMode); }
#ifdef ZT_TCP_NODELAY
{ BOOL f = TRUE; setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#else
int s = ::socket(to.isV4() ? AF_INET : AF_INET6,SOCK_STREAM,0);
if (s <= 0)
return false;
if (s >= FD_SETSIZE) {
::close(s);
return false;
}
fcntl(s,F_SETFL,O_NONBLOCK);
#ifdef ZT_TCP_NODELAY
{ int f = 1; setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#endif
bool connecting = false;
if (::connect(s,to.saddr(),to.saddrLen())) {
#ifdef __WINDOWS__
if (WSAGetLastError() != WSAEWOULDBLOCK) {
#else
if (errno != EINPROGRESS) {
#endif
CLOSE_SOCKET(s);
return false;
} else connecting = true;
}
ts = SharedPtr<Socket>(new NativeTcpSocket(this,s,Socket::ZT_SOCKET_TYPE_TCP_OUT,connecting,to));
if (!ts->send(to,msg,msglen)) {
_fdSetLock.lock();
FD_CLR(s,&_readfds);
FD_CLR(s,&_writefds);
_fdSetLock.unlock();
return false;
}
{
Mutex::Lock _l(_tcpSockets_m);
_tcpSockets[to] = ts;
}
_fdSetLock.lock();
FD_SET(s,&_readfds);
if (connecting)
FD_SET(s,&_writefds);
_fdSetLock.unlock();
_updateNfds();
whack();
return true;
} else if (to.isV4()) {
if (_udpV4Socket)
return _udpV4Socket->send(to,msg,msglen);
} else if (to.isV6()) {
if (_udpV6Socket)
return _udpV6Socket->send(to,msg,msglen);
}
return false;
}
void NativeSocketManager::poll(unsigned long timeout,void (*handler)(const SharedPtr<Socket> &,void *,const InetAddress &,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> &),void *arg)
{
fd_set rfds,wfds,efds;
struct timeval tv;
std::vector< SharedPtr<Socket> > ts;
#ifdef __WINDOWS__
SOCKET sockfd;
#else
int sockfd;
#endif
Mutex::Lock _l(_pollLock);
_fdSetLock.lock();
memcpy(&rfds,&_readfds,sizeof(rfds));
memcpy(&wfds,&_writefds,sizeof(wfds));
_fdSetLock.unlock();
FD_ZERO(&efds);
#ifdef __WINDOWS__
// Windows signals failed connects in exceptfds
{
Mutex::Lock _l2(_tcpSockets_m);
for(std::map< InetAddress,SharedPtr<Socket> >::iterator s(_tcpSockets.begin());s!=_tcpSockets.end();++s) {
if (((TcpSocket *)s->second.ptr())->_connecting)
FD_SET(s->second->_sock,&efds);
}
}
#endif
tv.tv_sec = (long)(timeout / 1000);
tv.tv_usec = (long)((timeout % 1000) * 1000);
select(_nfds + 1,&rfds,&wfds,&efds,(timeout > 0) ? &tv : (struct timeval *)0);
if (FD_ISSET(_whackReceivePipe,&rfds)) {
char tmp[16];
#ifdef __WINDOWS__
::recv(_whackReceivePipe,tmp,16,0);
#else
::read(_whackReceivePipe,tmp,16);
#endif
}
if ((_tcpV4ListenSocket != INVALID_SOCKET)&&(FD_ISSET(_tcpV4ListenSocket,&rfds))) {
struct sockaddr_in from;
socklen_t fromlen = sizeof(from);
sockfd = accept(_tcpV4ListenSocket,(struct sockaddr *)&from,&fromlen);
#ifdef __WINDOWS__
if (sockfd != INVALID_SOCKET) {
#else
if (sockfd > 0) {
if (sockfd < FD_SETSIZE) {
#endif
InetAddress fromia((const struct sockaddr *)&from);
Mutex::Lock _l2(_tcpSockets_m);
try {
_tcpSockets[fromia] = SharedPtr<Socket>(new NativeTcpSocket(this,sockfd,Socket::ZT_SOCKET_TYPE_TCP_IN,false,fromia));
#ifdef __WINDOWS__
{ u_long iMode=1; ioctlsocket(sockfd,FIONBIO,&iMode); }
#ifdef ZT_TCP_NODELAY
{ BOOL f = TRUE; setsockopt(sockfd,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#else
fcntl(sockfd,F_SETFL,O_NONBLOCK);
#ifdef ZT_TCP_NODELAY
{ int f = 1; setsockopt(sockfd,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#endif
_fdSetLock.lock();
FD_SET(sockfd,&_readfds);
_fdSetLock.unlock();
if ((int)sockfd > (int)_nfds)
_nfds = (int)sockfd;
} catch ( ... ) {
CLOSE_SOCKET(sockfd);
}
#ifndef __WINDOWS__
} else {
CLOSE_SOCKET(sockfd);
}
#endif
}
}
if ((_tcpV6ListenSocket != INVALID_SOCKET)&&(FD_ISSET(_tcpV6ListenSocket,&rfds))) {
struct sockaddr_in6 from;
socklen_t fromlen = sizeof(from);
sockfd = accept(_tcpV6ListenSocket,(struct sockaddr *)&from,&fromlen);
#ifdef __WINDOWS__
if (sockfd != INVALID_SOCKET) {
#else
if (sockfd > 0) {
if (sockfd < FD_SETSIZE) {
#endif
InetAddress fromia((const struct sockaddr *)&from);
Mutex::Lock _l2(_tcpSockets_m);
try {
_tcpSockets[fromia] = SharedPtr<Socket>(new NativeTcpSocket(this,sockfd,Socket::ZT_SOCKET_TYPE_TCP_IN,false,fromia));
#ifdef __WINDOWS__
{ u_long iMode=1; ioctlsocket(sockfd,FIONBIO,&iMode); }
#ifdef ZT_TCP_NODELAY
{ BOOL f = TRUE; setsockopt(sockfd,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#else
fcntl(sockfd,F_SETFL,O_NONBLOCK);
#ifdef ZT_TCP_NODELAY
{ int f = 1; setsockopt(sockfd,IPPROTO_TCP,TCP_NODELAY,(char *)&f,sizeof(f)); }
#endif
#endif
_fdSetLock.lock();
FD_SET(sockfd,&_readfds);
_fdSetLock.unlock();
if ((int)sockfd > (int)_nfds)
_nfds = (int)sockfd;
} catch ( ... ) {
CLOSE_SOCKET(sockfd);
}
#ifndef __WINDOWS__
} else {
CLOSE_SOCKET(sockfd);
}
#endif
}
}
{
NativeUdpSocket *usock = (NativeUdpSocket *)_udpV4Socket.ptr();
if ((usock)&&(FD_ISSET(usock->_sock,&rfds))) {
usock->notifyAvailableForRead(_udpV4Socket,this,handler,arg);
}
usock = (NativeUdpSocket *)_udpV6Socket.ptr();
if ((usock)&&(FD_ISSET(usock->_sock,&rfds))) {
usock->notifyAvailableForRead(_udpV6Socket,this,handler,arg);
}
}
bool closedSockets = false;
{ // grab copy of TCP sockets list because _tcpSockets[] might be changed in a handler
Mutex::Lock _l2(_tcpSockets_m);
if (!_tcpSockets.empty()) {
ts.reserve(_tcpSockets.size());
uint64_t now = Utils::now();
for(std::map< InetAddress,SharedPtr<Socket> >::iterator s(_tcpSockets.begin());s!=_tcpSockets.end();) {
NativeTcpSocket *tsock = (NativeTcpSocket *)s->second.ptr();
#ifdef __WINDOWS__
if ( ((now - tsock->_lastActivity) < ZT_TCP_TUNNEL_ACTIVITY_TIMEOUT) && (! ((tsock->_connecting)&&(FD_ISSET(tsock->_sock,&efds))) ) ) {
#else
if ((now - tsock->_lastActivity) < ZT_TCP_TUNNEL_ACTIVITY_TIMEOUT) {
#endif
ts.push_back(s->second);
++s;
} else {
_fdSetLock.lock();
FD_CLR(tsock->_sock,&_readfds);
FD_CLR(tsock->_sock,&_writefds);
_fdSetLock.unlock();
_tcpSockets.erase(s++);
closedSockets = true;
}
}
}
}
for(std::vector< SharedPtr<Socket> >::iterator s(ts.begin());s!=ts.end();++s) {
NativeTcpSocket *tsock = (NativeTcpSocket *)s->ptr();
if (FD_ISSET(tsock->_sock,&wfds)) {
if (!tsock->notifyAvailableForWrite(*s,this)) {
{
Mutex::Lock _l2(_tcpSockets_m);
_tcpSockets.erase(tsock->_remote);
}
_fdSetLock.lock();
FD_CLR(tsock->_sock,&_readfds);
FD_CLR(tsock->_sock,&_writefds);
_fdSetLock.unlock();
closedSockets = true;
continue;
}
}
if (FD_ISSET(tsock->_sock,&rfds)) {
if (!tsock->notifyAvailableForRead(*s,this,handler,arg)) {
{
Mutex::Lock _l2(_tcpSockets_m);
_tcpSockets.erase(tsock->_remote);
}
_fdSetLock.lock();
FD_CLR(tsock->_sock,&_readfds);
FD_CLR(tsock->_sock,&_writefds);
_fdSetLock.unlock();
closedSockets = true;
continue;
}
}
}
if (closedSockets)
_updateNfds();
}
void NativeSocketManager::whack()
{
_whackSendPipe_m.lock();
#ifdef __WINDOWS__
::send(_whackSendPipe,(const char *)this,1,0);
#else
::write(_whackSendPipe,(const void *)this,1); // data is arbitrary, just send a byte
#endif
_whackSendPipe_m.unlock();
}
void NativeSocketManager::closeTcpSockets()
{
{
Mutex::Lock _l2(_tcpSockets_m);
_fdSetLock.lock();
for(std::map< InetAddress,SharedPtr<Socket> >::iterator s(_tcpSockets.begin());s!=_tcpSockets.end();++s) {
FD_CLR(((NativeTcpSocket *)s->second.ptr())->_sock,&_readfds);
FD_CLR(((NativeTcpSocket *)s->second.ptr())->_sock,&_writefds);
}
_fdSetLock.unlock();
_tcpSockets.clear();
}
_updateNfds();
}
void NativeSocketManager::_startNotifyWrite(const NativeSocket *sock)
{
_fdSetLock.lock();
FD_SET(sock->_sock,&_writefds);
_fdSetLock.unlock();
}
void NativeSocketManager::_stopNotifyWrite(const NativeSocket *sock)
{
_fdSetLock.lock();
FD_CLR(sock->_sock,&_writefds);
_fdSetLock.unlock();
}
void NativeSocketManager::_closeSockets()
{
#ifdef __WINDOWS__
if (_whackSendPipe != INVALID_SOCKET)
::closesocket(_whackSendPipe);
if (_whackReceivePipe != INVALID_SOCKET)
::closesocket(_whackReceivePipe);
if (_tcpV4ListenSocket != INVALID_SOCKET)
::closesocket(_tcpV4ListenSocket);
if (_tcpV6ListenSocket != INVALID_SOCKET)
::closesocket(_tcpV6ListenSocket);
#else
if (_whackSendPipe > 0)
::close(_whackSendPipe);
if (_whackReceivePipe > 0)
::close(_whackReceivePipe);
if (_tcpV4ListenSocket > 0)
::close(_tcpV4ListenSocket);
if (_tcpV4ListenSocket > 0)
::close(_tcpV6ListenSocket);
#endif
}
void NativeSocketManager::_updateNfds()
{
#ifdef __WINDOWS__
SOCKET nfds = _whackSendPipe;
#else
int nfds = _whackSendPipe;
#endif
if (_whackReceivePipe > nfds)
nfds = _whackReceivePipe;
if (_tcpV4ListenSocket > nfds)
nfds = _tcpV4ListenSocket;
if (_tcpV6ListenSocket > nfds)
nfds = _tcpV6ListenSocket;
if ((_udpV4Socket)&&(((NativeUdpSocket *)_udpV4Socket.ptr())->_sock > nfds))
nfds = ((NativeUdpSocket *)_udpV4Socket.ptr())->_sock;
if ((_udpV6Socket)&&(((NativeUdpSocket *)_udpV6Socket.ptr())->_sock > nfds))
nfds = ((NativeUdpSocket *)_udpV6Socket.ptr())->_sock;
Mutex::Lock _l(_tcpSockets_m);
for(std::map< InetAddress,SharedPtr<Socket> >::const_iterator s(_tcpSockets.begin());s!=_tcpSockets.end();++s) {
if (((NativeTcpSocket *)s->second.ptr())->_sock > nfds)
nfds = ((NativeTcpSocket *)s->second.ptr())->_sock;
}
_nfds = (int)nfds;
}
} // namespace ZeroTier
Reference in New Issue View Git Blame Copy Permalink