Merge pull request #41 from Sendspin/fix/hook-serialisation

Run stream hooks one at a time, newest event winning the wait

Author: chrisuthe

README.md

@@ -1011,10 +1011,22 @@ the device refused is still audio arriving, so the amplifier is on for exactly a
 long as `status` says `stream: receiving`. Nothing waits on it — a hook that blocks
 cannot stall the audio path. It is reaped from the main loop; its output lands in
 the log (stdout deliberately re-pointed at stderr, since `-o stdout` may be
-carrying PCM); and a non-zero exit is a `W hook:` line, not a player failure. A
-hook still running at shutdown is left to finish: an amplifier half-switched is
+carrying PCM); and a non-zero exit is a `W hook:` line, not a player failure.
+
+Hooks run one at a time, in event order, so a start hook that runs long cannot
+finish after its own stream's stop hook and leave the amplifier on. While one
+runs, the newest event waits — and only the newest: a stop superseded by a start
+while an earlier hook is still running is skipped entirely (a `D hook:` line
+records it), because the hardware should end in the final state, not replay a
+stale one. A hook that *counts* events rather than setting state will see such
+flapping coalesced away. However a hook misbehaves, at most one runs and one
+waits — shutdown, below, is the one exception.
+
+A hook still running at shutdown is left to finish: an amplifier half-switched is
 worse than an orphan. A player stopped while a stream is playing runs its stop
-hook on the way out, so `systemctl stop` leaves the amplifier off rather than on.
+hook on the way out — beside a start hook that has still not finished, if it comes
+to that (a `W hook:` line says so) — so `systemctl stop` leaves the amplifier off
+rather than on.
 
 The hook is handed nothing of the player's but that output stream: every other
 descriptor is closed and SIGPIPE is back at its default, so `something | head -1`

docs/wiki/Controlling-the-Player.md

@@ -217,7 +217,14 @@ tell from its own — the library reports that one is up, not where it came from
 
 The hook never blocks playback: it is spawned and reaped from the main loop, its output
 goes to the log, and a non-zero exit is a `W hook:` warning rather than a player failure.
-Stopping the player while a stream is playing runs the stop hook before it exits, so
+Hooks run one at a time, in event order — a start hook that runs long cannot finish after
+its own stream's stop hook and leave the amplifier on. While one runs, only the newest
+event waits: a stop superseded by a start while an earlier hook is still running is
+skipped entirely, since the hardware should end in the final state rather than replay a
+stale one — a hook that *counts* events rather than setting state will see such flapping
+coalesced away, and each skipped event is a `D hook:` line in the log.
+Stopping the player while a stream is playing runs the stop hook before it
+exits — beside a start hook that has still not finished, if it comes to that — so
 `systemctl stop` switches the amplifier off rather than leaving it on. The hook is handed
 nothing of the player's but that output stream — every other descriptor is closed and
 SIGPIPE is back at its default — so `something | head -1` behaves as it would in any other

src/hooks.cpp

@@ -93,6 +93,85 @@ std::vector<std::string> hook_environment(const char* event, const HookContext&
 }  // namespace
 
 void HookRunner::run(const std::string& command, const char* event, const HookContext& context) {
+    if (!this->running_.empty()) {
+        // One at a time: spawned side by side, two events race in the scheduler, and a start
+        // hook finishing after its own stream's stop hook leaves the amplifier on with the
+        // player idle. The newest event wins the slot -- the hardware should end in the
+        // final state, not replay a stale intermediate on the way there.
+        if (this->pending_.has_value()) {
+            cli_log(LogLevel::DEBUG, "The queued %s hook was superseded by the %s event",
+                    this->pending_->event.c_str(), event);
+        }
+        this->pending_ = PendingHook{command, event, context};
+        return;
+    }
+    this->spawn(command, event, context);
+}
+
+void HookRunner::poll() {
+    for (size_t index = 0; index < this->running_.size();) {
+        const RunningHook& hook = this->running_[index];
+        int status = 0;
+        const pid_t reaped = waitpid(hook.pid, &status, WNOHANG);
+        if (reaped == 0) {
+            ++index;
+            continue;
+        }
+        if (reaped < 0 && errno == EINTR) {
+            // The signal landed on the call, not on the child: the hook is still running, and
+            // the next poll() asks again. Every handler this daemon installs restarts its call
+            // -- std::signal()'s BSD semantics for SIGINT and SIGTERM, SA_RESTART for SIGHUP --
+            // so this is a guard rather than a path taken. Erasing here instead would leak a
+            // zombie per stream, with nothing in the log to say why.
+            ++index;
+            continue;
+        }
+        if (reaped < 0) {
+            // ECHILD is the one that reaches this, and a child that cannot be waited on can
+            // only be leaked, not re-polled -- so the entry goes. Said out loud because it
+            // means something else reaped the hook, which is worth a breadcrumb; DEBUG because
+            // the hook itself ran and there is nothing an operator can do about it.
+            cli_log(LogLevel::DEBUG, "The %s hook [%d] could not be waited on (%s)",
+                    hook.event.c_str(), static_cast<int>(hook.pid), std::strerror(errno));
+        }
+        if (reaped == hook.pid && WIFEXITED(status) && WEXITSTATUS(status) != 0) {
+            cli_log(LogLevel::WARN, "The %s hook [%d] exited %d", hook.event.c_str(),
+                    static_cast<int>(hook.pid), WEXITSTATUS(status));
+        } else if (reaped == hook.pid && WIFSIGNALED(status)) {
+            cli_log(LogLevel::WARN, "The %s hook [%d] was killed by signal %d",
+                    hook.event.c_str(), static_cast<int>(hook.pid), WTERMSIG(status));
+        } else if (reaped == hook.pid) {
+            cli_log(LogLevel::DEBUG, "The %s hook [%d] finished", hook.event.c_str(),
+                    static_cast<int>(hook.pid));
+        }
+        this->running_.erase(this->running_.begin() + static_cast<ptrdiff_t>(index));
+    }
+
+    if (this->running_.empty() && this->pending_.has_value()) {
+        // Its turn: everything ahead of it has been reaped. Taken out of the slot before the
+        // spawn, which fills running_ again.
+        const PendingHook next = *this->pending_;
+        this->pending_.reset();
+        this->spawn(next.command, next.event.c_str(), next.context);
+    }
+}
+
+void HookRunner::flush() {
+    if (!this->pending_.has_value()) {
+        return;
+    }
+    if (!this->running_.empty()) {
+        // Ordering is knowingly given up, and said at WARN because it means a hook has hung:
+        // the pending hook -- the stop the daemon promised on the way out -- must run at all.
+        cli_log(LogLevel::WARN, "The %s hook has not finished -- running the %s hook beside it",
+                this->running_.front().event.c_str(), this->pending_->event.c_str());
+    }
+    const PendingHook next = *this->pending_;
+    this->pending_.reset();
+    this->spawn(next.command, next.event.c_str(), next.context);
+}
+
+void HookRunner::spawn(const std::string& command, const char* event, const HookContext& context) {
     // Built before the fork, because after it almost nothing is legal: this process has the
     // library's background threads, so the child may only call async-signal-safe functions
     // between fork() and execve() -- which allocation, and so std::string, is not.
@@ -157,44 +236,4 @@ void HookRunner::run(const std::string& command, const char* event, const HookCo
     this->running_.push_back({pid, event});
 }
 
-void HookRunner::poll() {
-    for (size_t index = 0; index < this->running_.size();) {
-        const RunningHook& hook = this->running_[index];
-        int status = 0;
-        const pid_t reaped = waitpid(hook.pid, &status, WNOHANG);
-        if (reaped == 0) {
-            ++index;
-            continue;
-        }
-        if (reaped < 0 && errno == EINTR) {
-            // The signal landed on the call, not on the child: the hook is still running, and
-            // the next poll() asks again. Every handler this daemon installs restarts its call
-            // -- std::signal()'s BSD semantics for SIGINT and SIGTERM, SA_RESTART for SIGHUP --
-            // so this is a guard rather than a path taken. Erasing here instead would leak a
-            // zombie per stream, with nothing in the log to say why.
-            ++index;
-            continue;
-        }
-        if (reaped < 0) {
-            // ECHILD is the one that reaches this, and a child that cannot be waited on can
-            // only be leaked, not re-polled -- so the entry goes. Said out loud because it
-            // means something else reaped the hook, which is worth a breadcrumb; DEBUG because
-            // the hook itself ran and there is nothing an operator can do about it.
-            cli_log(LogLevel::DEBUG, "The %s hook [%d] could not be waited on (%s)",
-                    hook.event.c_str(), static_cast<int>(hook.pid), std::strerror(errno));
-        }
-        if (reaped == hook.pid && WIFEXITED(status) && WEXITSTATUS(status) != 0) {
-            cli_log(LogLevel::WARN, "The %s hook [%d] exited %d", hook.event.c_str(),
-                    static_cast<int>(hook.pid), WEXITSTATUS(status));
-        } else if (reaped == hook.pid && WIFSIGNALED(status)) {
-            cli_log(LogLevel::WARN, "The %s hook [%d] was killed by signal %d",
-                    hook.event.c_str(), static_cast<int>(hook.pid), WTERMSIG(status));
-        } else if (reaped == hook.pid) {
-            cli_log(LogLevel::DEBUG, "The %s hook [%d] finished", hook.event.c_str(),
-                    static_cast<int>(hook.pid));
-        }
-        this->running_.erase(this->running_.begin() + static_cast<ptrdiff_t>(index));
-    }
-}
-
 }  // namespace sendspin_cli

src/hooks.h

@@ -19,6 +19,7 @@
 
 #include <sys/types.h>
 
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -48,8 +49,18 @@ struct HookContext {
 ///
 /// A hook is `/bin/sh -c <command>` with `SENDSPIN_EVENT` and the context above added to
 /// this process's own environment -- a shell rather than an argv split, so `amixer set
-/// Master unmute && relay on` is one hook. The spawn returns as soon as the child is
-/// forked: a hook that blocks must not stall the audio path, so nothing here waits.
+/// Master unmute && relay on` is one hook. run() returns without waiting either way: a
+/// hook that blocks must not stall the audio path.
+///
+/// Hooks run one at a time, in event order. Two events for the same stream would otherwise
+/// race in the scheduler, and a start hook that runs long finishing *after* the stop hook
+/// of its own stream leaves the amplifier on with the player idle -- the exact state the
+/// hooks exist to prevent. While one runs, the newest event waits in a single pending
+/// slot; a newer event replaces whatever waits there, because the hardware should end in
+/// the *final* state, not replay a stale intermediate. A hook that counts events rather
+/// than setting state will see such flapping coalesced away -- each replacement is a
+/// `D hook:` line. The slot is also the bound: one event waiting at most, and one child at
+/// most until the shutdown flush(), no matter how a hook misbehaves.
 ///
 /// The child's stdout and stderr both go where the player's stderr goes -- under -f, the
 /// logfile -- so whatever a hook prints lands beside the player's own lines. Its stdout is
@@ -61,24 +72,38 @@ struct HookContext {
 /// holding them keeps a restart from binding its port -- and SIGPIPE goes back to its default,
 /// which the player ignores and an exec would otherwise carry through.
 ///
-/// THREAD SAFETY: run() and poll() must both be called on the main loop thread. That is
-/// where the stream callbacks that trigger hooks already fire, and it is what lets the
-/// bookkeeping below go unsynchronised.
+/// THREAD SAFETY: run(), poll() and flush() must all be called on the main loop thread.
+/// That is where the stream callbacks that trigger hooks already fire and where the
+/// shutdown path runs, and it is what lets the bookkeeping below go unsynchronised.
 class HookRunner {
 public:
-    /// @brief Runs `command` with `SENDSPIN_EVENT=<event>` and `context` in its environment.
+    /// @brief Runs `command` with `SENDSPIN_EVENT=<event>` and `context` in its environment,
+    /// or holds it in the pending slot while an earlier hook is still running.
     ///
     /// Failure to spawn is a WARN, not an error: the stream the event describes is fine,
     /// and the player must keep playing it.
     /// @param event What SENDSPIN_EVENT carries: "start" or "stop".
+    /// @param context Copied when the event has to wait: it describes this event's stream,
+    /// and the caller's object will already describe the next one by the time the slot is
+    /// spawned.
     void run(const std::string& command, const char* event, const HookContext& context);
 
-    /// @brief Reaps any hooks that have finished, logging the ones that failed.
+    /// @brief Reaps any hooks that have finished, logging the ones that failed, and spawns
+    /// the pending event once the running hook is out.
     ///
     /// Call from the main loop. A hook still running when the daemon exits is left to
     /// finish on its own -- an amplifier half-switched-off is worse than an orphan.
     void poll();
 
+    /// @brief Spawns the pending event now, beside the running hook if there still is one.
+    ///
+    /// For the shutdown path, after the stream-end drain: the daemon promises that
+    /// stopping it runs the stop hook, and a start hook that never finishes must not be
+    /// allowed to turn that promise into an amplifier left on. Ordering is knowingly given
+    /// up here -- there is no later event left to order against, and the alternative is
+    /// the hook never running at all.
+    void flush();
+
     /// How many hooks have been spawned and not yet reaped. For tests, which need to know
     /// when poll() has seen a child out, and for nothing else.
     size_t running() const {
@@ -92,7 +117,18 @@ class HookRunner {
         std::string event;
     };
 
+    /// The event waiting for the running hook to finish, newest wins.
+    struct PendingHook {
+        std::string command;
+        std::string event;
+        HookContext context;
+    };
+
+    /// Forks and execs one hook, unconditionally. run() decides whether now is the time.
+    void spawn(const std::string& command, const char* event, const HookContext& context);
+
     std::vector<RunningHook> running_;
+    std::optional<PendingHook> pending_;
 };
 
 }  // namespace sendspin_cli

src/main.cpp

@@ -1015,6 +1015,10 @@ int main(int argc, char* argv[]) {
         hooks.poll();
         std::this_thread::sleep_for(std::chrono::milliseconds(LOOP_INTERVAL_MS));
     }
+    // The stop hook may be sitting in the pending slot behind a start hook that never
+    // finished, and no more polls are coming. Spawned now regardless: this is the last
+    // chance to keep the promise that stopping the player switches the amplifier off.
+    hooks.flush();
     // The lambda holds references to locals declared after the listener, so it outlives them by
     // exactly the width of this scope's teardown. Nothing calls it there -- stream events only
     // arrive inside client.loop(), and the last one has run -- and dropping it here is what

tests/hooks_test.cpp

@@ -24,7 +24,6 @@
 #include <gtest/gtest.h>
 #include <unistd.h>
 
-#include <algorithm>
 #include <chrono>
 #include <csignal>
 #include <cstdio>
@@ -225,18 +224,99 @@ TEST(HookRunner, AFailingHookIsReapedRatherThanLeaked) {
     EXPECT_EQ(runner.running(), 0U);
 }
 
-TEST(HookRunner, SeveralHooksInFlightAreEachReaped) {
+// ---------------------------------------------------------------------------
+// One at a time, newest event wins the wait
+// ---------------------------------------------------------------------------
+
+/// A hook command that spins until `gate` exists, then runs `then`.
+///
+/// What every ordering test below hangs off: the gated hook is deterministically still
+/// running until the test opens the gate, with no scheduler timing assumed anywhere.
+std::string gated(const std::string& gate, const std::string& then) {
+    return "while [ ! -e " + gate + " ]; do sleep 0.01; done; " + then;
+}
+
+void open_gate(const std::string& path) {
+    const std::ofstream gate(path);
+}
+
+TEST(HookRunner, ASecondEventWaitsForTheRunningHook) {
+    ScratchFile out;
+    ScratchFile gate;
+    HookRunner runner;
+    // The start hook cannot write until the gate exists, so spawned side by side the stop
+    // hook's 'b' would deterministically land first. The order below is the contract.
+    runner.run(gated(gate.path(), "printf 'a' >> " + out.path()), "start", HookContext{});
+    runner.run("printf 'b' >> " + out.path(), "stop", HookContext{});
+
+    EXPECT_EQ(runner.running(), 1U);
+    open_gate(gate.path());
+    ASSERT_TRUE(drain(runner));
+    EXPECT_EQ(slurp(out.path()), "ab");
+}
+
+TEST(HookRunner, TheNewestEventReplacesTheWaitingOne) {
     ScratchFile out;
+    ScratchFile gate;
     HookRunner runner;
-    runner.run("printf 'a' >> " + out.path(), "start", HookContext{});
+    runner.run(gated(gate.path(), "printf 'a' >> " + out.path()), "start", HookContext{});
     runner.run("printf 'b' >> " + out.path(), "stop", HookContext{});
     runner.run("printf 'c' >> " + out.path(), "start", HookContext{});
 
+    open_gate(gate.path());
+    ASSERT_TRUE(drain(runner));
+    // The stop was superseded while it waited: the hardware ends in the final state, not
+    // replaying the intermediate on the way there.
+    EXPECT_EQ(slurp(out.path()), "ac");
+}
+
+TEST(HookRunner, AWaitingEventKeepsItsOwnContext) {
+    ScratchFile out;
+    ScratchFile gate;
+    HookRunner runner;
+    runner.run(gated(gate.path(), "true"), "start", HookContext{});
+
+    HookContext context;
+    context.server_id = "srv-b";
+    runner.run("printf '%s' \"$SENDSPIN_SERVER_ID\" > " + out.path(), "stop", context);
+    // What the caller does to its object between events must not reach into the slot: the
+    // waiting event describes the stream it was fired for.
+    context.server_id = "srv-c";
+
+    open_gate(gate.path());
     ASSERT_TRUE(drain(runner));
-    // Order is the scheduler's, so only the multiset of writes is promised.
-    std::string content = slurp(out.path());
-    std::sort(content.begin(), content.end());
-    EXPECT_EQ(content, "abc");
+    EXPECT_EQ(slurp(out.path()), "srv-b");
+}
+
+TEST(HookRunner, FlushRunsTheWaitingHookBesideAHungOne) {
+    ScratchFile out;
+    ScratchFile gate;
+    HookRunner runner;
+    runner.run(gated(gate.path(), "true"), "start", HookContext{});
+    runner.run("printf 'b' >> " + out.path(), "stop", HookContext{});
+
+    // What the shutdown path does when the drain ends: two children out at once,
+    // deliberately -- the promise that stopping the player runs the stop hook outranks
+    // ordering when no more events can come.
+    runner.flush();
+    EXPECT_EQ(runner.running(), 2U);
+
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
+    while (slurp(out.path()) != "b" && std::chrono::steady_clock::now() < deadline) {
+        runner.poll();
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+    EXPECT_EQ(slurp(out.path()), "b");
+
+    open_gate(gate.path());
+    EXPECT_TRUE(drain(runner));
+}
+
+TEST(HookRunner, FlushWithNothingWaitingDoesNothing) {
+    HookRunner runner;
+    runner.flush();
+
+    EXPECT_EQ(runner.running(), 0U);
 }
 
 TEST(HookRunner, DoesNotHandTheHookThePlayersIgnoredSIGPIPE) {