Seeking filesrc pipeline for continuous looping

[Frank]

I am trying to continuously loop an audio file for playback.

filesrc - queue - wavparse - audioconvert - audioresample - appsink (async = false)

doSeekToStartAsync() and doSeekToStart() are invoked correctly but immediately after seeking the next EOS event is triggered before the file could even be played the 2nd or nth time. What am I doing wrong here?

GstBusSyncReply BasePipeline::busCallback([[maybe_unused]] GstBus* bus, GstMessage* message, gpointer data)
{
[...]
  switch (GST_MESSAGE_TYPE(message))
  {
  case GST_MESSAGE_EOS:
    {
      if (pipeline)
        pipeline->doSeekToStartAsync();

      break;
    }
  default:
    break;
  }

  gst_message_unref(message);
  return GST_BUS_DROP; // Do not forward
}

void BasePipeline::doSeekToStart()
{
  auto isSeeked = gst_element_seek(m_pipeline, m_playbackRate, GST_FORMAT_TIME,
                                   static_cast<GstSeekFlags>(GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_KEY_UNIT),
                                   GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, GST_CLOCK_TIME_NONE);
  if (!isSeeked)
    m_eventBus.onMessage(pipeline_message::Error{getId(), "Failed to seek to start."});
}

[tpm]

Depending on your requirements and use case, the simplest implementation might be the following:

1. have two pipelines, one with interaudiosink and one with interaudiosrc for live streaming.
2. When you get EOS on the first pipeline you seek back to 0.
3. The second pipeline keeps running.
4. Downside of this is that there may be a short gap (read: glitch/silence) while the seek is executed. Which may or may not be a problem for your use case.
5. Upside is that the interaudiosrc will just continuously output data and fill gaps with silence, and you can also just change the input pipeline at any point if you like, e.g. to pull in a different file.

To make the looping work, you would need to do something like:

1. Preroll pipeline (set to PAUSED state), then do a flushing seek back to 0 with the SEGMENT flag set and set pipeline to PLAYING.
2. This will make sure that the demuxer/wavparse posts a SEGMENT_DONE message on the bus when it’s got to the end of the file (or segment). This is instead of an EOS. This means the demuxer/parser is now ready to execute the next seek.
3. At this point there may still be audio data in flight in the pipeline (e.g. in a queue between wavparse and appsink or a queue inside appsink).
4. When the app gets SEGMENT_DONE it can do a (non-flushing) seek back to 0 that queues more samples. The demuxer/parser will send a new segment event such that the running time of the buffers always increases.

This example shows the principle, but it steps through the file in 1-second chunks, whereas you want 0-end and then repeat.

[Frank]

I tried to do it with the GST_SEEK_FLAG_FLUSH initially and without FLUSH after the first GST_MESSAGE_SEGMENT_DONE event but when the pipeline reaches PAUSED the first gst_element_seek() call always returns false. Also the duration can not be queried. A GST_MESSAGE_SEGMENT_DONE is never published on the bus.
Do you have an idea whats wrong?

void BasePipeline::doSeekToStartAtStartup()
{
  // Force a sync check to guarantee preroll has completed and wavparse is active
  gst_element_get_state(m_pipeline, nullptr, nullptr, GST_CLOCK_TIME_NONE);

  gint64 dur;
  if (!gst_element_query_duration(m_pipeline, GST_FORMAT_TIME, &dur)) {
    DEV_LOG() << "Failed to query duration!"; // This fails as well
}
  DEV_LOG() << "Duration: " << dur;

  auto isSeeked = gst_element_seek(
    m_pipeline, m_playbackRate, GST_FORMAT_TIME,
    static_cast<GstSeekFlags>(GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_ACCURATE),
    GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_NONE, GST_CLOCK_TIME_NONE);

  if (!isSeeked)
    m_eventBus.onMessage(pipeline_message::Error{m_id, "Failed to seek to start at startup."});
}

[tpm]

Just to be sure, you shouldn’t / can’t do the seeks from a streaming thread, so probably shouldn’t be doing it from a message sync callback.

You should process messages in your main application thread and issue seeks from there (but unsure if that’s related to your problem of it returning false).

[Frank]

The sample code below worked for me. The key has been to place the queue behind the wav-parser element and to emit a flushing seek event at the beginning as tpm mentioned already.

#include <iostream>
#include <string>
#include <thread>
#include <chrono>
#include <atomic>

#include <gst/gst.h>
#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>

// Use seeking mechanism if true, otherwise use EOS mechanism.
constexpr bool doSeek{true};

struct OnNewSampleData
{
  GstElement* sinkPipeline;
  GstAppSrc* appsrc;
};

static GstFlowReturn on_new_sample_from_sink(GstAppSink* appsink, gpointer user_data)
{
  static std::atomic<uint64_t> lastBufferEndTime{0};

  // Pass the target appsrc as user_data
  auto* data = static_cast<OnNewSampleData*>(user_data);

  // Pull the sample from the appsink
  GstSample* sample = gst_app_sink_pull_sample(appsink);
  if (!sample)
  {
    return GST_FLOW_OK;
  }

  // Extract the buffer from the sample
  GstBuffer* buffer = gst_sample_get_buffer(sample);
  if (buffer)
  {
    GstCaps* caps = gst_sample_get_caps(sample);
    if (caps)
    {
      // Tell the appsrc exactly what kind of data it is pushing
      gst_app_src_set_caps(data->appsrc, caps);
    }

    // Get the clock from the pipeline. The clock is only available when the pipeline is playing.
    GstClock* clock = gst_pipeline_get_pipeline_clock(GST_PIPELINE(data->sinkPipeline));
    if (!clock)
    {
      std::cerr << "Error: Failed to get clock of pipeline.";
      return GST_FLOW_OK;
    }

    GstClockTime runningTime = gst_clock_get_time(clock) - gst_element_get_base_time(data->sinkPipeline);
    gst_object_unref(clock);

    static constexpr GstClockTime maxBufferDuration{static_cast<GstClockTime>(100e6)};

    // Update timestamp for next cycle
    if (!GST_BUFFER_DURATION_IS_VALID(buffer))
    {
      std::cerr << ": Ignore buffer: Invalid buffer duration.";
      return GST_FLOW_OK;
    }

    const auto bufferDuration = GST_BUFFER_DURATION(buffer);
    if (bufferDuration > maxBufferDuration)
    {
      std::cerr << ": Ignore buffer: Buffer duration (" << GST_TIME_AS_MSECONDS(bufferDuration)
                << " ms) too high and unreliable.";
      return GST_FLOW_OK;
    }

    // The minimum delay of a buffer in nanoseconds. If the buffer PTS is less than this delay there is a
    // significant risk that the buffer is dropped as the buffer has to reach the pipelines sink
    // element by the time the pipeline clock aligns with the current running-time accounting for all
    // intermediate processing steps.
    constexpr GstClockTime minDelay{static_cast<GstClockTime>(1e6)};
    // For hysteresis this maximum delay in ns is higher than the minimum delay above.
    constexpr GstClockTime maxDelay{static_cast<GstClockTime>(5e6)};

    GstClockTime assignedPTS{0U};
    uint64_t expected = lastBufferEndTime.load(std::memory_order_relaxed);

    while (true)
    {
      const auto observed = expected;
      auto candidate = observed;

      // If the last timestamp is too old relative to the running clock
      // because the source pipeline has been PAUSED or the processing can not keep up
      // reset to runningTime + maxDelay
      if (observed < (runningTime + minDelay))
      {
        candidate = static_cast<uint64_t>(runningTime + maxDelay);
        std::cerr << ": lastBufferEndTime (" << GST_TIME_AS_MSECONDS(observed)
                  << " ms) is too old -> reset to " << GST_TIME_AS_MSECONDS(candidate) << " ms";
      }
      assignedPTS = candidate;
      uint64_t desired = candidate + static_cast<uint64_t>(bufferDuration);

      // Try to atomically replace expected (the value we observed) with desired.
      // On success the atomic was still equal expected and now equals desired.
      // On failure: the atomic has been changed in the meantime and does not equal expected.
      // Expected is updated with the current atomic value and we loop.
      if (lastBufferEndTime.compare_exchange_weak(expected, desired, std::memory_order_acq_rel,
                                                  std::memory_order_relaxed))
      {
        break; // success: assignedPTS is valid for this buffer
      }
    }
    // Set new timestamp to current buffer
    GST_BUFFER_PTS(buffer) = assignedPTS;


    // Push the buffer to the appsrc element. Be aware: appsrc takes ownership of buffer!
    GstBuffer* push_buffer = gst_buffer_make_writable(gst_buffer_ref(buffer));
    GstFlowReturn ret = gst_app_src_push_buffer(data->appsrc, push_buffer);
    if (ret != GST_FLOW_OK)
    {
      // we do not have to unref the buffer here because push_buffer has been called previously
      std::cerr << "Failed to push buffer to appsrc: ";
    }
  }

  // Properly release the sample container
  gst_sample_unref(sample);

  return GST_FLOW_OK;
}

static GstBusSyncReply busCallback(GstBus* bus, GstMessage* message, gpointer data)
{
  std::string logMessage = "[BUS MESSAGE]";

  GError* err = nullptr;
  gchar* debug_info = nullptr;

  switch (GST_MESSAGE_TYPE(message))
  {
  case GST_MESSAGE_ERROR:
  {
    gst_message_parse_error(message, &err, &debug_info);
    logMessage += "Error from pipeline: " + std::string(err->message);
    if (debug_info)
    {
      logMessage += "\n (Debug info: " + std::string(debug_info) + ")\n";
      std::cerr << logMessage;
    }

    g_error_free(err);
    g_free(debug_info);
    break;
  }
  case GST_MESSAGE_WARNING:
  {
    gst_message_parse_warning(message, &err, &debug_info);
    logMessage += "Warning from pipeline: " + std::string(err->message);
    if (debug_info)
    {
      logMessage += "\n (Debug info: " + std::string(debug_info) + ")\n";
      std::cout << logMessage;
    }

    g_error_free(err);
    g_free(debug_info);
    break;
  }
  case GST_MESSAGE_EOS:
  {
    logMessage += "End-Of-Stream reached in pipeline: ";
    std::cout << logMessage << std::endl;
    GstElement* pipeline = GST_ELEMENT(data);
    if (pipeline)
    {
      // Dispatch fire-and-forget thread to execute the seek safely outside the streaming context
      std::thread(
        [pipeline]()
        {
          // std::this_thread::sleep_for(std::chrono::milliseconds(1000));
          gint64 dur;
          if (!gst_element_query_duration(pipeline, GST_FORMAT_TIME, &dur))
            std::cout << "Failed to query duration!\n";

          gboolean isSeeked =
            gst_element_seek(pipeline, 1.0, GST_FORMAT_TIME,
                             static_cast<GstSeekFlags>(GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE),
                             GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, dur);

          if (!isSeeked)
          {
            std::cout << "Failed to seek to start asynchronously.\n";
          }
          else
          {
            std::cout << "Loop flushing seek successful.\n";
          }
        })
        .detach();
    }
    break;
  }
  case GST_MESSAGE_SEGMENT_DONE:
  {
    std::cout << "Segment DONE.\n";

    GstElement* pipeline = GST_ELEMENT(data);
    if (pipeline)
    {
      // Dispatch fire-and-forget thread to execute the seek safely outside the streaming context
      std::thread(
        [pipeline]()
        {
          // std::this_thread::sleep_for(std::chrono::milliseconds(1000)); error
          //  Re-query duration if needed, or stick to your layout:
          gint64 dur = GST_CLOCK_TIME_NONE;
          gst_element_query_duration(pipeline, GST_FORMAT_TIME, &dur);

          gboolean isSeeked =
            gst_element_seek(pipeline, 1.0, GST_FORMAT_TIME,
                             static_cast<GstSeekFlags>(GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_ACCURATE),
                             GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, GST_CLOCK_TIME_NONE);

          if (!isSeeked)
          {
            std::cout << "Failed to seek to start asynchronously.\n";
          }
          else
          {
            std::cout << "Loop seek successful.\n";
          }
        })
        .detach();
    }
    break;
  }
  default:
    break;
  }

  gst_message_unref(message);
  return GST_BUS_DROP; // Do not forward to async handler -> unnecessary
}

int main()
{
  gst_init(nullptr, nullptr);

  GstElement* srcPipeline = gst_pipeline_new("audio-player");
  GstBus* bus = gst_element_get_bus(srcPipeline);
  GstElement* appsink = gst_element_factory_make("appsink", "appsink");
  {
    GstElement* source = gst_element_factory_make("filesrc", "file-source");
    GstElement* queue = gst_element_factory_make("queue", "queue");
    GstElement* parser = gst_element_factory_make("wavparse", "wav-parser");
    GstElement* convert = gst_element_factory_make("audioconvert", "audio-converter");
    GstElement* resample = gst_element_factory_make("audioresample", "audio-resampler");


    g_object_set(G_OBJECT(appsink), "async", true, "sync", true, nullptr);
    gst_app_sink_set_max_buffers(GST_APP_SINK(appsink), 2);
    // drop old samples
    gst_app_sink_set_drop(GST_APP_SINK(appsink), true);
    // The GstAppSink has to emit the "new-sample" signal
    gst_app_sink_set_emit_signals(GST_APP_SINK(appsink), true);

    if (!srcPipeline || !source || !queue|| !parser || !convert || !resample || !appsink)
    {
      std::cerr << "Not all elements could be created.\n";
      return -1;
    }

    g_object_set(G_OBJECT(source), "location", "D:\\ring.wav", nullptr);

    gst_bus_set_sync_handler(bus, busCallback, srcPipeline, nullptr);

    gst_bin_add_many(GST_BIN(srcPipeline), source,queue, parser, convert, resample, appsink, nullptr);
    if (!gst_element_link_many(source, parser,queue, convert, resample, appsink, nullptr))
    {
      std::cerr << "Link failure.\n";
      return -1;
    }

    // Preroll
    std::cout << "Preroll...\n";
    if (auto ret = gst_element_set_state(srcPipeline, GST_STATE_PAUSED); ret == GST_STATE_CHANGE_FAILURE)
    {
      std::cerr << "Unable to set the pipeline to the GST_STATE_PAUSED state.\n";
      gst_object_unref(bus);
      gst_object_unref(srcPipeline);
      return -1;
    }

    // Explicitly block until the pipeline completes its asynchronous transition to PAUSED
    // This fills the internal audio buffers and evaluates file properties.
    if (auto ret = gst_element_get_state(srcPipeline, nullptr, nullptr, GST_CLOCK_TIME_NONE);
        ret == GST_STATE_CHANGE_FAILURE)
    {
      std::cerr << "Pipeline failed during preroll validation.\n";
      return -1;
    }

    if (doSeek)
    {
      // Seek at startup
      std::cout << "Seek at startup...\n";
      gint64 dur;
      if (!gst_element_query_duration(srcPipeline, GST_FORMAT_TIME, &dur))
        std::cout << "Failed to query duration!\n";

      if (auto isSeeked = gst_element_seek(
            srcPipeline, 1.0, GST_FORMAT_TIME,
            static_cast<GstSeekFlags>(GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE),
            GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_SET, GST_CLOCK_TIME_NONE);
          !isSeeked)
        std::cout << "Failed to seek to start at startup.\n";
    }
  }


  GstElement* sinkPipeline = gst_pipeline_new("audio-player2");
  GstElement* appsrc = gst_element_factory_make("appsrc", "appsrc");
  {
    gst_app_src_set_emit_signals(GST_APP_SRC(appsrc), false);
    g_object_set(G_OBJECT(appsrc), "format", GST_FORMAT_TIME, "is-live", true, "do-timestamp", false,
                 "min-latency", 0, "max-latency", 5e6, "max-time", 5e6, "stream-type",
                 GST_APP_STREAM_TYPE_STREAM, "leaky-type", GST_APP_LEAKY_TYPE_DOWNSTREAM, nullptr);

    GstElement* convert = gst_element_factory_make("audioconvert", "audio-converter2");
    GstElement* resample = gst_element_factory_make("audioresample", "audio-resampler2");
    GstElement* sink = gst_element_factory_make("wasapi2sink", "wasapi-sink");

    if (!sinkPipeline || !appsrc || !convert || !resample || !sink)
    {
      std::cerr << "Not all elements could be created.\n";
      return -1;
    }

    gst_bin_add_many(GST_BIN(sinkPipeline), appsrc, convert, resample, sink, nullptr);
    if (!gst_element_link_many(appsrc, convert, resample, sink, nullptr))
    {
      std::cerr << "Link failure.\n";
      return -1;
    }

    // Start playing
    std::cout << "Starting playback...\n";
    if (auto ret = gst_element_set_state(sinkPipeline, GST_STATE_PLAYING); ret == GST_STATE_CHANGE_FAILURE)
    {
      std::cerr << "Unable to set the pipeline to the playing state.\n";
      gst_object_unref(sinkPipeline);
      return -1;
    }
  }

  OnNewSampleData data{sinkPipeline, GST_APP_SRC(appsrc)};
  g_signal_connect(GST_APP_SINK(appsink), "new-sample", G_CALLBACK(on_new_sample_from_sink), &data);

  // Start playing
  std::cout << "Starting playback...\n";
  if (auto ret = gst_element_set_state(srcPipeline, GST_STATE_PLAYING); ret == GST_STATE_CHANGE_FAILURE)
  {
    std::cerr << "Unable to set the pipeline to the playing state.\n";
    gst_object_unref(bus);
    gst_object_unref(srcPipeline);
    return -1;
  }

  // Wait until error or EOS (End of Stream)
  GstMessage* msg = gst_bus_timed_pop_filtered(
    bus, GST_CLOCK_TIME_NONE, static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));

  // Free resources
  gst_element_set_state(srcPipeline, GST_STATE_NULL);
  gst_element_set_state(sinkPipeline, GST_STATE_NULL);
  gst_object_unref(bus);
  gst_object_unref(srcPipeline);
  gst_object_unref(sinkPipeline);

  std::cout << "done\n";
  return 0;
}