diff options
Diffstat (limited to 'src/libsink')
| -rw-r--r-- | src/libsink/sink.c | 478 | ||||
| -rw-r--r-- | src/libsink/sink.h | 3 |
2 files changed, 229 insertions, 252 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c index 0db5da2..80f40cd 100644 --- a/src/libsink/sink.c +++ b/src/libsink/sink.c @@ -1,6 +1,7 @@ #define AL_LOG_SECTION "sink" //#define AL_LOG_ENABLE_TRACE #include <al/log.h> +#include <al/random.h> #include <nnwt/multiplex.h> #include "../server/common.h" @@ -10,13 +11,11 @@ #include "sink.h" #include "common.h" -//#define CAMU_SINK_LOCAL //#define CAMU_SINK_ONESHOT // Requested state of the sinks outputs. enum { - SINK_EMPTY = 0, - SINK_PAUSED, + SINK_PAUSED = 0, SINK_PLAYING }; @@ -75,7 +74,7 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); // printf format for entries. #ifdef AL_DEBUG -#define ENTRY_FMT "#%u(%p)" +#define ENTRY_FMT "#%u(%p)" // @TODO: Put static and ended here at least. #define ENTRY_ARG(entry) (ENTRY_IS_VALID(entry) ? REMOTE_ENTRY_ID((entry)->id) : 0), ((entry) ? (entry) : 0x0) #else #define ENTRY_FMT "#%u" @@ -93,12 +92,15 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); #define AUDIO_ENDED(entry) (AUDIO_STATE(entry) >= BUFFER_ENDED) #define VIDEO_ENDED(entry) (VIDEO_STATE(entry) >= BUFFER_ENDED) -/* -#define ENTRY_ENDED(entry) \ + +#define AUDIO_STREAM(entry) ((al_assert(!AUDIO_EMPTY(entry)), (entry)->audio.buf.stream)) +#define VIDEO_STREAM(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.stream)) +#define VIDEO_IS_SINGLE_FRAME(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.single_frame)) + +#define ENTRY_EVAL_ENDED(entry) \ ((AUDIO_ENDED(entry) && (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry))) || \ (AUDIO_EMPTY(entry) && VIDEO_ENDED(entry))) -*/ -#define ENTRY_ENDED(entry) entry->ended +#define ENTRY_ENDED(entry) (al_assert(entry->ended == ENTRY_EVAL_ENDED(entry)), entry->ended) // IGNORED = ENDED or EMPTY. #define AUDIO_ADDED_OR_IGNORED(entry) (AUDIO_STATE(entry) >= BUFFER_ADDED || AUDIO_EMPTY(entry)) @@ -107,14 +109,10 @@ AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX); #define AUDIO_ENDED_OR_EMPTY(entry) (AUDIO_ENDED(entry) || AUDIO_EMPTY(entry)) #define VIDEO_ENDED_OR_EMPTY(entry) (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry)) -#define AUDIO_STREAM(entry) ((al_assert(!AUDIO_EMPTY(entry)), (entry)->audio.buf.stream)) -#define VIDEO_STREAM(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.stream)) -#define VIDEO_IS_SINGLE_FRAME(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.single_frame)) - #if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED -#define BLOCKING_SLEEP(delay) nn_thread_sleep(delay) +#define BLOCKING_SLEEP(sink, delay) ((void)sink, nn_thread_sleep(delay)) #else -#define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay) +#define BLOCKING_SLEEP(sink, delay) nn_event_loop_sleep(sink->loop, delay) #endif #define CMD(cmd, ...) ((struct camu_sink_cmd){ .op = cmd, __VA_ARGS__ }) @@ -178,6 +176,7 @@ static void refresh_video_output(struct camu_sink *sink) static inline void add_entry_audio_buffer(struct camu_sink_entry *entry) { + log_trace("add_entry_audio_buffer("ENTRY_FMT").", ENTRY_ARG(entry)); struct camu_sink *sink = entry->sink; #ifdef CAMU_MIXER_THREADED_START_STOP sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf); @@ -189,6 +188,7 @@ static inline void add_entry_audio_buffer(struct camu_sink_entry *entry) static inline void add_entry_video_buffer(struct camu_sink_entry *entry) { #ifndef CAMU_SINK_NO_VIDEO + log_trace("add_entry_video_buffer("ENTRY_FMT").", ENTRY_ARG(entry)); struct camu_sink *sink = entry->sink; sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf); #else @@ -200,6 +200,8 @@ static void remove_entry_audio_buffer(struct camu_sink_entry *entry) { al_assert(!AUDIO_ENDED(entry)); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); + log_trace("remove_entry_audio_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry), + BOOLSTR(AUDIO_STATE(entry) == BUFFER_ADDED)); switch (AUDIO_STATE(entry)) { case BUFFER_ADDED: AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED; @@ -224,6 +226,8 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry) // Don't assert !entry->ended here because of single frame handling. al_assert(!VIDEO_ENDED(entry)); al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + log_trace("remove_entry_video_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry), + BOOLSTR(VIDEO_STATE(entry) == BUFFER_ADDED)); switch (VIDEO_STATE(entry)) { case BUFFER_ADDED: VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED; @@ -251,32 +255,12 @@ static void remove_entry_video_buffer(struct camu_sink_entry *entry) // switch_to() -> maybe_add_to_previous() -> (possibly delayed)remove_entry_buffers(). static void remove_entry_buffers(struct camu_sink_entry *entry) { - log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); + log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); if (!AUDIO_ENDED(entry)) remove_entry_audio_buffer(entry); if (!VIDEO_ENDED(entry)) remove_entry_video_buffer(entry); } -static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry); -static void add_video_if_set_and_buffered(struct camu_sink_entry *entry); - -static void add_or_queue_entry(struct camu_sink_entry *entry) -{ - log_trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); - al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); - al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); - // Either buffer could be DETACHED. - if (AUDIO_STATE(entry) == BUFFER_INIT) { - AUDIO_STATE(entry) = BUFFER_QUEUED; - } else if (!AUDIO_ENDED_OR_EMPTY(entry)) { - add_audio_if_set_and_buffered(entry); - } - if (VIDEO_STATE(entry) == BUFFER_INIT) { - VIDEO_STATE(entry) = BUFFER_QUEUED; - } else if (!VIDEO_ENDED_OR_EMPTY(entry)) { - add_video_if_set_and_buffered(entry); - } -} - // Disconnecting a packet stream twice before a reconnect is an error. static void maybe_disconnect_entry(struct camu_sink_entry *entry) { @@ -287,57 +271,20 @@ static void maybe_disconnect_entry(struct camu_sink_entry *entry) } } -#ifdef CAMU_SINK_LOCAL -static void local_entry_pause(struct camu_sink *sink, struct camu_sink_entry *entry) -{ - if (!camu_clock_is_paused(&entry->clock)) { - entry->paused = true; - camu_clock_pause(&entry->clock, 0); - log_info("Clock paused."); - // Audio stop will be handled by a BUFFER_PAUSED callback. - if (!VIDEO_ENDED_OR_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) { -#ifndef CAMU_SINK_NO_VIDEO - sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL); -#endif - sink->video.state = SINK_PAUSED; - } - } else { - entry->paused = false; - camu_clock_resume(&entry->clock, 0); - log_info("Clock resumed."); - if (!VIDEO_ENDED_OR_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) { -#ifndef CAMU_SINK_NO_VIDEO - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL); -#endif - sink->video.state = SINK_PLAYING; - } - if (!AUDIO_ENDED_OR_EMPTY(entry) && sink->audio.state == SINK_PAUSED) { - sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL); - sink->audio.state = SINK_PLAYING; - } - } -} -#endif - // sink->current could be NULL. static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink) { return ENTRY_IS_VALID(sink->target) ? sink->target : sink->current; } -static inline s32 get_sequence_for_command(struct camu_sink_entry *entry) +static inline s32 get_sequence_for_command(struct camu_sink *sink, struct camu_sink_entry *entry) { + // If local, using entry->sequence could only lead to feeling like your inputs were eaten. + if (!sink->local && entry) { + return entry->sequence; + } // SEQUENCE_ANY resolves order on the server. - s32 sequence = LIANA_SEQUENCE_ANY; - // If local, this could only lead to feeling like your inputs were eaten. -#ifndef CAMU_SINK_LOCAL - // Entry is sent as an argument via get_entry_for_command(), so it can only - // be a valid entry or NULL and not a dangling target. - if (entry) sequence = entry->sequence; -#else - (void)entry; -#endif - return sequence; + return LIANA_SEQUENCE_ANY; } static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) @@ -433,33 +380,29 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) return; } case SKIP: { - if (!sink->conn) return; + if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_SKIP); - nn_packet_write_s32(packet, get_sequence_for_command(entry)); + nn_packet_write_s32(packet, get_sequence_for_command(sink, entry)); nn_packet_write_s32(packet, (s32)cmd->v.i); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); break; } case TOGGLE_PAUSE: { + if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; -#ifdef CAMU_SINK_LOCAL - if (!entry->held) local_entry_pause(sink, entry); -#else - if (!sink->conn) return; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE); - nn_packet_write_s32(packet, get_sequence_for_command(entry)); + nn_packet_write_s32(packet, get_sequence_for_command(sink, entry)); nn_packet_write_f64(packet, cmd->v.f); nn_rpc_connection_command(sink->conn, packet, NULL, NULL); -#endif break; } case SEEK: { - if (!sink->conn) return; + if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -471,13 +414,12 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case RESEEK: { - if (!sink->conn) return; // Crude way to trigger "re-add sink to list". - nn_rpc_conn_disconnect(sink->conn); + if (sink->conn) nn_rpc_conn_disconnect(sink->conn); break; } case SHUFFLE: { - if (!sink->conn) return; + if (!sink->connected) return; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); nn_packet_write_u8(packet, CAMU_LIST_SHUFFLE); @@ -485,7 +427,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd) break; } case END: { - if (!sink->conn) return; + if (!sink->connected) return; struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque; struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION); nn_packet_write_str(packet, &sink->default_list); @@ -513,14 +455,13 @@ static void queue_signal_callback(void *userdata) static void mixer_callback(void *userdata, u8 op) { struct camu_sink *sink = (struct camu_sink *)userdata; - // @TODO: Isn't the idea of MIXER_EMPTY to not explicitly STOP the audio in places we - // expect the mixer to be empty (rely on MIXER_EMPTY). if (op == CAMU_MIXER_EMPTY) { log_info("Mixer empty."); - // Regardless of if we are checking an entry's state here, we have to sync with - // do_add_entry() because the order of START/STOPs in the queue matters. + // Regardless of if we are checking an entry's state, we have to sync + // with do_add_entry() because the order of START/STOPs in the queue matters. nn_mutex_lock(&sink->lock); - // This feels a bit too loose. + // This check is too loose. Still about as good as any naive implementation + // of SINK_EMPTY, though. if (!(sink->current && AUDIO_STATE(sink->current) == BUFFER_ADDED)) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); } @@ -572,13 +513,13 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink) } } -static void maybe_remove_previous(struct camu_sink *sink) +static void maybe_run_previous(struct camu_sink *sink) { - log_trace("maybe_remove_previous(), previous_count: %u.", sink->previous.count); + log_trace("maybe_run_previous(), previous_count: %u.", sink->previous.count); struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { remove_entry_buffers(previous); - // Cleanup entries from old connections. This is especially important to + // Cleanup entries from old connections. Especially important to // keep reseek() from being overly wasteful. if (CONNECTION_NUMBER(previous->id) != sink->connection_number) { queue_cmd(sink, CMD(EJECT_ENTRY, .opaque = previous)); @@ -590,29 +531,43 @@ static void maybe_remove_previous(struct camu_sink *sink) // Due to the looseness of the previous queue, we may have to explicitly remove an // entry if it becomes incorrect to attempt removing it's buffers. The most obvious example // being at the point it's freed. -static void remove_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) +static void run_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key) { bool removed = false; struct camu_sink_entry *previous; al_array_foreach(sink->previous, i, previous) { if (previous == key) { - maybe_remove_previous(sink); + maybe_run_previous(sink); removed = true; break; } } - log_trace("remove_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); + log_trace("run_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed)); +} + +static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry) +{ + bool removed = false; + struct camu_sink_entry *previous; + al_array_foreach(sink->previous, i, previous) { + if (previous == entry) { + al_array_remove_at(sink->previous, i); + removed = true; + break; + } + } + log_trace("maybe_remove_from_previous("ENTRY_FMT"), removed: %s.", ENTRY_ARG(entry), BOOLSTR(removed)); + return removed; } // Every call to maybe_add_to_previous() must map to a remove_entry_buffers(). static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target) { - log_trace("maybe_add_to_previous("ENTRY_FMT", "ENTRY_FMT").", ENTRY_ARG(previous), ENTRY_ARG(target)); + log_trace("maybe_add_to_previous("ENTRY_FMT"), target: "ENTRY_FMT".", ENTRY_ARG(previous), ENTRY_ARG(target)); al_assert(previous != target); - // If none of an entry's buffers are added, we don't care about adding it to previous. - bool dangling_target = target == (struct camu_sink_entry *)0xb00b; - if (dangling_target || ENTRY_ENDED(target) || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) { + bool ignore_target = target == (struct camu_sink_entry *)0xb00b || ENTRY_ENDED(target); + if (ignore_target || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) { remove_entry_buffers(previous); return; } @@ -621,15 +576,14 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool skip_video) { - maybe_remove_previous(entry->sink); + struct camu_sink *sink = entry->sink; + maybe_run_previous(sink); queue_cmds(entry->sink, 2, CMD((skip_video || entry->paused) ? STOP : START, .v.u = CAMU_SINK_VIDEO), CMD((skip_audio || entry->paused) ? STOP : START, .v.u = CAMU_SINK_AUDIO) ); - if (VIDEO_ENDED_OR_EMPTY(entry)) { - // Clear the screen if skipping from a video to an audio-only entry. - refresh_video_output(entry->sink); - } + // Clear the screen if skipping from a video to an audio-only entry. + if (VIDEO_ENDED_OR_EMPTY(entry)) refresh_video_output(sink); } // Call this after setting state to ADDED because this entry might be in previous. @@ -638,13 +592,16 @@ static void do_add_entry(struct camu_sink_entry *entry) bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry); // Single frames are unconditionally added in add_video_if_set_and_buffered(). bool skip_video = VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); + if (!skip_audio && !skip_video) { + al_assert(VIDEO_STATE(entry) == AUDIO_STATE(entry)); + } // Whether to add audio or video first could be a consideration for responsiveness. if (!skip_video) add_entry_video_buffer(entry); if (!skip_audio) add_entry_audio_buffer(entry); after_add_entry(entry, skip_audio, skip_video); } -void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) +static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) { al_assert(!ENTRY_ENDED(entry)); al_assert(AUDIO_STATE(entry) != BUFFER_INIT); @@ -664,7 +621,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry) } } -void add_video_if_set_and_buffered(struct camu_sink_entry *entry) +static void add_video_if_set_and_buffered(struct camu_sink_entry *entry) { // Single frame entries will be added/removed with ended set. if (ENTRY_ENDED(entry)) al_assert(VIDEO_IS_SINGLE_FRAME(entry)); @@ -691,6 +648,35 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry) } } +static void add_or_queue_entry(struct camu_sink_entry *entry) +{ + log_trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry)); + al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED); + al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED); + // Either buffer could be DETACHED. + if (AUDIO_STATE(entry) == BUFFER_INIT) { + AUDIO_STATE(entry) = BUFFER_QUEUED; + } else if (!AUDIO_ENDED_OR_EMPTY(entry)) { + add_audio_if_set_and_buffered(entry); + } + if (VIDEO_STATE(entry) == BUFFER_INIT) { + VIDEO_STATE(entry) = BUFFER_QUEUED; + } else if (!VIDEO_ENDED_OR_EMPTY(entry)) { + add_video_if_set_and_buffered(entry); + } +} + +static void ensure_single_frame_removed(struct camu_sink_entry *entry) +{ + if (!VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { + remove_entry_video_buffer(entry); + struct camu_sink *sink = entry->sink; + while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); } + } +} + +// This is the only function that sets sink->current. static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) { struct camu_sink_entry *current = sink->current; @@ -703,8 +689,13 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) if (suspended) { al_assert(suspended == current); // It should be impossible for a buffer to be QUEUED while it's entry is suspended. + // Even for a single frame video buffer because we wouldn't even know if it only + // has a single frame yet. al_assert(AUDIO_STATE(suspended) != BUFFER_QUEUED); al_assert(VIDEO_STATE(suspended) != BUFFER_QUEUED); + ensure_single_frame_removed(suspended); + al_assert(AUDIO_STATE(suspended) != BUFFER_ADDED); + al_assert(VIDEO_STATE(suspended) != BUFFER_ADDED); sink->suspended = NULL; log_warn("Unset suspended entry as a substitute for remove."); } else { @@ -713,46 +704,34 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target) } bool dangling_target = target == (struct camu_sink_entry *)0xb00b; + if (dangling_target) log_trace("Ignoring dangling target."); bool stop_video = dangling_target; - if (dangling_target) { - log_trace("Ignored dangling target."); - } else { - remove_previous_if_contains(sink, target); - add_or_queue_entry(target); - stop_video = VIDEO_ENDED_OR_EMPTY(target) || VIDEO_IS_SINGLE_FRAME(target); - } - - if (dangling_target) { - sink->current = NULL; - } else { + if (!dangling_target) { target->audio.ignore_paused = false; - if (!target->paused) { + if (!sink->local && !target->paused) { camu_audio_buffer_resync(&target->audio.buf); } - sink->current = target; + if (!maybe_remove_from_previous(sink, target)) { + add_or_queue_entry(target); + } + al_assert(AUDIO_STATE(target) != BUFFER_INIT); + al_assert(VIDEO_STATE(target) != BUFFER_INIT); + // If target was just created, it's AUDIO/VIDEO_STATE() will be QUEUED. + // Meaning, at this point, it's video buffer would be considered empty + // as well as being too early to tell if it's static or not. + stop_video = VIDEO_ENDED_OR_EMPTY(target) || VIDEO_IS_SINGLE_FRAME(target); } if (stop_video) { queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO)); - refresh_video_output(sink); + if (dangling_target || VIDEO_ENDED_OR_EMPTY(target)) { + refresh_video_output(sink); + } } + + sink->current = dangling_target ? NULL : target; } -// @TODO: Could entry->ended be redundant? What about entry->held? -// - The catalyst for this is having to check entry->ended in CLIENT_REMOVE_BUFFERS breaking the ability -// to maintain the same state if called consecutively. -// - As long as end_entry_and_advance_queue() evaluates target, what is the point of -// ended except to say all of an entry's buffers are ended. -// (except to guess what the server thinks, which was it's original purpose). -// - Knowing if the outputs are paused is clearly necessary but what about it's relationship to held? -// - entry->paused basically means preempt any queued skip action. -// - On video or audio paused, check if SINK_EMPTY or PAUSED and run target (currently not handled at all). -// - Fix SINK_LOCAL relying on entry->paused hint from server (rely on clock state?). -// entry->paused reduced to nothing but a hint about if the entry is paused when skipping. -// Only consideration is for the sink outputs to never stay paused when playing and vise versa -// - Then entry->paused actually doesn't matter (sink state is above entry anyway and we should be able to -// rely on the clock per-entry). -#ifndef CAMU_SINK_LOCAL static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at) { struct camu_sink_entry *current = sink->current; @@ -760,34 +739,29 @@ static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *ta ENTRY_ARG(target), at / 1000000.0, ENTRY_ARG(current)); al_assert(target != current); al_assert(!sink->target); - // This is extra verbose because the order is important. - // 1. sink->target has to be set before calling clock_pause(). - if (current && !ENTRY_ENDED(current)) { - sink->target = target; - } - // 2. current must still be paused, even if it's ended. + sink->target = target; + bool immediate = !current || ENTRY_ENDED(current); if (current) { current->audio.ignore_paused = true; - camu_clock_pause(¤t->clock, at); + immediate |= camu_clock_pause(¤t->clock, at); } - // 3. In the immediate swap case, switch_to() has to come last. - if (!current || ENTRY_ENDED(current)) { - switch_to(sink, target); + if (immediate) { + switch_to(sink, sink->target); + sink->target = NULL; } } -#endif static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry) { log_debug("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry)); - remove_previous_if_contains(sink, entry); + run_previous_if_contains(sink, entry); #ifdef CAMU_SINK_ONESHOT sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL); return false; #endif // This entry's buffers cannot be added again until after a reset. entry->ended = true; - al_assert(ENTRY_ENDED(entry)); + al_assert(ENTRY_EVAL_ENDED(entry)); queue_cmd(sink, CMD(END, .v.u = entry->reset_token, .opaque = entry)); #ifdef LIANA_LIST_SCUFFED_LOOP log_info("Looping."); @@ -819,8 +793,13 @@ static void audio_buffer_callback(void *userdata, u8 op) case CAMU_BUFFER_UNCORK: lia_vcr_uncork(entry->audio.track); break; - case CAMU_BUFFER_PAUSED: + case CAMU_BUFFER_PAUSED: // Comes from audio read() thread. nn_mutex_lock(&sink->lock); + // entry->paused could plausibly be false here if the lock was held + // by pause_command_callback() to resume. This can be simulated by + // calling list_toggle_pause() twice in server/list_action_callback() + // for each sink request. Spaced by an nn_event_loop_sleep(~15500us) + // (no_video, MINIAUDIO_LOW_LATENCY mode). if (!entry->audio.ignore_paused && entry->paused) { log_info("Audio buffer paused."); queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO)); @@ -929,30 +908,31 @@ static void clock_callback(void *userdata, u8 op) static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry) { bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); - f64 avg_frame_duration = entry->video.buf.avg_frame_duration; -#ifdef CAMU_SINK_LOCAL - if (!AUDIO_EMPTY(entry) && !ignore_video) { - f64 audio = camu_mixer_get_latency(sink->audio.mixer); - struct camu_renderer *renderer = sink->video.renderer; - f64 video = renderer->get_latency(renderer) * avg_frame_duration; - camu_audio_buffer_set_latency(&entry->audio.buf, -audio); - camu_video_buffer_set_latency(&entry->video.buf, -video); - } - // If we're local we don't have to worry about syncing audio-only entries. - camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video); - camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); -#else - // When trying to sync clients with different audio/video latencies (common case), - // our only option is to shift each buffer forward directly by their latency. f64 audio = camu_mixer_get_latency(sink->audio.mixer); + u32 frames = 0; + f64 video = 0.0; if (!ignore_video) { struct camu_renderer *renderer = sink->video.renderer; - f64 video = renderer->get_latency(renderer) * avg_frame_duration; - camu_video_buffer_set_latency(&entry->video.buf, video); + if (renderer) { + f64 avg_frame_duration = entry->video.buf.avg_frame_duration; + frames = renderer->get_latency(renderer); + video = frames * avg_frame_duration; + } } camu_audio_buffer_set_latency(&entry->audio.buf, audio); camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); -#endif + camu_video_buffer_set_latency(&entry->video.buf, video); + log_info("video_latency: %f (%u frames), audio_latency: %f.", video, frames, audio); + if (sink->local) { + camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video); + // When the video buffer starts the clock, we have to consider the audio + // buffer is treating the last period of silence sent during a paused clock + // as part of the stream. This poses an issue for sync because it requires more + // than a period-length offset to not skip data at the start. Combined with the + // fact that the timing of a request for the next period doesn't have be uniform. + // Could make a diagram of this. // @TODO + camu_clock_offset(&entry->clock, MAX(audio, video)); + } } static void run_queue_by_opaque(struct camu_sink *sink, void *opaque) @@ -1074,29 +1054,33 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str case LIANA_CLIENT_REMOVE_BUFFERS: { struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); - log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s.", ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current)); + log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s, single_frame: %s.", + ENTRY_ARG(entry), BOOLSTR(rec->reconnect), + BOOLSTR(rec->unconfigured), BOOLSTR(entry == sink->current), + BOOLSTR(!VIDEO_EMPTY(entry) ? VIDEO_IS_SINGLE_FRAME(entry) : false)); if (entry == sink->current) { - al_assert(AUDIO_STATE(entry) != BUFFER_INIT); - al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + if (rec->reconnect) { + // Else, possible failed or aborted reconnect (Empty buffer could be INIT). + al_assert(AUDIO_STATE(entry) != BUFFER_INIT); + al_assert(VIDEO_STATE(entry) != BUFFER_INIT); + } if (sink->target) { // This is necessary to avoid re-adding an entry with an in-between clock state. // See note in clock.c::camu_clock_seek(). switch_to(sink, sink->target); sink->target = NULL; + } else if (rec->reconnect) { + // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers. + sink->suspended = entry; } } - // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers. - if (entry == sink->current && rec->reconnect) { - sink->suspended = entry; - } - // This entry might be in previous if it was added to previous then, // 1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered. // 2. it was seeked. - remove_previous_if_contains(sink, entry); - // AUDIO/VIDEO_STATE() could be INIT after this point, even if entry = current. + run_previous_if_contains(sink, entry); + // AUDIO/VIDEO_STATE() could be INIT at this point, even if entry = current. // We should treat CLIENT_REMOVE_BUFFERS as a function that removes an entry's buffers and // resets it's buffered state. For a non-empty entry that means remove_entry_audio/video_buffer() @@ -1104,7 +1088,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str if (AUDIO_STATE(entry) == BUFFER_QUEUED) AUDIO_STATE(entry) = BUFFER_INIT; if (VIDEO_STATE(entry) == BUFFER_QUEUED) VIDEO_STATE(entry) = BUFFER_INIT; - // Ignore unconfigured entries. + // Don't consider unconfigured entries past this point. if (rec->unconfigured) { al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry)); nn_mutex_unlock(&sink->lock); @@ -1136,7 +1120,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str nn_mutex_unlock(&sink->lock); while ((!skip_audio && entry_audio_buffer_held(entry)) || (!skip_video && entry_video_buffer_held(entry))) { - BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); + BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); } // At this point we can be sure that the entry's buffers are no longer in use. @@ -1150,13 +1134,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } if (VIDEO_STATE(entry) == BUFFER_ERRORED) { al_array_push(rec->detached, VIDEO_STREAM(entry)); - } else if (!VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { - // Don't request a duplicate single_frame. + } else if (VIDEO_STATE(entry) == BUFFER_ADDED && VIDEO_IS_SINGLE_FRAME(entry)) { + // Try to not request a duplicate frame. We can only be sure the frame wasn't + // dropped by the VCR if it's ADDED. rec->mask &= ~(1 << VIDEO_STREAM(entry)->index); } } - // Finalize the removal of the buffers by handling ended (ENDED or ERRORED) buffers. + // Finalize the removal of the buffers by handling "ended" (ENDED or ERRORED) buffers. // ENDED: Set to CONFIGURED to emulate the two removes earlier in this case. // ERRORED: Set to DETACHED and they will now be considered empty. if (AUDIO_STATE(entry) == BUFFER_ENDED) { @@ -1178,13 +1163,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str break; } - case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED. + case LIANA_CLIENT_RECOVER_TO: { struct lia_timing *time = (struct lia_timing *)opaque; -#ifdef CAMU_SINK_LOCAL - // List has no concept of a local sink, ignore it's request. + time->pos = camu_clock_get_last_pts(&entry->clock) * (u64)1000000; time->at = 0; -#endif - log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), time->pos / 1000000.0, entry->clock.paused_at); + break; + } + case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED. + struct lia_timing *time = (struct lia_timing *)opaque; + if (sink->local) time->at = 0; + f64 pos = time->pos / 1000000.0; + log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), pos, entry->clock.paused_at); nn_mutex_lock(&sink->lock); bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry); if (!AUDIO_EMPTY(entry)) { @@ -1193,24 +1182,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video); } if (!ignore_video) { - camu_video_buffer_reset(&entry->video.buf, time->pos / 1000000.0); - } -#ifdef LIANA_LIST_SCUFFED_LOOP - if (time->pos == 0) { - camu_clock_loop(&entry->clock, camu_clock_get_last_pts(&entry->clock)); - } else { -#endif - camu_clock_seek(&entry->clock, time->pos / 1000000.0, time->at); -#ifdef LIANA_LIST_SCUFFED_LOOP + camu_video_buffer_reset(&entry->video.buf, pos); } -#endif + camu_clock_seek(&entry->clock, pos, time->at); nn_mutex_unlock(&sink->lock); break; } case LIANA_CLIENT_RECONNECTED: { struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque; nn_mutex_lock(&sink->lock); - log_trace("reconnected("ENTRY_FMT"), suspended: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), ENTRY_ARG(sink->suspended), AUDIO_STATE(entry), VIDEO_STATE(entry)); + log_trace("reconnected("ENTRY_FMT"), suspended: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.", + ENTRY_ARG(entry), ENTRY_ARG(sink->suspended), AUDIO_STATE(entry), VIDEO_STATE(entry)); if (entry == sink->suspended) { al_assert(entry == sink->current); // Let this be the only other explicit BUFFER_ENDED check, or this will @@ -1262,11 +1244,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str nn_mutex_lock(&sink->lock); // We can be assured that CLIENT_REMOVE_BUFFERS has been called on this entry. - // Only possible case that a buffer could still be added. - if (rec->reconnect && !VIDEO_EMPTY(entry) && VIDEO_IS_SINGLE_FRAME(entry)) { - remove_entry_video_buffer(entry); - while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); } - } + // If a client is closed after a failed reconnect, a single frame + // video buffer could still be added + if (rec->reconnect) ensure_single_frame_removed(entry); al_assert(AUDIO_STATE(entry) != BUFFER_ADDED); al_assert(VIDEO_STATE(entry) != BUFFER_ADDED); @@ -1291,7 +1271,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str } } // If current was never fully added we need to call this here. - maybe_remove_previous(sink); + maybe_run_previous(sink); } nn_mutex_unlock(&sink->lock); @@ -1316,7 +1296,6 @@ static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id) entry->ended = false; camu_clock_init(&entry->clock, clock_callback, entry); - entry->held = false; entry->client.callback = client_callback; entry->client.userdata = entry; @@ -1412,25 +1391,8 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, goto out; } -#ifdef CAMU_SINK_LOCAL - (void)at; - al_assert(entry != current); - log_trace("set("ENTRY_FMT"), %s(local), created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create)); - if (current) { - current->audio.ignore_paused = true; - if (!current->paused) { - // paused has to map directly to the clock state here. - al_assert(!camu_clock_is_paused(¤t->clock)); - camu_clock_pause(¤t->clock, 0); - } - current->held = true; - } - entry->held = false; - if (!entry->paused) { - camu_clock_resume(&entry->clock, 0); - } - switch_to(sink, entry); -#else + if (sink->local) at = 0; + // For target to be set that must mean current is set, armed to pause, and not ended. struct camu_sink_entry *prev_target = sink->target; log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry), @@ -1501,7 +1463,6 @@ static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn, camu_clock_resume(&entry->clock, at); break; } -#endif out: nn_mutex_unlock(&sink->lock); @@ -1532,10 +1493,7 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con al_assert(entry->sequence == sequence); log_trace("pause("ENTRY_FMT"), %s, audio_state: %hhu, video_state: %hhu.", ENTRY_ARG(entry), lia_pause_op_name(pause), AUDIO_STATE(entry), VIDEO_STATE(entry)); -#ifdef CAMU_SINK_LOCAL - (void)at; - if (!entry->held) local_entry_pause(sink, entry); -#else + if (sink->local) at = 0; switch (pause) { case LIANA_PAUSE_PAUSE: { entry->paused = true; @@ -1553,13 +1511,14 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_VIDEO)); } if (!AUDIO_ENDED_OR_EMPTY(entry)) { - camu_audio_buffer_resync(&entry->audio.buf); + if (!sink->local) { + camu_audio_buffer_resync(&entry->audio.buf); + } queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_AUDIO)); } break; } } -#endif nn_mutex_unlock(&sink->lock); out: @@ -1629,8 +1588,12 @@ static void identify_on_connection(struct camu_sink *sink) static void connection_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; + nn_timer_stop(&sink->reconnect_timer); + if (sink->conn) al_assert(sink->conn == conn); sink->conn = conn; - sink->connection_number++; + sink->connected = true; + sink->connection_number = al_u16_inc_wrap(sink->connection_number); + if (sink->connection_number == 0) sink->connection_number = 1; identify_on_connection(sink); } @@ -1638,25 +1601,31 @@ static void reconnect_timer_callback(void *userdata, struct nn_timer *timer) { struct camu_sink *sink = (struct camu_sink *)userdata; (void)timer; - nn_timer_stop(&sink->reconnect_timer); - nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); + if (sink->conn) { + // If the client was connecting, reconnect() will force a disconnect before reconnecting. + log_warn("Forcing reconnect due to timeout."); + } + sink->conn = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); } static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn) { struct camu_sink *sink = (struct camu_sink *)userdata; - bool reconnect = !sink->reconnect_timer.disabled; + bool reconnect = sink->conn != NULL; bool disconnected = sink->conn && sink->connection_number > 0; if (sink->conn) { al_assert(sink->conn == conn); sink->conn = NULL; + sink->connected = false; + } else { + al_assert(!reconnect || !sink->connected); } if (reconnect) { if (disconnected) { - log_info("Connection to server closed, attempting reconnect..."); + log_warn("Connection to server closed, attempting reconnect..."); nn_rpc_reconnect(&sink->client, &sink->addr, sink->port); } else { - log_info("Failed to connect to server, trying again..."); + log_warn("Failed to connect to server, trying again..."); nn_timer_again(&sink->reconnect_timer); } } @@ -1671,7 +1640,8 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, sink->connection_number = 0; nn_mutex_init(&sink->lock); nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink); - nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1500000)); + // This is also effectively a timeout for attempted reconnects. + nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000)); nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink); nn_signal_start(&sink->queue_signal); camu_queue_init(sink->queue); @@ -1681,7 +1651,7 @@ bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop, al_array_init(sink->previous); al_array_init(sink->entries); // Start high to exercise the wrapping path. - sink->lru = SINK_LRU_MAX - 2; + sink->lru = SINK_LRU_MAX - al_random_int(0, ENTRY_MAX_AGE); mixer->callback = mixer_callback; mixer->userdata = sink; sink->audio.state = SINK_PAUSED; @@ -1704,6 +1674,7 @@ bool camu_sink_connect(struct camu_sink *sink, str *name, u8 type, str *addr, u1 nn_rpc_add_command(&sink->client, &commands[i]); } #ifdef CAMU_DIRECT_MODE + // Note that direct_connect() runs connection_callback() directly. nn_multiplex_direct_connect(sink->conn->stream, CAMU_MULTIPLEX_RPC); #else nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port); @@ -1736,7 +1707,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink) struct camu_sink_entry *current = get_entry_for_command(sink); nn_mutex_unlock(&sink->lock); if (!current) return; - f64 pts = camu_clock_get_pts(¤t->clock, 0.0, false); + bool armed_for_pause = false; + camu_clock_external_pause(¤t->clock); + f64 pts = camu_clock_get_pts(¤t->clock, 0.0, false, &armed_for_pause); queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = current)); } @@ -1791,7 +1764,7 @@ void camu_sink_shuffle(struct camu_sink *sink) void camu_sink_stop(struct camu_sink *sink) { queue_cmds(sink, 3, - CMD(STOP, .v.u = CAMU_SINK_AUDIO), + CMD(STOP, .v.u = CAMU_SINK_AUDIO), CMD(CLEAR_BUFFERS, .v.u = CAMU_SINK_AUDIO), CMD(CLOSE) ); @@ -1800,8 +1773,11 @@ void camu_sink_stop(struct camu_sink *sink) void camu_sink_close(struct camu_sink *sink) { nn_timer_stop(&sink->reconnect_timer); - nn_timer_disable(&sink->reconnect_timer); - if (sink->conn) nn_rpc_conn_disconnect(sink->conn); + if (sink->conn) { + struct nn_rpc_connection *conn = sink->conn; + sink->conn = NULL; // Signal to connection_closed_callback() we're done. + nn_rpc_conn_disconnect(conn); + } struct camu_sink_entry *entry; al_array_foreach_rev(sink->entries, i, entry) { al_array_remove_at(sink->entries, i); diff --git a/src/libsink/sink.h b/src/libsink/sink.h index 1e8bbf1..acb6a51 100644 --- a/src/libsink/sink.h +++ b/src/libsink/sink.h @@ -52,7 +52,6 @@ struct camu_sink_entry { bool ended; u32 reset_token; struct camu_clock clock; - bool held; // Only used in SINK_LOCAL. bool paused; struct { u8 state; @@ -83,7 +82,9 @@ struct camu_sink { u16 port; struct nn_rpc client; struct nn_rpc_connection *conn; + bool connected; u16 connection_number; + bool local; struct nn_mutex lock; struct nn_timer reconnect_timer; struct nn_signal queue_signal; |