Multiple streams = single or multiple scripts?

[adworks]

Thanks for supporting liquidsoap!

We're fairly new to liquidsoap. We are looking into developing a system that outputs multiple audio streams. The system will generate the necessary liquidsoap scripts and start them upon user request. We plan to support up to 50-100 audio streams on a single machine.

We have two options: either use %include to include all streams in a single script (and restart that script whenever a new stream is added or removed, restarting all the streams at once), or use separate .liq scripts and start/stop/restart each of them.

How much more resource-intensive is the separate scripts option compared to the single script one? What option would you recommend?

Thank you!

[vitoyucepi]

Hi @adworks,
I have never had so many audio streams running at once.

We have two options: either use %include to include all streams in a single script (and restart that script whenever a new stream is added or removed, restarting all the streams at once), or use separate .liq scripts and start/stop/restart each of them.

It's possible to dynamically create and destroy new stream pipelines without restart. But I would prefer to create a separate namespace for each process.
You might want to take a look at azuracast. The single instance of azuracast can host a bunch of radio stations at the same time.

How much more resource-intensive is the separate scripts option compared to the single script one? What option would you recommend?

The absolute minimum for each liquidsoap instance is 200 MB of RAM. Processor usage may vary, but for me it's 4% CPU per encoder.

[adworks]

Hi @vitoyucepi,
Thanks for the info.

I have never had so many audio streams running at once.

We're planning to spin up multiple short playlists of 10-20 tracks each that play in a loop. So they're not really radio stations per-se, but rather streamed playlists. This explains the large number of audio streams we're looking at and why we're rather careful with the hardware resources.

The single instance of azuracast can host a bunch of radio stations at the same time.

I'm not really sure, but I believe that AzuraCast spins up a liquidsoap script for each station, which can then be controlled via telnet. But new stations get new liquidsoap instances, right?

The absolute minimum for each liquidsoap instance is 200 MB of RAM. Processor usage may vary, but for me it's 4% CPU per encoder.

We're seeing the same 200-250 MB of RAM on a single instance. But if we output 3 streams from a single liquidsoap instance, we only see a marginal increase in RAM usage - like up to 400 MB of RAM. This is why we're leaning towards a single liquidsoap instance for the entire system.

CPU should not be a problem, as we're using servers with 24-32 CPU cores, provided that liquidsoap can use multiple cores with a single instance. This we haven't tested.

So, do you know if there is a way to start a new stream (i.e. %include a new script at runtime) without restarting the whole instance? Or any other way to dynamically add a new stream (playlist + output) to the instance without restarting it?

Thanks!

[Moonbase59]

I'm not really sure, but I believe that AzuraCast spins up a liquidsoap script for each station, which can then be controlled via telnet. But new stations get new liquidsoap instances, right?

Yes, every station gets a separate LS instance.

I think core LS currently only uses one core, but thread runners can use separate threads, so that might be an option. Since it’s "only" looping playlists, and if dynamic was possible, I’d probably go for one instance per so many playlists.

You might be creating a single point of failure with only one instance. Then again, LS is usually quite robust.

[vitoyucepi]

So, do you know if there is a way to start a new stream (i.e. %include a new script at runtime) without restarting the whole instance? Or any other way to dynamically add a new stream (playlist + output) to the instance without restarting it?

I think you could try to implement something like https://www.liquidsoap.info/doc-dev/dynamic_sources.html.

As for me, I don't want to use telnet, so I would create http endpoints. This is just an example and will not work.

harbor.http.register.simple("/station/new", api.create_station, method="POST")
harbor.http.register.simple("/station/new", api.delete_station, method="DELETE")

def api.create_station(request)
  json.parse something = request.body()
  if is_request_valid(something) then
    create_station(options=something.options)
    http.response.json.success()
  else
    http.response.json.error()
  end
end

def api.delete_station(request)
  json.parse something = request.body()
  if is_request_valid(something) then
    delete_station(options=something.options)
    http.response.json.success()
  else
    http.response.json.error()
  end
end

[adworks]

Just to follow-up on my previous questions, we finally went with a separate LS instance for each stream due to higher flexibility and easier implementation. We also realized that we can support enough streams with separate instances as we have 128GB RAM on our servers and the average RAM usage per instance is 200MB.

Also, we noticed that RAM usage varies with the number of songs in the playlist - 375 songs / up to 250MB RAM, 20 songs / up to 190MB RAM (which is probably to be expected). Processor usage is around 3.5% per stream on an older Intel(R) Xeon(R) CPU E5-2620 v3 @ 2.40GHz and somewhat surprisingly between 1.9-4.5% on a newer AMD EPYC 7313P 16-Core Processor. Maybe these figures help someone.

One rookie mistake that we did was to push the metadata via http to a web backend on the same thread as the main LS script. When the web backend responded a bit late, the stream would cut out for a second and we'd see a We must catchup 10.82 seconds! message in the logs.

We solved this by pushing the metadata asynchronously with thread.run like this:

def stream_3_send_meta(m)
    thread.run({
        print("playing now: " ^ m["initial_uri"])
        if (m["artist"] == "" and m["title"] == "") then
            ignore(http.get(
                normalize_url=false,
                stream_3_metadata_url
                ^ "?eventtype=UNSPECIFIED"
                ^ "&cid=" ^ stream_3_metadata_cid
                ))
        else
            ignore(http.get(
                normalize_url=false,
                stream_3_metadata_url
                ^ "?eventtype=SONG"
                ^ "&cid=" ^ stream_3_metadata_cid
                ^ "&artist=" ^ url.encode(m["artist"])
                ^ "&title=" ^ url.encode(m["title"])
                ))
        end
    })
end

We now rarely see We must catchup 1.05 seconds! messages, never above 5 seconds and without any stream interruptions.

Thank you both for your support!

Bogdan

[vitoyucepi]

Hi @adworks,

Thanks for the code snippet.
However, it's possible to make the thread.run a bit better by creating inner functions.

def stream_3_send_meta(m)
  id = "stream_3_send_meta"

  def api_post(data)
    # consider sending auth token in header
    response = http.post(stream_3_metadata_url, data=json.stringify(data, compact=true))
    # handle response
  end

  def post_song(artist, title)
    api_post({
      artist = artist,
      title = title,
      eventtype = "SONG",
      cid = stream_3_metadata_cid,
    })
  end

  def post_unspecified()
    api_post({
      eventtype = "UNSPECIFIED",
      cid = stream_3_metadata_cid,
    })
  end

  def send_metadata(metadata)
    log.important('playing now: #{metadata["initial_uri"]}', label=id)
    artist = metadata["artist"]
    title = metadata["title"]
    if (artist == "" and title == "") then
      post_unspecified()
    else
      post_song(artist, title)
    end
  end

  thread.run({send_metadata(m)}, fast=false)
end

It is possible to use put verb instead of post, by the way.

[adworks]

Hi @vitoyucepi ,

Thanks again for your suggestions! As usual, they are very elegant! :)

I know that a POST request with an auth header would be better, but unfortunately our web backend needs to ingest metadata from other streaming systems as well, so we need to stick to GET requests as those other systems don't support POST.

Also, good point with the fast=false flag in the thread.run function! We'll set settings.scheduler.non_blocking_queues := 2 and add that as well. Thanks!

[vitoyucepi]

I know that a POST request with an auth header would be better, but unfortunately our web backend needs to ingest metadata from other streaming systems as well, so we need to stick to GET requests as those other systems don't support POST.

You may still want to encode the URL and then send the request separately.

def stream_3_send_meta(m)
  id = "stream_3_send_meta"

  def api_send(params)
    args = list.map(fun(pair) -> "#{url.encode(fst(pair))}=#{url.encode(snd(pair))}", params)
    response = http.get('#{stream_3_metadata_url}?#{string.concat(args, separator="&")}')
    # handle response
  end

  def send_song(artist, title)
    api_send([
      ("artist", artist),
      ("title", title),
      ("eventtype", "SONG"),
      ("cid", stream_3_metadata_cid),
    ])
  end

  def send_unspecified()
    api_send([
      ("eventtype", "UNSPECIFIED"),
      ("cid", stream_3_metadata_cid),
    ])
  end

  def send_metadata(metadata)
    log.important('playing now: #{metadata["initial_uri"]}', label=id)
    artist = metadata["artist"]
    title = metadata["title"]
    if (artist == "" and title == "") then
      send_unspecified()
    else
      send_song(artist, title)
    end
  end

  thread.run({send_metadata(m)}, fast=false)
end

[vitoyucepi]

Hi @toots,

One rookie mistake that we did was to push the metadata via http to a web backend on the same thread as the main LS script. When the web backend responded a bit late, the stream would cut out for a second and we'd see a We must catchup 10.82 seconds! message in the logs.

I didn't know that the http.<func> could block the main thread, here's the reference example from the book

radio = playlist("~/Music")
def handle_metadata(m)
  h = http.post(
    headers=[("Content-Type", "application/json; charset=UTF-8")],
    data=metadata.json.stringify(m),
    "http://our.website.com/update_metadata.php"
  )
  if h.status_code >= 400 then
    log.important("Failed to update metadata.")
  end
end
radio.on_track(handle_metadata)