external/uhub
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' of git://github.com/janvidar/uhub

Browse Source
This commit is contained in:
root 2010-01-26 13:05:56 +03:00
commit 6bbdd24cb7
23 changed files with 343 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
ChangeLog
View File

@ -1,3 +1,15 @@
0.3.0:
- More user commands: ban, broadcast, mute, rules, history, myip, whoip, log
- Experimental SSL support
- Large rewrite of the network stack in order to support SSL.
- Added rule file for defining hub rules.
- Many crash fixes and other important bug fixes.
- Optimizations: O(1) timeout scheduler
- New sid allocation code.
- Added configurable server_listen_backlog (default 50).
- Added init.d scripts for RedHat/CentOS
0.2.8:
- Fix bug #13: getsockname() failure, use sockaddr from accept() instead.
- Fix bug #10: Improve logging, ensure logs are machine readable.

2
README
View File

@ -1,5 +1,5 @@
Welcome and thanks for downloading uHub, a high performance ADC p2p hub.
For the official documentation, bugs and other information, please visit:
http://www.extatic.org/uhub/
http://www.uhub.org/

1
autotest/test_hub.tcc
View File

@ -23,6 +23,7 @@ EXO_TEST(hub_net_startup, {
EXO_TEST(hub_config_initialize, {
config_defaults(&g_config);
g_config.server_port = 15111;
return 1;
});

4
src/core/commands.c
View File

@ -399,8 +399,8 @@ static int command_whoip(struct hub_info* hub, struct hub_user* user, struct hub
static int command_broadcast(struct hub_info* hub, struct hub_user* user, struct hub_command* cmd)
{
struct adc_message* command = adc_msg_construct(ADC_CMD_IMSG, strlen((cmd->message + 11)) + 6);
adc_msg_add_argument(command, (cmd->message + 11));
struct adc_message* command = adc_msg_construct(ADC_CMD_IMSG, strlen((cmd->message + 12)) + 6);
adc_msg_add_argument(command, (cmd->message + 12));
route_to_all(hub, command);
adc_msg_free(command);
return 0;

54
src/core/hub.c
View File

@ -450,37 +450,28 @@ static void hub_event_dispatcher(void* callback_data, struct event_data* message
}
}
static int start_listening_socket(const char* bind_addr, uint16_t port, int backlog, char* address_out)
static struct net_connection* start_listening_socket(const char* bind_addr, uint16_t port, int backlog, struct hub_info* hub)
{
struct net_connection* server;
struct sockaddr_storage addr;
socklen_t sockaddr_size;
int af, sd, ret;
int sd, ret;
if (ip_convert_address(bind_addr, port, (struct sockaddr*) &addr, &sockaddr_size) == -1)
{
return -1;
return 0;
}
af = addr.ss_family;
if (af == AF_INET)
{
net_address_to_string(AF_INET, &((struct sockaddr_in*) &addr)->sin_addr, address_out, INET6_ADDRSTRLEN);
}
else if (af == AF_INET6)
{
net_address_to_string(AF_INET6, &((struct sockaddr_in6*) &addr)->sin6_addr, address_out, INET6_ADDRSTRLEN);
}
sd = net_socket_create(af, SOCK_STREAM, IPPROTO_TCP);
sd = net_socket_create(addr.ss_family, SOCK_STREAM, IPPROTO_TCP);
if (sd == -1)
{
return -1;
return 0;
}
if ((net_set_reuseaddress(sd, 1) == -1) || (net_set_nonblocking(sd, 1) == -1))
{
net_close(sd);
return -1;
return 0;
}
ret = net_bind(sd, (struct sockaddr*) &addr, sockaddr_size);
@ -488,7 +479,7 @@ static int start_listening_socket(const char* bind_addr, uint16_t port, int back
{
LOG_ERROR("hub_start_service(): Unable to bind to TCP local address. errno=%d, str=%s", net_error(), net_error_string(net_error()));
net_close(sd);
return -1;
return 0;
}
ret = net_listen(sd, backlog);
@ -496,16 +487,19 @@ static int start_listening_socket(const char* bind_addr, uint16_t port, int back
{
LOG_ERROR("hub_start_service(): Unable to listen to socket");
net_close(sd);
return -1;
return 0;
}
return sd;
server = net_con_create();
net_con_initialize(server, sd, net_on_accept, hub, NET_EVENT_READ);
return server;
}
struct hub_info* hub_start_service(struct hub_config* config)
{
struct hub_info* hub = 0;
int server_tcp, ipv6_supported;
char address_buf[INET6_ADDRSTRLEN+1];
int ipv6_supported;
hub = hub_malloc_zero(sizeof(struct hub_info));
if (!hub)
@ -521,14 +515,14 @@ struct hub_info* hub_start_service(struct hub_config* config)
else
LOG_DEBUG("IPv6 not supported.");
server_tcp = start_listening_socket(config->server_bind_addr, config->server_port, config->server_listen_backlog, address_buf);
if (server_tcp == -1)
hub->server = start_listening_socket(config->server_bind_addr, config->server_port, config->server_listen_backlog, hub);
if (!hub->server)
{
hub_free(hub);
LOG_FATAL("Unable to start hub service");
return 0;
}
LOG_INFO("Starting " PRODUCT "/" VERSION ", listening on %s:%d...", address_buf, config->server_port);
LOG_INFO("Starting " PRODUCT "/" VERSION ", listening on %s:%d...", net_get_local_address(hub->server->sd), config->server_port);
#ifdef SSL_SUPPORT
if (config->tls_enable)
@ -558,22 +552,21 @@ struct hub_info* hub_start_service(struct hub_config* config)
}
#endif
hub->config = config;
hub->users = NULL;
if (uman_init(hub) == -1)
{
net_con_close(hub->server);
hub_free(hub);
net_close(server_tcp);
return 0;
}
if (event_queue_initialize(&hub->queue, hub_event_dispatcher, (void*) hub) == -1)
{
net_con_close(hub->server);
uman_shutdown(hub);
hub_free(hub);
net_close(server_tcp);
return 0;
}
@ -581,11 +574,11 @@ struct hub_info* hub_start_service(struct hub_config* config)
hub->sendbuf = hub_malloc(MAX_SEND_BUF);
if (!hub->recvbuf || !hub->sendbuf)
{
net_con_close(hub->server);
hub_free(hub->recvbuf);
hub_free(hub->sendbuf);
uman_shutdown(hub);
hub_free(hub);
net_close(server_tcp);
return 0;
}
@ -593,21 +586,18 @@ struct hub_info* hub_start_service(struct hub_config* config)
hub->logout_info = (struct linked_list*) list_create();
if (!hub->chat_history)
{
net_con_close(hub->server);
list_destroy(hub->chat_history);
list_destroy(hub->logout_info);
hub_free(hub->recvbuf);
hub_free(hub->sendbuf);
uman_shutdown(hub);
hub_free(hub);
net_close(server_tcp);
return 0;
}
hub->status = hub_status_running;
hub->server = net_con_create();
net_con_initialize(hub->server, server_tcp, net_on_accept, hub, NET_EVENT_READ);
g_hub = hub;
return hub;
}

32
src/core/main.c
View File

@ -33,13 +33,13 @@ static const char* arg_pid = 0;
static int arg_log_syslog = 0;
#if !defined(WIN32) && defined(USE_LIBEVENT)
void hub_handle_signal(int fd, short events, void* arg)
#if !defined(WIN32)
extern struct hub_info* g_hub;
void hub_handle_signal(int sig)
{
struct hub_info* hub = (struct hub_info*) arg;
int signal = fd;
struct hub_info* hub = g_hub;
switch (signal)
switch (sig)
{
case SIGINT:
LOG_INFO("Interrupted. Shutting down...");
@ -65,7 +65,6 @@ void hub_handle_signal(int fd, short events, void* arg)
}
}
static struct event signal_events[10];
static int signals[] =
{
SIGINT, /* Interrupt the application */
@ -77,11 +76,16 @@ static int signals[] =
void setup_signal_handlers(struct hub_info* hub)
{
sigset_t sig_set;
sigemptyset(&sig_set);
struct sigaction act;
act.sa_mask = sig_set;
act.sa_flags = SA_ONSTACK | SA_RESTART;
act.sa_handler = hub_handle_signal;
int i = 0;
for (i = 0; signals[i]; i++)
{
signal_set(&signal_events[i], signals[i], hub_handle_signal, hub);
if (signal_add(&signal_events[i], NULL))
if (sigaction(signals[i], &act, 0) != 0)
{
LOG_ERROR("Error setting signal handler %d", signals[i]);
}
@ -90,14 +94,8 @@ void setup_signal_handlers(struct hub_info* hub)
void shutdown_signal_handlers(struct hub_info* hub)
{
int i = 0;
for (i = 0; signals[i]; i++)
{
signal_del(&signal_events[i]);
}
}
#endif /* !WIN32 && USE_LIBEVENT*/
#endif /* !WIN32 */
int main_loop()
@ -133,7 +131,7 @@ int main_loop()
hub = hub_start_service(&configuration);
if (!hub)
return -1;
#if !defined(WIN32) && defined(USE_LIBEVENT)
#if !defined(WIN32)
setup_signal_handlers(hub);
#endif
}
@ -148,7 +146,7 @@ int main_loop()
} while (hub->status == hub_status_restart);
#if !defined(WIN32) && defined(USE_LIBEVENT)
#if !defined(WIN32)
shutdown_signal_handlers(hub);
#endif

22
src/core/netevent.c
View File

@ -27,7 +27,7 @@ extern struct hub_info* g_hub;
#ifdef DEBUG_SENDQ
void debug_sendq_send(struct hub_user* user, int sent, int total)
{
LOG_DUMP("SEND: sd=%d, %d/%d bytes\n", user->net.connection.sd, sent, total);
LOG_DUMP("SEND: sd=%d, %d/%d bytes\n", user->connection->sd, sent, total);
if (sent == -1)
{
int err = net_error();
@ -162,26 +162,16 @@ void net_event(struct net_connection* con, int event, void *arg)
int flag_close = 0;
#ifdef DEBUG_SENDQ
LOG_TRACE("net_event() : fd=%d, ev=%d, arg=%p", fd, (int) event, arg);
LOG_TRACE("net_event() : fd=%d, ev=%d, arg=%p", con->sd, (int) event, arg);
#endif
if (event == NET_EVENT_SOCKERROR)
{
hub_disconnect_user(g_hub, user, quit_socket_error);
return;
}
else if (event == NET_EVENT_CLOSED)
{
hub_disconnect_user(g_hub, user, quit_disconnected);
return;
}
else if (event == NET_EVENT_TIMEOUT)
if (event == NET_EVENT_TIMEOUT)
{
if (user_is_connecting(user))
{
hub_disconnect_user(g_hub, user, quit_timeout);
return;
}
return;
}
if (event & NET_EVENT_READ)
@ -238,7 +228,7 @@ void net_on_accept(struct net_connection* con, int event, void *arg)
if (acl_is_ip_banned(hub->acl, addr))
{
LOG_INFO("Denied [%s] (IP banned)", addr);
net_close(fd);
net_con_close(con);
continue;
}
@ -246,7 +236,7 @@ void net_on_accept(struct net_connection* con, int event, void *arg)
if (!probe)
{
LOG_ERROR("Unable to create probe after socket accepted. Out of memory?");
net_close(fd);
net_con_close(con);
break;
}
}

3
src/core/probe.c
View File

@ -26,8 +26,7 @@ static char probe_recvbuf[PROBE_RECV_SIZE];
static void probe_net_event(struct net_connection* con, int events, void *arg)
{
struct hub_probe* probe = (struct hub_probe*) net_con_get_ptr(con);
if (events == NET_EVENT_SOCKERROR || events == NET_EVENT_CLOSED || events == NET_EVENT_TIMEOUT)
if (events == NET_EVENT_TIMEOUT)
{
probe_destroy(probe);
return;

20
src/core/usermanager.c
View File

@ -19,8 +19,6 @@
#include "uhub.h"
#define USERMANAGER_TIMER
/*
* This callback function is used to clear user objects from the userlist.
* Should only be used in uman_shutdown().
@ -40,7 +38,6 @@ static void clear_user_list_callback(void* ptr)
}
}
void uman_update_stats(struct hub_info* hub)
{
const int factor = TIMEOUT_STATS;
@ -58,7 +55,6 @@ void uman_update_stats(struct hub_info* hub)
net_stats_reset();
}
void uman_print_stats(struct hub_info* hub)
{
LOG_INFO("Statistics users=" PRINTF_SIZE_T " (peak_users=" PRINTF_SIZE_T "), net_tx=%d KB/s, net_rx=%d KB/s (peak_tx=%d KB/s, peak_rx=%d KB/s)",
@ -70,6 +66,13 @@ void uman_print_stats(struct hub_info* hub)
(int) hub->stats.net_rx_peak / 1024);
}
static void timer_statistics(struct timeout_evt* t)
{
struct hub_info* hub = (struct hub_info*) t->ptr;
uman_update_stats(hub);
timeout_queue_reschedule(net_backend_get_timeout_queue(), hub->users->timeout, TIMEOUT_STATS);
}
int uman_init(struct hub_info* hub)
{
struct hub_user_manager* users = NULL;
@ -90,6 +93,10 @@ int uman_init(struct hub_info* hub)
return -1;
}
users->timeout = hub_malloc_zero(sizeof(struct timeout_evt));
timeout_evt_initialize(users->timeout, timer_statistics, hub);
timeout_queue_insert(net_backend_get_timeout_queue(), users->timeout, TIMEOUT_STATS);
hub->users = users;
return 0;
}
@ -100,6 +107,9 @@ int uman_shutdown(struct hub_info* hub)
if (!hub || !hub->users)
return -1;
timeout_queue_remove(net_backend_get_timeout_queue(), hub->users->timeout);
hub_free(hub->users->timeout);
if (hub->users->list)
{
list_clear(hub->users->list, &clear_user_list_callback);
@ -108,6 +118,8 @@ int uman_shutdown(struct hub_info* hub)
sid_pool_destroy(hub->users->sids);
hub_free(hub->users);
hub->users = 0;
return 0;
}

1
src/core/usermanager.h
View File

@ -28,6 +28,7 @@ struct hub_user_manager
uint64_t shared_size; /**<< "The total number of shared bytes among fully connected users." */
uint64_t shared_files; /**<< "The total number of shared files among fully connected users." */
struct linked_list* list; /**<< "Contains all logged in users" */
struct timeout_evt* timeout; /**<< "Timeout handler for statistics" */
};
/**

2
src/network/common.h
View File

@ -28,7 +28,6 @@
#define NET_CLEANUP 0x8000
#define NET_CON_STRUCT_BASIC \
int sd; /** socket descriptor */ \
uint32_t flags; /** Connection flags */ \
@ -38,6 +37,7 @@
#define NET_CON_STRUCT_SSL \
SSL* ssl; /** SSL handle */ \
uint32_t ssl_state; /** SSL state */ \
size_t write_len; /** Length of last SSL_write(), only used if flags is NET_WANT_SSL_READ. */ \
#ifdef SSL_SUPPORT

148
src/network/connection.c
View File

@ -21,6 +21,17 @@
#include "network/common.h"
#ifdef SSL_SUPPORT
enum uhub_tls_state
{
tls_st_none,
tls_st_error,
tls_st_accepting,
tls_st_connecting,
tls_st_connected,
tls_st_disconnecting,
};
static int handle_openssl_error(struct net_connection* con, int ret)
{
uhub_assert(con);
@ -30,30 +41,19 @@ static int handle_openssl_error(struct net_connection* con, int ret)
{
case SSL_ERROR_ZERO_RETURN:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_ZERO_RETURN", ret, error);
con->ssl_state = tls_st_error;
return -1;
case SSL_ERROR_WANT_READ:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_WANT_READ", ret, error);
net_con_update(con, NET_EVENT_READ | NET_WANT_SSL_READ);
con->flags |= NET_WANT_SSL_READ;
net_con_update(con, NET_EVENT_READ);
return 0;
case SSL_ERROR_WANT_WRITE:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_WANT_WRITE", ret, error);
net_con_update(con, NET_EVENT_READ | NET_EVENT_WRITE | NET_WANT_SSL_WRITE);
return 0;
case SSL_ERROR_WANT_CONNECT:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_WANT_CONNECT", ret, error);
net_con_update(con, NET_EVENT_READ | NET_EVENT_WRITE | NET_WANT_SSL_CONNECT);
return 0;
case SSL_ERROR_WANT_ACCEPT:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_WANT_ACCEPT", ret, error);
net_con_update(con, NET_EVENT_READ | NET_EVENT_WRITE | NET_WANT_SSL_ACCEPT);
return 0;
case SSL_ERROR_WANT_X509_LOOKUP:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_WANT_X509_LOOKUP", ret, error);
con->flags |= NET_WANT_SSL_WRITE;
net_con_update(con, NET_EVENT_READ | NET_EVENT_WRITE);
return 0;
case SSL_ERROR_SYSCALL:
@ -67,6 +67,7 @@ static int handle_openssl_error(struct net_connection* con, int ret)
case SSL_ERROR_SSL:
LOG_PROTO("SSL_get_error: ret=%d, error=%d: SSL_ERROR_SSL", ret, error);
/* internal openssl error */
con->ssl_state = tls_st_error;
return -1;
}
@ -76,7 +77,7 @@ static int handle_openssl_error(struct net_connection* con, int ret)
ssize_t net_con_ssl_accept(struct net_connection* con)
{
uhub_assert(con);
con->ssl_state = tls_st_accepting;
ssize_t ret = SSL_accept(con->ssl);
#ifdef NETWORK_DUMP_DEBUG
LOG_PROTO("SSL_accept() ret=%d", ret);
@ -84,6 +85,7 @@ ssize_t net_con_ssl_accept(struct net_connection* con)
if (ret > 0)
{
net_con_update(con, NET_EVENT_READ);
con->ssl_state = tls_st_connected;
}
else
{
@ -96,12 +98,14 @@ ssize_t net_con_ssl_connect(struct net_connection* con)
{
uhub_assert(con);
con->ssl_state = tls_st_connecting;
ssize_t ret = SSL_connect(con->ssl);
#ifdef NETWORK_DUMP_DEBUG
LOG_PROTO("SSL_connect() ret=%d", ret);
#endif
if (ret > 0)
{
con->ssl_state = tls_st_connected;
net_con_update(con, NET_EVENT_READ);
}
else
@ -136,28 +140,42 @@ ssize_t net_con_ssl_handshake(struct net_connection* con, enum net_con_ssl_mode
ssize_t net_con_send(struct net_connection* con, const void* buf, size_t len)
{
int ret = net_send(con->sd, buf, len, UHUB_SEND_SIGNAL);
int ret;
#ifdef SSL_SUPPORT
if (!con->ssl)
{
#endif
ret = net_send(con->sd, buf, len, UHUB_SEND_SIGNAL);
if (ret == -1)
{
if (net_error() == EWOULDBLOCK || net_error() == EINTR)
return 0;
return -1;
}
#ifdef SSL_SUPPORT
}
else
{
con->write_len = len;
ret = SSL_write(con->ssl, buf, len);
LOG_PROTO("SSL_write(con=%p, buf=%p, len=" PRINTF_SIZE_T ") => %d", con, buf, len, ret);
if (ret <= 0)
{
return -handle_openssl_error(con, ret);
}
}
#endif
return ret;
}
ssize_t net_con_recv(struct net_connection* con, void* buf, size_t len)
{
uhub_assert(con);
int ret;
#ifdef SSL_SUPPORT
if (!net_con_is_ssl(con))
{
#endif
int ret = net_recv(con->sd, buf, len, 0);
#ifdef NETWORK_DUMP_DEBUG
LOG_PROTO("net_recv: ret=%d", ret);
#endif
ret = net_recv(con->sd, buf, len, 0);
if (ret == -1)
{
if (net_error() == EWOULDBLOCK || net_error() == EINTR)
@ -168,16 +186,15 @@ ssize_t net_con_recv(struct net_connection* con, void* buf, size_t len)
{
return -1;
}
return ret;
#ifdef SSL_SUPPORT
}
else
{
int ret = SSL_read(con->ssl, buf, len);
#ifdef NETWORK_DUMP_DEBUG
LOG_PROTO("net_recv: ret=%d", ret);
#endif
if (con->ssl_state == tls_st_error)
return -1;
ret = SSL_read(con->ssl, buf, len);
LOG_PROTO("SSL_read(con=%p, buf=%p, len=" PRINTF_SIZE_T ") => %d", con, buf, len, ret);
if (ret > 0)
{
net_con_update(con, NET_EVENT_READ);
@ -186,9 +203,9 @@ ssize_t net_con_recv(struct net_connection* con, void* buf, size_t len)
{
return -handle_openssl_error(con, ret);
}
return ret;
}
#endif
return ret;
}
ssize_t net_con_peek(struct net_connection* con, void* buf, size_t len)
@ -234,10 +251,73 @@ void* net_con_get_ptr(struct net_connection* con)
void net_con_callback(struct net_connection* con, int events)
{
if (!(con->flags & NET_CLEANUP))
if (con->flags & NET_CLEANUP)
return;
if (events == NET_EVENT_TIMEOUT)
{
LOG_TRACE("net_con_callback(%p, %s%s%s)", con, (events & NET_EVENT_READ ? "R" : ""), (events & NET_EVENT_WRITE ? "W" : ""), events == NET_EVENT_TIMEOUT ? "TIMEOUT" : "");
LOG_TRACE("net_con_callback(%p, TIMEOUT", con);
con->callback(con, events, con->ptr);
}
return;
}
#ifdef SSL_SUPPORT
if (!con->ssl)
{
#endif
con->callback(con, events, con->ptr);
#ifdef SSL_SUPPORT
}
else
{
#ifdef NETWORK_DUMP_DEBUG
LOG_PROTO("net_con_event: events=%d, con=%p, state=%d", events, con, con->ssl_state);
#endif
switch (con->ssl_state)
{
case tls_st_none:
con->callback(con, events, con->ptr);
break;
case tls_st_error:
con->callback(con, NET_EVENT_READ, con->ptr);
break;
case tls_st_accepting:
if (net_con_ssl_accept(con) < 0)
{
con->callback(con, NET_EVENT_READ, con->ptr);
}
break;
case tls_st_connecting:
if (net_con_ssl_connect(con) < 0)
{
con->callback(con, NET_EVENT_READ, con->ptr);
}
break;
case tls_st_connected:
LOG_PROTO("tls_st_connected, events=%s%s, ssl_flags=%s%s", (events & NET_EVENT_READ ? "R" : ""), (events & NET_EVENT_WRITE ? "W" : ""), con->flags & NET_WANT_SSL_READ ? "R" : "", con->flags & NET_WANT_SSL_WRITE ? "W" : "");
if (events & NET_EVENT_WRITE && con->flags & NET_WANT_SSL_READ)
{
con->callback(con, events & NET_EVENT_READ, con->ptr);
return;
}
if (events & NET_EVENT_READ && con->flags & NET_WANT_SSL_WRITE)
{
con->callback(con, events & NET_EVENT_READ, con->ptr);
return;
}
con->callback(con, events, con->ptr);
break;
case tls_st_disconnecting:
return;
}
}
#endif
}

3
src/network/connection.h
View File

@ -27,8 +27,6 @@
#define NET_EVENT_TIMEOUT 0x0001
#define NET_EVENT_READ 0x0002
#define NET_EVENT_WRITE 0x0004
#define NET_EVENT_SOCKERROR 0x1000 /* Socket error, closed */
#define NET_EVENT_CLOSED 0x2000 /* Socket closed */
struct net_connection;
@ -53,7 +51,6 @@ extern void net_con_callback(struct net_connection* con, int events);
* Close the connection.
* This will ensure a connection is closed properly and will generate a NET_EVENT_DESTROYED event which indicates
* that the con can safely be deleted (or set to NULL).
* @returns 1 if the memory pointed to by con can be freed immediately, or 0 if it needs to go through the NET_EVENT_DESTROYED event.
*/
extern void net_con_close(struct net_connection* con);

51
src/network/epoll.c
View File

@ -95,6 +95,7 @@ int net_backend_initialize()
void net_backend_shutdown()
{
close(g_backend->epfd);
timeout_queue_shutdown(&g_backend->timeout_queue);
net_cleanup_shutdown(g_backend->cleaner);
hub_free(g_backend->conns);
hub_free(g_backend);
@ -106,8 +107,13 @@ void net_backend_shutdown()
int net_backend_process()
{
int n;
LOG_TRACE("epoll_wait: fd=%d, events=%x, max=%zu", g_backend->epfd, g_backend->events, MIN(g_backend->num, EPOLL_EVBUFFER));
int res = epoll_wait(g_backend->epfd, g_backend->events, MIN(g_backend->num, EPOLL_EVBUFFER), 1000);
size_t secs = timeout_queue_get_next_timeout(&g_backend->timeout_queue, g_backend->now);
LOG_TRACE("epoll_wait: fd=%d, events=%x, max=%zu, seconds=%d", g_backend->epfd, g_backend->events, MIN(g_backend->num, EPOLL_EVBUFFER), (int) secs);
int res = epoll_wait(g_backend->epfd, g_backend->events, MIN(g_backend->num, EPOLL_EVBUFFER), secs * 1000);
g_backend->now = time(0);
timeout_queue_process(&g_backend->timeout_queue, g_backend->now);
if (res == -1)
{
LOG_WARN("epoll_wait returned -1");
@ -181,32 +187,14 @@ void net_con_reinitialize(struct net_connection* con, net_connection_cb callback
void net_con_update(struct net_connection* con_, int events)
{
struct net_connection_epoll* con = (struct net_connection_epoll*) con_;
con->ev.events = 0;
if (events & NET_EVENT_READ) con->ev.events |= EPOLLIN;
if (events & NET_EVENT_WRITE) con->ev.events |= EPOLLOUT;
int newev = 0;
if (events & NET_EVENT_READ) newev |= EPOLLIN;
if (events & NET_EVENT_WRITE) newev |= EPOLLOUT;
#ifdef SSL_SUPPORT
if (events & NET_WANT_SSL_WRITE)
con->flags |= NET_WANT_SSL_WRITE;
else
con->flags &= ~NET_WANT_SSL_WRITE;
if (events & NET_WANT_SSL_READ)
con->flags |= NET_WANT_SSL_READ;
else
con->flags &= ~NET_WANT_SSL_READ;
if (events & NET_WANT_SSL_ACCEPT)
con->flags |= NET_WANT_SSL_ACCEPT;
else
con->flags &= ~NET_WANT_SSL_ACCEPT;
if (events & NET_WANT_SSL_CONNECT)
con->flags |= NET_WANT_SSL_CONNECT;
else
con->flags &= ~NET_WANT_SSL_CONNECT;
#endif /* SSL_SUPPORT */
if (newev == con->ev.events)
return;
con->ev.events = newev;
if (epoll_ctl(g_backend->epfd, EPOLL_CTL_MOD, con->sd, &con->ev) == -1)
{
LOG_TRACE("epoll_ctl() modify failed.");
@ -226,17 +214,16 @@ void net_con_close(struct net_connection* con_)
g_backend->num--;
}
if (timeout_evt_is_scheduled(con->timeout))
{
timeout_queue_remove(&g_backend->timeout_queue, con->timeout);
hub_free(con->timeout);
con->timeout = 0;
}
net_con_clear_timeout(con_);
if (epoll_ctl(g_backend->epfd, EPOLL_CTL_DEL, con->sd, &con->ev) == -1)
{
LOG_WARN("epoll_ctl() delete failed.");
}
net_close(con->sd);
con->sd = -1;
net_con_print("DEL", con);
net_cleanup_delayed_free(g_backend->cleaner, con_);
}

67
src/network/network.c
View File

@ -376,16 +376,16 @@ int net_accept(int fd, struct ip_addr_encap* ipaddr)
if (ipaddr)
{
memset(ipaddr, 0, sizeof(struct ip_addr_encap));
ipaddr->af = addr4->sin_family;
ipaddr->af = addr.ss_family;;
if (ipaddr->af == AF_INET6)
{
char address[INET6_ADDRSTRLEN+1] = { 0, };
net_address_to_string(AF_INET6, (void*) &addr6->sin6_addr, address, INET6_ADDRSTRLEN+1);
if (!strncmp(address, "::ffff:", 7))
if (strchr(address, '.'))
{
/* Hack to convert IPv6 mapped IPv4 addresses to true IPv4 addresses */
net_string_to_address(AF_INET, address + 7, (void*) &ipaddr->internal_ip_data.in);
ipaddr->af = AF_INET;
net_string_to_address(AF_INET, address, (void*) &ipaddr->internal_ip_data.in);
}
else
{
@ -526,7 +526,15 @@ const char* net_address_to_string(int af, const void* src, char* dst, socklen_t
return NULL;
#else
return inet_ntop(af, src, dst, cnt);
if (inet_ntop(af, src, dst, cnt))
{
if (af == AF_INET6 && strncmp(dst, "::ffff:", 7) == 0) /* IPv6 mapped IPv4 address. */
{
memmove(dst, dst + 7, cnt - 7);
}
return dst;
}
return NULL;
#endif
}
@ -573,9 +581,6 @@ int net_string_to_address(int af, const char* src, void* dst)
#endif
}
const char* net_get_peer_address(int fd)
{
static char address[INET6_ADDRSTRLEN+1];
@ -594,21 +599,16 @@ const char* net_get_peer_address(int fd)
if (getpeername(fd, (struct sockaddr*) name, &namelen) != -1)
{
int af = name4->sin_family;
int af = storage.ss_family;
if (af == AF_INET6)
{
net_address_to_string(af, (void*) &name6->sin6_addr, address, INET6_ADDRSTRLEN);
if (strncmp(address, "::ffff:", 7) == 0) /* IPv6 mapped IPv4 address. */
{
return &address[7];
}
return address;
}
else
{
net_address_to_string(af, (void*) &name4->sin_addr, address, INET6_ADDRSTRLEN);
return address;
}
return address;
}
else
{
@ -619,6 +619,45 @@ const char* net_get_peer_address(int fd)
return "0.0.0.0";
}
const char* net_get_local_address(int fd)
{
static char address[INET6_ADDRSTRLEN+1];
struct sockaddr_storage storage;
struct sockaddr_in6* name6;
struct sockaddr_in* name4;
struct sockaddr* name;
memset(address, 0, INET6_ADDRSTRLEN);
socklen_t namelen = sizeof(struct sockaddr_storage);
memset(&storage, 0, namelen);
name6 = (struct sockaddr_in6*) &storage;
name4 = (struct sockaddr_in*) &storage;
name = (struct sockaddr*) &storage;
if (getsockname(fd, (struct sockaddr*) name, &namelen) != -1)
{
int af = storage.ss_family;
if (af == AF_INET6)
{
net_address_to_string(af, (void*) &name6->sin6_addr, address, INET6_ADDRSTRLEN);
}
else
{
net_address_to_string(af, (void*) &name4->sin_addr, address, INET6_ADDRSTRLEN);
}
return address;
}
else
{
net_error_out(fd, "net_get_local_address");
net_stats_add_error();
}
return "0.0.0.0";
}
ssize_t net_recv(int fd, void* buf, size_t len, int flags)
{

2
src/network/network.h
View File

@ -224,6 +224,8 @@ extern int net_is_ipv6_supported();
*/
extern const char* net_get_peer_address(int fd);
extern const char* net_get_local_address(int fd);
/**
* See man(3) inet_ntop.
*/

21
src/network/select.c
View File

@ -84,6 +84,7 @@ int net_backend_initialize()
*/
void net_backend_shutdown()
{
timeout_queue_shutdown(&g_backend->timeout_queue);
net_cleanup_shutdown(g_backend->cleaner);
hub_free(g_backend->conns);
hub_free(g_backend);
@ -95,9 +96,14 @@ void net_backend_shutdown()
int net_backend_process()
{
int n, found, maxfd;
struct timeval tval = { 1, 0 };
struct timeval tval;
FD_ZERO(&g_backend->rfds);
FD_ZERO(&g_backend->wfds);
size_t secs = timeout_queue_get_next_timeout(&g_backend->timeout_queue, g_backend->now);
tval.tv_sec = secs;
tval.tv_usec = 0;
for (n = 0, found = 0; found < g_backend->num && n < g_backend->max; n++)
{
struct net_connection_select* con = g_backend->conns[n];
@ -111,6 +117,9 @@ int net_backend_process()
}
int res = select(maxfd+1, &g_backend->rfds, &g_backend->wfds, 0, &tval);
g_backend->now = time(0);
timeout_queue_process(&g_backend->timeout_queue, g_backend->now);
if (res == -1)
{
LOG_WARN("select returned -1");
@ -195,12 +204,10 @@ void net_con_close(struct net_connection* con)
g_backend->num--;
}
if (timeout_evt_is_scheduled(con->timeout))
{
timeout_queue_remove(&g_backend->timeout_queue, con->timeout);
hub_free(con->timeout);
con->timeout = 0;
}
net_con_clear_timeout(con);
net_close(con->sd);
con->sd = -1;
net_con_print("DEL", (struct net_connection_select*) con);
net_cleanup_delayed_free(g_backend->cleaner, con);

2
src/network/timer.c
View File

@ -41,7 +41,7 @@ void net_con_set_timeout(struct net_connection* con, int seconds)
void net_con_clear_timeout(struct net_connection* con)
{
if (timeout_evt_is_scheduled(con->timeout))
if (con->timeout && timeout_evt_is_scheduled(con->timeout))
{
timeout_queue_remove(net_backend_get_timeout_queue(), con->timeout);
hub_free(con->timeout);

8
src/tools/adcclient.c
View File

@ -75,19 +75,13 @@ static void event_callback(struct net_connection* con, int events, void *arg)
{
struct ADC_client* client = (struct ADC_client*) net_con_get_ptr(con);
if (events == NET_EVENT_SOCKERROR || events == NET_EVENT_CLOSED)
{
printf("NET_EVENT_SOCKERROR || NET_EVENT_CLOSED\n");
client->callback(client, ADC_CLIENT_DISCONNECTED, 0);
return;
}
if (events == NET_EVENT_TIMEOUT)
{
if (client->state == ps_conn)
{
client->callback(client, ADC_CLIENT_DISCONNECTED, 0);
}
return;
}
if (events & NET_EVENT_READ)

52
src/util/timeout.c
View File

@ -36,7 +36,6 @@ void timeout_evt_reset(struct timeout_evt* t)
int timeout_evt_is_scheduled(struct timeout_evt* t)
{
if (!t) return 0;
return !!t->prev;
}
@ -59,12 +58,11 @@ size_t timeout_queue_process(struct timeout_queue* t, time_t now)
size_t pos;
size_t events = 0;
struct timeout_evt* evt = 0;
for (pos = t->last; pos < now; pos++)
for (pos = t->last; pos <= now; pos++)
{
while ((evt = t->events[pos % t->max]))
{
timeout_queue_remove(t, evt);
timeout_evt_reset(evt);
evt->callback(evt);
events++;
}
@ -73,6 +71,19 @@ size_t timeout_queue_process(struct timeout_queue* t, time_t now)
return events;
}
size_t timeout_queue_get_next_timeout(struct timeout_queue* t, time_t now)
{
size_t seconds = 0;
while (t->events[(now + seconds) % t->max] == NULL && seconds < t->max)
{
seconds++;
}
if (seconds == 0)
return 1;
return seconds;
}
void timeout_queue_insert(struct timeout_queue* t, struct timeout_evt* evt, size_t seconds)
{
struct timeout_evt* first;
@ -82,16 +93,18 @@ void timeout_queue_insert(struct timeout_queue* t, struct timeout_evt* evt, size
first = t->events[pos];
if (!first)
if (first)
{
first->prev->next = evt;
evt->prev = first->prev;
first->prev = evt;
}
else
{
t->events[pos] = evt;
evt->prev = evt;
}
else
{
evt->prev = first->prev;
first->prev = evt;
}
evt->next = 0;
}
void timeout_queue_remove(struct timeout_queue* t, struct timeout_evt* evt)
@ -99,23 +112,32 @@ void timeout_queue_remove(struct timeout_queue* t, struct timeout_evt* evt)
size_t pos = (evt->timestamp % t->max);
struct timeout_evt* first = t->events[pos];
if (!first || !evt)
if (!first || !evt->prev)
return;
if (first == evt)
{
if (first->next)
first->next->prev = first->prev;
if (first->prev != first)
{
t->events[pos] = first->next;
t->events[pos]->prev = evt->prev;
}
else
{
t->events[pos] = 0;
}
}
else if (evt == first->prev)
{
first->prev = evt->prev;
evt->prev->next = 0;
}
else
{
evt->prev->next = evt->next;
if (evt->next)
evt->next->prev = evt->prev;
else
first->prev = evt->prev;
}
timeout_evt_reset(evt);
}
void timeout_queue_reschedule(struct timeout_queue* t, struct timeout_evt* evt, size_t seconds)

2
src/util/timeout.h
View File

@ -53,4 +53,6 @@ void timeout_queue_insert(struct timeout_queue*, struct timeout_evt*, size_t sec
void timeout_queue_remove(struct timeout_queue*, struct timeout_evt*);
void timeout_queue_reschedule(struct timeout_queue*, struct timeout_evt*, size_t seconds);
size_t timeout_queue_get_next_timeout(struct timeout_queue*, time_t now);
#endif /* HAVE_UHUB_TIMEOUT_HANDLER_H */

16
tools/create_certificate.sh Executable file
View File

@ -0,0 +1,16 @@
#!/bin/sh
OPENSSL=/usr/bin/openssl
NAME=certificate
if [ ! -x ${OPENSSL} ]; then
echo "Cannot locate the openssl utility: ${OPENSSL}"
exit 1
fi
${OPENSSL} genrsa -out ${NAME}.key 1024 &&
${OPENSSL} req -new -x509 -nodes -sha1 -days 365 -key ${NAME}.key > ${NAME}.crt &&
cat ${NAME}.key ${NAME}.crt > ${NAME}.pem && rm -f ${NAME}.key ${NAME}.crt
echo "Created certificate ${NAME}.pem"

2
version.h
View File

@ -9,7 +9,7 @@
#endif
#ifndef VERSION
#define VERSION "0.3.0-rc4" REVISION
#define VERSION "0.3.0" REVISION
#endif
#ifndef COPYRIGHT