hyung-hwan/qse
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

adding multiple listeners into TcpServer

Browse Source
This commit is contained in:
hyung-hwan
2018-06-26 14:43:43 +00:00
parent eb2755fa6b
commit d7bdc63690
7 changed files with 395 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
qse/lib/si/Socket.cpp
View File

@ -126,20 +126,6 @@ void Socket::close () QSE_CPP_NOEXCEPT
}
}
int Socket::shutdown (int how) QSE_CPP_NOEXCEPT
{
QSE_ASSERT (this->handle != QSE_INVALID_SCKHND);
if (::shutdown(this->handle, how) == -1)
{
this->setErrorCode (syserr_to_errnum(errno));
return -1;
}
return 0;
}
int Socket::getOption (int level, int optname, void* optval, qse_sck_len_t* optlen) QSE_CPP_NOEXCEPT
{
QSE_ASSERT (this->handle != QSE_INVALID_SCKHND);
@ -156,6 +142,118 @@ int Socket::setOption (int level, int optname, const void* optval, qse_sck_len_t
return n;
}
int Socket::setDebug (int n) QSE_CPP_NOEXCEPT
{
return this->setOption (SOL_SOCKET, SO_DEBUG, (char*)&n, QSE_SIZEOF(n));
};
int Socket::setReuseAddr (int n) QSE_CPP_NOEXCEPT
{
return this->setOption (SOL_SOCKET, SO_REUSEADDR, (char*)&n, QSE_SIZEOF(n));
}
int Socket::setReusePort (int n) QSE_CPP_NOEXCEPT
{
#if defined(SO_REUSEPORT)
return this->setOption (SOL_SOCKET, SO_REUSEPORT, (char*)&n, QSE_SIZEOF(n));
#else
this->setErrorCode (E_ENOIMPL);
return -1;
#endif
}
int Socket::setKeepAlive (int n, int keepidle, int keepintvl, int keepcnt) QSE_CPP_NOEXCEPT
{
if (this->setOption (SOL_SOCKET, SO_KEEPALIVE, (char*)&n, QSE_SIZEOF(n)) <= -1) return -1;
// the following values are just hints.
// i don't care about success and failure
#if defined(TCP_KEEPIDLE) && defined(SOL_TCP)
if (keepidle > 0) this->setOption (SOL_TCP, TCP_KEEPIDLE, (char*)&keepidle, QSE_SIZEOF(keepidle));
#endif
#if defined(TCP_KEEPINTVL) && defined(SOL_TCP)
if (keepintvl > 0) this->setOption (SOL_TCP, TCP_KEEPINTVL, (char*)&keepintvl, QSE_SIZEOF(keepintvl));
#endif
#if defined(TCP_KEEPCNT) && defined(SOL_TCP)
if (keepcnt > 0) this->setOption (SOL_TCP, TCP_KEEPCNT, (char*)&keepcnt, QSE_SIZEOF(keepcnt));
#endif
return 0;
}
int Socket::setBroadcast (int n) QSE_CPP_NOEXCEPT
{
return this->setOption (SOL_SOCKET, SO_BROADCAST, (char*)&n, QSE_SIZEOF(n));
}
int Socket::setSendBuf (unsigned int size) QSE_CPP_NOEXCEPT
{
return this->setOption (SOL_SOCKET, SO_SNDBUF, (char*)&size, QSE_SIZEOF(size));
}
int Socket::setRecvBuf (unsigned int size) QSE_CPP_NOEXCEPT
{
return this->setOption (SOL_SOCKET, SO_RCVBUF, (char*)&size, QSE_SIZEOF(size));
}
int Socket::setLingerOn (int sec) QSE_CPP_NOEXCEPT
{
struct linger lng;
lng.l_onoff = 1;
lng.l_linger = sec;
return this->setOption (SOL_SOCKET, SO_LINGER, (char*)&lng, QSE_SIZEOF(lng));
}
int Socket::setLingerOff () QSE_CPP_NOEXCEPT
{
struct linger lng;
lng.l_onoff = 0;
lng.l_linger = 0;
return this->setOption (SOL_SOCKET, SO_LINGER, (char*)&lng, QSE_SIZEOF(lng));
}
int Socket::setTcpNodelay (int n) QSE_CPP_NOEXCEPT
{
#if defined(TCP_NODELAY)
return this->setOption (IPPROTO_TCP, TCP_NODELAY, (char*)&n, QSE_SIZEOF(n));
#else
this->setErrorCode (E_ENOIMPL);
return -1;
#endif
}
int Socket::setOobInline (int n) QSE_CPP_NOEXCEPT
{
#if defined(SO_OOBINLINE)
return this->setOption (SOL_SOCKET, SO_OOBINLINE, (char*)&n, QSE_SIZEOF(n));
#else
this->setErrorCode (E_ENOIMPL);
return -1;
#endif
}
int Socket::setIpv6Only (int n) QSE_CPP_NOEXCEPT
{
#if defined(IPV6_V6ONLY)
return this->setOption (IPPROTO_IPV6, IPV6_V6ONLY, (char*)&n, QSE_SIZEOF(n));
#else
this->setErrorCode (E_ENOIMPL);
return -1;
#endif
}
int Socket::shutdown (int how) QSE_CPP_NOEXCEPT
{
QSE_ASSERT (this->handle != QSE_INVALID_SCKHND);
if (::shutdown(this->handle, how) == -1)
{
this->setErrorCode (syserr_to_errnum(errno));
return -1;
}
return 0;
}
int Socket::connect (const SocketAddress& target) QSE_CPP_NOEXCEPT
{
QSE_ASSERT (this->handle != QSE_INVALID_SCKHND);

40
qse/lib/si/SocketAddress.cpp
View File

@ -90,34 +90,34 @@ QSE_BEGIN_NAMESPACE(QSE)
/////////////////////////////////
SocketAddress::SocketAddress ()
SocketAddress::SocketAddress () QSE_CPP_NOEXCEPT
{
QSE_MEMSET (&this->skad, 0, QSE_SIZEOF(this->skad));
}
SocketAddress::SocketAddress (int family)
SocketAddress::SocketAddress (int family) QSE_CPP_NOEXCEPT
{
QSE_MEMSET (&this->skad, 0, QSE_SIZEOF(this->skad));
FAMILY(&this->skad) = family;
}
SocketAddress::SocketAddress (const qse_skad_t* skad)
SocketAddress::SocketAddress (const qse_skad_t* skad) QSE_CPP_NOEXCEPT
{
this->set (skad);
}
SocketAddress::SocketAddress (const qse_nwad_t* nwad)
SocketAddress::SocketAddress (const qse_nwad_t* nwad) QSE_CPP_NOEXCEPT
{
this->set (nwad);
}
int SocketAddress::getFamily () const
int SocketAddress::getFamily () const QSE_CPP_NOEXCEPT
{
return FAMILY(&this->skad);
//return qse_skadfamily (&this->skad);
}
void SocketAddress::setIpaddr (const qse_ip4ad_t* ipaddr)
void SocketAddress::setIpaddr (const qse_ip4ad_t* ipaddr) QSE_CPP_NOEXCEPT
{
#if defined(AF_INET)
if (FAMILY(&this->skad) == AF_INET)
@ -128,7 +128,7 @@ void SocketAddress::setIpaddr (const qse_ip4ad_t* ipaddr)
#endif
}
void SocketAddress::setIpaddr (const qse_ip6ad_t* ipaddr)
void SocketAddress::setIpaddr (const qse_ip6ad_t* ipaddr) QSE_CPP_NOEXCEPT
{
#if defined(AF_INET6)
if (FAMILY(&this->skad) == AF_INET6)
@ -139,7 +139,7 @@ void SocketAddress::setIpaddr (const qse_ip6ad_t* ipaddr)
#endif
}
qse_uint16_t SocketAddress::getPort () const
qse_uint16_t SocketAddress::getPort () const QSE_CPP_NOEXCEPT
{
switch (FAMILY(&this->skad))
{
@ -163,7 +163,7 @@ qse_uint16_t SocketAddress::getPort () const
return 0;
}
void SocketAddress::setPort (qse_uint16_t port)
void SocketAddress::setPort (qse_uint16_t port) QSE_CPP_NOEXCEPT
{
switch (FAMILY(&this->skad))
{
@ -187,32 +187,46 @@ void SocketAddress::setPort (qse_uint16_t port)
}
}
int SocketAddress::set (const qse_skad_t* skad)
int SocketAddress::set (const qse_skad_t* skad) QSE_CPP_NOEXCEPT
{
this->skad = *skad;
return 0;
}
int SocketAddress::set (const qse_nwad_t* nwad)
int SocketAddress::set (const qse_nwad_t* nwad) QSE_CPP_NOEXCEPT
{
return qse_nwadtoskad(nwad, &this->skad);
}
int SocketAddress::set (const qse_mchar_t* str)
int SocketAddress::set (const qse_mchar_t* str) QSE_CPP_NOEXCEPT
{
qse_nwad_t nwad;
if (qse_mbstonwad(str, &nwad) <= -1) return -1;
return qse_nwadtoskad(&nwad, &this->skad);
}
int SocketAddress::set (const qse_wchar_t* str)
int SocketAddress::set (const qse_wchar_t* str) QSE_CPP_NOEXCEPT
{
qse_nwad_t nwad;
if (qse_wcstonwad(str, &nwad) <= -1) return -1;
return qse_nwadtoskad(&nwad, &this->skad);
}
int SocketAddress::set (const qse_mchar_t* str, qse_size_t len) QSE_CPP_NOEXCEPT
{
qse_nwad_t nwad;
if (qse_mbsntonwad(str, len, &nwad) <= -1) return -1;
return qse_nwadtoskad(&nwad, &this->skad);
}
int SocketAddress::set (const qse_wchar_t* str, qse_size_t len) QSE_CPP_NOEXCEPT
{
qse_nwad_t nwad;
if (qse_wcsntonwad(str, len, &nwad) <= -1) return -1;
return qse_nwadtoskad(&nwad, &this->skad);
}
/////////////////////////////////
QSE_END_NAMESPACE(QSE)
/////////////////////////////////

229
qse/lib/si/TcpServer.cpp
View File

@ -26,9 +26,20 @@
#include <qse/si/TcpServer.hpp>
#include <qse/si/os.h>
#include <qse/cmn/str.h>
#include "../cmn/mem-prv.h"
#include <sys/epoll.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
QSE_BEGIN_NAMESPACE(QSE)
#include "../cmn/syserr.h"
IMPLEMENT_SYSERR_TO_ERRNUM (TcpServer::ErrorCode, TcpServer::)
TcpServer::Client::Client (TcpServer* server)
{
this->server = server;
@ -75,22 +86,21 @@ int TcpServer::Client::stop () QSE_CPP_NOEXCEPT
return 0;
}
TcpServer::TcpServer ():
TcpServer::TcpServer () QSE_CPP_NOEXCEPT:
errcode(E_ENOERR),
stop_requested(false),
server_serving(false),
max_connections(0),
thread_stack_size (0),
reopen_socket_upon_error(false)
thread_stack_size (0)
{
}
TcpServer::TcpServer (const SocketAddress& address):
TcpServer::TcpServer (const SocketAddress& address) QSE_CPP_NOEXCEPT:
binding_address(address),
stop_requested(false),
server_serving(false),
max_connections(0),
thread_stack_size (0),
reopen_socket_upon_error(false)
thread_stack_size (0)
{
}
@ -100,38 +110,159 @@ TcpServer::~TcpServer () QSE_CPP_NOEXCEPT
this->delete_all_clients ();
}
int TcpServer::open_tcp_socket (Socket& socket, int* err_code) QSE_CPP_NOEXCEPT
void TcpServer::free_all_listeners () QSE_CPP_NOEXCEPT
{
if (socket.open(this->binding_address.getFamily(), QSE_SOCK_STREAM, Socket::T_CLOEXEC | Socket::T_NONBLOCK) <= -1)
Listener* lp;
struct epoll_event dummy_ev;
::epoll_ctl (this->listener.ep_fd, EPOLL_CTL_DEL, this->listener.mux_pipe[0], &dummy_ev);
while (this->listener.head)
{
if (err_code) *err_code = ERR_OPEN;
return -1;
lp = this->listener.head;
this->listener.head = lp->next_listener;
this->listener.count--;
::epoll_ctl (this->listener.ep_fd, EPOLL_CTL_DEL, lp->getHandle(), &dummy_ev);
lp->close ();
delete lp;
}
//socket.setReuseAddr (true);
//socket.setReusePort (true);
if (socket.bind(this->binding_address) <= -1)
{
if (err_code) *err_code = ERR_BIND;
return -1;
}
if (socket.listen() <= -1)
{
if (err_code) *err_code = ERR_LISTEN;
return -1;
}
//socket.enableTimeout (1000);
return 0;
QSE_ASSERT (this->listener.ep_fd >= 0);
::close (this->listener.ep_fd);
this->listener.ep_fd = -1;
}
int TcpServer::start (int* err_code) QSE_CPP_NOEXCEPT
int TcpServer::setup_listeners (const qse_char_t* addrs) QSE_CPP_NOEXCEPT
{
const qse_char_t* addr_ptr, * comma;
int ep_fd = -1, fcv;
struct epoll_event ev;
int pfd[2] = { -1, - 1 };
SocketAddress sockaddr;
ep_fd = ::epoll_create(1024);
if (ep_fd <= -1)
{
this->setErrorCode (syserr_to_errnum(errno));
return -1;
}
#if defined(O_CLOEXEC)
fcv = ::fcntl(ep_fd, F_GETFD, 0);
if (fcv >= 0) ::fcntl(ep_fd, F_SETFD, fcv | O_CLOEXEC);
#endif
if (::pipe(pfd) <= -1)
{
this->setErrorCode (syserr_to_errnum(errno));
goto oops;
}
#if defined(O_CLOEXEC)
fcv = ::fcntl(pfd[0], F_GETFD, 0);
if (fcv >= 0) ::fcntl(pfd[0], F_SETFD, fcv | O_CLOEXEC);
fcv = ::fcntl(pfd[1], F_GETFD, 0);
if (fcv >= 0) ::fcntl(pfd[1], F_SETFD, fcv | O_CLOEXEC);
#endif
#if defined(O_NONBLOCK)
fcv = ::fcntl(pfd[0], F_GETFL, 0);
if (fcv >= 0) ::fcntl(pfd[0], F_SETFL, fcv | O_NONBLOCK);
fcv = ::fcntl(pfd[1], F_GETFL, 0);
if (fcv >= 0) ::fcntl(pfd[1], F_SETFL, fcv | O_NONBLOCK);
#endif
QSE_MEMSET (&ev, 0, QSE_SIZEOF(ev));
ev.events = EPOLLIN | EPOLLHUP | EPOLLERR;
ev.data.fd = pfd[0];
if (::epoll_ctl(ep_fd, EPOLL_CTL_ADD, pfd[0], &ev) <= -1)
{
this->setErrorCode (syserr_to_errnum(errno));
goto oops;
}
addr_ptr = addrs;
while (1)
{
qse_size_t addr_len;
Listener* lsck;
Socket sock;
comma = qse_strchr(addr_ptr, QSE_T(','));
addr_len = comma? comma - addr_ptr: qse_strlen(addr_ptr);
/* [NOTE] no whitespaces are allowed before and after a comma */
if (sockaddr.set(addr_ptr, addr_len) <= -1)
{
/* TOOD: logging */
goto next_segment;
}
try
{
lsck = new Listener();
}
catch (...)
{
/* TODO: logging... */
goto next_segment;
}
if (lsck->open(sockaddr.getFamily(), QSE_SOCK_STREAM, Socket::T_CLOEXEC | Socket::T_NONBLOCK) <= -1)
{
/* TODO: logging */
goto next_segment;
}
lsck->setReuseAddr (1);
lsck->setReusePort (1);
if (lsck->bind(sockaddr) <= -1 || lsck->listen() <= -1)
{
/* TODO: logging */
lsck->close ();
goto next_segment;
}
QSE_MEMSET (&ev, 0, QSE_SIZEOF(ev));
ev.events = EPOLLIN | EPOLLHUP | EPOLLERR;
ev.data.fd = lsck->getHandle();
if (::epoll_ctl(ep_fd, EPOLL_CTL_ADD, lsck->getHandle(), &ev) <= -1)
{
/* TODO: logging */
lsck->close ();
goto next_segment;
}
lsck->address = sockaddr;
lsck->next_listener = this->listener.head;
this->listener.head = lsck;
this->listener.count++;
next_segment:
if (!comma) break;
addr_ptr = comma + 1;
}
if (!this->listener.head) goto oops;
this->listener.ep_fd = ep_fd;
this->listener.mux_pipe[0] = pfd[0];
this->listener.mux_pipe[1] = pfd[1];
return 0;
oops:
if (this->listener.head) this->free_all_listeners ();
if (pfd[0] >= 0) close (pfd[0]);
if (pfd[1] >= 0) close (pfd[1]);
if (ep_fd >= 0) close (ep_fd);
return -1;
}
int TcpServer::start (const qse_char_t* addrs) QSE_CPP_NOEXCEPT
{
this->server_serving = true;
if (err_code != QSE_NULL) *err_code = ERR_NONE;
this->setStopRequested (false);
Client* client = QSE_NULL;
@ -140,7 +271,7 @@ int TcpServer::start (int* err_code) QSE_CPP_NOEXCEPT
{
Socket socket;
if (this->open_tcp_socket(socket, err_code) <= -1)
if (this->setup_listeners(addrs) <= -1)
{
this->server_serving = false;
this->setStopRequested (false);
@ -183,33 +314,7 @@ int TcpServer::start (int* err_code) QSE_CPP_NOEXCEPT
// don't "delete client" here as i want it to be reused
// in the next iteration after "continue"
if (this->reopen_socket_upon_error)
{
// closing the listeing socket causes the
// the pending connection to be dropped.
// this poses the risk of reopening failure.
// imagine the case of EMFILE(too many open files).
// accept() will fail until an open file is closed.
qse_size_t reopen_count = 0;
socket.close ();
reopen:
if (this->open_tcp_socket (socket, err_code) <= -1)
{
if (reopen_count >= 100)
{
qse_ntime_t interval;
if (reopen_count >= 200) qse_inittime (&interval, 0, 100000000); // 100 milliseconds
else qse_inittime (&interval, 0, 10000000); // 10 milliseconds
qse_sleep (&interval);
}
if (this->isStopRequested()) break;
reopen_count++;
goto reopen;
}
}
/* TODO: logging */
continue;
}
@ -237,7 +342,7 @@ int TcpServer::start (int* err_code) QSE_CPP_NOEXCEPT
this->delete_all_clients ();
if (client != QSE_NULL) delete client;
if (err_code) *err_code = ERR_EXCEPTION;
this->setErrorCode (E_EEXCEPT);
this->server_serving = false;
this->setStopRequested (false);
@ -298,11 +403,11 @@ void TcpServer::delete_all_clients () QSE_CPP_NOEXCEPT
p = np->value;
QSE_ASSERT (p != QSE_NULL);
#if defined(_WIN32)
#if defined(_WIN32)
while (p->state() == Thread::RUNNING) qse_sleep (300);
#else
#else
p->join ();
#endif
#endif
delete p;
np2 = np; np = np->getNextNode();
this->client_list.remove (np2);