/* GStreamer

 * Copyright (C) 2006 Edward Hervey <edward@fluendo.com>

 * Copyright (C) 2007 Jan Schmidt <jan@fluendo.com>

 * Copyright (C) 2007 Wim Taymans <wim@fluendo.com>

 *

 * gstmultiqueue.c:

 *

 * This library is free software; you can redistribute it and/or

 * modify it under the terms of the GNU Library General Public

 * License as published by the Free Software Foundation; either

 * version 2 of the License, or (at your option) any later version.

 *

 * This library is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

 * Library General Public License for more details.

 *

 * You should have received a copy of the GNU Library General Public

 * License along with this library; if not, write to the

 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,

 * Boston, MA 02111-1307, USA.

 */

/**

 * SECTION:element-multiqueue

 * @see_also: #GstQueue

 *

 * <refsect2>

 * <para>

 * Multiqueue is similar to a normal #GstQueue with the following additional

 * features:

 * <orderedlist>

 * <listitem>

 *   <itemizedlist><title>Multiple streamhandling</title>

 *   <listitem><para>

 *     The element handles queueing data on more than one stream at once. To

 *   </para><para>

 *     When requesting a given sinkpad with gst_element_get_request_pad(),

 *     the associated srcpad for that stream will be created.

 *   </para></listitem>

 *   </itemizedlist>

 * </listitem>

 * <listitem>

 *   <itemizedlist><title>Non-starvation on multiple streams</title>

 *   <listitem><para>

 *     If more than one stream is used with the element, the streams' queues

 *     will be dynamically grown (up to a limit), in order to ensure that no

 *     stream is risking data starvation. This guarantees that at any given

 *     time there are at least N bytes queued and available for each individual

 *     stream.

 *   </para><para>

 *     If an EOS event comes through a srcpad, the associated queue will be

 *     considered as 'not-empty' in the queue-size-growing algorithm.

 *   </para></listitem>

 *   </itemizedlist>

 * </listitem>

 * <listitem>

 *   <itemizedlist><title>Non-linked srcpads graceful handling</title>

 *   <listitem><para>

 *     In order to better support dynamic switching between streams, the multiqueue

 *     (unlike the current GStreamer queue) continues to push buffers on non-linked

 *     pads rather than shutting down.

 *   </para><para>

 *     In addition, to prevent a non-linked stream from very quickly consuming all

 *     available buffers and thus 'racing ahead' of the other streams, the element

 *     must ensure that buffers and inlined events for a non-linked stream are pushed

 *     in the same order as they were received, relative to the other streams

 *     controlled by the element. This means that a buffer cannot be pushed to a

 *     non-linked pad any sooner than buffers in any other stream which were received

 *     before it.

 *   </para></listitem>

 *   </itemizedlist>

 * </listitem>

 * </orderedlist>

 * </para>

 * <para>

 *   Data is queued until one of the limits specified by the

 *   #GstMultiQueue:max-size-buffers, #GstMultiQueue:max-size-bytes and/or

 *   #GstMultiQueue:max-size-time properties has been reached. Any attempt to push

 *   more buffers into the queue will block the pushing thread until more space

 *   becomes available. #GstMultiQueue:extra-size-buffers,

 * </para>

 * <para>

 *   #GstMultiQueue:extra-size-bytes and #GstMultiQueue:extra-size-time are

 *   currently unused.

 * </para>

 * <para>

 *   The default queue size limits are 5 buffers, 10MB of data, or

 *   two second worth of data, whichever is reached first. Note that the number

 *   of buffers will dynamically grow depending on the fill level of 

 *   other queues.

 * </para>

 * <para>

 *   The #GstMultiQueue::underrun signal is emitted when all of the queues

 *   are empty. The #GstMultiQueue::overrun signal is emitted when one of the

 *   queues is filled.

 *   Both signals are emitted from the context of the streaming thread.

 * </para>

 * </refsect2>

 *

 * Last reviewed on 2008-01-25 (0.10.17)

 */

#ifdef HAVE_CONFIG_H

#  include "config.h"

#endif

#include <gst/gst.h>

#include "gstmultiqueue.h"

/**

 * GstSingleQueue:

 * @sinkpad: associated sink #GstPad

 * @srcpad: associated source #GstPad

 *

 * Structure containing all information and properties about

 * a single queue.

 */

typedef struct _GstSingleQueue GstSingleQueue;

struct _GstSingleQueue

{

  /* unique identifier of the queue */

  guint id;

  GstMultiQueue *mqueue;

  GstPad *sinkpad;

  GstPad *srcpad;

  /* flowreturn of previous srcpad push */

  GstFlowReturn srcresult;

  GstSegment sink_segment;

  GstSegment src_segment;

  /* queue of data */

  GstDataQueue *queue;

  GstDataQueueSize max_size, extra_size;

  GstClockTime cur_time;

  gboolean is_eos;

  gboolean inextra;             /* TRUE if the queue is currently in extradata mode */

  /* Protected by global lock */

  guint32 nextid;               /* ID of the next object waiting to be pushed */

  guint32 oldid;                /* ID of the last object pushed (last in a series) */

  GCond *turn;                  /* SingleQueue turn waiting conditional */

};

/* Extension of GstDataQueueItem structure for our usage */

typedef struct _GstMultiQueueItem GstMultiQueueItem;

struct _GstMultiQueueItem

{

  GstMiniObject *object;

  guint size;

  guint64 duration;

  gboolean visible;

  GDestroyNotify destroy;

  guint32 posid;

};

static GstSingleQueue *gst_single_queue_new (GstMultiQueue * mqueue);

static void gst_single_queue_free (GstSingleQueue * squeue);

static void wake_up_next_non_linked (GstMultiQueue * mq);

static void compute_high_id (GstMultiQueue * mq);

static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink%d",

    GST_PAD_SINK,

    GST_PAD_REQUEST,

    GST_STATIC_CAPS_ANY);

static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src%d",

    GST_PAD_SRC,

    GST_PAD_SOMETIMES,

    GST_STATIC_CAPS_ANY);

GST_DEBUG_CATEGORY_STATIC (multi_queue_debug);

#define GST_CAT_DEFAULT (multi_queue_debug)

/* default limits, we try to keep up to 2 seconds of data and if there is not

 * time, up to 10 MB. The number of buffers is dynamically scaled to make sure

 * there is data in the queues. Normally, the byte and time limits are not hit

 * in theses conditions. */

#define DEFAULT_MAX_SIZE_BYTES 10 * 1024 * 1024 /* 10 MB */

#define DEFAULT_MAX_SIZE_BUFFERS 5

#define DEFAULT_MAX_SIZE_TIME 2 * GST_SECOND

/* second limits. When we hit one of the above limits we are probably dealing

 * with a badly muxed file and we scale the limits to these emergency values.

 * This is currently not yet implemented. */

#define DEFAULT_EXTRA_SIZE_BYTES 10 * 1024 * 1024       /* 10 MB */

#define DEFAULT_EXTRA_SIZE_BUFFERS 5

#define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND

/* Signals and args */

enum

{

  SIGNAL_UNDERRUN,

  SIGNAL_OVERRUN,

  LAST_SIGNAL

};

enum

{

  ARG_0,

  ARG_EXTRA_SIZE_BYTES,

  ARG_EXTRA_SIZE_BUFFERS,

  ARG_EXTRA_SIZE_TIME,

  ARG_MAX_SIZE_BYTES,

  ARG_MAX_SIZE_BUFFERS,

  ARG_MAX_SIZE_TIME,

};

#define GST_MULTI_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \

  g_mutex_lock (q->qlock);                                              \

} G_STMT_END

#define GST_MULTI_QUEUE_MUTEX_UNLOCK(q) G_STMT_START {                        \

  g_mutex_unlock (q->qlock);                                            \

} G_STMT_END

static void gst_multi_queue_finalize (GObject * object);

static void gst_multi_queue_set_property (GObject * object,

    guint prop_id, const GValue * value, GParamSpec * pspec);

static void gst_multi_queue_get_property (GObject * object,

    guint prop_id, GValue * value, GParamSpec * pspec);

static GstPad *gst_multi_queue_request_new_pad (GstElement * element,

    GstPadTemplate * temp, const gchar * name);

static void gst_multi_queue_release_pad (GstElement * element, GstPad * pad);

static void gst_multi_queue_loop (GstPad * pad);

#define _do_init(bla) \

  GST_DEBUG_CATEGORY_INIT (multi_queue_debug, "multiqueue", 0, "multiqueue element");

GST_BOILERPLATE_FULL (GstMultiQueue, gst_multi_queue, GstElement,

    GST_TYPE_ELEMENT, _do_init);

static guint gst_multi_queue_signals[LAST_SIGNAL] = { 0 };

static void

gst_multi_queue_base_init (gpointer g_class)

{

  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);

  gst_element_class_set_details_simple (gstelement_class,

      "MultiQueue",

      "Generic", "Multiple data queue", "Edward Hervey <edward@fluendo.com>");

  gst_element_class_add_pad_template (gstelement_class,

      gst_static_pad_template_get (&sinktemplate));

  gst_element_class_add_pad_template (gstelement_class,

      gst_static_pad_template_get (&srctemplate));

}

static void

gst_multi_queue_class_init (GstMultiQueueClass * klass)

{

  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);

  gobject_class->set_property =

      GST_DEBUG_FUNCPTR (gst_multi_queue_set_property);

  gobject_class->get_property =

      GST_DEBUG_FUNCPTR (gst_multi_queue_get_property);

  /* SIGNALS */

  gst_multi_queue_signals[SIGNAL_UNDERRUN] =

      g_signal_new ("underrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstMultiQueueClass, underrun), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  gst_multi_queue_signals[SIGNAL_OVERRUN] =

      g_signal_new ("overrun", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST,

      G_STRUCT_OFFSET (GstMultiQueueClass, overrun), NULL, NULL,

      g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);

  /* PROPERTIES */

  g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BYTES,

      g_param_spec_uint ("max-size-bytes", "Max. size (kB)",

          "Max. amount of data in the queue (bytes, 0=disable)",

  g_object_class_install_property (gobject_class, ARG_MAX_SIZE_BUFFERS,

      g_param_spec_uint ("max-size-buffers", "Max. size (buffers)",

  g_object_class_install_property (gobject_class, ARG_MAX_SIZE_TIME,

      g_param_spec_uint64 ("max-size-time", "Max. size (ns)",

  g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BYTES,

      g_param_spec_uint ("extra-size-bytes", "Extra Size (kB)",

          "Amount of data the queues can grow if one of them is empty (bytes, 0=disable)",

  g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_BUFFERS,

      g_param_spec_uint ("extra-size-buffers", "Extra Size (buffers)",

          "Amount of buffers the queues can grow if one of them is empty (0=disable)",

  g_object_class_install_property (gobject_class, ARG_EXTRA_SIZE_TIME,

      g_param_spec_uint64 ("extra-size-time", "Extra Size (ns)",

          "Amount of time the queues can grow if one of them is empty (in ns, 0=disable)",

  gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_multi_queue_finalize);

  gstelement_class->request_new_pad =

      GST_DEBUG_FUNCPTR (gst_multi_queue_request_new_pad);

  gstelement_class->release_pad =

      GST_DEBUG_FUNCPTR (gst_multi_queue_release_pad);

}

static void

gst_multi_queue_init (GstMultiQueue * mqueue, GstMultiQueueClass * klass)

{

  mqueue->nbqueues = 0;

  mqueue->queues = NULL;

  mqueue->max_size.bytes = DEFAULT_MAX_SIZE_BYTES;

  mqueue->max_size.visible = DEFAULT_MAX_SIZE_BUFFERS;

  mqueue->max_size.time = DEFAULT_MAX_SIZE_TIME;

  mqueue->extra_size.bytes = DEFAULT_EXTRA_SIZE_BYTES;

  mqueue->extra_size.visible = DEFAULT_EXTRA_SIZE_BUFFERS;

  mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;

  mqueue->counter = 1;

  mqueue->highid = -1;

  mqueue->nextnotlinked = -1;

  mqueue->qlock = g_mutex_new ();

}

static void

gst_multi_queue_finalize (GObject * object)

{

  GstMultiQueue *mqueue = GST_MULTI_QUEUE (object);

  g_list_foreach (mqueue->queues, (GFunc) gst_single_queue_free, NULL);

  g_list_free (mqueue->queues);

  mqueue->queues = NULL;

  /* free/unref instance data */

  g_mutex_free (mqueue->qlock);

  G_OBJECT_CLASS (parent_class)->finalize (object);

}

#define SET_CHILD_PROPERTY(mq,format) G_STMT_START {	        \

    GList * tmp = mq->queues;					\

    while (tmp) {						\

      GstSingleQueue *q = (GstSingleQueue*)tmp->data;		\

      q->max_size.format = mq->max_size.format;                 \

      tmp = g_list_next(tmp);					\

    };								\

} G_STMT_END

static void

gst_multi_queue_set_property (GObject * object, guint prop_id,

    const GValue * value, GParamSpec * pspec)

{

  GstMultiQueue *mq = GST_MULTI_QUEUE (object);

  switch (prop_id) {

    case ARG_MAX_SIZE_BYTES:

      GST_MULTI_QUEUE_MUTEX_LOCK (mq);

      mq->max_size.bytes = g_value_get_uint (value);

      SET_CHILD_PROPERTY (mq, bytes);

      GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);

      break;

    case ARG_MAX_SIZE_BUFFERS:

      GST_MULTI_QUEUE_MUTEX_LOCK (mq);

      mq->max_size.visible = g_value_get_uint (value);

      SET_CHILD_PROPERTY (mq, visible);

      GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);

      break;

    case ARG_MAX_SIZE_TIME:

      GST_MULTI_QUEUE_MUTEX_LOCK (mq);

      mq->max_size.time = g_value_get_uint64 (value);

      SET_CHILD_PROPERTY (mq, time);

      GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);

      break;

    case ARG_EXTRA_SIZE_BYTES:

      mq->extra_size.bytes = g_value_get_uint (value);

      break;

    case ARG_EXTRA_SIZE_BUFFERS:

      mq->extra_size.visible = g_value_get_uint (value);

      break;

    case ARG_EXTRA_SIZE_TIME:

      mq->extra_size.time = g_value_get_uint64 (value);

      break;

    default:

      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);

      break;

  }

}

static void

gst_multi_queue_get_property (GObject * object, guint prop_id,

    GValue * value, GParamSpec * pspec)

{

  GstMultiQueue *mq = GST_MULTI_QUEUE (object);

  GST_MULTI_QUEUE_MUTEX_LOCK (mq);

  switch (prop_id) {

    case ARG_EXTRA_SIZE_BYTES:

      g_value_set_uint (value, mq->extra_size.bytes);

      break;

    case ARG_EXTRA_SIZE_BUFFERS:

      g_value_set_uint (value, mq->extra_size.visible);

      break;

    case ARG_EXTRA_SIZE_TIME:

      g_value_set_uint64 (value, mq->extra_size.time);

      break;

    case ARG_MAX_SIZE_BYTES:

      g_value_set_uint (value, mq->max_size.bytes);

      break;

    case ARG_MAX_SIZE_BUFFERS:

      g_value_set_uint (value, mq->max_size.visible);

      break;

    case ARG_MAX_SIZE_TIME:

      g_value_set_uint64 (value, mq->max_size.time);

      break;

    default:

      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);

      break;

  }

  GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);

}

static GList *

gst_multi_queue_get_internal_links (GstPad * pad)

{

  GList *res = NULL;

  GstMultiQueue *mqueue;

  GstSingleQueue *sq = NULL;

  GList *tmp;

  g_return_val_if_fail (GST_IS_PAD (pad), NULL);

  mqueue = GST_MULTI_QUEUE (GST_PAD_PARENT (pad));

  if (!mqueue)

    goto no_parent;

  GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);

  /* Find which single queue it belongs to */

  for (tmp = mqueue->queues; tmp && !res; tmp = g_list_next (tmp)) {

    sq = (GstSingleQueue *) tmp->data;

    if (sq->sinkpad == pad)

      res = g_list_prepend (res, sq->srcpad);

    if (sq->srcpad == pad)

      res = g_list_prepend (res, sq->sinkpad);

  }

  if (!res)

    GST_WARNING_OBJECT (mqueue, "That pad doesn't belong to this element ???");

  GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);

  return res;

no_parent:

  {

    GST_DEBUG_OBJECT (pad, "no parent");

    return NULL;

  }

}

/*

 * GstElement methods

 */

static GstPad *

gst_multi_queue_request_new_pad (GstElement * element, GstPadTemplate * temp,

    const gchar * name)

{

  GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);

  GstSingleQueue *squeue;

  GST_LOG_OBJECT (element, "name : %s", GST_STR_NULL (name));

  /* Create a new single queue, add the sink and source pad and return the sink pad */

  squeue = gst_single_queue_new (mqueue);

  GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);

  mqueue->queues = g_list_append (mqueue->queues, squeue);

  GST_MULTI_QUEUE_MUTEX_UNLOCK (mqueue);

  GST_DEBUG_OBJECT (mqueue, "Returning pad %s:%s",

      GST_DEBUG_PAD_NAME (squeue->sinkpad));

  return squeue->sinkpad;

}

static void

gst_multi_queue_release_pad (GstElement * element, GstPad * pad)

{

  GstMultiQueue *mqueue = GST_MULTI_QUEUE (element);

  GstSingleQueue *sq = NULL;

  GList *tmp;

  GST_LOG_OBJECT (element, "pad %s:%s", GST_DEBUG_PAD_NAME (pad));

  GST_MULTI_QUEUE_MUTEX_LOCK (mqueue);

  /* Find which single queue it belongs to, knowing that it should be a sinkpad */

  for (tmp = mqueue->queues; tmp; tmp = g_list_next (tmp)) {

    sq = (GstSingleQueue *) tmp->data;

    if (sq->sinkpad == pad)

      break;