Commit 515481e6 authored by Philip Withnall's avatar Philip Withnall Committed by Olivier Crête

socket: Add vectored I/O support for sending on sockets

Replace the send() API with a send_messages() API, which supports
sending multiple messages, each with multiple buffers rather than a
single monolithic buffer.

This doesn’t break API, as the socket API is not exposed outside
libnice. It does introduce a new struct: NiceOutputMessage, which is
analogous to struct mmsghdr and NiceInputMessage.

This includes updates to the test-bsd test to cover the changed API.

The existing nice_socket_send() API has been retained as a thin wrapper
around nice_socket_send_messages(), for convenience only. It’s hoped
that internal usage of this API will decline to the point where it can
be removed.
parent 55e53a9c
......@@ -217,4 +217,7 @@ memcpy_buffer_to_input_message (NiceInputMessage *message,
guint8 *
compact_input_message (NiceInputMessage *message, gsize *buffer_length);
guint8 *
compact_output_message (const NiceOutputMessage *message, gsize *buffer_length);
#endif /*_NICE_AGENT_PRIV_H */
......@@ -2781,6 +2781,20 @@ memcpy_buffer_to_input_message (NiceInputMessage *message,
return message->length;
}
/* Concatenate all the buffers in the given @message into a single, newly
* allocated, monolithic buffer which is returned. The length of the new buffer
* is returned in @buffer_length, and should be equal to the length field of
* @recv_message.
*
* The return value must be freed with g_free(). */
guint8 *
compact_output_message (const NiceOutputMessage *message, gsize *buffer_length)
{
/* This works as long as NiceInputMessage and NiceOutputMessage are layed out
* identically. */
return compact_input_message ((NiceInputMessage *) message, buffer_length);
}
/**
* nice_input_message_iter_reset:
* @iter: a #NiceInputMessageIter
......
......@@ -170,6 +170,39 @@ typedef struct {
gsize length; /* sum of the lengths of @buffers */
} NiceInputMessage;
/**
* NiceOutputMessage:
* @buffers: (array length=n_buffers): unowned array of #GOutputVector buffers
* which contain data to transmit for this message
* @n_buffers: number of #GOutputVectors in @buffers, or -1 to indicate @buffers
* is %NULL-terminated
* @to: (allow-none): address of the peer to transmit the message to, or %NULL
* to use the default address for the outbound socket
* @length: total number of valid bytes contiguously stored in @buffers
*
* Represents a single message to transmit on the network. For reliable
* connections, this is essentially just an array of buffers (specifically,
* @to can be ignored). for non-reliable connections, it represents a single
* packet to send to the OS.
*
* @n_buffers may be -1 to indicate that @buffers is terminated by a
* #GOutputVector with a %NULL buffer pointer.
*
* By providing arrays of #NiceOutputMessages to functions like
* nice_agent_send_messages_nonblocking(), multiple messages may be transmitted
* with a single call, which is more efficient than making multiple calls in a
* loop. In this manner, nice_agent_send_messages_nonblocking() is analogous to
* sendmmsg(); and #NiceOutputMessage to struct mmsghdr.
*
* Since: 0.1.5
*/
typedef struct {
GOutputVector *buffers;
gint n_buffers;
const NiceAddress *to;
gsize length;
} NiceOutputMessage;
#define NICE_TYPE_AGENT nice_agent_get_type()
......
......@@ -92,12 +92,12 @@ struct to_be_sent {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len);
static void add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages);
static void free_to_be_sent (struct to_be_sent *tbs);
......@@ -125,7 +125,7 @@ nice_http_socket_new (NiceSocket *base_socket,
sock->fileno = priv->base_socket->fileno;
sock->addr = priv->base_socket->addr;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -136,6 +136,9 @@ nice_http_socket_new (NiceSocket *base_socket,
gchar *credential = NULL;
gchar host[INET6_ADDRSTRLEN];
gint port = nice_address_get_port (&priv->addr);
GOutputVector local_bufs;
NiceOutputMessage local_messages;
nice_address_to_string (&priv->addr, host);
if (username) {
......@@ -158,7 +161,14 @@ nice_http_socket_new (NiceSocket *base_socket,
credential? credential : "" );
g_free (credential);
nice_socket_send (priv->base_socket, NULL, strlen (msg), msg);
local_bufs.buffer = msg;
local_bufs.size = strlen (msg);
local_messages.buffers = &local_bufs;
local_messages.n_buffers = 1;
local_messages.to = NULL;
local_messages.length = local_bufs.size;
nice_socket_send_messages (priv->base_socket, &local_messages, 1);
priv->state = HTTP_STATE_INIT;
g_free (msg);
}
......@@ -565,23 +575,25 @@ retry:
return -1;
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
HttpPriv *priv = sock->priv;
if (priv->state == HTTP_STATE_CONNECTED) {
if (priv->base_socket)
return nice_socket_send (priv->base_socket, to, len, buf);
else
return FALSE;
/* Fast path. */
if (!priv->base_socket)
return -1;
return nice_socket_send_messages (priv->base_socket, messages, n_messages);
} else if (priv->state == HTTP_STATE_ERROR) {
return FALSE;
return -1;
} else {
add_to_be_sent (sock, to, buf, len);
add_to_be_sent (sock, messages, n_messages);
}
return TRUE;
return n_messages;
}
......@@ -593,22 +605,46 @@ socket_is_reliable (NiceSocket *sock)
static void
add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len)
add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
HttpPriv *priv = sock->priv;
struct to_be_sent *tbs = NULL;
guint i;
if (len <= 0)
if (n_messages == 0)
return;
tbs = g_slice_new0 (struct to_be_sent);
tbs->buf = g_memdup (buf, len);
tbs->length = len;
if (to)
tbs->to = *to;
g_queue_push_tail (&priv->send_queue, tbs);
/* Compact the message’s buffers before queueing. */
for (i = 0; i < n_messages; i++) {
const NiceOutputMessage *message = &messages[i];
struct to_be_sent *tbs = NULL;
guint j;
gsize message_len_remaining = message->length;
gsize offset = 0;
if (message->length == 0)
continue;
tbs = g_slice_new0 (struct to_be_sent);
tbs->buf = g_malloc (message->length);
tbs->length = message->length;
if (message->to)
tbs->to = *message->to;
g_queue_push_tail (&priv->send_queue, tbs);
for (j = 0;
(message->n_buffers >= 0 && j < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[j].buffer != NULL);
j++) {
const GOutputVector *buffer = &message->buffers[j];
gsize len;
len = MIN (buffer->size, message_len_remaining);
memcpy (tbs->buf + offset, buffer->buffer, len);
message_len_remaining -= len;
offset += len;
}
}
}
......
......@@ -58,8 +58,8 @@ typedef struct {
struct to_be_sent {
guint length;
gchar *buf;
guint8 *buf; /* owned */
gsize length;
NiceAddress to;
};
......@@ -91,12 +91,12 @@ static const gchar SSL_CLIENT_HANDSHAKE[] = {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len);
static void add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages);
static void free_to_be_sent (struct to_be_sent *tbs);
......@@ -112,7 +112,7 @@ nice_pseudossl_socket_new (NiceSocket *base_socket)
sock->fileno = priv->base_socket->fileno;
sock->addr = priv->base_socket->addr;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -173,7 +173,8 @@ socket_recv_messages (NiceSocket *sock,
struct to_be_sent *tbs = NULL;
priv->handshaken = TRUE;
while ((tbs = g_queue_pop_head (&priv->send_queue))) {
nice_socket_send (priv->base_socket, &tbs->to, tbs->length, tbs->buf);
nice_socket_send (priv->base_socket, &tbs->to, tbs->length,
(const gchar *) tbs->buf);
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
}
......@@ -188,19 +189,21 @@ socket_recv_messages (NiceSocket *sock,
return 0;
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
PseudoSSLPriv *priv = sock->priv;
if (priv->handshaken) {
if (priv->base_socket)
return nice_socket_send (priv->base_socket, to, len, buf);
else
/* Fast path: pass directly through to the base socket once the handshake is
* complete. */
if (priv->base_socket == NULL)
return FALSE;
return nice_socket_send_messages (priv->base_socket, messages, n_messages);
} else {
add_to_be_sent (sock, to, buf, len);
add_to_be_sent (sock, messages, n_messages);
}
return TRUE;
}
......@@ -214,22 +217,41 @@ socket_is_reliable (NiceSocket *sock)
static void
add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len)
add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
PseudoSSLPriv *priv = sock->priv;
struct to_be_sent *tbs = NULL;
if (len <= 0)
return;
tbs = g_slice_new0 (struct to_be_sent);
tbs->buf = g_memdup (buf, len);
tbs->length = len;
if (to)
tbs->to = *to;
g_queue_push_tail (&priv->send_queue, tbs);
guint i;
for (i = 0; i < n_messages; i++) {
struct to_be_sent *tbs;
const NiceOutputMessage *message = &messages[i];
guint j;
gsize offset = 0;
tbs = g_slice_new0 (struct to_be_sent);
/* Compact the buffer. */
tbs->buf = g_malloc (message->length);
tbs->length = message->length;
if (message->to != NULL)
tbs->to = *message->to;
g_queue_push_tail (&priv->send_queue, tbs);
for (j = 0;
(message->n_buffers >= 0 && j < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[j].buffer != NULL);
j++) {
const GOutputVector *buffer = &message->buffers[j];
gsize len;
len = MIN (message->length - offset, buffer->size);
memcpy (tbs->buf + offset, buffer->buffer, len);
offset += len;
}
g_assert_cmpuint (offset, ==, message->length);
}
}
static void
......
......@@ -93,11 +93,66 @@ nice_socket_recv_messages (NiceSocket *sock,
return sock->recv_messages (sock, recv_messages, n_recv_messages);
}
gboolean
nice_socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
/**
* nice_socket_send_messages:
* @sock: a #NiceSocket
* @messages: (array length=n_messages) (in caller-allocates):
* array of #NiceOutputMessages containing the messages to send
* @n_messages: number of elements in the @messages array
*
* Send up to @n_messages on the socket, in a non-reliable, non-blocking
* fashion. The total size of the buffers in each #NiceOutputMessage
* must be at most the maximum UDP payload size (65535 bytes), or excess
* bytes will be silently dropped.
*
* On success, the number of messages transmitted from @messages is returned,
* which may be less than @n_messages if the call would have blocked
* part-way through. If the socket would have blocked to begin with, or if
* @n_messages is zero, zero is returned. On failure, a negative value is
* returned, but no further error information is available. Calling this
* function on a socket which has closed is an error, and a negative value is
* returned.
*
* If a positive N is returned, the first N messages in @messages have been
* sent in full, and the remaining messages have not been sent at all.
*
* If #NiceOutputMessage::to is specified for a message, that will be used as
* the destination address for the message. Otherwise, if %NULL, the default
* destination for @sock will be used.
*
* Every field of every #NiceOutputMessage is guaranteed to be unmodified when
* this function returns.
*
* Returns: number of messages successfully sent from @messages, or a negative
* value on error
*
* Since: 0.1.5
*/
gint
nice_socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
return sock->send (sock, to, len, buf);
g_return_val_if_fail (sock != NULL, -1);
g_return_val_if_fail (n_messages == 0 || messages != NULL, -1);
return sock->send_messages (sock, messages, n_messages);
}
/* Convenience wrapper around nice_socket_send_messages(). Returns the number of
* bytes sent on success (which will be @len), zero if sending would block, or
* -1 on error. */
gssize
nice_socket_send (NiceSocket *sock, const NiceAddress *to, gsize len,
const gchar *buf)
{
GOutputVector local_buf = { buf, len };
NiceOutputMessage local_message = { &local_buf, 1, to, len };
gint ret;
ret = nice_socket_send_messages (sock, &local_message, 1);
if (ret == 1)
return len;
return ret;
}
gboolean
......
......@@ -63,8 +63,9 @@ struct _NiceSocket
* n_recv_messages is 0, recv_messages may be NULL. */
gint (*recv_messages) (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
gboolean (*send) (NiceSocket *sock, const NiceAddress *to, guint len,
const gchar *buf);
/* As above, @n_messages may be zero. Iff so, @messages may be %NULL. */
gint (*send_messages) (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages);
gboolean (*is_reliable) (NiceSocket *sock);
void (*close) (NiceSocket *sock);
void *priv;
......@@ -75,9 +76,12 @@ gint
nice_socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
gboolean
nice_socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
gint
nice_socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages);
gssize
nice_socket_send (NiceSocket *sock, const NiceAddress *addr, gsize len,
const gchar *buf);
gboolean
nice_socket_is_reliable (NiceSocket *sock);
......
......@@ -69,8 +69,8 @@ typedef struct {
struct to_be_sent {
guint length;
gchar *buf;
guint8 *buf; /* owned */
gsize length;
NiceAddress to;
};
......@@ -78,12 +78,12 @@ struct to_be_sent {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
static void add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len);
static void add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages);
static void free_to_be_sent (struct to_be_sent *tbs);
......@@ -105,7 +105,7 @@ nice_socks5_socket_new (NiceSocket *base_socket,
sock->fileno = priv->base_socket->fileno;
sock->addr = priv->base_socket->addr;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -330,7 +330,7 @@ socket_recv_messages (NiceSocket *sock,
}
while ((tbs = g_queue_pop_head (&priv->send_queue))) {
nice_socket_send (priv->base_socket, &tbs->to,
tbs->length, tbs->buf);
tbs->length, (const gchar *) tbs->buf);
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
}
......@@ -418,21 +418,22 @@ socket_recv_messages (NiceSocket *sock,
return -1;
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
Socks5Priv *priv = sock->priv;
if (priv->state == SOCKS_STATE_CONNECTED) {
if (priv->base_socket)
return nice_socket_send (priv->base_socket, to, len, buf);
else
/* Fast path: pass through to the base socket once connected. */
if (priv->base_socket == NULL)
return FALSE;
return nice_socket_send_messages (priv->base_socket, messages, n_messages);
} else if (priv->state == SOCKS_STATE_ERROR) {
return FALSE;
} else {
add_to_be_sent (sock, to, buf, len);
add_to_be_sent (sock, messages, n_messages);
}
return TRUE;
}
......@@ -446,22 +447,41 @@ socket_is_reliable (NiceSocket *sock)
static void
add_to_be_sent (NiceSocket *sock, const NiceAddress *to,
const gchar *buf, guint len)
add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
Socks5Priv *priv = sock->priv;
struct to_be_sent *tbs = NULL;
if (len <= 0)
return;
guint i;
tbs = g_slice_new0 (struct to_be_sent);
tbs->buf = g_memdup (buf, len);
tbs->length = len;
if (to)
tbs->to = *to;
g_queue_push_tail (&priv->send_queue, tbs);
for (i = 0; i < n_messages; i++) {
struct to_be_sent *tbs;
const NiceOutputMessage *message = &messages[i];
guint j;
gsize offset = 0;
tbs = g_slice_new0 (struct to_be_sent);
/* Compact the buffer. */
tbs->buf = g_malloc (message->length);
tbs->length = message->length;
if (message->to != NULL)
tbs->to = *message->to;
g_queue_push_tail (&priv->send_queue, tbs);
for (j = 0;
(message->n_buffers >= 0 && j < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[j].buffer != NULL);
j++) {
const GOutputVector *buffer = &message->buffers[j];
gsize len;
len = MIN (message->length - offset, buffer->size);
memcpy (tbs->buf + offset, buffer->buffer, len);
offset += len;
}
g_assert_cmpuint (offset, ==, message->length);
}
}
......
......@@ -62,8 +62,8 @@ typedef struct {
} TcpPriv;
struct to_be_sent {
guint length;
gchar *buf;
guint8 *buf;
gsize length;
gboolean can_drop;
};
......@@ -72,13 +72,13 @@ struct to_be_sent {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
static void add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len,
gboolean head);
static void add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *message,
gsize message_offset, gboolean head);
static void free_to_be_sent (struct to_be_sent *tbs);
static gboolean socket_send_more (GSocket *gsocket, GIOCondition condition,
gpointer data);
......@@ -169,7 +169,7 @@ nice_tcp_bsd_socket_new (GMainContext *ctx, NiceAddress *addr)
priv->error = FALSE;
sock->fileno = gsock;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -252,59 +252,88 @@ socket_recv_messages (NiceSocket *sock,
return i;
}
/* Data sent to this function must be a single entity because buffers can be
* dropped if the bandwidth isn't fast enough. So do not send a message in
* multiple chunks. */
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gssize
socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
{
TcpPriv *priv = sock->priv;
int ret;
gssize ret;
GError *gerr = NULL;
/* Don't try to access the socket if it had an error, otherwise we risk a
crash with SIGPIPE (Broken pipe) */
* crash with SIGPIPE (Broken pipe) */
if (priv->error)
return -1;
/* First try to send the data, don't send it later if it can be sent now
this way we avoid allocating memory on every send */
* this way we avoid allocating memory on every send */
if (g_queue_is_empty (&priv->send_queue)) {
ret = g_socket_send (sock->fileno, buf, len, NULL, &gerr);
ret = g_socket_send_message (sock->fileno, NULL, message->buffers,
message->n_buffers, NULL, 0, G_SOCKET_MSG_NONE, NULL, &gerr);
if (ret < 0) {
if(g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)
|| g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_FAILED)) {
add_to_be_sent (sock, buf, len, FALSE);
g_error_free (gerr);
return TRUE;
} else {
g_error_free (gerr);
return FALSE;
if (g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK) ||
g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_FAILED)) {
/* Queue the message and send it later. */
add_to_be_sent (sock, message, 0, FALSE);
ret = message->length;
}
} else if ((guint)ret < len) {
add_to_be_sent (sock, buf + ret, len - ret, TRUE);
return TRUE;
g_error_free (gerr);
} else if ((gsize) ret < message->length) {
/* Partial send. */
add_to_be_sent (sock, message, ret, TRUE);
ret = message->length;
}
} else {
if (g_queue_get_length(&priv->send_queue) >= MAX_QUEUE_LENGTH) {
int peek_idx = 0;
/* If the queue is too long, drop whatever packets we can. */
if (g_queue_get_length (&priv->send_queue) >= MAX_QUEUE_LENGTH) {
guint peek_idx = 0;
struct to_be_sent *tbs = NULL;
while ((tbs = g_queue_peek_nth (&priv->send_queue, peek_idx)) != NULL) {
if (tbs->can_drop) {
tbs = g_queue_pop_nth (&priv->send_queue, peek_idx);
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
free_to_be_sent (tbs);
break;
} else {
peek_idx++;
}
}
}
add_to_be_sent (sock, buf, len, FALSE);
/* Queue the message and send it later. */
add_to_be_sent (sock, message, 0, FALSE);
ret = message->length;
}
return TRUE;
return ret;
}
/* Data sent to this function must be a single entity because buffers can be
* dropped if the bandwidth isn't fast enough. So do not send a message in
* multiple chunks. */
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
guint i;
for (i = 0; i < n_messages; i++) {
const NiceOutputMessage *message = &messages[i];
gssize len;
len = socket_send_message (sock, message);
if (len < 0) {
/* Error. */
return len;
} else if (len == 0) {
/* EWOULDBLOCK. */
break;
}
}
return i;
}
static gboolean
......@@ -348,28 +377,37 @@ socket_send_more (
/* connection hangs up */
ret = -1;
} else {
ret = g_socket_send (sock->fileno, tbs->buf, tbs->length, NULL, &gerr);
GOutputVector local_bufs = { tbs->buf, tbs->length };
ret = g_socket_send_message (sock->fileno, NULL, &local_bufs, 1, NULL, 0,
G_SOCKET_MSG_NONE, NULL, &gerr);
}
if (ret < 0) {
if(gerr != NULL &&
if (gerr != NULL &&
g_error_matches (gerr, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
add_to_be_sent (sock, tbs->buf, tbs->length, TRUE);
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
GOutputVector local_buf = { tbs->buf, tbs->length };
NiceOutputMessage local_message = {
&local_buf, 1, NULL, local_buf.size
};
add_to_be_sent (sock, &local_message, 0, TRUE);
free_to_be_sent (tbs);
g_error_free (gerr);
break;
}
g_error_free (gerr);
} else if (ret < (int) tbs->length) {
add_to_be_sent (sock, tbs->buf + ret, tbs->length - ret, TRUE);
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
GOutputVector local_buf = { tbs->buf + ret, tbs->length - ret };
NiceOutputMessage local_message = {
&local_buf, 1, NULL, local_buf.size
};
add_to_be_sent (sock, &local_message, 0, TRUE);
free_to_be_sent (tbs);
break;
}
g_free (tbs->buf);
g_slice_free (struct to_be_sent, tbs);
free_to_be_sent (tbs);
}
if (g_queue_is_empty (&priv->send_queue)) {
......@@ -386,19 +424,25 @@ socket_send_more (
}
/* Queue data starting at byte offset @message_offset from @message’s
* buffers. */
static void
add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len, gboolean head)
add_to_be_sent (NiceSocket *sock, const NiceOutputMessage *message,
gsize message_offset, gboolean head)
{
TcpPriv *priv = sock->priv;
struct to_be_sent *tbs = NULL;
struct to_be_sent *tbs;
guint j;
gsize offset = 0;
if (len <= 0)
if (message_offset >= message->length)
return;
tbs = g_slice_new0 (struct to_be_sent);
tbs->buf = g_memdup (buf, len);
tbs->length = len;
tbs->buf = g_malloc (message->length - message_offset);
tbs->length = message->length - message_offset;
tbs->can_drop = !head;
if (head)
g_queue_push_head (&priv->send_queue, tbs);
else
......@@ -410,6 +454,26 @@ add_to_be_sent (NiceSocket *sock, const gchar *buf, guint len, gboolean head)
sock, NULL);
g_source_attach (priv->io_source, priv->context);
}
/* Move the data into the buffer. */
for (j = 0;
(message->n_buffers >= 0 && j < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[j].buffer != NULL);
j++) {
const GOutputVector *buffer = &message->buffers[j];
gsize len;
/* Skip this buffer if it’s within @message_offset. */
if (buffer->size <= message_offset) {
message_offset -= buffer->size;
continue;
}
len = MIN (tbs->length - offset, buffer->size - message_offset);
memcpy (tbs->buf + offset, (guint8 *) buffer->buffer + message_offset, len);
offset += len;
message_offset -= len;
}
}
......
......@@ -69,8 +69,8 @@ typedef struct {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
NiceSocket *
......@@ -86,7 +86,7 @@ nice_tcp_turn_socket_new (NiceSocket *base_socket,
sock->fileno = priv->base_socket->fileno;
sock->addr = priv->base_socket->addr;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -222,36 +222,99 @@ socket_recv_messages (NiceSocket *socket,
return i;
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gssize
socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
{
TurnTcpPriv *priv = sock->priv;
gchar padbuf[3] = {0, 0, 0};
int padlen = (len%4) ? 4 - (len%4) : 0;
gchar buffer[MAX_UDP_MESSAGE_SIZE + sizeof(guint16) + sizeof(padbuf)];
guint buffer_len = 0;
guint8 padbuf[3] = {0, 0, 0};
GOutputVector *local_bufs;
NiceOutputMessage local_message;
guint j;
gint ret;
guint n_bufs;
guint16 header_buf;
/* Count the number of buffers. */
if (message->n_buffers == -1) {
n_bufs = 0;
for (j = 0; message->buffers[j].buffer != NULL; j++)
n_bufs++;
} else {
n_bufs = message->n_buffers;
}
if (priv->compatibility != NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9 &&
priv->compatibility != NICE_TURN_SOCKET_COMPATIBILITY_RFC5766)
padlen = 0;
/* Allocate a new array of buffers, covering all the buffers in the input
* @message, but with an additional one for a header and one for a footer. */
local_bufs = g_malloc_n (n_bufs + 2, sizeof (GOutputVector));
local_message.buffers = local_bufs;
local_message.n_buffers = n_bufs + 2;
local_message.to = message->to;
local_message.length = message->length;
/* Copy the existing buffers across. */
for (j = 0; j < n_bufs; j++) {
local_bufs[j + 1].buffer = message->buffers[j].buffer;
local_bufs[j + 1].size = message->buffers[j].size;
}
/* Header buffer. */
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_GOOGLE) {
guint16 tmpbuf = htons (len);
memcpy (buffer + buffer_len, (gchar *)&tmpbuf, sizeof(guint16));
buffer_len += sizeof(guint16);
header_buf = htons (message->length);
local_bufs[0].buffer = &header_buf;
local_bufs[0].size = sizeof (header_buf);
local_message.length += sizeof (header_buf);
} else {
/* Skip over the allocated header buffer. */
local_message.buffers++;
local_message.n_buffers--;
}
memcpy (buffer + buffer_len, buf, len);
buffer_len += len;
/* Tail buffer. */
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9 ||
priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_RFC5766) {
memcpy (buffer + buffer_len, padbuf, padlen);
buffer_len += padlen;
gsize padlen = (message->length % 4) ? 4 - (message->length % 4) : 0;
local_bufs[n_bufs].buffer = &padbuf;
local_bufs[n_bufs].size = padlen;
local_message.length += padlen;
} else {
/* Skip over the allocated tail buffer. */
local_message.n_buffers--;
}
ret = nice_socket_send_messages (priv->base_socket, &local_message, 1);
g_free (local_bufs);
if (ret == 1)
return local_message.length;
return ret;
}
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
guint i;
for (i = 0; i < n_messages; i++) {
const NiceOutputMessage *message = &messages[i];
gssize len;
len = socket_send_message (sock, message);
if (len < 0) {
/* Error. */
return len;
} else if (len == 0) {
/* EWOULDBLOCK. */
break;
}
}
return nice_socket_send (priv->base_socket, to, buffer_len, buffer);
return i;
}
......
......@@ -116,8 +116,8 @@ typedef struct {
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
static void priv_process_pending_bindings (TurnPriv *priv);
......@@ -229,7 +229,7 @@ nice_turn_socket_new (GMainContext *ctx, NiceAddress *addr,
priv_send_data_queue_destroy);
sock->addr = *addr;
sock->fileno = base_socket->fileno;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -547,9 +547,8 @@ socket_dequeue_all_data (TurnPriv *priv, const NiceAddress *to)
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gssize
socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
{
TurnPriv *priv = (TurnPriv *) sock->priv;
StunMessage msg;
......@@ -561,34 +560,67 @@ socket_send (NiceSocket *sock, const NiceAddress *to,
} sa;
GList *i = priv->channels;
ChannelBinding *binding = NULL;
gint ret;
for (; i; i = i->next) {
ChannelBinding *b = i->data;
if (nice_address_equal (&b->peer, to)) {
if (nice_address_equal (&b->peer, message->to)) {
binding = b;
break;
}
}
nice_address_copy_to_sockaddr (to, &sa.addr);
nice_address_copy_to_sockaddr (message->to, &sa.addr);
if (binding) {
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9 ||
priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_RFC5766) {
if (len + sizeof(uint32_t) <= sizeof(buffer)) {
uint16_t len16 = htons ((uint16_t) len);
uint16_t channel16 = htons (binding->channel);
if (message->length + sizeof(uint32_t) <= sizeof(buffer)) {
guint j;
uint16_t len16, channel16;
gsize message_offset = 0;
len16 = htons ((uint16_t) message->length);
channel16 = htons (binding->channel);
memcpy (buffer, &channel16, sizeof(uint16_t));
memcpy (buffer + sizeof(uint16_t), &len16,sizeof(uint16_t));
memcpy (buffer + sizeof(uint32_t), buf, len);
msg_len = len + sizeof(uint32_t);
/* FIXME: Slow path! This should be replaced by code which manipulates
* the GOutputVector array, rather than the buffer contents
* themselves. */
for (j = 0;
(message->n_buffers >= 0 && j < (guint) message->n_buffers) ||
(message->n_buffers < 0 && message->buffers[j].buffer != NULL);
j++) {
const GOutputVector *out_buf = &message->buffers[j];
gsize out_len;
out_len = MIN (message->length - message_offset, out_buf->size);
memcpy (buffer + sizeof (uint32_t) + message_offset,
out_buf->buffer, out_len);
message_offset += out_len;
}
msg_len = message->length + sizeof(uint32_t);
} else {
return 0;
}
} else {
return nice_socket_send (priv->base_socket, &priv->server_addr, len, buf);
NiceOutputMessage local_message = {
message->buffers, message->n_buffers, &priv->server_addr,
message->length
};
ret = nice_socket_send_messages (priv->base_socket, &local_message, 1);
if (ret == 1)
return message->length;
return ret;
}
} else {
guint8 *compacted_buf;
gsize compacted_buf_len;
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_DRAFT9 ||
priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_RFC5766) {
if (!stun_agent_init_indication (&priv->agent, &msg,
......@@ -619,7 +651,7 @@ socket_send (NiceSocket *sock, const NiceAddress *to,
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_GOOGLE &&
priv->current_binding &&
nice_address_equal (&priv->current_binding->peer, to)) {
nice_address_equal (&priv->current_binding->peer, message->to)) {
stun_message_append32 (&msg, STUN_ATTRIBUTE_OPTIONS, 1);
}
}
......@@ -634,10 +666,20 @@ socket_send (NiceSocket *sock, const NiceAddress *to,
stun_message_ensure_ms_realm(&msg, priv->ms_realm);
}
/* Slow path! We have to compact the buffers to append them to the message.
* FIXME: This could be improved by adding vectored I/O support to
* stun_message_append_bytes(). */
compacted_buf = compact_output_message (message, &compacted_buf_len);
if (stun_message_append_bytes (&msg, STUN_ATTRIBUTE_DATA,
buf, len) != STUN_MESSAGE_RETURN_SUCCESS)
compacted_buf, compacted_buf_len) != STUN_MESSAGE_RETURN_SUCCESS) {
g_free (compacted_buf);
goto send;
}
g_free (compacted_buf);
/* Finish the message. */
msg_len = stun_agent_finish_message (&priv->agent, &msg,
priv->password, priv->password_len);
if (msg_len > 0 && stun_message_get_class (&msg) == STUN_REQUEST) {
......@@ -653,22 +695,59 @@ socket_send (NiceSocket *sock, const NiceAddress *to,
if (msg_len > 0) {
if (priv->compatibility == NICE_TURN_SOCKET_COMPATIBILITY_RFC5766 &&
!priv_has_permission_for_peer (priv, to)) {
if (!priv_has_sent_permission_for_peer (priv, to)) {
priv_send_create_permission (priv, NULL, to);
!priv_has_permission_for_peer (priv, message->to)) {
if (!priv_has_sent_permission_for_peer (priv, message->to)) {
priv_send_create_permission (priv, NULL, message->to);
}
/* enque data */
nice_debug ("enqueuing data");
socket_enqueue_data(priv, to, msg_len, (gchar *)buffer);
return TRUE;
socket_enqueue_data(priv, message->to, msg_len, (gchar *)buffer);
return msg_len;
} else {
return nice_socket_send (priv->base_socket, &priv->server_addr,
msg_len, (gchar *)buffer);
GOutputVector local_buf = { buffer, msg_len };
NiceOutputMessage local_message = {
&local_buf, 1, &priv->server_addr, msg_len
};
ret = nice_socket_send_messages (priv->base_socket, &local_message, 1);
if (ret == 1)
return msg_len;
return ret;
}
}
send:
return nice_socket_send (priv->base_socket, to, len, buf);
send:
/* Error condition pass through to the base socket. */
ret = nice_socket_send_messages (priv->base_socket, message, 1);
if (ret == 1)
return message->length;
return ret;
}
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
guint i;
for (i = 0; i < n_messages; i++) {
const NiceOutputMessage *message = &messages[i];
gssize len;
len = socket_send_message (sock, message);
if (len < 0) {
/* Error. */
return len;
} else if (len == 0) {
/* EWOULDBLOCK. */
break;
}
}
return i;
}
static gboolean
......
......@@ -59,8 +59,8 @@
static void socket_close (NiceSocket *sock);
static gint socket_recv_messages (NiceSocket *sock,
NiceInputMessage *recv_messages, guint n_recv_messages);
static gboolean socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf);
static gint socket_send_messages (NiceSocket *sock,
const NiceOutputMessage *messages, guint n_messages);
static gboolean socket_is_reliable (NiceSocket *sock);
struct UdpBsdSocketPrivate
......@@ -142,7 +142,7 @@ nice_udp_bsd_socket_new (NiceAddress *addr)
nice_address_init (&priv->niceaddr);
sock->fileno = gsock;
sock->send = socket_send;
sock->send_messages = socket_send_messages;
sock->recv_messages = socket_recv_messages;
sock->is_reliable = socket_is_reliable;
sock->close = socket_close;
......@@ -224,34 +224,71 @@ socket_recv_messages (NiceSocket *sock,
return i;
}
static gboolean
socket_send (NiceSocket *sock, const NiceAddress *to,
guint len, const gchar *buf)
static gssize
socket_send_message (NiceSocket *sock, const NiceOutputMessage *message)
{
struct UdpBsdSocketPrivate *priv = sock->priv;
ssize_t sent;
GError *child_error = NULL;
gssize len;
if (!nice_address_is_valid (&priv->niceaddr) ||
!nice_address_equal (&priv->niceaddr, to)) {
!nice_address_equal (&priv->niceaddr, message->to)) {
union {
struct sockaddr_storage storage;
struct sockaddr addr;
} sa;
GSocketAddress *gaddr;
g_assert (message->to != NULL);
if (priv->gaddr)
g_object_unref (priv->gaddr);
nice_address_copy_to_sockaddr (to, &sa.addr);
nice_address_copy_to_sockaddr (message->to, &sa.addr);
gaddr = g_socket_address_new_from_native (&sa.addr, sizeof(sa));
priv->gaddr = gaddr;
if (gaddr == NULL)
return -1;
priv->niceaddr = *to;
priv->niceaddr = *message->to;
}
sent = g_socket_send_to (sock->fileno, priv->gaddr, buf, len, NULL, NULL);
len = g_socket_send_message (sock->fileno, priv->gaddr, message->buffers,
message->n_buffers, NULL, 0, G_SOCKET_MSG_NONE, NULL, &child_error);
if (len < 0) {
if (g_error_matches (child_error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
len = 0;
return sent == (ssize_t)len;
g_error_free (child_error);
}
return len;
}
static gint
socket_send_messages (NiceSocket *sock, const NiceOutputMessage *messages,
guint n_messages)
{
guint i;
for (i = 0; i < n_messages; i++) {
const NiceOutputMessage *message = &messages[i];
gssize len;
len = socket_send_message (sock, message);
if (len < 0) {
/* Error. */
return len;
} else if (len == 0) {
/* EWOULDBLOCK. */
break;
}
}
return i;
}
static gboolean
......
......@@ -104,12 +104,12 @@ test_simple_send_recv (void)
nice_address_set_port (&tmp, nice_address_get_port (&server->addr));
/* Send and receive stuff. */
g_assert (nice_socket_send (client, &tmp, 5, "hello"));
g_assert_cmpint (nice_socket_send (client, &tmp, 5, "hello"), ==, 5);
g_assert_cmpint (socket_recv (server, &tmp, 5, buf), ==, 5);
g_assert_cmpint (strncmp (buf, "hello", 5), ==, 0);
g_assert (nice_socket_send (server, &tmp, 5, "uryyb"));
g_assert_cmpint (nice_socket_send (server, &tmp, 5, "uryyb"), ==, 5);
g_assert_cmpint (socket_recv (client, &tmp, 5, buf), ==, 5);
g_assert_cmpint (strncmp (buf, "uryyb", 5), ==, 0);
......@@ -126,6 +126,8 @@ test_zero_send_recv (void)
NiceSocket *sock;
NiceAddress tmp;
gchar buf[5];
NiceOutputMessage local_out_message;
NiceInputMessage local_in_message;
sock = nice_udp_bsd_socket_new (NULL);
g_assert (sock != NULL);
......@@ -135,12 +137,21 @@ test_zero_send_recv (void)
nice_address_set_port (&tmp, nice_address_get_port (&sock->addr));
g_assert_cmpuint (nice_address_get_port (&tmp), !=, 0);
g_assert (nice_socket_send (sock, &tmp, 0, "some-buffer-to-be-ignored"));
g_assert (nice_socket_send (sock, &tmp, 0, NULL));
g_assert_cmpint (nice_socket_send (sock, &tmp, 0, "ignore-me"), ==, 0);
g_assert_cmpint (nice_socket_send (sock, &tmp, 0, NULL), ==, 0);
g_assert_cmpint (socket_recv (sock, &tmp, 0, buf), ==, 0);
g_assert_cmpint (socket_recv (sock, &tmp, 0, NULL), ==, 0);
/* And again with messages. */
g_assert_cmpint (nice_socket_send_messages (sock,
&local_out_message, 0), ==, 0);
g_assert_cmpint (nice_socket_send_messages (sock, NULL, 0), ==, 0);
g_assert_cmpint (nice_socket_recv_messages (sock,
&local_in_message, 0), ==, 0);
g_assert_cmpint (nice_socket_recv_messages (sock, NULL, 0), ==, 0);
nice_socket_free (sock);
}
......@@ -181,7 +192,7 @@ test_multi_buffer_recv (void)
memset (dummy_buf, 0xaa, sizeof (dummy_buf));
/* Send and receive. */
g_assert (nice_socket_send (client, &tmp, 11, "hello-world"));
g_assert_cmpint (nice_socket_send (client, &tmp, 11, "hello-world"), ==, 11);
g_assert_cmpuint (nice_socket_recv_messages (server, &message, 1), ==, 1);
g_assert_cmpuint (message.length, ==, 11);
......@@ -219,7 +230,7 @@ fill_send_buf (guint8 *buf, gsize buf_len, guint seed)
static void
test_multi_message_recv (guint n_sends, guint n_receives,
guint n_bufs_per_message, gsize send_buf_size, gsize recv_buf_size,
guint expected_n_received_messages)
guint expected_n_received_messages, guint expected_n_sent_messages)
{
NiceSocket *server;
NiceSocket *client;
......@@ -236,45 +247,66 @@ test_multi_message_recv (guint n_sends, guint n_receives,
/* Send and receive stuff. */
{
GInputVector *bufs;
NiceInputMessage *messages;
GInputVector *recv_bufs;
NiceInputMessage *recv_messages;
GOutputVector *send_bufs;
NiceOutputMessage *send_messages;
guint i, j;
guint8 *_expected_recv_buf, *send_buf;
guint8 *_expected_recv_buf;
gsize expected_recv_buf_len;
/* Set up the send buffers. */
send_bufs = g_malloc0_n (n_sends * n_bufs_per_message,
sizeof (GOutputVector));
send_messages = g_malloc0_n (n_sends, sizeof (NiceOutputMessage));
for (i = 0; i < n_sends; i++) {
for (j = 0; j < n_bufs_per_message; j++) {
guint8 *buf = g_slice_alloc (send_buf_size);
send_bufs[i * n_bufs_per_message + j].buffer = buf;
send_bufs[i * n_bufs_per_message + j].size = send_buf_size;
/* Set up the buffer data. */
fill_send_buf (buf, send_buf_size, i);
}
send_messages[i].buffers = send_bufs + i * n_bufs_per_message;
send_messages[i].n_buffers = n_bufs_per_message;
send_messages[i].to = &tmp;
send_messages[i].length = 0;
}
/* Set up the receive buffers. Yay for dynamic tests! */
bufs = g_malloc0_n (n_receives * n_bufs_per_message, sizeof (GInputVector));
messages = g_malloc0_n (n_receives, sizeof (NiceInputMessage));
recv_bufs = g_malloc0_n (n_receives * n_bufs_per_message,
sizeof (GInputVector));
recv_messages = g_malloc0_n (n_receives, sizeof (NiceInputMessage));
for (i = 0; i < n_receives; i++) {
for (j = 0; j < n_bufs_per_message; j++) {
bufs[i * n_bufs_per_message + j].buffer = g_slice_alloc (recv_buf_size);
bufs[i * n_bufs_per_message + j].size = recv_buf_size;
recv_bufs[i * n_bufs_per_message + j].buffer =
g_slice_alloc (recv_buf_size);
recv_bufs[i * n_bufs_per_message + j].size = recv_buf_size;
/* Initialise the buffer to try to catch out-of-bounds accesses. */
memset (bufs[i * n_bufs_per_message + j].buffer, 0xaa, recv_buf_size);
memset (recv_bufs[i * n_bufs_per_message + j].buffer, 0xaa,
recv_buf_size);
}
messages[i].buffers = bufs + i * n_bufs_per_message;
messages[i].n_buffers = n_bufs_per_message;
messages[i].from = NULL;
messages[i].length = 0;
recv_messages[i].buffers = recv_bufs + i * n_bufs_per_message;
recv_messages[i].n_buffers = n_bufs_per_message;
recv_messages[i].from = NULL;
recv_messages[i].length = 0;
}
/* Send multiple packets. */
send_buf = g_slice_alloc (send_buf_size);
for (i = 0; i < n_sends; i++) {
fill_send_buf (send_buf, send_buf_size, i);
g_assert (nice_socket_send (client, &tmp, send_buf_size,
(gchar *) send_buf));
}
g_slice_free1 (send_buf_size, send_buf);
g_assert_cmpint (
nice_socket_send_messages (client, send_messages, n_sends), ==,
expected_n_sent_messages);
/* Receive things. */
g_assert_cmpuint (
nice_socket_recv_messages (server, messages, n_receives), ==,
g_assert_cmpint (
nice_socket_recv_messages (server, recv_messages, n_receives), ==,
expected_n_received_messages);
/* Check all of the things. The sizes should not have been modified. */
......@@ -282,20 +314,18 @@ test_multi_message_recv (guint n_sends, guint n_receives,
_expected_recv_buf = g_slice_alloc (expected_recv_buf_len);
for (i = 0; i < expected_n_received_messages; i++) {
NiceInputMessage *message = &messages[i];
NiceInputMessage *message = &recv_messages[i];
guint8 *expected_recv_buf = _expected_recv_buf;
gsize expected_len;
expected_len = MIN (send_buf_size, expected_recv_buf_len);
expected_len = MIN (send_buf_size * n_bufs_per_message,
expected_recv_buf_len);
g_assert_cmpuint (message->length, ==, expected_len);
/* Build the expected buffer as a concatenation of the expected values of
* all receive buffers in the message. */
memset (expected_recv_buf, 0xaa, expected_recv_buf_len);
fill_send_buf (expected_recv_buf, expected_len, i);
if (expected_recv_buf_len > send_buf_size) {
memset (expected_recv_buf + expected_len, 0xaa,
expected_recv_buf_len - send_buf_size);
}
for (j = 0; j < n_bufs_per_message; j++) {
g_assert_cmpuint (message->buffers[j].size, ==, recv_buf_size);
......@@ -329,25 +359,34 @@ main (void)
{
guint i;
struct {
guint n_sends;
guint n_receives;
guint n_sends; /* messages */
guint expected_n_sent_messages;
guint n_receives; /* messages */
guint expected_n_received_messages;
guint n_bufs_per_message;
gsize send_buf_size;
gsize recv_buf_size;
guint expected_n_received_messages;
} test_cases[] = {
{ 2, 2, 1, 100, 100, 2 }, /* same number of sends and receives */
{ 4, 2, 2, 100, 77, 2 }, /* more sends than receives */
{ 1, 4, 4, 10, 100, 1 }, /* more receives than sends */
{ 100, 100, 1, 100, 64, 100 }, /* small receive buffer (data loss) */
{ 100, 100, 10, 100, 8, 100 }, /* small receive buffers (data loss) */
/* same number of sends and receives */
{ 2, 2, 2, 2, 1, 100, 100 }, /* send 200B, receive 200B */
/* more sends than receives */
{ 4, 4, 2, 2, 2, 100, 77 }, /* send 800B, receive 308B */
/* more receives than sends */
{ 1, 1, 4, 1, 4, 10, 100 }, /* send 40B, receive 1600B */
/* small receive buffer (data loss) */
{ 100, 100, 100, 100, 1, 100, 64 }, /* send 10000B, receive 6400B */
/* small receive buffers (data loss) */
{ 50, 50, 50, 50, 10, 100, 8 }, /* send 50000B, receive 4000B */
};
for (i = 0; i < G_N_ELEMENTS (test_cases); i++) {
test_multi_message_recv (test_cases[i].n_sends, test_cases[i].n_receives,
test_cases[i].n_bufs_per_message, test_cases[i].send_buf_size,
test_cases[i].recv_buf_size,
test_cases[i].expected_n_received_messages);
test_cases[i].expected_n_received_messages,
test_cases[i].expected_n_sent_messages);
}
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment