Allow packet_stream_reconnect() to cut offHEADmaster

 a connecting stream. Cleanup timer.

Signed-off-by: Andrew Opalach <andrew@akon.city>

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;
+}