@@ -59,6 +59,7 @@ AC_CHECK_HEADER(pthread.h, [], [echo "pthread.h not found"; exit1]) # Checks for library functions. AC_CHECK_FUNC(gethostbyname_r, [AC_DEFINE([HAVE_GETHOSTBYNAME_R], [], [have gethostbyname_r() or not])]) +AC_CHECK_FUNC(epoll_create, [AC_DEFINE([HAVE_EPOLL], [], [have epoll functions or not])]) AC_MSG_CHECKING(for gnu specific strerror_r) AC_TRY_COMPILE([ configure.ac @@ -134,6 +134,7 @@ int flared::startup(int argc, char **argv) { log_notice(" reconstruction_inteval: %d", ini_option_object().get_reconstruction_interval()); log_notice(" server_name: %s", ini_option_object().get_server_name().c_str()); log_notice(" server_port: %d", ini_option_object().get_server_port()); + log_notice(" server_socket: %s", ini_option_object().get_server_socket().c_str()); log_notice(" stack_size: %d", ini_option_object().get_stack_size()); log_notice(" storage_ap: %u", ini_option_object().get_storage_ap()); log_notice(" storage_bucket_size: %llu", ini_option_object().get_storage_bucket_size()); @@ -161,9 +162,15 @@ int flared::startup(int argc, char **argv) { // application objects connection::read_timeout = ini_option_object().get_net_read_timeout() * 1000; // -> msec this->_server = _new_ server(); - if (this->_server->listen(ini_option_object().get_server_port(), ini_option_object().get_back_log()) < 0) { + this->_server->set_back_log(ini_option_object().get_back_log()); + if (this->_server->listen(ini_option_object().get_server_port()) < 0) { return -1; } + if (ini_option_object().get_server_socket().empty() == false) { + if (this->_server->listen(ini_option_object().get_server_socket()) < 0) { + return -1; + } + } this->_thread_pool = _new_ thread_pool(ini_option_object().get_thread_pool_size(), ini_option_object().get_stack_size()); src/flared/flared.cc @@ -42,6 +42,7 @@ ini_option::ini_option(): _reconstruction_interval(default_reconstruction_interval), _server_name(""), _server_port(default_server_port), + _server_socket(""), _stack_size(default_stack_size), _storage_ap(default_storage_ap), _storage_bucket_size(default_storage_bucket_size), @@ -215,6 +216,10 @@ int ini_option::load() { this->_server_port = opt_var_map["server-port"].as<int>(); } + if (opt_var_map.count("server-socket")) { + this->_server_socket = opt_var_map["server-socket"].as<string>(); + } + if (opt_var_map.count("stack-size")) { this->_stack_size = opt_var_map["stack-size"].as<int>(); } @@ -388,6 +393,7 @@ int ini_option::_setup_config_option(program_options::options_description& optio ("reconstruction-interval", program_options::value<int>(), "master/slave dump interval in usec (dynamic)") ("server-name", program_options::value<string>(), "my server name") ("server-port", program_options::value<int>(), "my server port") + ("server-socket", program_options::value<string>(), "my server unix domain socket (optional)") ("stack-size", program_options::value<int>(), "thread stack size (kb)") ("storage-ap", program_options::value<uint32_t>(), "storage size of record alignment by power of 2 (tch)") ("storage-bucket-size", program_options::value<uint64_t>(), "number of elements of the bucket array (tch)") src/flared/ini_option.cc @@ -45,6 +45,7 @@ private: int _reconstruction_interval; string _server_name; int _server_port; + string _server_socket; int _stack_size; uint32_t _storage_ap; uint64_t _storage_bucket_size; @@ -102,6 +103,7 @@ public: int get_reconstruction_interval() { return this->_reconstruction_interval; }; string get_server_name() { return this->_server_name; }; int get_server_port() { return this->_server_port; }; + string get_server_socket() { return this->_server_socket; }; int get_stack_size() { return this->_stack_size; }; uint32_t get_storage_ap() { return this->_storage_ap; }; uint64_t get_storage_bucket_size() { return this->_storage_bucket_size; }; src/flared/ini_option.h @@ -115,6 +115,7 @@ int flarei::startup(int argc, char **argv) { log_notice(" net_read_timeout: %d", ini_option_object().get_net_read_timeout()); log_notice(" server_name: %s", ini_option_object().get_server_name().c_str()); log_notice(" server_port: %d", ini_option_object().get_server_port()); + log_notice(" server_socket: %s", ini_option_object().get_server_socket().c_str()); log_notice(" stack_size: %d", ini_option_object().get_stack_size()); log_notice(" thread_pool_size: %d", ini_option_object().get_thread_pool_size()); @@ -139,6 +140,11 @@ int flarei::startup(int argc, char **argv) { if (this->_server->listen(ini_option_object().get_server_port()) < 0) { return -1; } + if (ini_option_object().get_server_socket().empty() == false) { + if (this->_server->listen(ini_option_object().get_server_socket()) < 0) { + return -1; + } + } this->_thread_pool = _new_ thread_pool(ini_option_object().get_thread_pool_size(), ini_option_object().get_stack_size()); src/flarei/flarei.cc @@ -32,6 +32,7 @@ ini_option::ini_option(): _net_read_timeout(default_net_read_timeout), _server_name(""), _server_port(default_server_port), + _server_socket(""), _stack_size(default_stack_size), _thread_pool_size(default_thread_pool_size) { } @@ -157,6 +158,10 @@ int ini_option::load() { this->_server_port = opt_var_map["server-port"].as<int>(); } + if (opt_var_map.count("server-socket")) { + this->_server_socket = opt_var_map["server-socket"].as<string>(); + } + if (opt_var_map.count("stack-size")) { this->_stack_size = opt_var_map["stack-size"].as<int>(); } @@ -291,6 +296,7 @@ int ini_option::_setup_config_option(program_options::options_description& optio ("monitor-read-timeout", program_options::value<int>(), "node server monitoring read timeout (millisec) (dynamic)") ("server-name", program_options::value<string>(), "my server name") ("server-port", program_options::value<int>(), "my server port") + ("server-socket", program_options::value<string>(), "my server unix domain socket (optional)") ("stack-size", program_options::value<int>(), "thread stack size (kb)") ("thread-pool-size", program_options::value<int>(), "thread pool size (dynamic)"); src/flarei/ini_option.cc @@ -34,6 +34,7 @@ private: int _net_read_timeout; string _server_name; int _server_port; + string _server_socket; int _stack_size; int _thread_pool_size; @@ -66,6 +67,7 @@ public: int get_net_read_timeout() { return this->_net_read_timeout; }; string get_server_name() { return this->_server_name; }; int get_server_port() { return this->_server_port; }; + string get_server_socket() { return this->_server_socket; }; int get_stack_size() { return this->_stack_size; }; int get_thread_pool_size() { return this->_thread_pool_size; }; src/flarei/ini_option.h @@ -20,8 +20,10 @@ int connection::read_timeout = 10*60*1000; * ctor for connection */ connection::connection(): + _addr_family(AF_INET), _host(""), _port(-1), + _path(""), _read_timeout(read_timeout), _read_buf(NULL), _read_buf_p(NULL), @@ -36,9 +38,30 @@ connection::connection(): */ connection::connection(int sock, struct sockaddr_in addr): net(sock), - _addr(addr), + _addr_family(AF_INET), + _addr_inet(addr), _host(""), _port(-1), + _path(""), + _read_timeout(read_timeout), + _read_buf(NULL), + _read_buf_p(NULL), + _read_buf_len(0), + _write_buf(NULL), + _write_buf_len(0), + _write_buf_chunk_size(0) { +} + +/** + * ctor for connection + */ +connection::connection(int sock, struct sockaddr_un addr): + net(sock), + _addr_family(AF_UNIX), + _addr_unix(addr), + _host(""), + _port(-1), + _path(""), _read_timeout(read_timeout), _read_buf(NULL), _read_buf_p(NULL), @@ -66,7 +89,7 @@ connection::~connection() { // {{{ public methods /** - * open connection + * open tcp connection */ int connection::open(string host, int port) { this->_errno = 0; @@ -79,9 +102,10 @@ int connection::open(string host, int port) { return -1; } - memset(&this->_addr, 0, sizeof(this->_addr)); - this->_addr.sin_family = AF_INET; - this->_addr.sin_port = htons(port); + this->_addr_family = AF_INET; + memset(&this->_addr_inet, 0, sizeof(this->_addr_inet)); + this->_addr_inet.sin_family = AF_INET; + this->_addr_inet.sin_port = htons(port); struct hostent he; int he_errno; @@ -90,11 +114,11 @@ int connection::open(string host, int port) { this->close(); return -1; } - memcpy(&this->_addr.sin_addr, he.h_addr, he.h_length); + memcpy(&this->_addr_inet.sin_addr, he.h_addr, he.h_length); int i; for (i = 0; i < connection::connect_retry_limit; i++) { - if (connect(this->_sock, (struct sockaddr*)&this->_addr, sizeof(this->_addr)) < 0) { + if (connect(this->_sock, (struct sockaddr*)&this->_addr_inet, sizeof(this->_addr_inet)) < 0) { log_warning("connect() failed: %s (%d) -> wait for %d usec", util::strerror(errno), errno, connection::connect_retry_wait); usleep(connection::connect_retry_wait); continue; @@ -536,10 +560,12 @@ int connection::writeline(const char* p) { string connection::get_host() { if (this->_host.empty() == false) { return this->_host; + } else if (this->_addr_family == AF_UNIX) { + return ""; } char buf[BUFSIZ]; - util::inet_ntoa(this->_addr.sin_addr, buf); + util::inet_ntoa(this->_addr_inet.sin_addr, buf); string s = buf; return s; @@ -548,9 +574,21 @@ string connection::get_host() { int connection::get_port() { if (this->_port >= 0) { return this->_port; + } else if (this->_addr_family == AF_UNIX) { + return 0; + } + + return ntohs(this->_addr_inet.sin_port); +} + +string connection::get_path() { + if (this->_path.empty() == false) { + return this->_path; + } else if (this->_addr_family != AF_UNIX) { + return ""; } - return ntohs(this->_addr.sin_port); + return this->_addr_unix.sun_path; } // }}} src/lib/connection.cc @@ -9,6 +9,7 @@ #define __NET_CONNECTION_H__ #include <poll.h> +#include <sys/un.h> #include "net.h" @@ -23,9 +24,12 @@ typedef shared_ptr<connection> shared_connection; */ class connection : public net { protected: - struct sockaddr_in _addr; + sa_family_t _addr_family; + struct sockaddr_in _addr_inet; + struct sockaddr_un _addr_unix; string _host; int _port; + string _path; int _read_timeout; char* _read_buf; char* _read_buf_p; @@ -45,6 +49,7 @@ public: connection(); connection(int sock, struct sockaddr_in addr); + connection(int sock, struct sockaddr_un addr); virtual ~connection(); int open(string host, int port); @@ -59,6 +64,7 @@ public: bool is_error() { return this->_errno != 0 ? true : false; }; string get_host(); int get_port(); + string get_path(); int get_read_timeout() { return this->_read_timeout; }; int set_read_timeout(int timeout) { this->_read_timeout = timeout; return 0; }; src/lib/connection.h @@ -17,13 +17,20 @@ namespace flare { * ctor for server */ server::server(): - _port(0) { + _listen_socket_index(0), +#ifdef HAVE_EPOLL + _epoll_socket(0), +#endif + _back_log(SOMAXCONN) { } /** * dtor for server */ server::~server() { + if (this->close() < 0) { + // ... + } } // }}} @@ -32,53 +39,161 @@ server::~server() { // {{{ public methods /** + * close listening socket(s) + */ +int server::close() { + int r = 0; + + struct sockaddr addr; + struct sockaddr_un* addr_unix = NULL; + +#ifdef HAVE_EPOLL + if (this->_epoll_socket > 0) { + if (::close(this->_epoll_socket) < 0) { + log_err("close() failed: %s (%d) (sock=epoll)", util::strerror(errno), errno); + } + } +#endif + + for (int i = 0; i < this->_listen_socket_index; i++) { + int sock = this->_listen_socket[i]; + + if (sock >= 0) { + // this will be cost but i do not care about it in this situation + socklen_t addr_len = sizeof(addr); + if (getsockname(sock, &addr, &addr_len) < 0) { + log_notice("getsockname() failed: %s (%d) -> assuming sa_family = AF_INET", util::strerror(errno), errno); + addr.sa_family = AF_INET; + } + + if (::close(sock) < 0) { + r = -1; + log_err("close() failed: %s (%d) (index=%d)", util::strerror(errno), errno, i); + } else { + log_debug("socket closed (sock=%d, index=%d)", sock, i); + } + + if (addr.sa_family == AF_UNIX) { + addr_unix = reinterpret_cast<struct sockaddr_un*>(&addr); + if (unlink(addr_unix->sun_path) < 0) { + log_err("unlink() failed: %s (%d) (index=%d)", util::strerror(errno), errno, i); + } else { + log_debug("unix domain socket removed (path=%s)", addr_unix->sun_path); + } + } + + this->_listen_socket[i] = -1; + } else { + log_info("close() called but socket seems to be already closed (sock=%d, index=%d)", sock, i); + } + } + this->_listen_socket_index = 0; + + return r; +} + +/** * setup server socket */ -int server::listen(int port, int backlog) { - this->_port = port; +int server::listen(int port) { + if (this->_listen_socket_index >= this->max_listen_socket) { + log_err("listen socket limit exceeded (max_listen_socket=%d)", this->max_listen_socket); + return -1; + } // creating a socket - this->_sock = socket(PF_INET, SOCK_STREAM, 0); - if (this->_sock < 0) { + int sock = socket(PF_INET, SOCK_STREAM, 0); + if (sock < 0) { log_err("socket() failed: %s (%d)", util::strerror(errno), errno); return -1; } // socket option/attr - int tmp = 1; - if (setsockopt(this->_sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char *>(&tmp), sizeof(tmp)) < 0) { - log_err("setsockopt() for SO_REUSEADDR failed: %s (%d)", util::strerror(errno), errno); + if (this->_set_listen_socket_option(sock) < 0) { return -1; } - if (setsockopt(this->_sock, SOL_SOCKET, SO_KEEPALIVE, reinterpret_cast<char *>(&tmp), sizeof(tmp)) < 0) { - log_err("setsockopt() SO_KEEPALIVE failed: %s (%d)", util::strerror(errno), errno); + + // bind -> listen + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = INADDR_ANY; + addr.sin_port = htons(port); + if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) { + log_err("bind() failed: %s (%d)", util::strerror(errno), errno); return -1; } - if (setsockopt(this->_sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char*>(&tmp), sizeof(tmp)) < 0) { - log_err("setsockopt() TCP_NODELAY failed: %s (%d)", util::strerror(errno), errno); + if (::listen(sock, this->_back_log) < 0) { + log_err("listen() failed: %s (%d)", util::strerror(errno), errno); return -1; } - int flag = fcntl(this->_sock, F_GETFL, 0); - if (fcntl(this->_sock, F_SETFL, flag | O_NONBLOCK) < 0) { - log_err("fcntl() failed: %s (%d)", util::strerror(errno), errno); + + this->_listen_socket[this->_listen_socket_index] = sock; + this->_listen_socket_index++; + + log_info("server socket (type=inet) is ready to accept (port: %d)", port); + +#ifdef HAVE_EPOLL + if (this->_add_epoll_socket(sock) < 0) { + return -1; + } +#endif + + return 0; +} + +/** + * setup server socket + */ +int server::listen(string uds) { + if (this->_listen_socket_index >= this->max_listen_socket) { + log_err("listen socket limit exceeded (max_listen_socket=%d)", this->max_listen_socket); + return -1; + } + + // creating a socket + int sock = socket(PF_UNIX, SOCK_STREAM, 0); + if (sock < 0) { + log_err("socket() failed: %s (%d)", util::strerror(errno), errno); + return -1; + } + + // socket option/attr + if (this->_set_listen_socket_option(sock, PF_UNIX) < 0) { return -1; } // bind -> listen - struct sockaddr_in addr; - addr.sin_family = AF_INET; - addr.sin_addr.s_addr = INADDR_ANY; - addr.sin_port = htons(this->_port); - if (bind(this->_sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) { + struct sockaddr_un addr; + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, uds.c_str(), sizeof(addr.sun_path)-1); + if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) { log_err("bind() failed: %s (%d)", util::strerror(errno), errno); return -1; } - if (::listen(this->_sock, backlog) < 0) { + if (::listen(sock, this->_back_log) < 0) { log_err("listen() failed: %s (%d)", util::strerror(errno), errno); return -1; } - log_info("server socket is ready to accept (port: %d)", this->_port); + if (chmod(addr.sun_path, 0777) < 0) { + log_err("chmod() failed: %s (%d) (path=%s)", util::strerror(errno), errno, addr.sun_path); + return -1; + } else { + log_debug("setting permission of unix domain socket as world writable (path=%s)", addr.sun_path); + } + + this->_listen_socket[this->_listen_socket_index] = sock; + this->_listen_socket_index++; + + log_info("server socket (type=unix) is ready to accept (path: %s)", uds.c_str()); + +#ifdef HAVE_EPOLL + if (this->_add_epoll_socket(sock) < 0) { + return -1; + } +#endif return 0; } @@ -89,70 +204,171 @@ int server::listen(int port, int backlog) { vector<shared_connection> server::wait() { vector<shared_connection> connection_list; +#ifdef HAVE_EPOLL + const char* poll_type = "epoll_wait"; // just for logging + struct epoll_event ev_list[this->max_listen_socket]; + int n = epoll_wait(this->_epoll_socket, ev_list, this->max_listen_socket, -1); +#else + const char* poll_type = "select"; // just for logging fd_set fds; FD_ZERO(&fds); - FD_SET(this->_sock, &fds); + for (int i = 0; i < this->_listen_socket_index; i++) { + FD_SET(this->_listen_socket[i], &fds); + } int n = select(FD_SETSIZE, &fds, 0, 0, 0); +#endif if (n <= 0) { if (errno != EINTR) { - log_err("select() failed: %s (%d)", util::strerror(errno), errno); + log_err("%s() failed: %s (%d)", poll_type, util::strerror(errno), errno); } else { - log_notice("select() failed: %s (%d)", util::strerror(errno), errno); + log_notice("%s() failed: %s (%d)", poll_type, util::strerror(errno), errno); } return connection_list; } // accpet anyway - struct sockaddr_in addr_remote; - socklen_t addr_remote_len; - int sock, j; - for (j = 0; j < server::accept_retry_limit; j++) { - addr_remote_len = sizeof(addr_remote); - sock = accept(this->_sock, (struct sockaddr *)&addr_remote, &addr_remote_len); - if (sock >= 0) { - break; - } - if (errno == EAGAIN || errno == EINTR) { - log_info("eagain or eintr -> retrying accept() (%d times left)", server::accept_retry_limit - j - 1); +#ifdef HAVE_EPOLL + for (int i = 0; i < n; i++) { + int listen_socket = ev_list[i].data.fd; +#else + for (int i = 0; i < this->_listen_socket_index; i++) { + if (!FD_ISSET(this->_listen_socket[i], &fds)) { continue; + } + int listen_socket = this->_listen_socket[i]; +#endif + + struct sockaddr addr; + struct sockaddr_in* addr_inet = NULL; + struct sockaddr_un* addr_unix = NULL; + socklen_t addr_len; + int sock, j; + for (j = 0; j < server::accept_retry_limit; j++) { + addr_len = sizeof(addr); + sock = accept(listen_socket, &addr, &addr_len); + if (sock >= 0) { + break; + } + if (errno == EAGAIN || errno == EINTR) { + log_info("eagain or eintr -> retrying accept() (%d times left)", server::accept_retry_limit - j - 1); + continue; + } else { + break; + } + } + if (sock < 0) { + log_err("accept() failed: %s (%d)", util::strerror(errno), errno); + return connection_list; + } + + if (addr.sa_family == AF_UNIX) { + addr_unix = reinterpret_cast<sockaddr_un*>(&addr); } else { - break; + addr_inet = reinterpret_cast<sockaddr_in*>(&addr); + } + + // socket option/attr + int flag = fcntl(sock, F_GETFL, 0); + if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) { + log_err("fcntl() failed: %s (%d)", util::strerror(errno), errno); + ::close(sock); // try + return connection_list; + } + + flag = 1; + if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, reinterpret_cast<char*>(&flag), sizeof(flag)) < 0) { + log_err("setsockopt() failed: %s (%d) - SO_KEEPALIVE", util::strerror(errno), errno); + return connection_list; } + + if (addr.sa_family != AF_UNIX) { + flag = 1; + if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char*>(&flag), sizeof(flag)) < 0) { + log_err("setsockopt() failed: %s (%d) - TCP_NODELAY", util::strerror(errno), errno); + return connection_list; + } + } + + log_info("socket accepted (fd=%d remote=%s)", sock, addr.sa_family == AF_UNIX ? "unix domain socket" : inet_ntoa(addr_inet->sin_addr)); + shared_connection c; + try { + if (addr.sa_family == AF_UNIX) { + c = shared_connection(new connection(sock, *addr_unix)); + } else { + c = shared_connection(new connection(sock, *addr_inet)); + } + } catch (int e) { + return connection_list; + } + connection_list.push_back(c); } - if (sock < 0) { - log_err("accept() failed: %s (%d)", util::strerror(errno), errno); - return connection_list; + + return connection_list; +} +// }}} + +// {{{ protected methods +/** + * setup sever socket option + */ +int server::_set_listen_socket_option(int sock, int domain) { + int tmp = 1; + + if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char *>(&tmp), sizeof(tmp)) < 0) { + log_err("setsockopt() for SO_REUSEADDR failed: %s (%d)", util::strerror(errno), errno); + return -1; + } + + if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, reinterpret_cast<char *>(&tmp), sizeof(tmp)) < 0) { + log_err("setsockopt() SO_KEEPALIVE failed: %s (%d)", util::strerror(errno), errno); + return -1; + } + + if (domain != PF_UNIX) { + if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char*>(&tmp), sizeof(tmp)) < 0) { + log_err("setsockopt() TCP_NODELAY failed: %s (%d)", util::strerror(errno), errno); + return -1; + } } + int flag = fcntl(sock, F_GETFL, 0); if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) { log_err("fcntl() failed: %s (%d)", util::strerror(errno), errno); - ::close(sock); // try - return connection_list; - } - flag = 1; - if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, reinterpret_cast<char*>(&flag), sizeof(flag)) < 0) { - log_err("setsockopt() failed: %s (%d) - SO_KEEPALIVE", util::strerror(errno), errno); - return connection_list; + return -1; } - flag = 1; - if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<char*>(&flag), sizeof(flag)) < 0) { - log_err("setsockopt() failed: %s (%d) - TCP_NODELAY", util::strerror(errno), errno); - return connection_list; + + return 0; +} + +#ifdef HAVE_EPOLL +/** + * add listen socket to epoll + */ +int server::_add_epoll_socket(int sock) { + if (this->_epoll_socket <= 0) { + this->_epoll_socket = epoll_create(this->max_listen_socket); + if (this->_epoll_socket < 0) { + log_err("epoll_create() failed: %s (%d)", util::strerror(errno), errno); + return -1; + } } - log_info("socket accepted (fd=%d remote=%s)", sock, inet_ntoa(addr_remote.sin_addr)); - shared_connection c; - try { - c = shared_connection(new connection(sock, addr_remote)); - } catch (int e) { - return connection_list; + + struct epoll_event ev; + memset(&ev, 0, sizeof(ev)); + + // we can use edge trigger here but use leve trigger for safe for a while... + ev.events = EPOLLIN; + ev.data.fd = sock; + if (epoll_ctl(this->_epoll_socket, EPOLL_CTL_ADD, sock, &ev) < 0) { + log_err("epoll_ctl() failed: %s (%d) (sock=%d)", util::strerror(errno), errno, sock); + return -1; + } else { + log_debug("added listen socket to epoll (epoll_socket=%d, listen_socket=%d)", this->_epoll_socket, sock); } - connection_list.push_back(c); - return connection_list; + return 0; } -// }}} - -// {{{ protected methods +#endif // }}} // {{{ private methods src/lib/server.cc @@ -8,7 +8,16 @@ #ifndef __NET_SERVER_H__ #define __NET_SERVER_H__ +#include "config.h" + #include <netinet/tcp.h> +#include <sys/un.h> +#include <sys/types.h> +#include <sys/stat.h> + +#ifdef HAVE_EPOLL +#include <sys/epoll.h> +#endif #include "connection.h" @@ -21,19 +30,35 @@ namespace flare { * @todo support multi-port-listening (w/ epoll, kequeue, etc...) */ class server : public net { -protected: - int _port; - public: + static const int max_listen_socket = 256; static const int accept_retry_limit = 16; +protected: + int _listen_socket[max_listen_socket]; + int _listen_socket_index; +#ifdef HAVE_EPOLL + int _epoll_socket; +#endif + int _back_log; + +public: server(); virtual ~server(); - int get_port() { return this->_port; }; + int get_back_log() { return this->_back_log; }; + int set_back_log(int back_log) { this->_back_log = back_log; return 0; }; - int listen(int port, int backlog = 10); + int close(); + int listen(int port); + int listen(string uds); vector<shared_connection> wait(); + +protected: + int _set_listen_socket_option(int sock, int domain = PF_INET); +#ifdef HAVE_EPOLL + int _add_epoll_socket(int sock); +#endif }; } // namespace flare src/lib/server.h cf45f8965ee6673a787915424c65adc5884c8275 Masaki Fujimoto fujimoto@php.net http://github.com/fujimoto/flare/commit/36e94778c5126a72a3db890dd9f0df847cdff96a 36e94778c5126a72a3db890dd9f0df847cdff96a 2009-09-29T00:57:20-07:00 2009-09-29T00:57:20-07:00 added unix domain socket support (option "server-socket" is now available) and try to use epoll if available 4c36cbe406a9cb83301a59f14e3f88cf4722ac7c Masaki Fujimoto fujimoto@php.net