Tau - add killswitch

Now that we have changed Tau's boot scripts to use the release scripts from Elixir, we no longer can get the OS pid from the spawn process. The pid is necessary so that we can explicitly kill Tau when we shutdown or something goes wrong. We attempted to circumnavigate this by explicitly passign the pid from Tau to Daemon via a TCP socket. However, there are still some edge cases where this pid isn't correctly sent/received. To mitigate this and to add extra reassurances that Tau will properly exit and not turn into a zombie process we now have an internal kill switch within Tau. This is a simple timer which when fired will bring the whole process down. The timer is reset by the Daemon sending an appropriate keep alive OSC message. If no keepalive message is received within 5 seconds Tau will shut itself down.

Author: samaaron

app/server/beam/tau/src/tau_server/tau_keepalive.erl

@@ -0,0 +1,66 @@
+%% Sonic Pi API server process
+%% --
+%% This file is part of Sonic Pi: http://sonic-pi.net
+%% Full project source: https://github.com/sonic-pi-net/sonic-pi
+%% License: https://github.com/sonic-pi-net/sonic-pi/blob/main/LICENSE.md
+%%
+%% Copyright 2021 by Sam Aaron (http://sam.aaron.name/)
+%% All rights reserved.
+%%
+%% Permission is granted for use, copying, modification, and
+%% distribution of modified versions of this work as long as this
+%% notice is included.
+%% ++
+
+-module(tau_keepalive).
+
+-export([start/1, init/1, loop/1]).
+
+-import(tau_server_util,
+        [log/1, log/2, debug/2, debug/3]).
+
+start(DaemonPortNum) ->
+    spawn(?MODULE, init, [DaemonPortNum]).
+
+init(DaemonPortNum) ->
+    log("connecting to Daemon via TCP...~n", []),
+    {ok, DaemonSocket} = gen_tcp:connect({127,0,0,1}, DaemonPortNum, [
+                                                                      binary,
+                                                                      {active, true},
+                                                                      {packet, 4},
+                                                                      {keepalive, true}
+                                                                     ]),
+    OSPid = os:getpid(),
+    PidMsg = osc:encode(["/tau_pid", OSPid]),
+    log("Sending Pid ~p to Daemon...~n", [OSPid]),
+    gen_tcp:send(DaemonSocket, PidMsg),
+    KillSwitch = erlang:start_timer(5000, self(), trigger_kill_switch),
+    log("Waiting for keepalive messages..."),
+    loop(KillSwitch).
+
+loop(KillSwitch) ->
+    log("KillSwitch loop ~n", []),
+    receive
+        {tcp, _Socket, Bin} ->
+            try osc:decode(Bin) of
+                {cmd, ["/system/keepalive"]} ->
+                    log("Received keepalive message from Daemon ~n", []),
+                    erlang:cancel_timer(KillSwitch),
+                    NewKillSwitch = erlang:start_timer(5000, self(), trigger_kill_switch),
+                    ?MODULE:loop(NewKillSwitch);
+                Other ->
+                    log("Unexpected message from Daemon:~p~n", [Other]),
+                    ?MODULE:loop(KillSwitch)
+            catch
+                Class:Term:Trace ->
+                    log("keepalive process: Error decoding OSC: ~p~n~p:~p~n~p~n",
+                        [Bin, Class, Term, Trace]),
+                    ?MODULE:loop(KillSwitch)
+            end;
+        {timeout, _Timer, trigger_kill_switch} ->
+            log("Tau kill switch activated. Shutting down....", []),
+            init:stop();
+        Any ->
+            log("Tau keepalive received unexpected message: ~p~n", [Any]),
+            ?MODULE:loop(KillSwitch)
+    end.

app/server/beam/tau/src/tau_server/tau_server_api.erl

@@ -90,18 +90,9 @@ init(Parent, CueServer, MIDIServer, LinkServer) ->
               [erlang:system_info(otp_release), APIPort]),
 
     {ok, APISocket} = gen_udp:open(APIPort, [binary, {ip, loopback}]),
-    io:format("connecting to Daemon via TCP...~n", []),
-    {ok, DaemonSocket} = gen_tcp:connect({127,0,0,1}, DaemonPort, [
-                                                                   binary,
-                                                                   {active, true},
-                                                                   {packet, 4},
-                                                                   {keepalive, true}
-                                                                  ]),
-
-    PidMsg = osc:encode(["/tau_pid", os:getpid()]),
-    io:format("Sending Pid to Daemon...~n", []),
-
-    gen_tcp:send(DaemonSocket, PidMsg),
+
+    _KeepAlivePid = tau_keepalive:start(DaemonPort),
+
     %% tell parent we have allocated resources and are up and running
     proc_lib:init_ack(Parent, {ok, self()}),
 

app/server/ruby/bin/daemon.rb

@@ -548,8 +548,11 @@ def initialize(ports)
         spider_port  = ports["listen-to-tau"]
         daemon_port  = ports["daemon-listen-to-tau"]
 
+        @osc_out_queue = SizedQueue.new(20)
+
         tau_comms    = TCPServer.new "127.0.0.1", daemon_port
         osc_decoder  = SonicPi::OSC::OscDecode.new
+        osc_encoder  = SonicPi::OSC::OscEncode.new
         comms_thread_started = Promise.new
 
         Thread.new do
@@ -561,6 +564,37 @@ def initialize(ports)
             client = tau_comms.accept    # Wait for a client to connect
             Util.log "----->   Connection accepted"
 
+            @tau_send_thread = Thread.new do
+              loop do
+                begin
+                  osc = @osc_out_queue.pop
+                  client.write([osc.bytesize].pack('N'))
+                  client.write(osc)
+                rescue StandardError => e
+                  Util.log "Critical Error, sending messages to Tau:"
+                  Util.log "Error Class: #{e.class}"
+                  Util.log "Error Message: #{e.message}"
+                  Util.log "Error Backtrace: #{e.backtrace.inspect}"
+                end
+              end
+            end
+
+            @tau_keep_alive_thread = Thread.new do
+              loop do
+                begin
+                  osc = osc_encoder.encode_single_message("/system/keepalive")
+                  @osc_out_queue << osc
+                rescue StandardError => e
+                  Util.log "Critical Error, encoding OSC keepalive message to Tau:"
+                  Util.log "Error Class: #{e.class}"
+                  Util.log "Error Message: #{e.message}"
+                  Util.log "Error Backtrace: #{e.backtrace.inspect}"
+                ensure
+                  Kernel.sleep 4
+                end
+              end
+            end
+
             recv_osc = lambda do
               size_str = client.recvfrom(4, Socket::MSG_WAITALL)[0].chomp
               size = size_str.unpack('N')[0]
@@ -569,7 +603,6 @@ def initialize(ports)
               osc_decoder.decode_single_message(data_raw)
             end
 
-
             begin
               data = recv_osc.call
               tau_pid = Integer(data[1][0])
@@ -624,6 +657,8 @@ def process_running?
       end
 
       def kill
+        @tau_send_thread.kill
+        @tau_keep_alive_thread.kill
         begin
           @pid = @tau_pid.get(30)
         rescue SonicPi::PromiseTimeoutError