Compare commits

...

3 Commits

Author SHA1 Message Date
Jan Vidar Krey d183c7b3dc Try to send data using writev() to reduce copying and system calls. 2014-12-15 14:08:38 +01:00
Jan Vidar Krey be098144db Cleanup the io queue interface by hiding the internal structures of the
send queue, and read queue.
2014-12-15 13:13:30 +01:00
Jan Vidar Krey 71cdf158e4 Detect writev, pwritev, readv, preadv functions. 2014-12-15 11:31:23 +01:00
7 changed files with 128 additions and 27 deletions

View File

@ -26,7 +26,8 @@ find_package(Git)
find_package(Sqlite3) find_package(Sqlite3)
include(TestBigEndian) include(TestBigEndian)
include(CheckSymbolExists) include(CheckSymbolExists)
include(CheckFunctionExists)
include(CheckIncludeFile) include(CheckIncludeFile)
include(CheckTypeSize) include(CheckTypeSize)
@ -69,9 +70,19 @@ check_include_file(sys/types.h HAVE_SYS_TYPES_H)
if (HAVE_SYS_TYPES_H) if (HAVE_SYS_TYPES_H)
set (CMAKE_EXTRA_INCLUDE_FILES ${CMAKE_EXTRA_INCLUDE_FILES} "sys/types.h") set (CMAKE_EXTRA_INCLUDE_FILES ${CMAKE_EXTRA_INCLUDE_FILES} "sys/types.h")
endif() endif()
check_type_size( ssize_t SSIZE_T )
check_include_file(sys/uio.h HAVE_SYS_UIO_H)
if (HAVE_SYS_UIO_H)
set (CMAKE_EXTRA_INCLUDE_FILES ${CMAKE_EXTRA_INCLUDE_FILES} "sys/uio.h")
endif()
check_type_size(ssize_t SSIZE_T)
check_symbol_exists(memmem string.h HAVE_MEMMEM) check_symbol_exists(memmem string.h HAVE_MEMMEM)
check_symbol_exists(strndup string.h HAVE_STRNDUP) check_symbol_exists(strndup string.h HAVE_STRNDUP)
check_function_exists(writev HAVE_FUNC_WRITEV)
check_function_exists(pwritev HAVE_FUNC_PWRITEV)
check_function_exists(readv HAVE_FUNC_WRITEV)
check_function_exists(preadv HAVE_FUNC_PWRITEV)
include_directories("${PROJECT_SOURCE_DIR}") include_directories("${PROJECT_SOURCE_DIR}")
include_directories(${SQLITE3_INCLUDE_DIRS}) include_directories(${SQLITE3_INCLUDE_DIRS})

View File

@ -19,6 +19,23 @@
#include "uhub.h" #include "uhub.h"
struct ioq_send
{
size_t size; /** Size of send queue (in bytes, not messages) */
size_t offset; /** Queue byte offset in the first message. Should be 0 unless a partial write. */
#ifdef SSL_SUPPORT
size_t last_send; /** When using SSL, one have to send the exact same buffer and length if a write cannot complete. */
#endif
struct linked_list* queue; /** List of queued messages (struct adc_message) */
};
struct ioq_recv
{
char* buf;
size_t size;
};
#ifdef DEBUG_SENDQ #ifdef DEBUG_SENDQ
static void debug_msg(const char* prefix, struct adc_message* msg) static void debug_msg(const char* prefix, struct adc_message* msg)
{ {
@ -143,10 +160,46 @@ static void ioq_send_remove(struct ioq_send* q, struct adc_message* msg)
int ioq_send_send(struct ioq_send* q, struct net_connection* con) int ioq_send_send(struct ioq_send* q, struct net_connection* con)
{ {
int ret; int ret;
struct adc_message* msg = list_get_first(q->queue); struct adc_message* msg = list_get_first(q->queue);
if (!msg) return 0; if (!msg) return 0;
uhub_assert(msg->cache && *msg->cache); uhub_assert(msg->cache && *msg->cache);
#ifdef HAVE_FUNC_WRITEV
#define MAX_IOVEC 32
struct iovec vec[MAX_IOVEC];
size_t n = 0;
vec[0].iov_base = msg->cache + q->offset;
vec[0].iov_len = msg->length - q->offset;
for (struct adc_message* tmp = (struct adc_message*) list_get_next(q->queue), n = 1; tmp && n < MAX_IOVEC; tmp = (struct adc_message*) list_get_next(q->queue), n++)
{
vec[n].iov_base = msg->cache;
vec[n].iov_len = msg->length;
}
ret = net_con_writev(con, &iovec, n);
if (ret > 0)
{
while (ret > 0)
{
if (ret >= (msg->length - q->offset))
{
ret -= (msg->length - q->offset);
q->offset = 0;
ioq_send_remove(q, msg);
}
else
{
q->offset += ret;
return 0;
}
}
return 1;
}
return ret;
#else
ret = net_con_send(con, msg->cache + q->offset, msg->length - q->offset); ret = net_con_send(con, msg->cache + q->offset, msg->length - q->offset);
if (ret > 0) if (ret > 0)
@ -159,8 +212,10 @@ int ioq_send_send(struct ioq_send* q, struct net_connection* con)
return 1; return 1;
} }
return ret; return ret;
#endif
} }
int ioq_send_is_empty(struct ioq_send* q) int ioq_send_is_empty(struct ioq_send* q)
{ {
return (q->size - q->offset) == 0; return (q->size - q->offset) == 0;

View File

@ -21,25 +21,8 @@
#define HAVE_UHUB_IO_QUEUE_H #define HAVE_UHUB_IO_QUEUE_H
struct adc_message; struct adc_message;
struct linked_list; struct ioq_send;
typedef int (*ioq_write)(void* desc, const void* buf, size_t len); struct ioq_recv;
typedef int (*ioq_read)(void* desc, void* buf, size_t len);
struct ioq_send
{
size_t size; /** Size of send queue (in bytes, not messages) */
size_t offset; /** Queue byte offset in the first message. Should be 0 unless a partial write. */
#ifdef SSL_SUPPORT
size_t last_send; /** When using SSL, one have to send the exact same buffer and length if a write cannot complete. */
#endif
struct linked_list* queue; /** List of queued messages (struct adc_message) */
};
struct ioq_recv
{
char* buf;
size_t size;
};
/** /**
* Create a send queue * Create a send queue

View File

@ -80,13 +80,13 @@ static int check_send_queue(struct hub_info* hub, struct hub_user* user, struct
if (user_flag_get(user, flag_user_list)) if (user_flag_get(user, flag_user_list))
return 1; return 1;
if ((user->send_queue->size + msg->length) > get_max_send_queue(hub)) if ((ioq_send_get_bytes(user->send_queue) + msg->length) > get_max_send_queue(hub))
{ {
LOG_WARN("send queue overflowed, message discarded."); LOG_WARN("send queue overflowed, message discarded.");
return -1; return -1;
} }
if (user->send_queue->size > get_max_send_queue_soft(hub)) if (ioq_send_get_bytes(user->send_queue) > get_max_send_queue_soft(hub))
{ {
LOG_WARN("send queue soft overflowed."); LOG_WARN("send queue soft overflowed.");
return 0; return 0;

View File

@ -57,6 +57,48 @@ ssize_t net_con_send(struct net_connection* con, const void* buf, size_t len)
return ret; return ret;
} }
#ifdef HAVE_FUNC_WRITEV
ssize_t net_con_writev(struct net_connection* con, const struct iovec* iov, size_t iocnt)
{
int ret;
#ifdef SSL_SUPPORT
if (!con->ssl)
{
#endif
ret = writev(con->sd, iovec, (int) iocnt);
if (ret == -1)
{
if (is_blocked_or_interrupted())
return 0;
return -1;
}
#ifdef SSL_SUPPORT
}
else
{
ssize_t total = 0;
ret = 0;
while (iocnt--)
{
ret = net_ssl_send(con, buf, len);
if (ret >= 0)
total += ret;
else
{
break;
}
}
if (total == 0 && !is_blocked_or_interrupted())
return -1;
return total;
}
#endif /* SSL_SUPPORT */
return ret;
}
#endif
ssize_t net_con_recv(struct net_connection* con, void* buf, size_t len) ssize_t net_con_recv(struct net_connection* con, void* buf, size_t len)
{ {
int ret; int ret;

View File

@ -92,6 +92,11 @@ extern void net_con_close(struct net_connection* con);
*/ */
extern ssize_t net_con_send(struct net_connection* con, const void* buf, size_t len); extern ssize_t net_con_send(struct net_connection* con, const void* buf, size_t len);
#ifdef HAVE_FUNC_WRITEV
extern ssize_t net_con_writev(struct net_connection* con, const struct iovec* iov, size_t iocnt);
#endif
/** /**
* Receive data * Receive data
* *

View File

@ -88,6 +88,11 @@
#cmakedefine HAVE_STRNDUP #cmakedefine HAVE_STRNDUP
#cmakedefine HAVE_MEMMEM #cmakedefine HAVE_MEMMEM
#cmakedefine HAVE_SYS_UIO_H
#ifdef HAVE_SYS_TYPES_H
#include <sys/uio.h>
#endif
/* printf support for size_t and uint64_t */ /* printf support for size_t and uint64_t */
#if defined(WIN32) #if defined(WIN32)
#define PRINTF_SIZE_T "%Iu" #define PRINTF_SIZE_T "%Iu"