Link - further work improving and optimising

* Reduce the amount of communication required between Spider and Tau for timing information (this enables much higher BPMs to work without issue)
* Add API plumbing for enabling/disabling link in addition to setting the BPM (this requires an update to sp_midi to work correctly)
* Handle changing BPM to and from :link mode with the same logic for both use_bpm and bpm_sync.
* Modify tau server API to receive API messages from other Erlang processes in addition to the current route via UDP/OSC (this should help in making it easier to implement a general way to handle scheduled OSC bundles).

Note: currently all time checks are disabled - this needs to be addressed and fixed before a release can be made.

Author: samaaron

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

@@ -106,91 +106,110 @@ init(Parent, CueServer, MIDIServer, LinkServer) ->
 
 loop(State) ->
     receive
+        {timeout, Timer, {call, Server, Msg, Tracker}} ->
+            Server ! Msg,
+            tau_server_tracker:forget(Timer, Tracker),
+            ?MODULE:loop(State);
+
         {udp, APISocket, Ip, Port, Bin} ->
             debug(3, "api server got UDP on ~p:~p~n", [Ip, Port]),
-            try osc:decode(Bin) of
-                {bundle, Time, X} ->
-                    debug("got bundle for time ~f~n", [Time]),
-                    NewState = do_bundle(Time, X, State),
-                    ?MODULE:loop(NewState);
+            case osc:decode(Bin) of
                 {cmd, ["/ping"]} ->
                     debug("sending! /pong to  ~p ~p ~n", [Ip, Port]),
                     PongBin = osc:encode(["/pong"]),
                     ok = gen_udp:send(APISocket, Ip, Port, PongBin),
                     ?MODULE:loop(State);
-                {cmd, ["/midi", OSC]=Cmd} ->
-                    debug_cmd(Cmd),
-                    MIDIServer = maps:get(midi_server, State),
-                    MIDIServer ! {send, OSC},
-                    ?MODULE:loop(State);
-                {cmd, ["/midi-flush"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    MIDIServer = maps:get(midi_server, State),
-                    MIDIServer ! {flush},
-                    ?MODULE:loop(State);
-                {cmd, ["/flush", Tag]=Cmd} ->
-                    debug_cmd(Cmd),
-                    {Tracker, NewState} = tracker_pid(Tag, State),
-                    tau_server_tracker:flush(all, Tracker),
-                    ?MODULE:loop(NewState);
-                {cmd, ["/internal-cue-port", Flag]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_cue({internal, Flag =:= 1}, State),
-                    ?MODULE:loop(State);
-                {cmd, ["/stop-start-cue-server", Flag]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_cue({enabled, Flag =:= 1}, State),
-                    ?MODULE:loop(State);
-                {cmd, ["/stop-start-midi-cues", Flag]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_cue({midi_enabled, Flag =:= 1}, State),
-                    ?MODULE:loop(State);
-                {cmd, ["/api-rpc", UUID, "/link-get-current-time"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_rpc, UUID, get_current_time}, State),
-                    ?MODULE:loop(State);
-
-                {cmd, ["/api-rpc", UUID, "/link-get-beat-at-time", Time, Quantum]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_rpc, UUID, get_beat_at_time, Time, Quantum}, State),
-                    ?MODULE:loop(State);
-
-                {cmd, ["/api-rpc", UUID, "/link-get-time-at-beat", Beat, Quantum]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_rpc, UUID, get_time_at_beat, Beat, Quantum}, State),
-                    ?MODULE:loop(State);
-
-                {cmd, ["/api-rpc", UUID, "/link-get-tempo"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_rpc, UUID, get_tempo}, State),
-                    ?MODULE:loop(State);
-
-                {cmd, ["/api-rpc", UUID, "/link-get-num-peers"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_rpc, UUID, get_num_peers}, State),
-                    ?MODULE:loop(State);
-
-
-                {cmd, ["/link-disable"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_disable}, State),
-                    ?MODULE:loop(State);
-
-                {cmd, ["/link-enable"]=Cmd} ->
-                    debug_cmd(Cmd),
-                    send_to_link({link_enable}, State),
-                    ?MODULE:loop(State);
-
+                Any -> self() ! Any
+            end,
+            ?MODULE:loop(State);
+
+        {bundle, Time, X} ->
+            debug("got bundle for time ~f~n", [Time]),
+            NewState = do_bundle(Time, X, State),
+            ?MODULE:loop(NewState);
+
+        {cmd, ["/midi", OSC]=Cmd} ->
+            debug_cmd(Cmd),
+            MIDIServer = maps:get(midi_server, State),
+            MIDIServer ! {send, OSC},
+            ?MODULE:loop(State);
+
+        {cmd, ["/midi-flush"]=Cmd} ->
+            debug_cmd(Cmd),
+            MIDIServer = maps:get(midi_server, State),
+            MIDIServer ! {flush},
+            ?MODULE:loop(State);
+
+        {cmd, ["/flush", Tag]=Cmd} ->
+            debug_cmd(Cmd),
+            {Tracker, NewState} = tracker_pid(Tag, State),
+            tau_server_tracker:flush(all, Tracker),
+            ?MODULE:loop(NewState);
+
+        {cmd, ["/internal-cue-port", Flag]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_cue({internal, Flag =:= 1}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/stop-start-cue-server", Flag]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_cue({enabled, Flag =:= 1}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/stop-start-midi-cues", Flag]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_cue({midi_enabled, Flag =:= 1}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-get-current-time"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, get_current_time}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-get-beat-at-time", Time, Quantum]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, get_beat_at_time, Time, Quantum}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-get-time-at-beat", Beat, Quantum]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, get_time_at_beat, Beat, Quantum}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-get-tempo"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, get_tempo}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-get-num-peers"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, get_num_peers}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/api-rpc", UUID, "/link-is-enabled"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_rpc, UUID, is_enabled}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/link-reset"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_reset}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/link-disable"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_disable}, State),
+            ?MODULE:loop(State);
+
+        {cmd, ["/link-enable"]=Cmd} ->
+            debug_cmd(Cmd),
+            send_to_link({link_enable}, State),
+            ?MODULE:loop(State);
+
+        {cmd, Cmd} ->
+            log("Unknown OSC command:: ~p~n", [Cmd]),
+            ?MODULE:loop(State);
 
-                {cmd, Cmd} ->
-                    log("Unknown command:: ~p~n", [Cmd]),
-                    ?MODULE:loop(State)
-            catch
-                Class:Term:Trace ->
-                    log("Error decoding OSC: ~p~n~p:~p~n~p~n",
-                        [Bin, Class, Term, Trace]),
-                    ?MODULE:loop(State)
-            end;
         {system, From, Request} ->
             %% handling system messages (like a gen_server does)
             sys:handle_system_msg(Request, From,
@@ -218,14 +237,18 @@ debug_cmd([Cmd|Args]) ->
 do_bundle(Time, [{_,Bin}|T], State) ->
     NewState =
         try osc:decode(Bin) of
-            {cmd, ["/send-after", Host, Port , OSC]} ->
-                schedule_cmd("default", Time, Host, Port, OSC, State);
+            {cmd, ["/send-after", Host, Port, OSC]} ->
+                schedule_cmd(Time, "default", State, {send_osc, Host, Port, OSC});
             {cmd, ["/send-after-tagged", Tag, Host, Port, OSC]} ->
-                schedule_cmd(Tag, Time, Host, Port, OSC, State);
-            {cmd, ["/midi-at", Cmd]} ->
-                schedule_midi("default", Time, Cmd, State);
-            {cmd, ["/midi-at-tagged", Tag, Cmd]} ->
-                schedule_midi(Tag, Time, Cmd, State);
+                schedule_cmd(Time, Tag, State, {send_osc, Host, Port, OSC});
+            {cmd, ["/midi-at", MIDI]} ->
+                schedule_midi(Time, "default", State, {send_midi, MIDI});
+            {cmd, ["/midi-at-tagged", Tag, MIDI]} ->
+                schedule_midi(Time, Tag, State, {send_midi, MIDI});
+            {cmd, ["/link-set-tempo", Tempo]} ->
+                schedule_link(Time, "default", State, {link_set_tempo, Tempo});
+            {cmd, ["/link-set-tempo-tagged", Tag, Tempo]} ->
+                schedule_link(Time, Tag, State, {link_set_tempo, Tempo});
             Other ->
                 log("Unexpected bundle content:~p~n", [Other]),
                 State
@@ -239,46 +262,37 @@ do_bundle(Time, [{_,Bin}|T], State) ->
 do_bundle(_Time, [], State) ->
     State.
 
-schedule_midi(Tag, Time, Data, State) ->
-   {Tracker, NewState} = tracker_pid(Tag, State),
+schedule_internal_call(Time, Tag, State, Server, Msg) ->
     Delay = Time - osc:now(),
     MsDelay = trunc(Delay*1000+0.5), %% nearest
-    MIDIServer = maps:get(midi_server, State),
+    {Tracker, NewState} = tracker_pid(Tag, State),
     if MsDelay > ?NODELAY_LIMIT ->
-            Msg = {send, Time, Data, Tracker},
+
             %% Note: lookup of the registered server name will happen
             %% when the timer triggers, and if no such process exists
             %% at that time, the message will be quietly dropped
-            Timer = erlang:start_timer(MsDelay, MIDIServer, Msg),
+            SchedMsg = {call, Server, Msg, Tracker},
+            Timer = erlang:start_timer(MsDelay, self(), SchedMsg),
             debug(2, "start (MIDI) timer of ~w ms for time ~f~n", [MsDelay, Time]),
             tau_server_tracker:track(Timer, Time, Tracker);
        true ->
-            MIDIServer ! {send, Time, Data},
-            debug(2, "directly forward (MIDI) message for delay ~f~n", [Delay])
+            Server ! Msg,
+            debug(2, "Directly sent scheduled call~n", [])
     end,
     NewState.
 
 
-%% Schedules a command for forwarding (or forwards immediately)
-schedule_cmd(Tag, Time, Host, Port, OSC, State) ->
-   {Tracker, NewState} = tracker_pid(Tag, State),
-    Data = {Host, Port, OSC},
-    Delay = Time - osc:now(),
-    MsDelay = trunc(Delay*1000+0.5), %% nearest
-    if MsDelay > ?NODELAY_LIMIT ->
-            Msg = {forward, Time, Data, Tracker},
-            %% Note: lookup of the registered server name will happen
-            %% when the timer triggers, and if no such process exists
-            %% at that time, the message will be quietly dropped
-            CueServer = maps:get(cue_server, State),
-            Timer = erlang:start_timer(MsDelay, CueServer, Msg),
-            debug(2, "start timer of ~w ms for time ~f~n", [MsDelay, Time]),
-            tau_server_tracker:track(Timer, Time, Tracker);
-       true ->
-            send_to_cue({forward, Time, Data}, NewState),
-            debug(2, "directly forward message for delay ~f~n", [Delay])
-    end,
-    NewState.
+schedule_link(Time, Tag, State, Msg) ->
+    LinkServer = maps:get(link_server, State),
+    schedule_internal_call(Time, Tag, State, LinkServer, Msg).
+
+schedule_midi(Time, Tag, State, Msg) ->
+    MIDIServer = maps:get(midi_server, State),
+    schedule_internal_call(Time, Tag, State, MIDIServer, Msg).
+
+schedule_cmd(Time, Tag, State, Msg) ->
+    CueServer = maps:get(cue_server, State),
+    schedule_internal_call(Time, Tag, State, CueServer, Msg).
 
 %% Get the pid for the tag group tracker, creating it if needed
 tracker_pid(Tag, State) ->

app/server/erlang/tau/src/tau_server/tau_server_cue.erl

@@ -236,13 +236,8 @@ loop(State) ->
             log("Disabling midi cue forwarding ~n"),
             ?MODULE:loop(State#{midi_enabled := false});
 
-        {timeout, Timer, {forward, Time, Data, Tracker}} ->
-            send_forward(maps:get(in_socket, State), Time, Data),
-            tau_server_tracker:forget(Timer, Tracker),
-            ?MODULE:loop(State);
-
-        {forward, Time, Data} ->
-            send_forward(maps:get(in_socket, State), Time, Data),
+        {send_osc, Host, Port, OSC} ->
+            send_udp(maps:get(in_socket, State), Host, Port, OSC),
             ?MODULE:loop(State);
 
         {udp_error, _Port, econnreset} ->
@@ -277,13 +272,6 @@ loop(State) ->
     end.
 
 
-send_forward(Socket, Time, {Host, Port, Bin}) ->
-    Now = osc:now(),
-    send_udp(Socket, Host, Port, Bin),
-    debug(1, Now, "sent message for time ~f with error ~f~n",
-          [Time, Now-Time]),
-    ok.
-
 send_udp(Socket, Host, Port, Bin)
   when is_port(Socket) ->
     %% check to see if host is correct and usable

app/server/erlang/tau/src/tau_server/tau_server_link.erl

@@ -88,6 +88,16 @@ loop(State) ->
             maps:get(cue_server, State) ! {link, stop},
             ?MODULE:loop(State);
 
+        {link_reset} ->
+            log("Resetting link~n", []),
+            case sp_link:is_enabled() of
+                true ->
+                    log("Link is currently enabled, now disabling then re-enabling it...", []),
+                    sp_link:enable(false),
+                    sp_link:enable(true);
+                _ -> ok
+            end,
+            ?MODULE:loop(State);
 
         {link_disable} ->
             log("Disabling link~n", []),
@@ -99,6 +109,11 @@ loop(State) ->
             sp_link:enable(true),
             ?MODULE:loop(State);
 
+        {link_set_tempo, Tempo} ->
+            TNow = sp_link:get_current_time_microseconds(),
+            sp_link:set_tempo(Tempo, TNow),
+            ?MODULE:loop(State);
+
         {link_rpc, UUID, get_current_time} ->
             Time = sp_link:get_current_time_microseconds(),
             log("Received link rpc current_time [~p]~n", [Time]),
@@ -129,15 +144,21 @@ loop(State) ->
             maps:get(cue_server, State) ! {api_reply, UUID, [NumPeers]},
             ?MODULE:loop(State);
 
+        {link_rpc, UUID, is_enabled} ->
+            Enabled = case sp_link:is_enabled() of
+                          true -> 1;
+                          false -> 0
+                      end,
+            log("Link is enabled:  [~p]~n", [Enabled]),
+            maps:get(cue_server, State) ! {api_reply, UUID, [Enabled]},
+            ?MODULE:loop(State);
+
         Any ->
             io:format("Received something unexpected", []),
             io:format("Unexpected value:  ~p~n", [Any]),
             ?MODULE:loop(State)
     end.
 
-
-
-
 %% sys module callbacks
 
 system_continue(_Parent, _Debug, State) ->

app/server/erlang/tau/src/tau_server/tau_server_midi.erl

@@ -85,12 +85,8 @@ mk_tau_str({tau, Kind, Event, Source, _}) ->
 
 loop(State) ->
     receive
-        {send, Time, Data} ->
-            midi_send(Time, Data),
-            ?MODULE:loop(State);
-        {timeout, Timer, {send, Time, Data, Tracker}} ->
-            midi_send(Time, Data),
-            tau_server_tracker:forget(Timer, Tracker),
+        {send_midi, Data} ->
+            midi_send(Data),
             ?MODULE:loop(State);
         {flush} ->
             sp_midi:midi_flush(),
@@ -143,7 +139,7 @@ update_midi_ports(State) ->
                    midi_outs := NewOuts}
     end.
 
-midi_send(_Time, <<Data/binary>>) ->
+midi_send(<<Data/binary>>) ->
     debug("sending MIDI: ~p~n", [Data]),
     case tau_server_midi_out:encode_midi_from_osc(Data) of
         {ok, multi_chan, _, PortName, MIDIBinaries} ->