subprojects/gst-plugins-good/ext/adaptivedemux2/gstadaptivedemuxutils.c

   1 /* GStreamer
   2  *
   3  * Copyright (C) 2014 Samsung Electronics. All rights reserved.
   4  *   Author: Thiago Santos <thiagoss@osg.samsung.com>
   5  * Copyright (C) 2021-2022 Jan Schmidt <jan@centricular.com>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25
  26 #include <stdio.h>
  27 #include <gst/gst.h>
  28 #include <gst/pbutils/pbutils.h>
  29
  30 #include "gstadaptivedemuxutils.h"
  31
  32 GST_DEBUG_CATEGORY_EXTERN (adaptivedemux2_debug);
  33 #define GST_CAT_DEFAULT adaptivedemux2_debug
  34
  35 struct _GstAdaptiveDemuxClock
  36 {
  37   gint ref_count;
  38
  39   GstClock *gst_clock;
  40   GstClockTimeDiff clock_offset;        /* offset between realtime_clock and UTC */
  41 };
  42
  43 struct _GstAdaptiveDemuxLoop
  44 {
  45   gint ref_count;
  46   GCond cond;
  47   GMutex lock;
  48
  49   GRecMutex context_lock;
  50
  51   GThread *thread;
  52   GMainLoop *loop;
  53   GMainContext *context;
  54
  55   gboolean stopped;
  56   gboolean paused;
  57 };
  58
  59 GstAdaptiveDemuxClock *
  60 gst_adaptive_demux_clock_new (void)
  61 {
  62   GstAdaptiveDemuxClock *clock = g_slice_new (GstAdaptiveDemuxClock);
  63   GstClockType clock_type = GST_CLOCK_TYPE_OTHER;
  64   GObjectClass *gobject_class;
  65
  66   g_atomic_int_set (&clock->ref_count, 1);
  67
  68   clock->gst_clock = gst_system_clock_obtain ();
  69   g_assert (clock->gst_clock != NULL);
  70
  71   gobject_class = G_OBJECT_GET_CLASS (clock->gst_clock);
  72   if (g_object_class_find_property (gobject_class, "clock-type")) {
  73     g_object_get (clock->gst_clock, "clock-type", &clock_type, NULL);
  74   } else {
  75     GST_WARNING ("System clock does not have clock-type property");
  76   }
  77
  78   if (clock_type == GST_CLOCK_TYPE_REALTIME) {
  79     clock->clock_offset = 0;
  80   } else {
  81     GDateTime *utc_now;
  82
  83     utc_now = g_date_time_new_now_utc ();
  84     gst_adaptive_demux_clock_set_utc_time (clock, utc_now);
  85     g_date_time_unref (utc_now);
  86   }
  87
  88   return clock;
  89 }
  90
  91 GstAdaptiveDemuxClock *
  92 gst_adaptive_demux_clock_ref (GstAdaptiveDemuxClock * clock)
  93 {
  94   g_return_val_if_fail (clock != NULL, NULL);
  95   g_atomic_int_inc (&clock->ref_count);
  96   return clock;
  97 }
  98
  99 void
 100 gst_adaptive_demux_clock_unref (GstAdaptiveDemuxClock * clock)
 101 {
 102   g_return_if_fail (clock != NULL);
 103   if (g_atomic_int_dec_and_test (&clock->ref_count)) {
 104     gst_object_unref (clock->gst_clock);
 105     g_slice_free (GstAdaptiveDemuxClock, clock);
 106   }
 107 }
 108
 109 GstClockTime
 110 gst_adaptive_demux_clock_get_time (GstAdaptiveDemuxClock * clock)
 111 {
 112   g_return_val_if_fail (clock != NULL, GST_CLOCK_TIME_NONE);
 113   return gst_clock_get_time (clock->gst_clock);
 114 }
 115
 116 GDateTime *
 117 gst_adaptive_demux_clock_get_now_utc (GstAdaptiveDemuxClock * clock)
 118 {
 119   GstClockTime rtc_now;
 120   GDateTime *unix_datetime;
 121   GDateTime *result_datetime;
 122   gint64 utc_now_in_us;
 123
 124   rtc_now = gst_clock_get_time (clock->gst_clock);
 125   utc_now_in_us = clock->clock_offset + GST_TIME_AS_USECONDS (rtc_now);
 126   unix_datetime =
 127       g_date_time_new_from_unix_utc (utc_now_in_us / G_TIME_SPAN_SECOND);
 128   result_datetime =
 129       g_date_time_add (unix_datetime, utc_now_in_us % G_TIME_SPAN_SECOND);
 130   g_date_time_unref (unix_datetime);
 131   return result_datetime;
 132 }
 133
 134 void
 135 gst_adaptive_demux_clock_set_utc_time (GstAdaptiveDemuxClock * clock,
 136     GDateTime * utc_now)
 137 {
 138   GstClockTime rtc_now = gst_clock_get_time (clock->gst_clock);
 139   GstClockTimeDiff clock_offset;
 140
 141   clock_offset =
 142       g_date_time_to_unix (utc_now) * G_TIME_SPAN_SECOND +
 143       g_date_time_get_microsecond (utc_now) - GST_TIME_AS_USECONDS (rtc_now);
 144
 145   GST_INFO ("Changing UTC clock offset to %" GST_STIME_FORMAT
 146       " was %" GST_STIME_FORMAT, GST_STIME_ARGS (clock_offset),
 147       GST_STIME_ARGS (clock->clock_offset));
 148
 149   clock->clock_offset = clock_offset;
 150 }
 151
 152 GstAdaptiveDemuxLoop *
 153 gst_adaptive_demux_loop_new (void)
 154 {
 155   GstAdaptiveDemuxLoop *loop = g_slice_new0 (GstAdaptiveDemuxLoop);
 156   g_atomic_int_set (&loop->ref_count, 1);
 157
 158   g_mutex_init (&loop->lock);
 159   g_rec_mutex_init (&loop->context_lock);
 160   g_cond_init (&loop->cond);
 161
 162   loop->stopped = TRUE;
 163   loop->paused = FALSE;
 164
 165   return loop;
 166 }
 167
 168 GstAdaptiveDemuxLoop *
 169 gst_adaptive_demux_loop_ref (GstAdaptiveDemuxLoop * loop)
 170 {
 171   g_return_val_if_fail (loop != NULL, NULL);
 172   g_atomic_int_inc (&loop->ref_count);
 173   return loop;
 174 }
 175
 176 void
 177 gst_adaptive_demux_loop_unref (GstAdaptiveDemuxLoop * loop)
 178 {
 179   g_return_if_fail (loop != NULL);
 180   if (g_atomic_int_dec_and_test (&loop->ref_count)) {
 181     gst_adaptive_demux_loop_stop (loop, TRUE);
 182
 183     g_mutex_clear (&loop->lock);
 184     g_rec_mutex_clear (&loop->context_lock);
 185     g_cond_clear (&loop->cond);
 186
 187     g_slice_free (GstAdaptiveDemuxLoop, loop);
 188   }
 189 }
 190
 191 static gpointer
 192 _gst_adaptive_demux_loop_thread (GstAdaptiveDemuxLoop * loop)
 193 {
 194   g_mutex_lock (&loop->lock);
 195
 196   loop->loop = g_main_loop_new (loop->context, FALSE);
 197
 198   while (!loop->stopped) {
 199     g_mutex_unlock (&loop->lock);
 200
 201     g_rec_mutex_lock (&loop->context_lock);
 202
 203     g_main_context_push_thread_default (loop->context);
 204     g_main_loop_run (loop->loop);
 205     g_main_context_pop_thread_default (loop->context);
 206
 207     g_rec_mutex_unlock (&loop->context_lock);
 208
 209     g_mutex_lock (&loop->lock);
 210     while (loop->paused)
 211       g_cond_wait (&loop->cond, &loop->lock);
 212   }
 213
 214   g_main_loop_unref (loop->loop);
 215   loop->loop = NULL;
 216
 217   g_cond_broadcast (&loop->cond);
 218   g_mutex_unlock (&loop->lock);
 219
 220   g_main_context_unref (loop->context);
 221   loop->context = NULL;
 222
 223   gst_adaptive_demux_loop_unref (loop);
 224
 225   return NULL;
 226 }
 227
 228 void
 229 gst_adaptive_demux_loop_start (GstAdaptiveDemuxLoop * loop)
 230 {
 231   g_mutex_lock (&loop->lock);
 232   if (loop->thread != NULL)
 233     goto done;                  /* Already running */
 234
 235   loop->stopped = FALSE;
 236   loop->context = g_main_context_new ();
 237
 238   loop->thread =
 239       g_thread_new ("AdaptiveDemux",
 240       (GThreadFunc) _gst_adaptive_demux_loop_thread,
 241       gst_adaptive_demux_loop_ref (loop));
 242
 243 done:
 244   g_mutex_unlock (&loop->lock);
 245 }
 246
 247 static gboolean
 248 do_quit_cb (GstAdaptiveDemuxLoop * loop)
 249 {
 250   g_main_loop_quit (loop->loop);
 251   return G_SOURCE_REMOVE;
 252 }
 253
 254 void
 255 gst_adaptive_demux_loop_stop (GstAdaptiveDemuxLoop * loop, gboolean wait)
 256 {
 257   g_mutex_lock (&loop->lock);
 258
 259   if (!loop->stopped) {
 260     loop->stopped = TRUE;
 261
 262     GSource *s = g_idle_source_new ();
 263     g_source_set_callback (s, (GSourceFunc) do_quit_cb,
 264         gst_adaptive_demux_loop_ref (loop),
 265         (GDestroyNotify) gst_adaptive_demux_loop_unref);
 266     g_source_attach (s, loop->context);
 267     g_source_unref (s);
 268
 269     if (wait) {
 270       while (loop->loop != NULL)
 271         g_cond_wait (&loop->cond, &loop->lock);
 272     }
 273
 274     if (loop->thread != NULL) {
 275       g_thread_unref (loop->thread);
 276       loop->thread = NULL;
 277     }
 278   }
 279
 280   g_mutex_unlock (&loop->lock);
 281 }
 282
 283 gboolean
 284 gst_adaptive_demux_loop_pause_and_lock (GstAdaptiveDemuxLoop * loop)
 285 {
 286   /* Try and acquire the context lock directly. This will succeed
 287    * if called when the loop is not running, and we can avoid
 288    * adding an unnecessary extra idle source to quit the loop. */
 289   if (!g_rec_mutex_trylock (&loop->context_lock)) {
 290     g_mutex_lock (&loop->lock);
 291
 292     if (loop->stopped) {
 293       g_mutex_unlock (&loop->lock);
 294       return FALSE;
 295     }
 296
 297     loop->paused = TRUE;
 298
 299     {
 300       GSource *s = g_idle_source_new ();
 301       g_source_set_callback (s, (GSourceFunc) do_quit_cb,
 302           gst_adaptive_demux_loop_ref (loop),
 303           (GDestroyNotify) gst_adaptive_demux_loop_unref);
 304       g_source_attach (s, loop->context);
 305       g_source_unref (s);
 306     }
 307
 308     g_mutex_unlock (&loop->lock);
 309
 310     g_rec_mutex_lock (&loop->context_lock);
 311   }
 312   g_main_context_push_thread_default (loop->context);
 313
 314   return TRUE;
 315 }
 316
 317 gboolean
 318 gst_adaptive_demux_loop_unlock_and_unpause (GstAdaptiveDemuxLoop * loop)
 319 {
 320   g_main_context_pop_thread_default (loop->context);
 321   g_rec_mutex_unlock (&loop->context_lock);
 322
 323   g_mutex_lock (&loop->lock);
 324   loop->paused = FALSE;
 325
 326   if (loop->stopped) {
 327     g_mutex_unlock (&loop->lock);
 328     return FALSE;
 329   }
 330
 331   /* Wake up the loop to run again */
 332   g_cond_broadcast (&loop->cond);
 333   g_mutex_unlock (&loop->lock);
 334
 335   return TRUE;
 336 }
 337
 338 guint
 339 gst_adaptive_demux_loop_call (GstAdaptiveDemuxLoop * loop, GSourceFunc func,
 340     gpointer data, GDestroyNotify notify)
 341 {
 342   guint ret = 0;
 343
 344   g_mutex_lock (&loop->lock);
 345   if (loop->context) {
 346     GSource *s = g_idle_source_new ();
 347     g_source_set_callback (s, func, data, notify);
 348     ret = g_source_attach (s, loop->context);
 349     g_source_unref (s);
 350   } else if (notify != NULL) {
 351     notify (data);
 352   }
 353
 354   g_mutex_unlock (&loop->lock);
 355
 356   return ret;
 357 }
 358
 359 guint
 360 gst_adaptive_demux_loop_call_delayed (GstAdaptiveDemuxLoop * loop,
 361     GstClockTime delay, GSourceFunc func, gpointer data, GDestroyNotify notify)
 362 {
 363   guint ret = 0;
 364
 365   g_mutex_lock (&loop->lock);
 366   if (loop->context) {
 367     GSource *s = g_timeout_source_new (GST_TIME_AS_MSECONDS (delay));
 368     g_source_set_callback (s, func, data, notify);
 369     ret = g_source_attach (s, loop->context);
 370     g_source_unref (s);
 371   } else if (notify != NULL) {
 372     notify (data);
 373   }
 374
 375   g_mutex_unlock (&loop->lock);
 376
 377   return ret;
 378 }
 379
 380 void
 381 gst_adaptive_demux_loop_cancel_call (GstAdaptiveDemuxLoop * loop, guint cb_id)
 382 {
 383   GSource *s;
 384
 385   g_mutex_lock (&loop->lock);
 386   s = g_main_context_find_source_by_id (loop->context, cb_id);
 387   if (s)
 388     g_source_destroy (s);
 389   g_mutex_unlock (&loop->lock);
 390 }
 391
 392 struct Rfc5322TimeZone
 393 {
 394   const gchar *name;
 395   gfloat tzoffset;
 396 };
 397
 398 /*
 399  Parse an RFC5322 (section 3.3) date-time from the Date: field in the
 400  HTTP response.
 401  See https://tools.ietf.org/html/rfc5322#section-3.3
 402 */
 403 GstDateTime *
 404 gst_adaptive_demux_util_parse_http_head_date (const gchar * http_date)
 405 {
 406   static const gchar *months[] = { NULL, "Jan", "Feb", "Mar", "Apr",
 407     "May", "Jun", "Jul", "Aug",
 408     "Sep", "Oct", "Nov", "Dec", NULL
 409   };
 410   static const struct Rfc5322TimeZone timezones[] = {
 411     {"Z", 0},
 412     {"UT", 0},
 413     {"GMT", 0},
 414     {"BST", 1},
 415     {"EST", -5},
 416     {"EDT", -4},
 417     {"CST", -6},
 418     {"CDT", -5},
 419     {"MST", -7},
 420     {"MDT", -6},
 421     {"PST", -8},
 422     {"PDT", -7},
 423     {NULL, 0}
 424   };
 425   gint ret;
 426   const gchar *pos;
 427   gint year = -1, month = -1, day = -1, hour = -1, minute = -1, second = -1;
 428   gchar zone[6];
 429   gchar monthstr[4];
 430   gfloat tzoffset = 0;
 431   gboolean parsed_tz = FALSE;
 432
 433   g_return_val_if_fail (http_date != NULL, NULL);
 434
 435   /* skip optional text version of day of the week */
 436   pos = strchr (http_date, ',');
 437   if (pos)
 438     pos++;
 439   else
 440     pos = http_date;
 441
 442   ret =
 443       sscanf (pos, "%02d %3s %04d %02d:%02d:%02d %5s", &day, monthstr, &year,
 444       &hour, &minute, &second, zone);
 445
 446   if (ret == 7) {
 447     gchar *z = zone;
 448     gint i;
 449
 450     for (i = 1; months[i]; ++i) {
 451       if (g_ascii_strncasecmp (months[i], monthstr, strlen (months[i])) == 0) {
 452         month = i;
 453         break;
 454       }
 455     }
 456     for (i = 0; timezones[i].name && !parsed_tz; ++i) {
 457       if (g_ascii_strncasecmp (timezones[i].name, z,
 458               strlen (timezones[i].name)) == 0) {
 459         tzoffset = timezones[i].tzoffset;
 460         parsed_tz = TRUE;
 461       }
 462     }
 463     if (!parsed_tz) {
 464       gint hh, mm;
 465       gboolean neg = FALSE;
 466       /* check if it is in the form +-HHMM */
 467       if (*z == '+' || *z == '-') {
 468         if (*z == '+')
 469           ++z;
 470         else if (*z == '-') {
 471           ++z;
 472           neg = TRUE;
 473         }
 474         ret = sscanf (z, "%02d%02d", &hh, &mm);
 475         if (ret == 2) {
 476           tzoffset = hh;
 477           tzoffset += mm / 60.0;
 478           if (neg)
 479             tzoffset = -tzoffset;
 480           parsed_tz = TRUE;
 481         }
 482       }
 483     }
 484     /* Accept year in both 2 digit or 4 digit format */
 485     if (year < 100)
 486       year += 2000;
 487   }
 488
 489   if (month < 1 || !parsed_tz)
 490     return NULL;
 491
 492   return gst_date_time_new (tzoffset, year, month, day, hour, minute, second);
 493 }
 494
 495 typedef struct
 496 {
 497   gboolean delivered;
 498   GstEvent *event;
 499 } PadEvent;
 500
 501 void
 502 gst_event_store_init (GstEventStore * store)
 503 {
 504   store->events = g_array_sized_new (FALSE, TRUE, sizeof (PadEvent), 16);
 505   store->events_pending = FALSE;
 506 }
 507
 508 void
 509 gst_event_store_flush (GstEventStore * store)
 510 {
 511   guint i, len;
 512   GArray *events = store->events;
 513
 514   len = events->len;
 515   for (i = 0; i < len; i++) {
 516     PadEvent *ev = &g_array_index (events, PadEvent, i);
 517     GstEvent *event = ev->event;
 518
 519     ev->event = NULL;
 520
 521     gst_event_unref (event);
 522   }
 523   g_array_set_size (events, 0);
 524
 525   store->events_pending = FALSE;
 526 }
 527
 528 void
 529 gst_event_store_deinit (GstEventStore * store)
 530 {
 531   gst_event_store_flush (store);
 532   g_array_free (store->events, TRUE);
 533 }
 534
 535 void
 536 gst_event_store_insert_event (GstEventStore * store, GstEvent * event,
 537     gboolean delivered)
 538 {
 539   guint i, len;
 540   GstEventType type;
 541   GArray *events;
 542   GQuark name_id = 0;
 543   gboolean insert = TRUE;
 544
 545   type = GST_EVENT_TYPE (event);
 546
 547   if (type & GST_EVENT_TYPE_STICKY_MULTI)
 548     name_id = gst_structure_get_name_id (gst_event_get_structure (event));
 549
 550   events = store->events;
 551
 552   len = events->len;
 553   for (i = 0; i < len; i++) {
 554     PadEvent *ev = &g_array_index (events, PadEvent, i);
 555
 556     if (ev->event == NULL)
 557       continue;
 558
 559     if (type == GST_EVENT_TYPE (ev->event)) {
 560       /* matching types, check matching name if needed */
 561       if (name_id && !gst_event_has_name_id (ev->event, name_id))
 562         continue;
 563
 564       /* overwrite if different */
 565       if (gst_event_replace (&ev->event, event)) {
 566         ev->delivered = delivered;
 567         /* If the event was not delivered, mark that we have a pending
 568          * undelivered event */
 569         if (!delivered)
 570           store->events_pending = TRUE;
 571       }
 572
 573       insert = FALSE;
 574       break;
 575     }
 576
 577     if (type < GST_EVENT_TYPE (ev->event) || (type != GST_EVENT_TYPE (ev->event)
 578             && GST_EVENT_TYPE (ev->event) == GST_EVENT_EOS)) {
 579       /* STREAM_START, CAPS and SEGMENT must be delivered in this order. By
 580        * storing the sticky ordered we can check that this is respected. */
 581       if (G_UNLIKELY (GST_EVENT_TYPE (ev->event) <= GST_EVENT_SEGMENT
 582               || GST_EVENT_TYPE (ev->event) == GST_EVENT_EOS))
 583         g_warning (G_STRLOC
 584             ":%s:<store %p> Sticky event misordering, got '%s' before '%s'",
 585             G_STRFUNC, store,
 586             gst_event_type_get_name (GST_EVENT_TYPE (ev->event)),
 587             gst_event_type_get_name (type));
 588       break;
 589     }
 590   }
 591   if (insert) {
 592     PadEvent ev;
 593     ev.event = gst_event_ref (event);
 594     ev.delivered = delivered;
 595     g_array_insert_val (events, i, ev);
 596
 597     /* If the event was not delivered, mark that we have a pending
 598      * undelivered event */
 599     if (!delivered)
 600       store->events_pending = TRUE;
 601     GST_LOG ("store %p stored sticky event %s", store,
 602         GST_EVENT_TYPE_NAME (event));
 603   }
 604 }
 605
 606 /* Find the first non-pending event and return a ref to it, owned by the caller */
 607 GstEvent *
 608 gst_event_store_get_next_pending (GstEventStore * store)
 609 {
 610   GArray *events;
 611   guint i, len;
 612
 613   if (!store->events_pending)
 614     return NULL;
 615
 616   events = store->events;
 617   len = events->len;
 618   for (i = 0; i < len; i++) {
 619     PadEvent *ev = &g_array_index (events, PadEvent, i);
 620
 621     if (ev->event == NULL || ev->delivered)
 622       continue;
 623
 624     /* Found an undelivered event, return it. The caller will mark it
 625      * as delivered once it has done so successfully by calling
 626      * gst_event_store_mark_delivered() */
 627     return gst_event_ref (ev->event);
 628   }
 629
 630   store->events_pending = FALSE;
 631   return NULL;
 632 }
 633
 634 void
 635 gst_event_store_mark_delivered (GstEventStore * store, GstEvent * event)
 636 {
 637   gboolean events_pending = FALSE;
 638   GArray *events;
 639   guint i, len;
 640
 641   events = store->events;
 642   len = events->len;
 643   for (i = 0; i < len; i++) {
 644     PadEvent *ev = &g_array_index (events, PadEvent, i);
 645
 646     if (ev->event == NULL)
 647       continue;
 648
 649     /* Check if there are any pending events other than
 650      * the passed one, so we can update the events_pending
 651      * flag at the end */
 652     if (ev->event != event && !ev->delivered) {
 653       events_pending = TRUE;
 654       continue;
 655     }
 656
 657     ev->delivered = TRUE;
 658   }
 659
 660   store->events_pending = events_pending;
 661 }
 662
 663 void
 664 gst_event_store_mark_all_undelivered (GstEventStore * store)
 665 {
 666   gboolean events_pending = FALSE;
 667   GArray *events;
 668   guint i, len;
 669
 670   events = store->events;
 671   len = events->len;
 672   for (i = 0; i < len; i++) {
 673     PadEvent *ev = &g_array_index (events, PadEvent, i);
 674
 675     if (ev->event == NULL)
 676       continue;
 677
 678     ev->delivered = FALSE;
 679     events_pending = TRUE;
 680   }
 681
 682   /* Only set the flag if there was at least
 683    * one sticky event in the store */
 684   store->events_pending = events_pending;
 685 }