Commit 0655aab0 authored by Philip Withnall's avatar Philip Withnall Committed by Olivier Crête

tests: Add tests for the I/O stream classes

parent d285b5a5
......@@ -129,9 +129,15 @@ tests/test
tests/test-add-remove-stream
tests/test-address
tests/test-bsd
tests/test-build-io-stream
tests/test-dribble
tests/test-fallback
tests/test-fullmode
tests/test-io-stream-cancelling
tests/test-io-stream-closing
tests/test-io-stream-thread
tests/test-io-stream-pollable
tests/test-send-recv
tests/test-mainloop
tests/test-priority
tests/test-pseudotcp
......
......@@ -26,6 +26,12 @@ check_PROGRAMS = \
test \
test-address \
test-add-remove-stream \
test-build-io-stream \
test-io-stream-thread \
test-io-stream-closing \
test-io-stream-cancelling \
test-io-stream-pollable \
test-send-recv \
test-priority \
test-mainloop \
test-fullmode \
......@@ -53,6 +59,23 @@ test_address_LDADD = $(COMMON_LDADD)
test_add_remove_stream_LDADD = $(COMMON_LDADD)
test_build_io_stream_LDADD = $(COMMON_LDADD)
test_io_stream_thread_SOURCES = test-io-stream-thread.c test-io-stream-common.c
test_io_stream_thread_LDADD = $(COMMON_LDADD)
test_io_stream_closing_SOURCES = test-io-stream-closing.c test-io-stream-common.c
test_io_stream_closing_LDADD = $(COMMON_LDADD)
test_io_stream_cancelling_SOURCES = test-io-stream-cancelling.c test-io-stream-common.c
test_io_stream_cancelling_LDADD = $(COMMON_LDADD)
test_io_stream_pollable_SOURCES = test-io-stream-pollable.c test-io-stream-common.c
test_io_stream_pollable_LDADD = $(COMMON_LDADD)
test_send_recv_SOURCES = test-send-recv.c test-io-stream-common.c
test_send_recv_LDADD = $(COMMON_LDADD)
test_priority_LDADD = $(COMMON_LDADD)
test_mainloop_LDADD = $(COMMON_LDADD)
......
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2013 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include <string.h>
#include "agent.h"
static void
test_invalid_stream (NiceAddress *addr)
{
NiceAgent *agent;
NiceIOStream *io_stream;
GInputStream *input_stream;
GOutputStream *output_stream;
uint8_t data[65536];
GError *error = NULL;
agent = nice_agent_new_reliable (NULL, NICE_COMPATIBILITY_RFC5245);
nice_agent_add_local_address (agent, addr);
/* Try building an I/O stream for an invalid stream. All its operations should
* return G_IO_ERROR_BROKEN_PIPE. */
io_stream = nice_agent_build_io_stream (agent, 5, 5);
g_assert (G_IS_IO_STREAM (io_stream));
g_assert (NICE_IS_IO_STREAM (io_stream));
input_stream = g_io_stream_get_input_stream (G_IO_STREAM (io_stream));
g_assert (
g_input_stream_read (input_stream, data, G_N_ELEMENTS (data),
NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_BROKEN_PIPE);
g_clear_error (&error);
output_stream = g_io_stream_get_output_stream (G_IO_STREAM (io_stream));
g_assert (
g_output_stream_write (output_stream, data, G_N_ELEMENTS (data),
NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_BROKEN_PIPE);
g_clear_error (&error);
g_object_unref (io_stream);
}
static void
test_io_stream_properties (NiceAddress *addr, gboolean add_stream_first)
{
NiceAgent *agent;
guint stream_id;
NiceIOStream *io_stream;
GInputStream *input_stream;
GOutputStream *output_stream;
agent = nice_agent_new_reliable (NULL, NICE_COMPATIBILITY_RFC5245);
nice_agent_add_local_address (agent, addr);
if (add_stream_first) {
/* Add a stream. */
stream_id = nice_agent_add_stream (agent, 1);
} else {
/* Guess at the stream ID, then check it later. This tests the case where
* the I/O stream is built before the stream is created in the NiceAgent
* (but no I/O operations are performed on it until after both have been
* created). */
stream_id = 1;
}
/* Try building an I/O stream around it. */
io_stream = nice_agent_build_io_stream (agent, stream_id, 1);
g_assert (G_IS_IO_STREAM (io_stream));
g_assert (NICE_IS_IO_STREAM (io_stream));
if (!add_stream_first) {
g_assert (stream_id == nice_agent_add_stream (agent, 1));
}
/* Check various initial properties. */
g_assert (!g_io_stream_is_closed (G_IO_STREAM (io_stream)));
g_assert (!g_io_stream_has_pending (G_IO_STREAM (io_stream)));
/* Check the input stream’s properties. */
input_stream = g_io_stream_get_input_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_INPUT_STREAM (input_stream));
g_assert (NICE_IS_INPUT_STREAM (input_stream));
g_assert (!g_input_stream_is_closed (input_stream));
g_assert (!g_input_stream_has_pending (input_stream));
/* Check the output stream’s properties. */
output_stream = g_io_stream_get_output_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_OUTPUT_STREAM (output_stream));
g_assert (NICE_IS_OUTPUT_STREAM (output_stream));
g_assert (!g_output_stream_is_closing (output_stream));
g_assert (!g_output_stream_is_closed (output_stream));
g_assert (!g_output_stream_has_pending (output_stream));
/* Remove the component and check that the I/O streams close. */
nice_agent_remove_stream (agent, stream_id);
g_assert (g_io_stream_is_closed (G_IO_STREAM (io_stream)));
g_assert (g_input_stream_is_closed (input_stream));
g_assert (g_output_stream_is_closed (output_stream));
g_object_unref (io_stream);
g_object_unref (agent);
}
static void
test_pollable_properties (NiceAddress *addr)
{
NiceAgent *agent;
guint stream_id;
NiceIOStream *io_stream;
GInputStream *input_stream;
GOutputStream *output_stream;
GPollableInputStream *pollable_input_stream;
GPollableOutputStream *pollable_output_stream;
guint8 buf[65536];
GError *error = NULL;
GSource *stream_source;
agent = nice_agent_new_reliable (NULL, NICE_COMPATIBILITY_RFC5245);
nice_agent_add_local_address (agent, addr);
/* Add a stream. */
stream_id = nice_agent_add_stream (agent, 1);
/* Try building an I/O stream around it. */
io_stream = nice_agent_build_io_stream (agent, stream_id, 1);
g_assert (G_IS_IO_STREAM (io_stream));
g_assert (NICE_IS_IO_STREAM (io_stream));
/* Check the input stream’s properties. */
input_stream = g_io_stream_get_input_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_POLLABLE_INPUT_STREAM (input_stream));
pollable_input_stream = G_POLLABLE_INPUT_STREAM (input_stream);
g_assert (g_pollable_input_stream_can_poll (pollable_input_stream));
g_assert (!g_pollable_input_stream_is_readable (pollable_input_stream));
g_assert (
g_pollable_input_stream_read_nonblocking (pollable_input_stream,
buf, sizeof (buf), NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
g_clear_error (&error);
stream_source =
g_pollable_input_stream_create_source (pollable_input_stream, NULL);
g_assert (stream_source != NULL);
g_source_unref (stream_source);
/* Check the output stream’s properties. */
output_stream = g_io_stream_get_output_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_POLLABLE_OUTPUT_STREAM (output_stream));
pollable_output_stream = G_POLLABLE_OUTPUT_STREAM (output_stream);
g_assert (g_pollable_output_stream_can_poll (pollable_output_stream));
g_assert (!g_pollable_output_stream_is_writable (pollable_output_stream));
g_assert (
g_pollable_output_stream_write_nonblocking (pollable_output_stream,
buf, sizeof (buf), NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
g_clear_error (&error);
stream_source =
g_pollable_output_stream_create_source (pollable_output_stream, NULL);
g_assert (stream_source != NULL);
g_source_unref (stream_source);
/* Remove the component and check that the I/O streams close. */
nice_agent_remove_stream (agent, stream_id);
g_assert (!g_pollable_input_stream_is_readable (pollable_input_stream));
g_assert (!g_pollable_output_stream_is_writable (pollable_output_stream));
g_assert (
g_pollable_input_stream_read_nonblocking (pollable_input_stream,
buf, sizeof (buf), NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED);
g_clear_error (&error);
g_assert (
g_pollable_output_stream_write_nonblocking (pollable_output_stream,
buf, sizeof (buf), NULL, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED);
g_clear_error (&error);
g_object_unref (io_stream);
g_object_unref (agent);
}
static gboolean
source_cancel_cb (gpointer user_data)
{
GCancellable *cancellable = user_data;
g_cancellable_cancel (cancellable);
return FALSE;
}
static gboolean
source_cancelled_cb (GObject *pollable_stream, gpointer user_data)
{
GMainLoop *main_loop = user_data;
/* Try and check that the callback was invoked due to cancellation rather than
* a poll() event on the socket itself. */
if (G_IS_POLLABLE_INPUT_STREAM (pollable_stream)) {
g_assert (
!g_pollable_input_stream_is_readable (
G_POLLABLE_INPUT_STREAM (pollable_stream)));
} else {
g_assert (
!g_pollable_output_stream_is_writable (
G_POLLABLE_OUTPUT_STREAM (pollable_stream)));
}
g_main_loop_quit (main_loop);
return FALSE;
}
static gboolean
source_timeout_cb (gpointer user_data)
{
g_error ("check_pollable_source_cancellation() took too long. Aborting.");
return FALSE;
}
/* Check that cancelling a GCancellable which is associated with a pollable
* stream’s GSource invokes a callback from that source in the main loop. This
* uses a main context with three sources: the pollable source, an idle source
* to trigger the cancellation, and a timeout source to fail the test if it
* takes too long. */
static void
check_pollable_source_cancellation (GSource *pollable_source,
GCancellable *cancellable)
{
GMainContext *main_context;
GMainLoop *main_loop;
GSource *idle_source, *timeout_source;
main_context = g_main_context_new ();
main_loop = g_main_loop_new (main_context, FALSE);
/* Set up the pollable source. */
g_source_set_callback (pollable_source, (GSourceFunc) source_cancelled_cb,
main_loop, NULL);
g_source_attach (pollable_source, main_context);
/* Idle source to cancel the cancellable. */
idle_source = g_idle_source_new ();
g_source_set_callback (idle_source, (GSourceFunc) source_cancel_cb,
cancellable, NULL);
g_source_attach (idle_source, main_context);
g_source_unref (idle_source);
/* Timeout. */
timeout_source = g_timeout_source_new (30000);
g_source_set_callback (timeout_source, (GSourceFunc) source_timeout_cb,
NULL, NULL);
g_source_attach (timeout_source, main_context);
g_source_unref (timeout_source);
/* Run the main loop and expect to quit it immediately as the pollable source
* is cancelled. */
g_main_loop_run (main_loop);
g_assert (g_cancellable_is_cancelled (cancellable));
g_main_loop_unref (main_loop);
g_main_context_unref (main_context);
}
static void
test_pollable_cancellation (NiceAddress *addr)
{
NiceAgent *agent;
guint stream_id;
NiceIOStream *io_stream;
GInputStream *input_stream;
GOutputStream *output_stream;
GPollableInputStream *pollable_input_stream;
GPollableOutputStream *pollable_output_stream;
guint8 buf[65536];
GError *error = NULL;
GSource *stream_source;
GCancellable *cancellable;
agent = nice_agent_new_reliable (NULL, NICE_COMPATIBILITY_RFC5245);
nice_agent_add_local_address (agent, addr);
/* Add a stream. */
stream_id = nice_agent_add_stream (agent, 1);
/* Try building an I/O stream around it. */
io_stream = nice_agent_build_io_stream (agent, stream_id, 1);
g_assert (G_IS_IO_STREAM (io_stream));
g_assert (NICE_IS_IO_STREAM (io_stream));
/* Grab the input and output streams. */
input_stream = g_io_stream_get_input_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_POLLABLE_INPUT_STREAM (input_stream));
pollable_input_stream = G_POLLABLE_INPUT_STREAM (input_stream);
output_stream = g_io_stream_get_output_stream (G_IO_STREAM (io_stream));
g_assert (G_IS_POLLABLE_OUTPUT_STREAM (output_stream));
pollable_output_stream = G_POLLABLE_OUTPUT_STREAM (output_stream);
/* Check the non-blocking read() and write() return immediately if called with
* a cancelled cancellable. */
cancellable = g_cancellable_new ();
g_cancellable_cancel (cancellable);
g_assert (
g_pollable_input_stream_read_nonblocking (pollable_input_stream,
buf, sizeof (buf), cancellable, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
g_clear_error (&error);
g_assert (
g_pollable_output_stream_write_nonblocking (pollable_output_stream,
buf, sizeof (buf), cancellable, &error) == -1);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
g_clear_error (&error);
g_object_unref (cancellable);
/* Check the GSources invoke a callback when run with the cancellable
* cancelled. */
cancellable = g_cancellable_new ();
stream_source =
g_pollable_input_stream_create_source (pollable_input_stream,
cancellable);
check_pollable_source_cancellation (stream_source, cancellable);
g_source_unref (stream_source);
g_object_unref (cancellable);
/* And for the output stream. */
cancellable = g_cancellable_new ();
stream_source =
g_pollable_output_stream_create_source (pollable_output_stream,
cancellable);
check_pollable_source_cancellation (stream_source, cancellable);
g_source_unref (stream_source);
g_object_unref (cancellable);
}
static void
test_zero_length_reads_writes (NiceAddress *addr)
{
NiceAgent *agent;
guint stream_id;
NiceIOStream *io_stream;
GInputStream *input_stream;
GOutputStream *output_stream;
GPollableInputStream *pollable_input_stream;
GPollableOutputStream *pollable_output_stream;
GError *error = NULL;
guint8 buf[1]; /* should never be accessed */
agent = nice_agent_new_reliable (NULL, NICE_COMPATIBILITY_RFC5245);
nice_agent_add_local_address (agent, addr);
/* Add a stream. */
stream_id = nice_agent_add_stream (agent, 1);
/* Try building an I/O stream around it. */
io_stream = nice_agent_build_io_stream (agent, stream_id, 1);
g_assert (G_IS_IO_STREAM (io_stream));
g_assert (NICE_IS_IO_STREAM (io_stream));
input_stream = g_io_stream_get_input_stream (G_IO_STREAM (io_stream));
output_stream = g_io_stream_get_output_stream (G_IO_STREAM (io_stream));
pollable_input_stream = G_POLLABLE_INPUT_STREAM (input_stream);
pollable_output_stream = G_POLLABLE_OUTPUT_STREAM (output_stream);
/* Check zero-length reads and writes complete immediately without error. */
g_assert (g_input_stream_read (input_stream, buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (g_output_stream_write (output_stream, buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (
g_pollable_input_stream_read_nonblocking (pollable_input_stream,
buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (
g_pollable_output_stream_write_nonblocking (pollable_output_stream,
buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
/* Remove the component and check that zero-length reads and writes still
* result in a 0 response, rather than any error. */
nice_agent_remove_stream (agent, stream_id);
g_assert (g_io_stream_is_closed (G_IO_STREAM (io_stream)));
g_assert (g_input_stream_read (input_stream, buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (g_output_stream_write (output_stream, buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (
g_pollable_input_stream_read_nonblocking (pollable_input_stream,
buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_assert (
g_pollable_output_stream_write_nonblocking (pollable_output_stream,
buf, 0, NULL, &error) == 0);
g_assert_no_error (error);
g_object_unref (io_stream);
g_object_unref (agent);
}
int
main (void)
{
NiceAddress addr;
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
nice_address_init (&addr);
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2, 31, 8)
g_thread_init (NULL);
#endif
g_assert (nice_address_set_from_string (&addr, "127.0.0.1"));
test_invalid_stream (&addr);
test_io_stream_properties (&addr, TRUE);
test_io_stream_properties (&addr, FALSE);
test_pollable_properties (&addr);
test_pollable_cancellation (&addr);
test_zero_length_reads_writes (&addr);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef struct {
GCancellable *cancellable; /* owned */
GCond cond;
GMutex mutex;
gboolean blocking; /* protected by @mutex */
} CancellationData;
static gpointer
cancellation_thread_cb (gpointer user_data)
{
CancellationData *data = user_data;
/* Wait to be signalled from read_thread_cb(). */
g_mutex_lock (&data->mutex);
while (!data->blocking)
g_cond_wait (&data->cond, &data->mutex);
g_mutex_unlock (&data->mutex);
/* Try and ensure we cancel part-way through the read, rather than before the
* read function is called. */
g_usleep (100000);
g_cancellable_cancel (data->cancellable);
return NULL;
}
static void
read_thread_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
CancellationData *user_data = data->user_data;
GError *error = NULL;
guint8 buf[MESSAGE_SIZE];
gssize len;
/* Block on receiving some data or cancellation. */
g_mutex_lock (&user_data->mutex);
user_data->blocking = TRUE;
g_cond_signal (&user_data->cond);
g_mutex_unlock (&user_data->mutex);
len = g_input_stream_read (input_stream, buf, sizeof (buf),
user_data->cancellable, &error);
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_CANCELLED);
g_assert (len == -1);
g_main_loop_quit (data->error_loop);
}
int main (void)
{
GThread *l_cancellation_thread, *r_cancellation_thread;
CancellationData *l_data, *r_data;
const TestIOStreamCallbacks callbacks = {
read_thread_cb,
NULL,
NULL,
NULL,
};
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2, 31, 8)
g_thread_init (NULL);
#endif
l_data = g_malloc0 (sizeof (CancellationData));
l_data->cancellable = g_cancellable_new ();
l_data->blocking = FALSE;
r_data = g_malloc0 (sizeof (CancellationData));
r_data->cancellable = g_cancellable_new ();
r_data->blocking = FALSE;
l_cancellation_thread = spawn_thread ("libnice L cancel",
cancellation_thread_cb, l_data);
r_cancellation_thread = spawn_thread ("libnice R cancel",
cancellation_thread_cb, r_data);
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
g_thread_join (l_cancellation_thread);
g_thread_join (r_cancellation_thread);
/* Free things. */
g_object_unref (r_data->cancellable);
g_free (r_data);
g_object_unref (l_data->cancellable);
g_free (l_data);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef struct {
gint recv_count;
gint *other_recv_count;
} ClosingData;
#define NUM_MESSAGES 10
static void
read_thread_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
ClosingData *user_data = data->user_data;
GOutputStream *output_stream;
gpointer tmp;
guint stream_id;
GError *error = NULL;
while (user_data->recv_count < NUM_MESSAGES) {
gchar expected_data[MESSAGE_SIZE];
guint8 buf[MESSAGE_SIZE];
gssize len;
gsize offset;
/* Block on receiving some data. */
len = g_input_stream_read (input_stream, buf, sizeof (buf), NULL, &error);
g_assert_no_error (error);
offset = 0;
while (len > 0) {
g_assert (len == MESSAGE_SIZE);
g_assert (user_data->recv_count < NUM_MESSAGES);
memset (expected_data, user_data->recv_count + '1', MESSAGE_SIZE);
g_assert (
memcmp (buf + offset, expected_data, sizeof (expected_data)) == 0);
user_data->recv_count++;
len -= MESSAGE_SIZE;
offset += MESSAGE_SIZE;
}
g_assert (len == 0);
}
/* Signal completion. */
output_stream = g_io_stream_get_output_stream (data->io_stream);
g_output_stream_write (output_stream, "Done", strlen ("Done"), NULL, &error);
g_assert_no_error (error);
/* Wait for a done packet. */
while (TRUE) {
guint8 buf[4];
gssize len;
len = g_input_stream_read (input_stream, buf, sizeof (buf), NULL, &error);
g_assert_no_error (error);
g_assert (len == 4);
g_assert (memcmp (buf, "Done", strlen ("Done")) == 0);
break;
}
user_data->recv_count = -1;
tmp = g_object_get_data (G_OBJECT (data->agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
nice_agent_remove_stream (data->agent, stream_id);
/* Have both threads finished? */
if (user_data->recv_count == -1 &&
*user_data->other_recv_count == -1) {
g_main_loop_quit (data->error_loop);
}
}
static void
write_thread_cb (GOutputStream *output_stream, TestIOStreamThreadData *data)
{
gchar buf[MESSAGE_SIZE];
guint i;
for (i = 0; i < NUM_MESSAGES; i++) {
GError *error = NULL;
memset (buf, i + '1', MESSAGE_SIZE);
g_output_stream_write (output_stream, buf, sizeof (buf), NULL, &error);
g_assert_no_error (error);
}
}
int main (void)
{
ClosingData *l_data, *r_data;
const TestIOStreamCallbacks callbacks = {
read_thread_cb,
write_thread_cb,
NULL,
NULL,
};
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2, 31, 8)
g_thread_init (NULL);
#endif
l_data = g_malloc0 (sizeof (ClosingData));
r_data = g_malloc0 (sizeof (ClosingData));
l_data->recv_count = 0;
l_data->other_recv_count = &r_data->recv_count;
r_data->recv_count = 0;
r_data->other_recv_count = &l_data->recv_count;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
g_free (r_data);
g_free (l_data);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
static gboolean timer_cb (gpointer pointer)
{
g_debug ("test-thread:%s: %p", G_STRFUNC, pointer);
/* note: should not be reached, abort */
g_debug ("ERROR: test has got stuck, aborting...");
exit (-1);
}
static gpointer
write_thread_cb (gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
GMainContext *main_context;
GOutputStream *output_stream = NULL;
main_context = g_main_context_new ();
g_main_context_push_thread_default (main_context);
/* Synchronise thread starting. */
while (!g_main_loop_is_running (data->error_loop));
/* Wait for the stream to be writeable. */
g_mutex_lock (&data->write_mutex);
while (!(data->stream_open && data->stream_ready))
g_cond_wait (&data->write_cond, &data->write_mutex);
g_mutex_unlock (&data->write_mutex);
if (data->reliable)
output_stream = g_io_stream_get_output_stream (data->io_stream);
data->callbacks->write_thread (output_stream, data);
g_main_context_pop_thread_default (main_context);
g_main_context_unref (main_context);
return NULL;
}
static gpointer
read_thread_cb (gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
GMainContext *main_context;
GInputStream *input_stream = NULL;
main_context = g_main_context_new ();
g_main_context_push_thread_default (main_context);
/* Synchronise thread starting. */
while (!g_main_loop_is_running (data->error_loop));
if (data->reliable)
input_stream = g_io_stream_get_input_stream (data->io_stream);
data->callbacks->read_thread (input_stream, data);
g_main_context_pop_thread_default (main_context);
g_main_context_unref (main_context);
return NULL;
}
static gpointer
main_thread_cb (gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
g_main_context_push_thread_default (data->main_context);
/* Synchronise thread starting. */
while (!g_main_loop_is_running (data->error_loop));
/* Run the main context. */
g_main_loop_run (data->main_loop);
g_main_context_pop_thread_default (data->main_context);
return NULL;
}
static void
candidate_gathering_done_cb (NiceAgent *agent, guint stream_id,
gpointer user_data)
{
NiceAgent *other = g_object_get_data (G_OBJECT (agent), "other-agent");
gchar *ufrag = NULL, *password = NULL;
GSList *cands, *i;
guint id, other_id;
gpointer tmp;
tmp = g_object_get_data (G_OBJECT (agent), "stream-id");
id = GPOINTER_TO_UINT (tmp);
tmp = g_object_get_data (G_OBJECT (other), "stream-id");
other_id = GPOINTER_TO_UINT (tmp);
nice_agent_get_local_credentials (agent, id, &ufrag, &password);
nice_agent_set_remote_credentials (other,
other_id, ufrag, password);
g_free (ufrag);
g_free (password);
cands = nice_agent_get_local_candidates (agent, id, 1);
g_assert (cands != NULL);
nice_agent_set_remote_candidates (other, other_id, 1, cands);
for (i = cands; i; i = i->next)
nice_candidate_free ((NiceCandidate *) i->data);
g_slist_free (cands);
}
static void
reliable_transport_writable_cb (NiceAgent *agent, guint stream_id,
guint component_id, gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
g_assert (data->reliable);
/* Signal writeability. */
g_mutex_lock (&data->write_mutex);
data->stream_open = TRUE;
g_cond_broadcast (&data->write_cond);
g_mutex_unlock (&data->write_mutex);
if (data->callbacks->reliable_transport_writable != NULL) {
GIOStream *io_stream;
GOutputStream *output_stream;
io_stream = g_object_get_data (G_OBJECT (agent), "io-stream");
g_assert (io_stream != NULL);
output_stream = g_io_stream_get_output_stream (io_stream);
data->callbacks->reliable_transport_writable (output_stream, agent,
stream_id, component_id, data);
}
}
static void
component_state_changed_cb (NiceAgent *agent, guint stream_id,
guint component_id, guint state, gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
if (state != NICE_COMPONENT_STATE_READY)
return;
/* Signal stream state. */
g_mutex_lock (&data->write_mutex);
data->stream_ready = TRUE;
g_cond_broadcast (&data->write_cond);
g_mutex_unlock (&data->write_mutex);
}
static void
new_selected_pair_cb (NiceAgent *agent, guint stream_id, guint component_id,
gchar *lfoundation, gchar *rfoundation, gpointer user_data)
{
TestIOStreamThreadData *data = user_data;
if (data->callbacks->new_selected_pair != NULL) {
data->callbacks->new_selected_pair (agent, stream_id, component_id,
lfoundation, rfoundation, data);
}
}
static NiceAgent *
create_agent (gboolean controlling_mode, TestIOStreamThreadData *data,
GMainContext **main_context, GMainLoop **main_loop)
{
NiceAgent *agent;
NiceAddress base_addr;
const gchar *stun_server, *stun_server_port;
/* Create main contexts. */
*main_context = g_main_context_new ();
*main_loop = g_main_loop_new (*main_context, FALSE);
/* Use Google compatibility to ignore credentials. */
if (data->reliable)
agent = nice_agent_new_reliable (*main_context, NICE_COMPATIBILITY_GOOGLE);
else
agent = nice_agent_new (*main_context, NICE_COMPATIBILITY_GOOGLE);
g_object_set (G_OBJECT (agent),
"controlling-mode", controlling_mode,
"upnp", FALSE,
NULL);
/* Specify which local interface to use. */
g_assert (nice_address_set_from_string (&base_addr, "127.0.0.1"));
nice_agent_add_local_address (agent, &base_addr);
/* Hook up signals. */
g_signal_connect (G_OBJECT (agent), "candidate-gathering-done",
(GCallback) candidate_gathering_done_cb,
GUINT_TO_POINTER (controlling_mode));
g_signal_connect (G_OBJECT (agent), "new-selected-pair",
(GCallback) new_selected_pair_cb, data);
g_signal_connect (G_OBJECT (agent), "component-state-changed",
(GCallback) component_state_changed_cb, data);
if (data->reliable) {
g_signal_connect (G_OBJECT (agent), "reliable-transport-writable",
(GCallback) reliable_transport_writable_cb, data);
} else {
data->stream_open = TRUE;
}
/* Configure the STUN server. */
stun_server = g_getenv ("NICE_STUN_SERVER");
stun_server_port = g_getenv ("NICE_STUN_SERVER_PORT");
if (stun_server != NULL) {
g_object_set (G_OBJECT (agent),
"stun-server", stun_server,
"stun-server-port", atoi (stun_server_port),
NULL);
}
return agent;
}
static void
add_stream (NiceAgent *agent)
{
guint stream_id;
stream_id = nice_agent_add_stream (agent, 2);
g_assert (stream_id > 0);
g_object_set_data (G_OBJECT (agent), "stream-id",
GUINT_TO_POINTER (stream_id));
}
static void
run_agent (TestIOStreamThreadData *data, NiceAgent *agent)
{
guint stream_id;
gpointer tmp;
tmp = g_object_get_data (G_OBJECT (agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
nice_agent_gather_candidates (agent, stream_id);
if (data->reliable) {
data->io_stream =
G_IO_STREAM (nice_agent_build_io_stream (agent, stream_id, 1));
g_object_set_data (G_OBJECT (agent), "io-stream", data->io_stream);
} else {
data->io_stream = NULL;
}
}
GThread *
spawn_thread (const gchar *thread_name, GThreadFunc thread_func,
gpointer user_data)
{
GThread *thread;
#if !GLIB_CHECK_VERSION(2, 31, 8)
thread = g_thread_create (thread_func, user_data, TRUE, NULL);
#else
thread = g_thread_new (thread_name, thread_func, user_data);
#endif
g_assert (thread);
return thread;
}
static void
join_main_loop (GMainLoop *main_loop)
{
while (!g_main_loop_is_running (main_loop));
while (g_main_loop_is_running (main_loop))
g_main_loop_quit (main_loop);
}
void
run_io_stream_test (guint deadlock_timeout, gboolean reliable,
const TestIOStreamCallbacks *callbacks,
gpointer l_user_data, GDestroyNotify l_user_data_free,
gpointer r_user_data, GDestroyNotify r_user_data_free)
{
GMainLoop *error_loop;
GThread *l_main_thread, *r_main_thread;
GThread *l_write_thread, *l_read_thread, *r_write_thread, *r_read_thread;
TestIOStreamThreadData l_data, r_data;
error_loop = g_main_loop_new (NULL, FALSE);
/* Set up data structures. */
l_data.reliable = reliable;
l_data.error_loop = error_loop;
l_data.callbacks = callbacks;
l_data.user_data = l_user_data;
l_data.user_data_free = l_user_data_free;
g_cond_init (&l_data.write_cond);
g_mutex_init (&l_data.write_mutex);
l_data.stream_open = FALSE;
l_data.stream_ready = FALSE;
r_data.reliable = reliable;
r_data.error_loop = error_loop;
r_data.callbacks = callbacks;
r_data.user_data = r_user_data;
r_data.user_data_free = r_user_data_free;
g_cond_init (&r_data.write_cond);
g_mutex_init (&r_data.write_mutex);
r_data.stream_open = FALSE;
r_data.stream_ready = FALSE;
/* Create the L and R agents. */
l_data.agent = create_agent (TRUE, &l_data,
&l_data.main_context, &l_data.main_loop);
r_data.agent = create_agent (FALSE, &r_data,
&r_data.main_context, &r_data.main_loop);
g_object_set_data (G_OBJECT (l_data.agent), "other-agent", r_data.agent);
g_object_set_data (G_OBJECT (r_data.agent), "other-agent", l_data.agent);
/* Add a timer to catch deadlocks. */
g_timeout_add_seconds (deadlock_timeout, timer_cb, NULL);
l_main_thread = spawn_thread ("libnice L main", main_thread_cb, &l_data);
r_main_thread = spawn_thread ("libnice R main", main_thread_cb, &r_data);
add_stream (l_data.agent);
add_stream (r_data.agent);
run_agent (&l_data, l_data.agent);
run_agent (&r_data, r_data.agent);
l_read_thread = spawn_thread ("libnice L read", read_thread_cb, &l_data);
r_read_thread = spawn_thread ("libnice R read", read_thread_cb, &r_data);
if (callbacks->write_thread != NULL) {
l_write_thread = spawn_thread ("libnice L write", write_thread_cb, &l_data);
r_write_thread = spawn_thread ("libnice R write", write_thread_cb, &r_data);
} else {
l_write_thread = NULL;
r_write_thread = NULL;
}
/* Run loop for error timer */
g_main_loop_run (error_loop);
/* Clean up the main loops and threads. */
join_main_loop (l_data.main_loop);
join_main_loop (r_data.main_loop);
g_thread_join (l_read_thread);
g_thread_join (r_read_thread);
if (l_write_thread != NULL)
g_thread_join (l_write_thread);
if (r_write_thread != NULL)
g_thread_join (r_write_thread);
g_thread_join (l_main_thread);
g_thread_join (r_main_thread);
/* Free things. */
if (r_data.user_data_free != NULL)
r_data.user_data_free (r_data.user_data);
if (l_data.user_data_free != NULL)
l_data.user_data_free (l_data.user_data);
g_cond_clear (&r_data.write_cond);
g_mutex_clear (&r_data.write_mutex);
g_cond_clear (&l_data.write_cond);
g_mutex_clear (&l_data.write_mutex);
if (r_data.io_stream != NULL)
g_object_unref (r_data.io_stream);
if (l_data.io_stream != NULL)
g_object_unref (l_data.io_stream);
g_object_unref (r_data.agent);
g_object_unref (l_data.agent);
g_main_loop_unref (r_data.main_loop);
g_main_loop_unref (l_data.main_loop);
g_main_context_unref (r_data.main_context);
g_main_context_unref (l_data.main_context);
g_main_loop_unref (error_loop);
}
/* Once we’ve received all the expected bytes, wait to finish sending all bytes,
* then send and wait for the close message. Finally, remove the stream.
*
* This must only be called from the read thread implementation. */
void
check_for_termination (TestIOStreamThreadData *data, gsize *recv_count,
gsize *other_recv_count, gsize *send_count, gsize expected_recv_count)
{
guint stream_id;
gpointer tmp;
guint8 buf[65536];
gsize buf_len;
gssize len;
GError *error = NULL;
g_assert_cmpuint (*recv_count, ==, expected_recv_count);
/* Wait for transmission to complete. */
while (*send_count < expected_recv_count);
/* Send a close message. */
tmp = g_object_get_data (G_OBJECT (data->agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
buf_len = strlen ("Done");
memcpy (buf, "Done", buf_len);
len = nice_agent_send (data->agent, stream_id, 1, buf_len, (gchar *) buf);
g_assert_cmpint (len, ==, buf_len);
/* Wait for a done packet. */
buf_len = data->reliable ? strlen ("Done") : sizeof (buf);
len = nice_agent_recv (data->agent, stream_id, 1, buf, buf_len, NULL,
&error);
g_assert_no_error (error);
g_assert_cmpint (len, ==, strlen ("Done"));
g_assert_cmpint (memcmp (buf, "Done", strlen ("Done")), ==, 0);
*recv_count = *recv_count + 1;
/* Remove the stream and run away. */
nice_agent_remove_stream (data->agent, stream_id);
/* If both sides have finished, quit the test main loop. */
if (*recv_count > expected_recv_count &&
*other_recv_count > expected_recv_count) {
g_main_loop_quit (data->error_loop);
}
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
/* Make the message sufficiently large to not hit Nagle’s algorithm in the
* pseudo-TCP implementation, and hence run really slowly. */
#define MESSAGE_SIZE 1284 /* bytes */
typedef struct _TestIOStreamThreadData TestIOStreamThreadData;
typedef struct {
void (*read_thread) (GInputStream *input_stream,
TestIOStreamThreadData *data);
void (*write_thread) (GOutputStream *output_stream,
TestIOStreamThreadData *data);
void (*reliable_transport_writable) (GOutputStream *output_stream,
NiceAgent *agent, guint stream_id, guint component_id,
TestIOStreamThreadData *data);
void (*new_selected_pair) (NiceAgent *agent, guint stream_id,
guint component_id, gchar *lfoundation, gchar *rfoundation,
TestIOStreamThreadData *data);
} TestIOStreamCallbacks;
struct _TestIOStreamThreadData {
NiceAgent *agent;
GIOStream *io_stream;
gboolean reliable;
GMainLoop *main_loop;
GMainLoop *error_loop;
GMainContext *main_context;
GMainContext *write_context;
GMainContext *read_context;
gpointer user_data;
GDestroyNotify user_data_free;
/*< private >*/
const TestIOStreamCallbacks *callbacks;
/* Condition signalling for the stream being open/writeable. */
gboolean stream_open;
gboolean stream_ready;
GCond write_cond;
GMutex write_mutex;
};
GThread *spawn_thread (const gchar *thread_name, GThreadFunc thread_func,
gpointer user_data);
void run_io_stream_test (guint deadlock_timeout, gboolean reliable,
const TestIOStreamCallbacks *callbacks,
gpointer l_user_data, GDestroyNotify l_user_data_free,
gpointer r_user_data, GDestroyNotify r_user_data_free);
void check_for_termination (TestIOStreamThreadData *data, gsize *recv_count,
gsize *other_recv_count, gsize *send_count, gsize expected_recv_count);
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef struct {
GMainLoop *read_loop; /* unowned */
gsize recv_count;
gsize *other_recv_count;
gsize send_count;
gsize *other_send_count;
} ThreadData;
static gboolean
read_stream_cb (GObject *pollable_stream, gpointer _user_data)
{
TestIOStreamThreadData *data = _user_data;
ThreadData *user_data = data->user_data;
gchar expected_data[MESSAGE_SIZE];
GError *error = NULL;
guint8 buf[MESSAGE_SIZE];
gssize len;
/* Try to receive some data. */
len = g_pollable_input_stream_read_nonblocking (
G_POLLABLE_INPUT_STREAM (pollable_stream), buf, sizeof (buf), NULL,
&error);
if (len == -1) {
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
return TRUE;
}
g_assert_cmpint (len, ==, MESSAGE_SIZE);
memset (expected_data, user_data->recv_count + '1', sizeof (expected_data));
g_assert (memcmp (buf, expected_data, sizeof (expected_data)) == 0);
user_data->recv_count++;
if (user_data->recv_count == 10) {
g_main_loop_quit (user_data->read_loop);
return FALSE;
}
return TRUE;
}
static void
read_thread_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
GMainContext *main_context;
GMainLoop *main_loop;
GSource *stream_source;
ThreadData *user_data = data->user_data;
main_context = g_main_context_new ();
main_loop = g_main_loop_new (main_context, FALSE);
g_main_context_push_thread_default (main_context);
stream_source =
g_pollable_input_stream_create_source (
G_POLLABLE_INPUT_STREAM (input_stream), NULL);
g_source_set_callback (stream_source, (GSourceFunc) read_stream_cb,
data, NULL);
g_source_attach (stream_source, main_context);
g_source_unref (stream_source);
/* Run the main loop. */
user_data->read_loop = main_loop;
g_main_loop_run (main_loop);
g_main_context_pop_thread_default (main_context);
g_main_loop_unref (main_loop);
g_main_context_unref (main_context);
check_for_termination (data, &user_data->recv_count,
user_data->other_recv_count, &user_data->send_count, 10);
}
static void
write_thread_cb (GOutputStream *output_stream, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
guint8 buf[MESSAGE_SIZE];
for (user_data->send_count = 0;
user_data->send_count < 10;
user_data->send_count++) {
GError *error = NULL;
memset (buf, user_data->send_count + '1', MESSAGE_SIZE);
g_pollable_output_stream_write_nonblocking (
G_POLLABLE_OUTPUT_STREAM (output_stream), buf, sizeof (buf), NULL,
&error);
g_assert_no_error (error);
}
}
int main (void)
{
ThreadData *l_data, *r_data;
const TestIOStreamCallbacks callbacks = {
read_thread_cb,
write_thread_cb,
NULL,
NULL,
};
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2,31,8)
g_thread_init (NULL);
#endif
l_data = g_malloc0 (sizeof (ThreadData));
r_data = g_malloc0 (sizeof (ThreadData));
l_data->recv_count = 0;
l_data->send_count = 0;
l_data->other_recv_count = &r_data->recv_count;
l_data->other_send_count = &r_data->send_count;
r_data->recv_count = 0;
r_data->send_count = 0;
r_data->other_recv_count = &l_data->recv_count;
r_data->other_send_count = &l_data->send_count;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
g_free (r_data);
g_free (l_data);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2013 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef struct {
guint cand_count;
guint *other_cand_count;
gsize recv_count;
gsize *other_recv_count;
gsize send_count;
gsize *other_send_count;
} ThreadData;
static void
read_thread_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
for (user_data->recv_count = 0;
user_data->recv_count < 10;
user_data->recv_count++) {
guint8 expected_data[MESSAGE_SIZE];
GError *error = NULL;
guint8 buf[MESSAGE_SIZE];
gssize len;
/* Block on receiving some data. */
len = g_input_stream_read (input_stream, buf, sizeof (buf), NULL, &error);
g_assert_no_error (error);
g_assert_cmpint (len, ==, sizeof (buf));
memset (expected_data, user_data->recv_count + '1', sizeof (expected_data));
g_assert (memcmp (buf, expected_data, sizeof (expected_data)) == 0);
}
check_for_termination (data, &user_data->recv_count,
user_data->other_recv_count, &user_data->send_count, 10);
}
static void
new_selected_pair_cb (NiceAgent *agent, guint stream_id, guint component_id,
gchar *lfoundation, gchar *rfoundation, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
g_atomic_int_inc (&user_data->cand_count);
}
static void
write_thread_cb (GOutputStream *output_stream, TestIOStreamThreadData *data)
{
ThreadData *user_data = data->user_data;
guint8 buf[MESSAGE_SIZE];
for (user_data->send_count = 0;
user_data->send_count < 10;
user_data->send_count++) {
GError *error = NULL;
memset (buf, user_data->send_count + '1', MESSAGE_SIZE);
g_output_stream_write (output_stream, buf, sizeof (buf), NULL, &error);
g_assert_no_error (error);
}
}
int main (void)
{
ThreadData *l_data, *r_data;
const TestIOStreamCallbacks callbacks = {
read_thread_cb,
write_thread_cb,
NULL,
new_selected_pair_cb,
};
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2,31,8)
g_thread_init (NULL);
#endif
l_data = g_malloc0 (sizeof (ThreadData));
r_data = g_malloc0 (sizeof (ThreadData));
l_data->cand_count = 0;
l_data->other_cand_count = &r_data->cand_count;
l_data->recv_count = 0;
l_data->other_recv_count = &r_data->recv_count;
l_data->send_count = 0;
l_data->other_send_count = &r_data->send_count;
r_data->cand_count = 0;
r_data->other_cand_count = &l_data->cand_count;
r_data->recv_count = 0;
r_data->other_recv_count = &l_data->recv_count;
r_data->send_count = 0;
r_data->other_send_count = &l_data->send_count;
run_io_stream_test (30, TRUE, &callbacks, l_data, NULL, r_data, NULL);
/* Verify that correct number of local candidates were reported. */
g_assert (l_data->cand_count == 1);
g_assert (r_data->cand_count == 1);
g_free (r_data);
g_free (l_data);
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
/*
* This file is part of the Nice GLib ICE library.
*
* (C) 2014 Collabora Ltd.
* Contact: Philip Withnall
*
* The contents of this file are subject to the Mozilla Public License Version
* 1.1 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* http://www.mozilla.org/MPL/
*
* Software distributed under the License is distributed on an "AS IS" basis,
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
* for the specific language governing rights and limitations under the
* License.
*
* The Original Code is the Nice GLib ICE library.
*
* The Initial Developers of the Original Code are Collabora Ltd and Nokia
* Corporation. All Rights Reserved.
*
* Contributors:
* Philip Withnall, Collabora Ltd.
*
* Alternatively, the contents of this file may be used under the terms of the
* the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
* case the provisions of LGPL are applicable instead of those above. If you
* wish to allow use of your version of this file only under the terms of the
* LGPL and not to allow others to use your version of this file under the
* MPL, indicate your decision by deleting the provisions above and replace
* them with the notice and other provisions required by the LGPL. If you do
* not delete the provisions above, a recipient may use your version of this
* file under either the MPL or the LGPL.
*/
/**
* This is a comprehensive unit test for send() and recv() behaviour in libnice,
* covering all APIs except the old nice_agent_attach_recv() one. It aims to
* test the correctness of reliable and non-reliable I/O through libnice, using
* a variety of data and a variety of buffer sizes.
*
* Abnormal features like error handling, zero-length buffer handling, stream
* closure and cancellation are not tested.
*
* This is *not* a performance test, and would require significant work to be
* useful as one. It allocates all of its buffers dynamically, and walks over
* them frequently to set and check data.
*
* Several of the strategies in the test make use of random numbers. The seed
* values for these are deterministically set (in main()), but may be specified
* on the command line to allow fuzzing.
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include "agent.h"
#include "test-io-stream-common.h"
#include <stdlib.h>
#include <string.h>
#ifndef G_OS_WIN32
#include <unistd.h>
#endif
typedef enum {
STREAM_AGENT, /* nice_agent_[send|recv]() */
STREAM_AGENT_NONBLOCKING, /* nice_agent_[send|recv]_nonblocking() */
STREAM_GIO, /* Nice[Input|Output]Stream */
STREAM_GSOURCE, /* GPollable[Input|Output]Stream */
} StreamApi;
#define STREAM_API_N_ELEMENTS (STREAM_GSOURCE + 1)
typedef enum {
BUFFER_SIZE_CONSTANT_LARGE, /* always 65535 bytes */
BUFFER_SIZE_CONSTANT_SMALL, /* always 1024 bytes */
BUFFER_SIZE_CONSTANT_TINY, /* always 1 byte */
BUFFER_SIZE_ASCENDING, /* ascending powers of 2 */
BUFFER_SIZE_RANDOM, /* random every time */
} BufferSizeStrategy;
#define BUFFER_SIZE_STRATEGY_N_ELEMENTS (BUFFER_SIZE_RANDOM + 1)
typedef enum {
BUFFER_DATA_CONSTANT, /* fill with 0xfe */
BUFFER_DATA_ASCENDING, /* ascending values for each byte */
BUFFER_DATA_PSEUDO_RANDOM, /* every byte is pseudo-random */
} BufferDataStrategy;
#define BUFFER_DATA_STRATEGY_N_ELEMENTS (BUFFER_DATA_PSEUDO_RANDOM + 1)
typedef struct {
/* Test configuration (immutable per test run). */
gboolean reliable;
StreamApi stream_api;
BufferSizeStrategy transmit_buffer_size_strategy;
BufferSizeStrategy receive_buffer_size_strategy;
BufferDataStrategy buffer_data_strategy;
gsize n_bytes;
/* Test state. */
GRand *transmit_size_rand;
GRand *receive_size_rand;
gsize transmitted_bytes;
gsize received_bytes;
gsize *other_received_bytes;
} TestData;
/* Whether @stream_api is blocking (vs. non-blocking). */
static gboolean
stream_api_is_blocking (StreamApi stream_api)
{
switch (stream_api) {
case STREAM_AGENT:
case STREAM_GIO:
return TRUE;
case STREAM_AGENT_NONBLOCKING:
case STREAM_GSOURCE:
return FALSE;
default:
g_assert_not_reached ();
}
}
/* Whether @stream_api only works for reliable NiceAgents. */
static gboolean
stream_api_is_reliable_only (StreamApi stream_api)
{
switch (stream_api) {
case STREAM_GSOURCE:
case STREAM_GIO:
return TRUE;
case STREAM_AGENT:
case STREAM_AGENT_NONBLOCKING:
return FALSE;
default:
g_assert_not_reached ();
}
}
/* Generate a size for the @buffer_index-th buffer. Guaranteed to be in
* the interval [1, 1 << 16). ((1 << 16) is the maximum message size.) */
static gsize
generate_buffer_size (BufferSizeStrategy strategy, GRand *rand,
guint buffer_index)
{
switch (strategy) {
case BUFFER_SIZE_CONSTANT_LARGE:
return (1 << 16) - 1;
case BUFFER_SIZE_CONSTANT_SMALL:
return 4096;
case BUFFER_SIZE_CONSTANT_TINY:
return 1;
case BUFFER_SIZE_ASCENDING:
return CLAMP (1L << buffer_index, 1, (1 << 16) - 1);
case BUFFER_SIZE_RANDOM:
return g_rand_int_range (rand, 1, 1 << 16);
default:
g_assert_not_reached ();
}
}
/* Fill the given @buf with @buf_len bytes of generated data. The data is
* deterministically generated, so that:
* generate_buffer_data(_, I, buf, 2)
* and
* generate_buffer_data(_, I+1, buf+1, 1)
* generate the same buf[I+1] byte, for all I.
*
* The generation strategies are generally chosen to produce data which makes
* send/receive errors (insertions, swaps, elisions) obvious. */
static void
generate_buffer_data (BufferDataStrategy strategy, gsize buffer_offset,
guint8 *buf, gsize buf_len)
{
switch (strategy) {
case BUFFER_DATA_CONSTANT:
memset (buf, 0xfe, buf_len);
break;
case BUFFER_DATA_ASCENDING: {
gsize i;
for (i = 0; i < buf_len; i++) {
buf[i] = (i + buffer_offset) & 0xff;
}
break;
}
case BUFFER_DATA_PSEUDO_RANDOM: {
gsize i;
/* This can’t use GRand, because then the number of calls to g_rand_*()
* methods would affect its output, and the bytes generated here have to be
* entirely deterministic on @buffer_offset.
*
* Instead, use something akin to a LCG, except without any feedback
* (because that would make it non-deterministic). The objective is to
* generate numbers which are sufficiently pseudo-random that it’s likely
* transpositions, elisions and insertions will be detected.
*
* The constants come from ‘ANSI C’ in:
* http://en.wikipedia.org/wiki/Linear_congruential_generator
*/
for (i = 0; i < buf_len; i++) {
buf[i] = (1103515245 * (buffer_offset + i) + 12345) & 0xff;
}
break;
}
default:
g_assert_not_reached ();
}
}
/* Choose a size and allocate a receive buffer in @buf, ready to receive bytes
* starting at @buffer_offset into the stream. Fill the buffer with poison
* values to hopefully make incorrect writes/reads more obvious.
*
* @buf must be freed with g_free(). */
static void
generate_buffer_to_receive (TestIOStreamThreadData *data, gsize buffer_offset,
guint8 **buf, gsize *buf_len)
{
TestData *test_data = data->user_data;
/* Allocate the buffer. */
*buf_len = generate_buffer_size (test_data->receive_buffer_size_strategy,
test_data->receive_size_rand, buffer_offset);
*buf = g_malloc (*buf_len);
/* Fill it with poison to try and detect incorrect writes. */
memset (*buf, 0xaa, *buf_len);
}
/* Validate the length and data of a received buffer of length @buf_len, filled
* with @len valid bytes. Updates the internal state machine to mark the bytes
* as received. This consumes @buf. */
static void
validate_received_buffer (TestIOStreamThreadData *data, gsize buffer_offset,
guint8 **buf, gsize buf_len, gssize len)
{
TestData *test_data = data->user_data;
guint8 *expected_buf;
g_assert_cmpint (len, <=, buf_len);
g_assert_cmpint (len, >=, 0);
if (stream_api_is_blocking (test_data->stream_api) && data->reliable)
g_assert_cmpint (len, ==, buf_len);
/* Validate the buffer contents. */
expected_buf = g_malloc (buf_len);
memset (expected_buf, 0xaa, buf_len);
generate_buffer_data (test_data->buffer_data_strategy, buffer_offset,
expected_buf, len);
g_assert (memcmp (*buf, expected_buf, buf_len) == 0);
g_free (expected_buf);
test_data->received_bytes += len;
g_free (*buf);
}
/* Determine a size for the next transmit buffer, allocate it, and fill it with
* data to be transmitted. */
static void
generate_buffer_to_transmit (TestIOStreamThreadData *data, gsize buffer_offset,
guint8 **buf, gsize *buf_len)
{
TestData *test_data = data->user_data;
/* Allocate the buffer. */
*buf_len = generate_buffer_size (test_data->transmit_buffer_size_strategy,
test_data->transmit_size_rand, buffer_offset);
*buf_len = MIN (*buf_len, test_data->n_bytes - test_data->transmitted_bytes);
*buf = g_malloc (*buf_len);
/* Fill it with data. */
generate_buffer_data (test_data->buffer_data_strategy, buffer_offset,
*buf, *buf_len);
}
/* Validate the number of bytes transmitted, and update the test’s internal
* state machine. Consumes @buf. */
static void
notify_transmitted_buffer (TestIOStreamThreadData *data, gsize buffer_offset,
guint8 **buf, gsize buf_len, gssize len)
{
TestData *test_data = data->user_data;
g_assert_cmpint (len, <=, buf_len);
g_assert_cmpint (len, >=, 0);
test_data->transmitted_bytes += len;
g_free (*buf);
}
/*
* Implementation using nice_agent_recv() and nice_agent_send().
*/
static void
read_thread_agent_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
guint stream_id, component_id;
gpointer tmp;
tmp = g_object_get_data (G_OBJECT (data->agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
component_id = 1;
while (test_data->received_bytes < test_data->n_bytes) {
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
/* Initialise a receive buffer. */
generate_buffer_to_receive (data, test_data->received_bytes, &buf,
&buf_len);
/* Trim the receive buffer to avoid blocking on bytes which will never
* appear. */
if (data->reliable)
buf_len = MIN (buf_len, test_data->n_bytes - test_data->received_bytes);
/* Block on receiving some data. */
len = nice_agent_recv (data->agent, stream_id, component_id, buf, buf_len,
NULL, &error);
g_assert_no_error (error);
/* Check the buffer and update the test’s state machine. */
validate_received_buffer (data, test_data->received_bytes, &buf, buf_len,
len);
}
check_for_termination (data, &test_data->received_bytes,
test_data->other_received_bytes, &test_data->transmitted_bytes,
test_data->n_bytes);
}
static void
write_thread_agent_cb (GOutputStream *output_stream,
TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
guint stream_id, component_id;
gpointer tmp;
tmp = g_object_get_data (G_OBJECT (data->agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
component_id = 1;
while (test_data->transmitted_bytes < test_data->n_bytes) {
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize _len;
gssize len = 0;
/* Generate a buffer to transmit. */
generate_buffer_to_transmit (data, test_data->transmitted_bytes, &buf,
&buf_len);
/* Transmit it. */
do {
_len = nice_agent_send_full (data->agent, stream_id, component_id,
buf + len, buf_len - len, NULL, &error);
/* Busy loop on EWOULDBLOCK. */
if (_len == -1 &&
g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
g_clear_error (&error);
continue;
} else if (_len > 0) {
len += _len;
} else {
len = _len;
}
g_assert_no_error (error);
} while (len != -1 && (gsize) len < buf_len);
/* Update the test’s buffer generation state machine. */
notify_transmitted_buffer (data, test_data->transmitted_bytes, &buf,
buf_len, len);
}
}
/*
* Implementation using nice_agent_recv_nonblocking() and
* nice_agent_send_nonblocking().
*/
static void
read_thread_agent_nonblocking_cb (GInputStream *input_stream,
TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
guint stream_id, component_id;
gpointer tmp;
tmp = g_object_get_data (G_OBJECT (data->agent), "stream-id");
stream_id = GPOINTER_TO_UINT (tmp);
component_id = 1;
while (test_data->received_bytes < test_data->n_bytes) {
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
/* Initialise a receive buffer. */
generate_buffer_to_receive (data, test_data->received_bytes, &buf,
&buf_len);
/* Trim the receive buffer to avoid consuming the ‘done’ message. */
if (data->reliable)
buf_len = MIN (buf_len, test_data->n_bytes - test_data->received_bytes);
/* Busy loop on receiving some data. */
do {
g_clear_error (&error);
len = nice_agent_recv_nonblocking (data->agent, stream_id, component_id,
buf, buf_len, NULL, &error);
} while (len == -1 &&
g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK));
g_assert_no_error (error);
/* Check the buffer and update the test’s state machine. */
validate_received_buffer (data, test_data->received_bytes, &buf, buf_len,
len);
}
check_for_termination (data, &test_data->received_bytes,
test_data->other_received_bytes, &test_data->transmitted_bytes,
test_data->n_bytes);
}
static void
write_thread_agent_nonblocking_cb (GOutputStream *output_stream,
TestIOStreamThreadData *data)
{
/* FIXME: There is no nice_agent_send_nonblocking(); nice_agent_send() is
* non-blocking by default. */
write_thread_agent_cb (output_stream, data);
}
/*
* Implementation using NiceInputStream and NiceOutputStream.
*/
static void
read_thread_gio_cb (GInputStream *input_stream, TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
while (test_data->received_bytes < test_data->n_bytes) {
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
/* Initialise a receive buffer. */
generate_buffer_to_receive (data, test_data->received_bytes, &buf,
&buf_len);
/* Trim the receive buffer to avoid blocking on bytes which will never
* appear. */
buf_len = MIN (buf_len, test_data->n_bytes - test_data->received_bytes);
/* Block on receiving some data. */
len = g_input_stream_read (input_stream, buf, buf_len, NULL, &error);
g_assert_no_error (error);
/* Check the buffer and update the test’s state machine. */
validate_received_buffer (data, test_data->received_bytes, &buf, buf_len,
len);
}
check_for_termination (data, &test_data->received_bytes,
test_data->other_received_bytes, &test_data->transmitted_bytes,
test_data->n_bytes);
}
static void
write_thread_gio_cb (GOutputStream *output_stream, TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
while (test_data->transmitted_bytes < test_data->n_bytes) {
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
gsize total_len = 0;
/* Generate a buffer to transmit. */
generate_buffer_to_transmit (data, test_data->transmitted_bytes, &buf,
&buf_len);
/* Transmit it. */
do {
len = g_output_stream_write (output_stream, buf + total_len,
buf_len - total_len, NULL, &error);
g_assert_no_error (error);
total_len += len;
} while (total_len < buf_len);
/* Update the test’s buffer generation state machine. */
notify_transmitted_buffer (data, test_data->transmitted_bytes, &buf,
buf_len, total_len);
}
}
/*
* Implementation using GPollableInputStream and GPollableOutputStream.
*
* GSourceData is effectively the closure for the ‘for’ loop in other stream API
* implementations.
*/
typedef struct {
TestIOStreamThreadData *data;
GMainLoop *main_loop;
} GSourceData;
static gboolean
read_stream_cb (GObject *pollable_stream, gpointer _user_data)
{
GSourceData *gsource_data = _user_data;
TestIOStreamThreadData *data = gsource_data->data;
TestData *test_data = data->user_data;
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
/* Initialise a receive buffer. */
generate_buffer_to_receive (data, test_data->received_bytes, &buf, &buf_len);
/* Trim the receive buffer to avoid consuming the ‘done’ message. */
buf_len = MIN (buf_len, test_data->n_bytes - test_data->received_bytes);
/* Try to receive some data. */
len = g_pollable_input_stream_read_nonblocking (
G_POLLABLE_INPUT_STREAM (pollable_stream), buf, buf_len, NULL, &error);
if (len == -1) {
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
g_free (buf);
return TRUE;
}
g_assert_no_error (error);
/* Check the buffer and update the test’s state machine. */
validate_received_buffer (data, test_data->received_bytes, &buf, buf_len,
len);
/* Termination time? */
if (test_data->received_bytes == test_data->n_bytes) {
g_main_loop_quit (gsource_data->main_loop);
return FALSE;
}
return TRUE;
}
static void
read_thread_gsource_cb (GInputStream *input_stream,
TestIOStreamThreadData *data)
{
TestData *test_data = data->user_data;
GSourceData gsource_data;
GMainContext *main_context;
GMainLoop *main_loop;
GSource *stream_source;
main_context = g_main_context_ref_thread_default ();
main_loop = g_main_loop_new (main_context, FALSE);
gsource_data.data = data;
gsource_data.main_loop = main_loop;
stream_source =
g_pollable_input_stream_create_source (
G_POLLABLE_INPUT_STREAM (input_stream), NULL);
g_source_set_callback (stream_source, (GSourceFunc) read_stream_cb,
&gsource_data, NULL);
g_source_attach (stream_source, main_context);
/* Run the main loop. */
g_main_loop_run (main_loop);
g_source_destroy (stream_source);
g_source_unref (stream_source);
g_main_loop_unref (main_loop);
g_main_context_unref (main_context);
/* Termination? */
check_for_termination (data, &test_data->received_bytes,
test_data->other_received_bytes, &test_data->transmitted_bytes,
test_data->n_bytes);
}
static gboolean
write_stream_cb (GObject *pollable_stream, gpointer _user_data)
{
GSourceData *gsource_data = _user_data;
TestIOStreamThreadData *data = gsource_data->data;
TestData *test_data = data->user_data;
GError *error = NULL;
guint8 *buf = NULL;
gsize buf_len = 0;
gssize len;
/* Initialise a receive buffer. */
generate_buffer_to_transmit (data, test_data->transmitted_bytes, &buf,
&buf_len);
/* Try to transmit some data. */
len = g_pollable_output_stream_write_nonblocking (
G_POLLABLE_OUTPUT_STREAM (pollable_stream), buf, buf_len, NULL, &error);
if (len == -1) {
g_assert_error (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK);
g_free (buf);
return TRUE;
}
g_assert_no_error (error);
/* Update the test’s buffer generation state machine. */
notify_transmitted_buffer (data, test_data->transmitted_bytes, &buf, buf_len,
len);
/* Termination time? */
if (test_data->transmitted_bytes == test_data->n_bytes) {
g_main_loop_quit (gsource_data->main_loop);
return FALSE;
}
return TRUE;
}
static void
write_thread_gsource_cb (GOutputStream *output_stream,
TestIOStreamThreadData *data)
{
GSourceData gsource_data;
GMainContext *main_context;
GMainLoop *main_loop;
GSource *stream_source;
main_context = g_main_context_ref_thread_default ();
main_loop = g_main_loop_new (main_context, FALSE);
gsource_data.data = data;
gsource_data.main_loop = main_loop;
stream_source =
g_pollable_output_stream_create_source (
G_POLLABLE_OUTPUT_STREAM (output_stream), NULL);
g_source_set_callback (stream_source, (GSourceFunc) write_stream_cb,
&gsource_data, NULL);
g_source_attach (stream_source, main_context);
/* Run the main loop. */
g_main_loop_run (main_loop);
g_source_destroy (stream_source);
g_source_unref (stream_source);
g_main_loop_unref (main_loop);
g_main_context_unref (main_context);
}
static void
test_data_init (TestData *data, gboolean reliable, StreamApi stream_api,
gsize n_bytes, BufferSizeStrategy transmit_buffer_size_strategy,
BufferSizeStrategy receive_buffer_size_strategy,
BufferDataStrategy buffer_data_strategy, guint32 transmit_seed,
guint32 receive_seed, gsize *other_received_bytes)
{
data->reliable = reliable;
data->stream_api = stream_api;
data->n_bytes = n_bytes;
data->transmit_buffer_size_strategy = transmit_buffer_size_strategy;
data->receive_buffer_size_strategy = receive_buffer_size_strategy;
data->buffer_data_strategy = buffer_data_strategy;
data->transmit_size_rand = g_rand_new_with_seed (transmit_seed);
data->receive_size_rand = g_rand_new_with_seed (receive_seed);
data->transmitted_bytes = 0;
data->received_bytes = 0;
data->other_received_bytes = other_received_bytes;
}
/*
* Test closures.
*/
static void
test_data_clear (TestData *data)
{
g_rand_free (data->receive_size_rand);
g_rand_free (data->transmit_size_rand);
}
static void
test (gboolean reliable, StreamApi stream_api, gsize n_bytes,
BufferSizeStrategy transmit_buffer_size_strategy,
BufferSizeStrategy receive_buffer_size_strategy,
BufferDataStrategy buffer_data_strategy,
guint32 transmit_seed, guint32 receive_seed,
guint deadlock_timeout)
{
TestData l_data, r_data;
/* Indexed by StreamApi. */
const TestIOStreamCallbacks callbacks[] = {
{ read_thread_agent_cb,
write_thread_agent_cb, NULL, NULL, }, /* STREAM_AGENT */
{ read_thread_agent_nonblocking_cb, write_thread_agent_nonblocking_cb,
NULL, NULL, }, /* STREAM_AGENT_NONBLOCKING */
{ read_thread_gio_cb, write_thread_gio_cb, NULL, NULL, }, /* STREAM_GIO */
{ read_thread_gsource_cb, write_thread_gsource_cb,
NULL, NULL }, /* STREAM_GSOURCE */
};
test_data_init (&l_data, reliable, stream_api, n_bytes,
transmit_buffer_size_strategy, receive_buffer_size_strategy,
buffer_data_strategy, transmit_seed, receive_seed,
&r_data.received_bytes);
test_data_init (&r_data, reliable, stream_api, n_bytes,
transmit_buffer_size_strategy, receive_buffer_size_strategy,
buffer_data_strategy, transmit_seed, receive_seed,
&l_data.received_bytes);
run_io_stream_test (deadlock_timeout, reliable, &callbacks[stream_api],
&l_data, NULL, &r_data, NULL);
test_data_clear (&r_data);
test_data_clear (&l_data);
}
/* Options with default values. */
guint32 option_transmit_seed = 0;
guint32 option_receive_seed = 0;
gsize option_n_bytes = 100000;
guint option_timeout = 1200; /* seconds */
gboolean option_long_mode = FALSE;
static GOptionEntry entries[] = {
{ "transmit-seed", 0, 0, G_OPTION_ARG_INT, &option_transmit_seed,
"Seed for transmission RNG", "S" },
{ "receive-seed", 0, 0, G_OPTION_ARG_INT, &option_receive_seed,
"Seed for reception RNG", "S" },
{ "n-bytes", 'n', 0, G_OPTION_ARG_INT64, &option_n_bytes,
"Number of bytes to send in each test (default 100000)", "N" },
{ "timeout", 't', 0, G_OPTION_ARG_INT, &option_timeout,
"Deadlock detection timeout length, in seconds (default: 1200)", "S" },
{ "long-mode", 'l', 0, G_OPTION_ARG_NONE, &option_long_mode,
"Enable all tests, rather than a fast subset", NULL },
{ NULL },
};
int
main (int argc, char *argv[])
{
gboolean reliable;
StreamApi stream_api;
BufferSizeStrategy transmit_buffer_size_strategy;
BufferSizeStrategy receive_buffer_size_strategy;
BufferDataStrategy buffer_data_strategy;
guint32 transmit_seed;
guint32 receive_seed;
gsize n_bytes;
guint deadlock_timeout;
gboolean long_mode;
GOptionContext *context;
GError *error = NULL;
/* Argument parsing. Allow some of the test parameters to be specified on the
* command line. */
context = g_option_context_new ("— test send()/recv() correctness");
g_option_context_add_main_entries (context, entries, NULL);
if (!g_option_context_parse (context, &argc, &argv, &error)) {
g_printerr ("Option parsing failed: %s\n", error->message);
g_error_free (error);
exit (1);
}
/* Set up the defaults. */
transmit_seed = option_transmit_seed;
receive_seed = option_receive_seed;
n_bytes = option_n_bytes;
deadlock_timeout = option_timeout;
long_mode = option_long_mode;
#ifdef G_OS_WIN32
WSADATA w;
WSAStartup (0x0202, &w);
#endif
#if !GLIB_CHECK_VERSION(2, 36, 0)
g_type_init ();
#endif
#if !GLIB_CHECK_VERSION(2, 31, 8)
g_thread_init (NULL);
#endif
if (!long_mode) {
/* Quick mode. Just test each of the stream APIs in reliable and
* non-reliable mode, with a single pair of buffer strategies, and a single
* data strategy. */
/* Reliability. */
for (reliable = 0; reliable < 2; reliable++) {
/* Stream API. */
for (stream_api = 0;
(guint) stream_api < STREAM_API_N_ELEMENTS;
stream_api++) {
/* GIO streams must always be reliable. */
if (!reliable && stream_api_is_reliable_only (stream_api))
continue;
/* Non-reliable socket receives require large buffers. */
if (reliable) {
receive_buffer_size_strategy = BUFFER_SIZE_RANDOM;
} else {
receive_buffer_size_strategy = BUFFER_SIZE_CONSTANT_LARGE;
}
transmit_buffer_size_strategy = BUFFER_SIZE_RANDOM;
buffer_data_strategy = BUFFER_DATA_PSEUDO_RANDOM;
g_debug ("Running test (%u, %u, %" G_GSIZE_FORMAT ", %u, "
"%u, %u, %u, %u)…",
reliable, stream_api, n_bytes, transmit_buffer_size_strategy,
receive_buffer_size_strategy, buffer_data_strategy,
transmit_seed, receive_seed);
test (reliable, stream_api, n_bytes, transmit_buffer_size_strategy,
receive_buffer_size_strategy, buffer_data_strategy,
transmit_seed, receive_seed,
deadlock_timeout / 20 /* arbitrary reduction */);
}
}
goto done;
}
/* Transmit buffer strategy. */
for (transmit_buffer_size_strategy = 0;
(guint) transmit_buffer_size_strategy < BUFFER_SIZE_STRATEGY_N_ELEMENTS;
transmit_buffer_size_strategy++) {
/* Receive buffer strategy. */
for (receive_buffer_size_strategy = 0;
(guint) receive_buffer_size_strategy < BUFFER_SIZE_STRATEGY_N_ELEMENTS;
receive_buffer_size_strategy++) {
/* Transmit data strategy. */
for (buffer_data_strategy = 0;
(guint) buffer_data_strategy < BUFFER_DATA_STRATEGY_N_ELEMENTS;
buffer_data_strategy++) {
/* Reliability. */
for (reliable = 0; reliable < 2; reliable++) {
/* Stream API. */
for (stream_api = 0;
(guint) stream_api < STREAM_API_N_ELEMENTS;
stream_api++) {
/* GIO streams must always be reliable. */
if (!reliable && stream_api_is_reliable_only (stream_api))
continue;
/* Non-reliable socket receives require large buffers. We don’t
* claim to support using them with small (<< 65535B) buffers, so
* don’t test them. */
if (!reliable &&
receive_buffer_size_strategy != BUFFER_SIZE_CONSTANT_LARGE)
continue;
/* Non-reliable socket transmits will always block with huge
* buffers. */
if (!reliable &&
transmit_buffer_size_strategy == BUFFER_SIZE_CONSTANT_LARGE)
continue;
g_debug ("Running test (%u, %u, %" G_GSIZE_FORMAT ", %u, "
"%u, %u, %u, %u)…",
reliable, stream_api, n_bytes, transmit_buffer_size_strategy,
receive_buffer_size_strategy, buffer_data_strategy,
transmit_seed, receive_seed);
test (reliable, stream_api, n_bytes, transmit_buffer_size_strategy,
receive_buffer_size_strategy, buffer_data_strategy,
transmit_seed, receive_seed,
deadlock_timeout);
}
}
}
}
}
done:
#ifdef G_OS_WIN32
WSACleanup ();
#endif
return 0;
}
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