New streaming model

.github/opam/liquidsoap-core-windows.opam

@@ -24,6 +24,7 @@ depends: [
   "mm-windows" {>= "0.8.4"}
   "re-windows" {>= "1.11.0"}
   "cry-windows" {>= "1.0.1"}
+  "saturn_lockfree-windows" {>= "0.4.1"}
   "sedlex" {>= "3.2"}
   "sedlex-windows" {>= "3.2"}
   "menhir"

.github/scripts/build-posix.sh

@@ -31,6 +31,9 @@ cd /tmp/liquidsoap-full/liquidsoap
 
 ./.github/scripts/checkout-deps.sh
 
+opam update
+opam install -y saturn_lockfree
+
 cd /tmp/liquidsoap-full
 
 sed -e 's@ocaml-gstreamer@#ocaml-gstreamer@' -i PACKAGES

doc/content/liq/radiopi.liq

@@ -159,7 +159,6 @@ def crossfade(
   ~medium=-32.,
   ~margin=4.,
   ~width=2.,
-  ~conservative=false,
   s
 ) =
   fade_out = fun (s) -> fade.out(type="sin", duration=fade_out, s)
@@ -280,9 +279,7 @@ def crossfade(
     end
   end
 
-  cross(
-    width=width, duration=start_next, conservative=conservative, transition, s
-  )
+  cross(width=width, duration=start_next, transition, s)
 end
 
 # Create a radiopilote-driven source
@@ -311,8 +308,7 @@ def channel_radiopilote(~skip=true, name) =
     request.create(ret)
   end
 
-  # Create the dynamic source. Set conservative to true to queue several songs
-  # in advance.
+  # Create the dynamic source.
   s = request.dynamic(id="dyn_" ^ name, req, timeout=60.)
 
   # Apply normalization using replaygain information.

dune-project

@@ -55,6 +55,7 @@
     uri
     fileutils
     menhirLib
+    (saturn_lockfree (>= 0.4.1))
     (metadata (>= 0.2.0))
     dune-build-info
     (liquidsoap-lang (= :version))

liquidsoap-core.opam

@@ -20,6 +20,7 @@ depends: [
   "uri"
   "fileutils"
   "menhirLib"
+  "saturn_lockfree" {>= "0.4.1"}
   "metadata" {>= "0.2.0"}
   "dune-build-info"
   "liquidsoap-lang" {= version}

src/core/builtins/builtins_ffmpeg_bitstream_filters.ml

@@ -308,7 +308,7 @@ let register_filters () =
                          (fun pos -> Generator.add_track_mark ~pos generator)
                          (List.filter
                             (fun x -> x < Lazy.force Frame.size)
-                            (Frame.breaks frame));
+                            (Frame.track_marks frame));
                        process frame
                    | `Flush -> flush ()
                  in

src/core/builtins/builtins_ffmpeg_decoder.ml

@@ -460,7 +460,7 @@ let mk_decoder mode =
                    (Frame.get_all_metadata frame);
                  List.iter
                    (fun pos -> Generator.add_track_mark ~pos generator)
-                   (List.filter (fun x -> x < size) (Frame.breaks frame));
+                   (List.filter (fun x -> x < size) (Frame.track_marks frame));
                  decode_frame (`Frame frame)
              | `Flush -> decode_frame `Flush
            in

src/core/builtins/builtins_ffmpeg_encoder.ml

@@ -512,7 +512,7 @@ let mk_encoder mode =
                    (Frame.get_all_metadata frame);
                  List.iter
                    (fun pos -> Generator.add_track_mark ~pos generator)
-                   (List.filter (fun x -> x < size) (Frame.breaks frame));
+                   (List.filter (fun x -> x < size) (Frame.track_marks frame));
                  encode_frame (`Frame frame)
              | `Flush -> encode_frame `Flush
            in

src/core/builtins/builtins_ffmpeg_filters.ml

@@ -116,7 +116,7 @@ let is_ready graph =
     | _ -> ());
   (not graph.failed)
   && Queue.fold
-       (fun cur (s : Source.source) -> cur && s#is_ready ())
+       (fun cur (s : Source.source) -> cur && s#is_ready)
        true graph.graph_inputs
 
 let pull graph =

src/core/builtins/builtins_prometheus.ml

@@ -170,12 +170,12 @@ let source_monitor ~prefix ~label_names ~labels ~window s =
     ()
   in
   let sleep () = () in
-  let get_frame ~start_time ~end_time ~start_position ~end_position
-      ~is_partial:_ ~metadata:_ =
+  let generate_frame ~start_time ~end_time ~length ~has_track_mark:_ ~metadata:_
+      =
     last_start_time := start_time;
     last_end_time := end_time;
     Prometheus.Gauge.set last_data end_time;
-    let encoded_time = Frame.seconds_of_main (end_position - start_position) in
+    let encoded_time = Frame.seconds_of_main length in
     let latency = (end_time -. start_time) /. encoded_time in
     add_input_latency latency
   in
@@ -188,7 +188,7 @@ let source_monitor ~prefix ~label_names ~labels ~window s =
     {
       Source.wake_up;
       sleep;
-      get_frame;
+      generate_frame;
       before_output = (fun _ -> ());
       after_output;
     }

src/core/builtins/builtins_source.ml

@@ -87,7 +87,7 @@ let _ =
        available), or currently streaming."
     [("", Lang.source_t (Lang.univ_t ()), None, None)]
     Lang.bool_t
-    (fun p -> Lang.bool ((Lang.to_source (List.assoc "" p))#is_ready ()))
+    (fun p -> Lang.bool (Lang.to_source (List.assoc "" p))#is_ready)
 
 let _ =
   Lang.add_builtin ~base:source "is_up" ~category:`System
@@ -138,8 +138,7 @@ let _ =
         else 0
       in
       let frame_position = Lazy.force Frame.duration *. float ticks in
-      let in_frame_position = Frame.seconds_of_main (Frame.position s#memo) in
-      Lang.float (frame_position +. in_frame_position))
+      Lang.float frame_position)
 
 let _ =
   Lang.add_builtin ~base:source "on_shutdown" ~category:(`Source `Liquidsoap)
@@ -200,9 +199,19 @@ let _ =
     (fun p ->
       let module Time = (val Clock.time_implementation () : Liq_time.T) in
       let open Time in
+      let stopped = ref false in
       let proto =
         let p = Pipe_output.file_proto (Lang.univ_t ()) in
-        List.filter_map (fun (l, _, v, _) -> Option.map (fun v -> (l, v)) v) p
+        List.filter_map
+          (fun (l, _, v, _) ->
+            if l <> "on_stop" then Option.map (fun v -> (l, v)) v
+            else
+              Some
+                ( "on_stop",
+                  Lang.val_fun [] (fun _ ->
+                      stopped := true;
+                      Lang.unit) ))
+          p
       in
       let proto = ("fallible", Lang.bool true) :: proto in
       let s = Lang.to_source (Lang.assoc "" 3 p) in
@@ -214,7 +223,7 @@ let _ =
       Clock.unify ~pos:fo#pos fo#clock (Clock.create_known clock);
       ignore (clock#start_outputs (fun _ -> true) ());
       log#info "Start dumping source (ratio: %.02fx)" ratio;
-      while (not (Atomic.get should_stop)) && fo#is_ready () do
+      while (not (Atomic.get should_stop)) && not !stopped do
         let start_time = Time.time () in
         clock#end_tick;
         sleep_until (start_time |+| latency)
@@ -242,11 +251,12 @@ let _ =
       let module Time = (val Clock.time_implementation () : Liq_time.T) in
       let open Time in
       let s = List.assoc "" p |> Lang.to_source in
+      let stopped = ref false in
       let o =
         new Output.dummy
           ~infallible:false
           ~on_start:(fun () -> ())
-          ~on_stop:(fun () -> ())
+          ~on_stop:(fun () -> stopped := true)
           ~register_telnet:false ~autostart:true (Lang.source s)
       in
       let ratio = Lang.to_float (List.assoc "ratio" p) in
@@ -255,7 +265,7 @@ let _ =
       Clock.unify ~pos:o#pos o#clock (Clock.create_known clock);
       ignore (clock#start_outputs (fun _ -> true) ());
       log#info "Start dropping source (ratio: %.02fx)" ratio;
-      while (not (Atomic.get should_stop)) && o#is_ready () do
+      while (not (Atomic.get should_stop)) && not !stopped do
         let start_time = Time.time () in
         clock#end_tick;
         sleep_until (start_time |+| latency)

src/core/clock.ml

@@ -209,7 +209,7 @@ module MkClock (Time : Liq_time.T) = struct
             | `Auto ->
                 List.exists
                   (fun (state, (s : Source.active_source)) ->
-                    state = `Active && snd s#self_sync && s#is_ready ())
+                    state = `Active && snd s#self_sync && s#is_ready)
                   outputs
             | `CPU -> false
             | `None -> true

src/core/conversions/conversion.ml

@@ -20,47 +20,13 @@
 
  *****************************************************************************)
 
-(* Conversion base class. contents marked as [true] are modified
- * by the converter using this class. Other contents are untouched. *)
-class base ?(audio = false) ?(video = false) ?(midi = false) ~converter
-  (source : Source.source) =
-  object (self)
+class base ~converter (source : Source.source) =
+  object
     method stype = source#stype
-    method private _is_ready = source#is_ready
+    method private can_generate_frame = source#is_ready
     method abort_track = source#abort_track
     method remaining = source#remaining
     method seek_source = source#seek_source
     method self_sync = source#self_sync
-    val mutable tmp_frame = None
-
-    (* The tmp_frame is intended to be filled by the underlying source. Content
-       untouched by the converter are replaced by by content from the calling
-       frame. Touched content get their own layer. *)
-    method private tmp_frame =
-      match tmp_frame with
-        | Some tmp_frame -> tmp_frame
-        | None ->
-            (* We need to delay the creation of the frame in order not to evaluate
-               the kind too early (the user has to have a chance of setting the
-               default number of channels). *)
-            let frame = Frame.create source#content_type in
-            tmp_frame <- Some frame;
-            frame
-
-    method private copy_frame src dst =
-      Frame.set_breaks dst (Frame.breaks src);
-      Frame.set_all_metadata dst (Frame.get_all_metadata src);
-      if not audio then (
-        try Frame.set_audio dst (Frame.audio src) with Not_found -> ());
-      if not video then (
-        try Frame.set_video dst (Frame.video src) with Not_found -> ());
-      if not midi then (
-        try Frame.set_video dst (Frame.midi src) with Not_found -> ())
-
-    method private get_frame frame =
-      let tmp_frame = self#tmp_frame in
-      self#copy_frame frame tmp_frame;
-      source#get tmp_frame;
-      converter ~frame tmp_frame;
-      self#copy_frame tmp_frame frame
+    method private generate_frame = converter source#get_frame
   end

src/core/conversions/mean.ml

@@ -21,31 +21,29 @@
  *****************************************************************************)
 
 open Source
+open Mm
 
 class mean ~field ~normalize source =
   object
     inherit operator [source] ~name:"mean"
 
     inherit
       Conversion.base
-        ~audio:true source
-        ~converter:(fun ~frame tmp_frame ->
+        ~converter:(fun frame ->
           (* Compute the mean of audio channels *)
-          let start = Frame.position frame in
-          let len = Frame.position tmp_frame - start in
-          let content = Content.Audio.get_data (Frame.get frame field) in
-          let tmp_content =
-            Content.Audio.get_data (Frame.get tmp_frame field)
-          in
+          let len = Frame.position frame in
+          let alen = Frame.audio_of_main len in
+          let src_content = Content.Audio.get_data (Frame.get frame field) in
+          let dst_content = Audio.Mono.create alen in
           let amp =
-            if normalize then 1. /. float (Array.length tmp_content) else 1.
+            if normalize then 1. /. float (Array.length src_content) else 1.
           in
-          let ( ! ) = Frame.audio_of_main in
-          for i = !start to !(start + len) - 1 do
-            content.(0).(i) <-
-              Array.fold_left (fun m b -> m +. b.(i)) 0. tmp_content *. amp
+          for i = 0 to alen - 1 do
+            dst_content.(i) <-
+              Array.fold_left (fun m b -> m +. b.(i)) 0. src_content *. amp
           done;
-          Frame.set frame field (Content.Audio.lift_data content))
+          Frame.set_data frame field Content.Audio.lift_data [| dst_content |])
+        source
   end
 
 let _ =

src/core/conversions/stereo.ml

@@ -38,11 +38,10 @@ class basic ~field source =
 
     inherit
       Conversion.base
-        ~audio:true source
-        ~converter:(fun ~frame tmp_frame ->
+        ~converter:(fun frame ->
           (* Set audio layer. *)
           let audio =
-            match Content.Audio.get_data (Frame.get tmp_frame field) with
+            match Content.Audio.get_data (Frame.get frame field) with
               | [||] ->
                   let len = AFrame.size () in
                   let buf = Audio.Mono.create len in
@@ -51,7 +50,8 @@ class basic ~field source =
               | [| chan |] -> [| chan; chan |]
               | audio -> Array.sub audio 0 2
           in
-          Frame.set frame field (Content.Audio.lift_data audio))
+          Frame.set_data frame field Content.Audio.lift_data audio)
+        source
   end
 
 let stereo =

src/core/conversions/swap.ml

@@ -25,29 +25,14 @@ open Source
 class swap ~field (source : source) =
   object
     inherit operator [source] ~name:"swap"
-    method stype = source#stype
-    method private _is_ready = source#is_ready
-    method remaining = source#remaining
-    method abort_track = source#abort_track
-    method seek_source = source#seek_source
-    method self_sync = source#self_sync
-
-    method private get_frame buf =
-      let offset = AFrame.position buf in
-      let buffer =
-        source#get buf;
-        Content.Audio.get_data (Frame.get buf field)
-      in
-      if offset = 0 then (
-        let tmp = buffer.(1) in
-        buffer.(1) <- buffer.(2);
-        buffer.(2) <- tmp)
-      else
-        for i = offset to AFrame.position buf - 1 do
-          let tmp = buffer.(0).(i) in
-          buffer.(0).(i) <- buffer.(1).(i);
-          buffer.(1).(i) <- tmp
-        done
+
+    inherit
+      Conversion.base
+        ~converter:(fun frame ->
+          let buffer = Content.Audio.get_data (Frame.get frame field) in
+          Frame.set_data frame field Content.Audio.lift_data
+            [| buffer.(1); buffer.(0) |])
+        source
   end
 
 let _ =