summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--meson_options.txt1
-rw-r--r--src/multiplex.c6
-rw-r--r--src/packet_stream.c19
-rw-r--r--src/packet_stream.h2
-rw-r--r--src/rpc.c3
-rw-r--r--src/rpc2.h4
-rw-r--r--src/timer.c15
-rw-r--r--src/timer.h3
-rw-r--r--src/util/thread/thread_linux.c4
-rw-r--r--tests/meson.build1
-rw-r--r--tests/timer_clear_pending.c158
11 files changed, 183 insertions, 33 deletions
diff --git a/meson_options.txt b/meson_options.txt
index 2c984a4..883cbc0 100644
--- a/meson_options.txt
+++ b/meson_options.txt
@@ -2,4 +2,5 @@ option('event-loop', type: 'feature', value: 'enabled')
option('curl', type: 'feature', value: 'disabled')
option('json', type: 'feature', value: 'disabled')
option('win32-compat', type: 'boolean', value: false, yield: true)
+option('release-asserts', type: 'boolean', value: true, yield: true)
option('tests', type: 'feature', value: 'auto', yield: true)
diff --git a/src/multiplex.c b/src/multiplex.c
index 2de4621..0911540 100644
--- a/src/multiplex.c
+++ b/src/multiplex.c
@@ -108,9 +108,9 @@ static void direct_connect(struct nn_packet_stream *client, u8 id)
multiplex_direct_global.connection_callback(multiplex_direct_global.userdata, id, server);
server->direct = client;
client->direct = server;
- // Connected happens on the server first.
- nn_packet_stream_connected(server);
- nn_packet_stream_connected(client);
+ // Explicitly signal connected on the server first.
+ nn_packet_stream_set_connected(server);
+ nn_packet_stream_set_connected(client);
}
void nn_multiplex_direct_connect(struct nn_packet_stream *client, u8 id)
diff --git a/src/packet_stream.c b/src/packet_stream.c
index 7868242..0a45b22 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -104,7 +104,7 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
}
}
-static bool stream_connected(struct nn_packet_stream *stream)
+static bool set_stream_connected(struct nn_packet_stream *stream)
{
stream->connect = PACKET_STREAM_CONNECTED;
al_assert(stream->corked);
@@ -134,7 +134,7 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
// Try again on next POLLOUT, not sure if this can actually happen.
return;
}
- if (stream_connected(stream)) {
+ if (set_stream_connected(stream)) {
ev_io_start(stream->loop->ev, &stream->revent);
}
}
@@ -201,7 +201,7 @@ void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_eve
s32 fd = nn_socket_get_fd(&stream->sock);
ev_io_init_n(&stream->revent, stream_read_callback, fd, EV_READ);
ev_io_init_n(&stream->wevent, stream_write_callback, fd, EV_WRITE);
- if (stream_connected(stream)) {
+ if (set_stream_connected(stream)) {
ev_io_start(stream->loop->ev, &stream->revent);
}
}
@@ -234,15 +234,20 @@ void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_l
do_connect_internal(stream, addr, port);
}
+// Keep in mind this stream will have had it's state reset in stop_internal().
void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port)
{
+ if (stream->connect == PACKET_STREAM_CONNECTING) {
+ stop_write_internal(stream);
+ stop_internal(stream);
+ }
al_assert(stream->connect == PACKET_STREAM_DISCONNECTED);
do_connect_internal(stream, addr, port);
}
-bool nn_packet_stream_connected(struct nn_packet_stream *stream)
+bool nn_packet_stream_set_connected(struct nn_packet_stream *stream)
{
- return stream_connected(stream);
+ return set_stream_connected(stream);
}
void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork)
@@ -314,10 +319,8 @@ void nn_packet_stream_return_packets(struct nn_packet_stream *stream, struct nn_
void nn_packet_stream_disconnect(struct nn_packet_stream *stream)
{
+ al_assert(stream->connect != PACKET_STREAM_DISCONNECTED);
al_assert(stream->connect != PACKET_STREAM_DISCONNECTING);
- // If reusing this stream, keep in mind it will have been set back
- // to a default state in stop_internal().
- if (stream->connect == PACKET_STREAM_DISCONNECTED) return;
if (stream->direct) {
struct nn_packet_stream *direct = stream->direct;
direct->connect = PACKET_STREAM_DISCONNECTED;
diff --git a/src/packet_stream.h b/src/packet_stream.h
index 5531d00..d60af16 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -43,7 +43,7 @@ void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_eve
void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop,
u8 id, u8 type, str *addr, u16 port);
void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port);
-bool nn_packet_stream_connected(struct nn_packet_stream *stream);
+bool nn_packet_stream_set_connected(struct nn_packet_stream *stream);
void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork);
bool nn_packet_stream_send_packet(struct nn_packet_stream *stream, struct nn_packet *packet);
void nn_packet_stream_discard_queue(struct nn_packet_stream *stream);
diff --git a/src/rpc.c b/src/rpc.c
index 1b00e63..7938341 100644
--- a/src/rpc.c
+++ b/src/rpc.c
@@ -125,10 +125,11 @@ void nn_rpc_connect(struct nn_rpc *rpc, u8 id, u8 type, str *addr, u16 port)
nn_packet_stream_connect(rpc->conn->stream, rpc->loop, id, type, addr, port);
}
-void nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port)
+struct nn_rpc_connection *nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port)
{
al_assert(rpc->conn);
nn_packet_stream_reconnect(rpc->conn->stream, addr, port);
+ return rpc->conn;
}
// @TODO: Pack opcode into a u32, reduce ID by 8 bits (make struct with bitmask)
diff --git a/src/rpc2.h b/src/rpc2.h
index e6c2098..bed8822 100644
--- a/src/rpc2.h
+++ b/src/rpc2.h
@@ -45,11 +45,11 @@ void nn_rpc_add_command(struct nn_rpc *rpc, struct nn_rpc_command *command);
void nn_rpc_add_stream(struct nn_rpc *rpc, struct nn_packet_stream *stream);
struct nn_rpc_connection *nn_rpc_prepare_client(struct nn_rpc *rpc);
void nn_rpc_connect(struct nn_rpc *rpc, u8 id, u8 type, str *addr, u16 port);
-void nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port);
+ struct nn_rpc_connection *nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port);
struct nn_packet *nn_rpc_get_packet(struct nn_rpc *rpc, s8 op);
void nn_rpc_free(struct nn_rpc *rpc);
-// The order of commands per connection will be respected.
+// The order of commands per-connection will be respected.
void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet *packet,
void (*callback)(void *, struct nn_rpc_connection *conn, struct nn_packet *), void *userdata);
void nn_rpc_conn_flush(struct nn_rpc_connection *conn);
diff --git a/src/timer.c b/src/timer.c
index c0759cc..1d14a38 100644
--- a/src/timer.c
+++ b/src/timer.c
@@ -5,7 +5,7 @@ static void timer_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
(void)loop;
struct nn_timer *timer = (struct nn_timer *)w->data;
(void)revents;
- if (!timer->disabled) timer->callback(timer->userdata, timer);
+ timer->callback(timer->userdata, timer);
}
void nn_timer_init(struct nn_timer *timer, struct nn_event_loop *loop,
@@ -15,7 +15,6 @@ void nn_timer_init(struct nn_timer *timer, struct nn_event_loop *loop,
timer->callback = callback;
timer->userdata = userdata;
timer->timer.data = timer;
- timer->disabled = false;
ev_init_n(&timer->timer, timer_callback);
}
@@ -26,17 +25,7 @@ void nn_timer_set_repeat(struct nn_timer *timer, nn_os_tstamp repeat)
void nn_timer_again(struct nn_timer *timer)
{
- if (!timer->disabled) ev_timer_again(timer->loop->ev, &timer->timer);
-}
-
-void nn_timer_disable(struct nn_timer *timer)
-{
- timer->disabled = true;
-}
-
-void nn_timer_enable(struct nn_timer *timer)
-{
- timer->disabled = false;
+ ev_timer_again(timer->loop->ev, &timer->timer);
}
void nn_timer_stop(struct nn_timer *timer)
diff --git a/src/timer.h b/src/timer.h
index 15b467f..29107df 100644
--- a/src/timer.h
+++ b/src/timer.h
@@ -7,7 +7,6 @@
struct nn_timer {
ev_timer timer;
struct nn_event_loop *loop;
- bool disabled;
void (*callback)(void *, struct nn_timer *);
void *userdata;
};
@@ -16,6 +15,4 @@ void nn_timer_init(struct nn_timer *timer, struct nn_event_loop *loop,
void (*callback)(void *, struct nn_timer *), void *userdata);
void nn_timer_set_repeat(struct nn_timer *timer, nn_os_tstamp repeat);
void nn_timer_again(struct nn_timer *timer);
-void nn_timer_disable(struct nn_timer *timer);
-void nn_timer_enable(struct nn_timer *timer);
void nn_timer_stop(struct nn_timer *timer);
diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c
index cd80d06..c9edce5 100644
--- a/src/util/thread/thread_linux.c
+++ b/src/util/thread/thread_linux.c
@@ -91,9 +91,9 @@ void nn_cond_init(struct nn_cond *cond)
void nn_cond_wait(struct nn_cond *cond, struct nn_mutex *mutex)
{
cond->condition = false;
- while (!cond->condition) {
+ do {
pthread_cond_wait(&cond->cond, &mutex->mutex);
- }
+ } while (!cond->condition);
}
bool nn_cond_is_waiting(struct nn_cond *cond)
diff --git a/tests/meson.build b/tests/meson.build
index 0ff6ffb..1074e40 100644
--- a/tests/meson.build
+++ b/tests/meson.build
@@ -3,6 +3,7 @@ executable('condition_variable', ['condition_variable.c'], dependencies: naunet)
executable('timer_thread', ['timer_thread.c'], dependencies: naunet)
if get_option('event-loop').enabled()
executable('timer_loop', ['timer_loop.c'], dependencies: naunet)
+ executable('timer_clear_pending', ['timer_clear_pending.c'], dependencies: naunet)
executable('loop_sleep', ['loop_sleep.c'], dependencies: naunet)
if not is_windows
executable('stdin_lines', ['stdin_lines.c'], dependencies: naunet)
diff --git a/tests/timer_clear_pending.c b/tests/timer_clear_pending.c
new file mode 100644
index 0000000..35eb3dd
--- /dev/null
+++ b/tests/timer_clear_pending.c
@@ -0,0 +1,158 @@
+#include <nnwt/common.h>
+#include <nnwt/thread.h>
+#include <nnwt/timer.h>
+#include <nnwt/signal.h>
+#include <nnwt/time.h>
+
+//#define TEST_WAIT_ON_SIGNAL
+
+static s32 s_quit = 0;
+static struct nn_event_loop s_loop;
+#ifndef TEST_WAIT_ON_SIGNAL
+static struct nn_timer s_timer;
+#endif
+static struct nn_signal s_signal;
+static struct nn_cond s_cond;
+static struct nn_mutex s_mutex;
+
+#ifndef TEST_WAIT_ON_SIGNAL
+static bool flip = false;
+static bool any_fail = false;
+static bool last_was_timeout = false;
+#endif
+static f64 last_tick;
+static u64 iter = 0;
+
+#define DELAY 0.002
+
+#ifndef TEST_WAIT_ON_SIGNAL
+static void timer_callback(void *userdata, struct nn_timer *timer)
+{
+ (void)userdata;
+ (void)timer;
+ if (!flip) {
+ al_printf("*******FAIL*******\n");
+ any_fail = true;
+ } else if (!any_fail && last_was_timeout) {
+ // Calling timer_stop() here could avoid this but it should
+ // realistically not happen at nearly any above 0 delay.
+ al_assert(false && "Unexpected repeat timeout, signal thread deadlocked or couldn't keep up?");
+ }
+ al_printf("Timeout iter #%llu (%.5f)\n", iter, nn_get_tick() - last_tick);
+ last_was_timeout = true;
+}
+#endif
+
+static void signal_callback(void *userdata)
+{
+ (void)userdata;
+ f64 tick = nn_get_tick();
+#ifndef TEST_WAIT_ON_SIGNAL
+ if (flip) {
+ if (last_was_timeout) {
+ last_was_timeout = false;
+ } else {
+ al_printf("Stopped iter #%llu (%.5f)\n", iter, tick - last_tick);
+ }
+ // The point of this test is to call timer_stop() right before it was set
+ // to fire. So we only care about the case where signal_callback() runs first.
+ // To do that we have to keep it close though, meaning they will trade back and forth.
+ nn_timer_stop(&s_timer);
+ // Now wait for ~DELAY seconds and if the timer still fires after being
+ // stopped, a fail case will trigger.
+ } else {
+ iter++;
+ nn_timer_again(&s_timer);
+ }
+ flip = !flip;
+#else
+ iter++;
+ al_printf("Signal(#%llu) (%.5f)\n", iter, tick - last_tick);
+#endif
+ nn_mutex_lock(&s_mutex);
+ nn_cond_signal(&s_cond);
+ last_tick = tick;
+ if (s_quit) {
+#ifndef TEST_WAIT_ON_SIGNAL
+ nn_timer_stop(&s_timer);
+#endif
+ nn_signal_stop(&s_signal);
+ }
+ nn_mutex_unlock(&s_mutex);
+}
+
+static nn_thread_result NNWT_THREADCALL signal_and_wait_thread(void *userdata)
+{
+ (void)userdata;
+ for (;;) {
+ // Locking before signal_send() is necessary. Otherwise there would be a race
+ // between this send being processed and calling cond_signal() on the loop thread
+ // before we get to cond_wait() in this thread.
+ nn_mutex_lock(&s_mutex);
+ nn_signal_send(&s_signal);
+ if (s_quit) break;
+ nn_cond_wait(&s_cond, &s_mutex);
+ f64 tick = last_tick;
+ nn_mutex_unlock(&s_mutex);
+#ifndef TEST_WAIT_ON_SIGNAL
+ nn_thread_sleep(MAX((tick + DELAY) - nn_get_tick(), 0.0));
+#else
+ (void)tick;
+#endif
+ }
+ nn_mutex_unlock(&s_mutex);
+ return 0;
+}
+
+static void sigint_handler(s32 signum)
+{
+ (void)signum;
+ nn_mutex_lock(&s_mutex);
+ if (s_quit) {
+ al_printf("If the console output was frozen (deadlock), test failed.\n");
+ exit(128 + SIGINT);
+ }
+ s_quit = 1;
+ al_printf("...^C again to force exit.\n");
+ nn_mutex_unlock(&s_mutex);
+}
+
+s32 main(void)
+{
+ if (!nn_common_init(NULL)) return EXIT_FAILURE;
+
+ signal(SIGINT, sigint_handler);
+
+ nn_event_loop_init(&s_loop);
+
+ nn_signal_init(&s_signal, &s_loop, signal_callback, NULL);
+ nn_signal_start(&s_signal);
+
+#ifndef TEST_WAIT_ON_SIGNAL
+ nn_timer_init(&s_timer, &s_loop, timer_callback, NULL);
+ nn_timer_set_repeat(&s_timer, DELAY);
+#endif
+ last_tick = nn_get_tick();
+
+ nn_mutex_init(&s_mutex);
+ nn_cond_init(&s_cond);
+
+ struct nn_thread thread;
+ nn_thread_create(&thread, signal_and_wait_thread, NULL);
+
+ nn_event_loop_run(&s_loop);
+ nn_event_loop_destroy(&s_loop);
+
+ nn_mutex_destroy(&s_mutex);
+ nn_cond_destroy(&s_cond);
+
+#ifndef TEST_WAIT_ON_SIGNAL
+ al_printf(any_fail ? "Test failed.\n" : "Test passed.\n");
+#else
+ al_printf("Test passed.\n");
+#endif
+
+ nn_common_close();
+
+ return EXIT_SUCCESS;
+}
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 10:48:54 +0000