RAM and CPU usage skyrocketting when sending FLAC to input.harbor (on v2.1.4 and 2.2.0)

gabsoftware commented 1 year ago

Hello,

My setup :

At home, I have a liquidsoap instance sending a FLAC stream to my server on its harbor input. This home instance works well.
On a remote server, I have another liquidsoap with input.harbor configured (see script below).
On this server Liquidsoap instance, I set 4 output.icecast (ogg, opus, mp3 and flac). I have a single set (a flac file).
The Icecast instance is on the same server and works well
The purpose was to send only one high quality stream from my home to my server and to transcode it to 4 icecast streams.
The issue is on the Liquidsoap of the server, not the one in my home.
I get the exact same issue on another server with the same script.
I tried using 2.2.0 on server but it solves nothing. So I reverted to 2.1.4.

Whatever I try, when I send FLAC to input.harbor, in less than 2 or 3 hours it's almost 100% cpu usage and 90% RAM used.

I tried disabling the internal FLAC decoder as seen in issue savonet/ocaml-flac#9 and force FFMPEG decoder for audio/ogg.
I tried setting the list of decoders to be only FFMPEG.
I tried sending FLAC through audio.raw from my home, changes nothing.
I tried disabling send_icy_metadata as I read there could be a potential memory leak with FLAC metadata, changes nothing.
I tried changing settings for the GC but I don't think it changes much.
I tried disabling the outputs, changes nothing (it reduced initial CPU and RAM usage but then it starts to grow again...)

And when I try to send WAV to my server, CPU and RAM usage is much lower. Still increasing but very slowly.

This is on Debian bullseye 11.6 AMD64. Liquidsoap 2.1.4 installed from the Github releases page. I also tried 2.2.0.

Here is my script on the server, it is very simple:

#!/usr/bin/liquidsoap

# Activate the live stream input
settings.harbor.bind_addrs.set(["0.0.0.0"])
log.file.path.set('/home/radio/radio.log')

# DOES NOT SOLVE ANYTHING
#settings.ffmpeg.log.verbosity.set("warning")
#settings.decoder.decoders.set(["FFMPEG"])
#settings.decoder.mime_types.ffmpeg.set(["application/ogg"])
#settings.decoder.mime_types.ogg.set([])

# DOES NOT SOLVE ANYTHING EITHER
#runtime.gc.set(runtime.gc.get().{
#  space_overhead = 20,
#   allocation_policy = 2
#})

input_icecast = input.harbor(
    '/master-stream.flac',
    port=9000,
    password='hackme',
)

security = single("/home/radio/default.flac")

radio = fallback(track_sensitive = false, [input_icecast, security])

output.icecast(
    %ffmpeg(
        format="ogg",
        %audio(
            codec="libvorbis",
            global_quality="9"
        )
    ),
    radio,
    host='localhost',
    port=8000,
    password='hackme',
    mount='stream.ogg',
    name='Radio (OGG)',
    genre='electronic',
    description='A radio',
    url='https://radio.server/stream.ogg',
    encoding="UTF-8"
)
output.icecast(
    %ffmpeg(
        format="opus",
        %audio(
            codec="libopus",
            b="327680",
            ar="48000"
        )
    ),
    radio,
    host='localhost',
    port=8000,
    password='hackme',
    mount='stream.opus',
    name='Radio (OPUS)',
    genre='electronic',
    description='A radio',
    url='https://radio.server/stream.opus',
    encoding="UTF-8"
)
output.icecast(
    %ffmpeg(
        format="mp3",
        %audio(
            codec="libmp3lame",
            b="320k"
        )
    ),
    radio,
    host='localhost',
    port=8000,
    password='hackme',
    mount='stream.mp3',
    name='Radio (MP3)',
    genre='electronic',
    description='A radio',
    url='https://radio.server/stream.mp3',
    icy_metadata="true",
    encoding="UTF-8"
)
output.icecast(
    %ffmpeg(
        format="ogg",
        %audio(
            codec="flac"
        )
    ),
    radio,
    host='localhost',
    port=8000,
    password='hackme',
    mount='stream.flac',
    name='Radio (FLAC)',
    genre='electronic',
    description='A radio',
    url='https://radio.server/stream.flac',
    encoding="UTF-8"
)

toots commented 1 year ago

Hi,

Thanks for this report. I'm running a test script and will report.

One remark that I would suggest trying is to test with the sending and receiving liquidsoap on the same machine to make sure the memory doesn't increase because input.harbor is receiving data too fast.

gabsoftware commented 1 year ago

Would be interested to know if you can had that issue as well :-) Anyway I switched to WAV streaming between the 2 servers. RAM and CPU usage is stable. But I have "end of stream" issues every few hours and then, out of the scope of the current issue.

toots commented 1 year ago

I haven't forgotten about this one.

gabsoftware commented 1 year ago

Using a minimal script. I send a %ogg(%flac) encoded stream to input.harbor:

settings.harbor.bind_addrs.set(["0.0.0.0"])
log.file.path.set(log_path)

# configure security input
security = single(
    id="security_single",
    default_wav_path
)

# configure harbor input
raw_harbor_input=input.harbor(
    id="input_harbor_master_stream",
    port=harbor_port,
    password=harbor_password,
    replay_metadata=true,
    metadata_charset="UTF-8",
    "/master-stream.flac"
)

fallbackswitch=fallback.skip(
    raw_harbor_input,
    fallback=security
)

output.dummy(
    id="output_dummy",
    fallbackswitch
)

I still see a memory leak and CPU usage grows also.

But, if I try setting the FLAC and OGG decoders at a higher priority than the FFMPEG one:

settings.decoder.priorities.flac.set( 11 )
settings.decoder.priorities.ogg.set( 12 )

Then it no longer seems to have a memory leak and CPU usage is stable.

So I'm pretty confident that the issue lies within the FFMPEG decoder for OGG/FLAC. Not sure if it's important that it comes from input.harbor or not.

toots commented 1 year ago

Great thanks. What are you using to send to the input.harbor?

gabsoftware commented 1 year ago

Great thanks. What are you using to send to the input.harbor?

Just another Liquidsoap flac stream using output.icecast

toots commented 1 year ago

Ok. I assume you mean ogg/flac? I haven't been able to reproduce so far.

gabsoftware commented 1 year ago

Ok. I assume you mean ogg/flac? I haven't been able to reproduce so far.

Yes, I send OGG/Flac to input.harbor.

The same thing happens on my two servers.

Here is the exact script (stripped of passwords and such) that I run on a NAS at home, and that sends the audio to my servers:

#!/usr/bin/liquidsoap

#settings
settings.sandbox.set(true)
settings.sandbox.network.set(true)
settings.sandbox.shell.set(true)
settings.sandbox.shell.path.set("/bin/bash")

# This function is called when
# a new metadata block is passed in
# the stream.
def apply_metadata(m) =

    log("calling apply_metadata")

    artist = url.encode(m["artist"])
    album  = url.encode(m["album"])
    title  = url.encode(m["title"])

    if(artist == "" and album == "" and title == "") then
        log("Artist, album and title all empty!")
    else
        query = "artist=#{artist}&album=#{album}&title=#{title}"
        icy   = "#{artist}+-+#{title}"
        log("query for metadata server: #{query}")
        log("icy for icecast metadata: #{icy}")

        # mise à jour metadata
        command = process.quote.command( args=["-m", "2", "https://server1.com/sendMetadata?#{query}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)
        command = process.quote.command( args=["-m", "2", "https://server2.com/sendMetadata?#{query}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)

        # mise à jour historique
        command = process.quote.command( args=["-m", "2", "https://server1.com/sendHistory?#{query}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)
        command = process.quote.command( args=["-m", "2", "https://server2.com/sendHistory?#{query}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)

        # mise à jour metadata des streams (flac et opus ne supportent pas pour l'instant)

        # mise à jour metadata du stream mp3
        command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server1.com:3333/admin/metadata?mount=%2Fstream.mp3&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)
        command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server2.com:3333/admin/metadata?mount=%2Fstream.mp3&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)

        # mise à jour metadata du stream ogg
        command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server1.com:3333/admin/metadata?mount=%2Fstream.ogg&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)
        command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server2.com:3333/admin/metadata?mount=%2Fstream.ogg&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        #print(command)
        process.run(timeout=4.0, command)

        # flac et opus ne supportent pas les maj de metadata

        # # mise à jour metadata du stream opus
        # command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server1.com:3333/admin/metadata?mount=%2Fstream.opus&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        # #print(command)
        # process.run(timeout=4.0, command)
        # command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server2.com:3333/admin/metadata?mount=%2Fstream.opus&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        # #print(command)
        # process.run(timeout=4.0, command)

        # # mise à jour metadata du stream flac
        # command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server1.com:3333/admin/metadata?mount=%2Fstream.flac&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        # #print(command)
        # process.run(timeout=4.0, command)
        # command = process.quote.command( args=["-m", "2", "-u", "adminuser:hackme", "https://server2.com:3333/admin/metadata?mount=%2Fstream.flac&mode=updinfo&charset=UTF-8&song=#{icy}"], "curl" )
        # #print(command)
        # process.run(timeout=4.0, command)

    end
end

# Log dir
log.file.path.set("/home/user/radio.log")

# Music
playlist_ambient   = playlist(id="radio_ambient"  , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_ambient.pls")
playlist_chillout  = playlist(id="radio_chillout" , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_chillout.pls")
playlist_dnb       = playlist(id="radio_dnb"      , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_dnb.pls")
playlist_dub       = playlist(id="radio_dub"      , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_dub.pls")
playlist_dubstep   = playlist(id="radio_dubstep"  , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_dubstep.pls")
playlist_electro   = playlist(id="radio_electro"  , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_electro.pls")
playlist_futurepop = playlist(id="radio_futurepop", mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_futurepop.pls")
playlist_idm       = playlist(id="radio_idm"      , mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_idm.pls")
playlist_nightcity = playlist(id="radio_nightcity", mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_nightcity.pls")
playlist_synthwave = playlist(id="radio_synthwave", mode="randomize", reload=3600, reload_mode="seconds", "/home/user/radio_synthwave.pls")
playlist_mix = random(
    id="playlist_mix",
    transitions=[],
    transition_length=0.0,
    weights = [3, 15, 18, 2, 1, 6, 21, 2, 3, 75],
    [
        playlist_ambient,
        playlist_chillout,
        playlist_dnb,
        playlist_dub,
        playlist_dubstep,
        playlist_electro,
        playlist_futurepop,
        playlist_idm,
        playlist_nightcity,
        playlist_synthwave
    ]
)

# If something goes wrong, we'll play this
security=single(
    id="security_single",
    "/home/user/default.flac"
)

# create the radio stream
radio_stream=blank.skip(
    id="radio_stream_blank_skipper",
    # "radio_to_stereo is deprecated in 2.2.0, use stereo instead"
    stereo(
        id="radio_stream_audio_to_stereo",
        clock(
            id="radio_stream_clock",
            blank.eat(
                id="playlist_mix_blank_eater",
                max_blank=5.0,
                playlist_mix
            )
        )
    )
)

# And finally the security
radio=fallback.skip(
    radio_stream,
    fallback=security
)

# action on metadata
radio.on_metadata(apply_metadata)

# icecast events radio1
def output_icecast_radio1_connect() =
    log("output_icecast_radio1_connect")
end
def output_icecast_radio1_disconnect() =
    log("output_icecast_radio1_disconnect")
end
def output_icecast_radio1_error(_) =
    log("output_icecast_radio1_error")
    3.0
end
def output_icecast_radio1_start() =
    log("output_icecast_radio1_start")
end
def output_icecast_radio1_stop() =
    log("output_icecast_radio1_stop")
end

# icecast events radio2
def output_icecast_radio2_connect() =
    log("output_icecast_radio2_connect")
end
def output_icecast_radio2_disconnect() =
    log("output_icecast_radio2_disconnect")
end
def output_icecast_radio2_error(_) =
    log("output_icecast_radio2_error")
    3.0
end
def output_icecast_radio2_start() =
    log("output_icecast_radio2_start")
end
def output_icecast_radio2_stop() =
    log("output_icecast_radio2_stop")
end

radio_radio1=mksafe(
    id="radio1_mksafe",
    buffer(
        id="radio1_buffer",
        fallible=false,
        radio
    )
)

radio_radio2=mksafe(
    id="radio2_mksafe",
    buffer(
        id="radio2_buffer",
        fallible=false,
        radio
    )
)

# output to icecast mounts
output.icecast(
    id="output_icecast_radio1",
    %ogg(
        %flac(
            samplerate=44100,
            channels=2,
            compression=8,
            bits_per_sample=16
        )
    ),
    host="server2.com",
    port=2222,
    password="hackme",
    mount="/master-stream.flac",
    on_connect=output_icecast_radio1_connect,
    on_disconnect=output_icecast_radio1_disconnect,
    on_error=output_icecast_radio1_error,
    on_start=output_icecast_radio1_start,
    on_stop=output_icecast_radio1_stop,
    radio_radio1
)

output.icecast(
    id="output_icecast_radio2",
    %ogg(
        %flac(
            samplerate=44100,
            channels=2,
            compression=8,
            bits_per_sample=16
        )
    ),
    host="server1.com",
    port=2222,
    password="hackme",
    mount="/master-stream.flac",
    on_connect=output_icecast_radio2_connect,
    on_disconnect=output_icecast_radio2_disconnect,
    on_error=output_icecast_radio2_error,
    on_start=output_icecast_radio2_start,
    on_stop=output_icecast_radio2_stop,
    radio_radio2
)

# called when accessing the HTTP "/skiptrack" API endpoint
def try_skiptrack(request, response) =
    log("Got a request on path #{request.path}, protocol version: #{request.http_version}, \
           method: #{request.method}, headers: #{request.headers}, query: #{request.query}, \
           body: #{request.body()}")

    # skip to next track
    log("Skipping to next track!")
    source.skip(playlist_mix)

    # write response for debug
    response.status_code(200)
    response.status_message("OK")
    response.content_type("application/json")
    response.http_version("1.1")
    response.json({status = "success"})
end

# register HTTP "/skiptrack" API endpoint
harbor.http.register(
    port=1111,
    method="GET",
    "/skiptrack",
    try_skiptrack
)

Hopefully you'll be able to replicate with this although the important part is probably the output.icecast part and preparaton made before.

The .PLS playlists link to hundreds of various .flac files that have nothing special.

toots commented 1 year ago

I'm still not sure what was causing the issue but it would be worth revisiting now that we are producing ogg/flac streams that conform to what ffmpeg is expecting.

gabsoftware commented 1 year ago

All that is needed is sending an %ogg(%flac) stream to input.harbor on this minimal script:

settings.harbor.bind_addrs.set(["0.0.0.0"])
log.file.path.set(log_path)

# configure security input
security = single(
   id="security_single",
   default_wav_path
)

# configure harbor input
raw_harbor_input=input.harbor(
   id="input_harbor_master_stream",
   port=harbor_port,
   password=harbor_password,
   replay_metadata=true,
   metadata_charset="UTF-8",
   "/master-stream.flac"
)

fallbackswitch=fallback.skip(
   raw_harbor_input,
   fallback=security
)

output.dummy(
   id="output_dummy",
   fallbackswitch
)

If no leak then it should be okay. I'll try soon when I add a switch with input.harbor for live mixes.

gabsoftware commented 1 year ago

Well Mixxx cannot stream in FLAC, damnit.

savonet / liquidsoap

RAM and CPU usage skyrocketting when sending FLAC to input.harbor (on v2.1.4 and 2.2.0) #3017