dom/media/Pacer.h

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim:set ts=2 sw=2 sts=2 et cindent: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
   5  * You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "MediaEventSource.h"
   8 #include "MediaTimer.h"
   9 #include "mozilla/TaskQueue.h"
  10 #include "nsDeque.h"
  11
  12 #ifndef DOM_MEDIA_PACER_H_
  13 #  define DOM_MEDIA_PACER_H_
  14
  15 namespace mozilla {
  16
  17 /**
  18  * Pacer<T> takes a queue of Ts tied to timestamps, and emits PacedItemEvents
  19  * for every T at its corresponding timestamp.
  20  *
  21  * The queue is ordered. Enqueing an item at time t will drop all items at times
  22  * later than T. This is because of how video sources work (some send out frames
  23  * in the future, some don't), and to allow swapping one source for another.
  24  *
  25  * It supports a duplication interval. If there is no new item enqueued within
  26  * the duplication interval since the last enqueued item, the last enqueud item
  27  * is emitted again.
  28  */
  29 template <typename T>
  30 class Pacer {
  31  public:
  32   NS_INLINE_DECL_THREADSAFE_REFCOUNTING(Pacer)
  33
  34   Pacer(RefPtr<TaskQueue> aTaskQueue, TimeDuration aDuplicationInterval)
  35       : mTaskQueue(std::move(aTaskQueue)),
  36         mDuplicationInterval(aDuplicationInterval),
  37         mTimer(MakeAndAddRef<MediaTimer>()) {}
  38
  39   /**
  40    * Enqueues an item and schedules a timer to pass it on to PacedItemEvent() at
  41    * t=aTime. Already queued items with t>=aTime will be dropped.
  42    */
  43   void Enqueue(T aItem, TimeStamp aTime) {
  44     MOZ_ALWAYS_SUCCEEDS(mTaskQueue->Dispatch(NS_NewRunnableFunction(
  45         __func__,
  46         [this, self = RefPtr<Pacer>(this), aItem = std::move(aItem), aTime] {
  47           MOZ_DIAGNOSTIC_ASSERT(!mIsShutdown);
  48           while (const auto* item = mQueue.Peek()) {
  49             if (item->mTime < aTime) {
  50               break;
  51             }
  52             RefPtr<QueueItem> dropping = mQueue.Pop();
  53           }
  54           mQueue.Push(MakeAndAddRef<QueueItem>(std::move(aItem), aTime));
  55           EnsureTimerScheduled(aTime);
  56         })));
  57   }
  58
  59   RefPtr<GenericPromise> Shutdown() {
  60     return InvokeAsync(
  61         mTaskQueue, __func__, [this, self = RefPtr<Pacer>(this)] {
  62           mIsShutdown = true;
  63           mTimer->Cancel();
  64           mQueue.Erase();
  65           mCurrentTimerTarget = Nothing();
  66           return GenericPromise::CreateAndResolve(true, "Pacer::Shutdown");
  67         });
  68   }
  69
  70   MediaEventSourceExc<T, TimeStamp>& PacedItemEvent() {
  71     return mPacedItemEvent;
  72   }
  73
  74  protected:
  75   ~Pacer() = default;
  76
  77   void EnsureTimerScheduled(TimeStamp aTime) {
  78     if (mCurrentTimerTarget && *mCurrentTimerTarget <= aTime) {
  79       return;
  80     }
  81
  82     if (mCurrentTimerTarget) {
  83       mTimer->Cancel();
  84       mCurrentTimerTarget = Nothing();
  85     }
  86
  87     mTimer->WaitUntil(aTime, __func__)
  88         ->Then(
  89             mTaskQueue, __func__,
  90             [this, self = RefPtr<Pacer>(this)] { OnTimerTick(); },
  91             [] {
  92               // Timer was rejected. This is fine.
  93             });
  94     mCurrentTimerTarget = Some(aTime);
  95   }
  96
  97   void OnTimerTick() {
  98     MOZ_ASSERT(mTaskQueue->IsOnCurrentThread());
  99
 100     mCurrentTimerTarget = Nothing();
 101
 102     while (RefPtr<QueueItem> item = mQueue.PopFront()) {
 103       auto now = TimeStamp::Now();
 104
 105       if (item->mTime <= now) {
 106         // It's time to process this item.
 107         if (const auto& next = mQueue.PeekFront();
 108             !next || next->mTime > (item->mTime + mDuplicationInterval)) {
 109           // No future frame within the duplication interval exists. Schedule
 110           // a copy.
 111           mQueue.PushFront(MakeAndAddRef<QueueItem>(
 112               item->mItem, item->mTime + mDuplicationInterval));
 113         }
 114         mPacedItemEvent.Notify(std::move(item->mItem), item->mTime);
 115         continue;
 116       }
 117
 118       // This item is in the future. Put it back.
 119       mQueue.PushFront(item.forget());
 120       break;
 121     }
 122
 123     if (const auto& next = mQueue.PeekFront(); next) {
 124       // The queue is not empty. Schedule the timer.
 125       EnsureTimerScheduled(next->mTime);
 126     }
 127   }
 128
 129  public:
 130   const RefPtr<TaskQueue> mTaskQueue;
 131   const TimeDuration mDuplicationInterval;
 132
 133  protected:
 134   struct QueueItem {
 135     NS_INLINE_DECL_THREADSAFE_REFCOUNTING(QueueItem)
 136
 137     QueueItem(T aItem, TimeStamp aTime)
 138         : mItem(std::forward<T>(aItem)), mTime(aTime) {}
 139
 140     T mItem;
 141     TimeStamp mTime;
 142
 143    private:
 144     ~QueueItem() = default;
 145   };
 146
 147   // Accessed on mTaskQueue.
 148   nsRefPtrDeque<QueueItem> mQueue;
 149
 150   // Accessed on mTaskQueue.
 151   RefPtr<MediaTimer> mTimer;
 152
 153   // Accessed on mTaskQueue.
 154   Maybe<TimeStamp> mCurrentTimerTarget;
 155
 156   // Accessed on mTaskQueue.
 157   bool mIsShutdown = false;
 158
 159   MediaEventProducerExc<T, TimeStamp> mPacedItemEvent;
 160 };
 161
 162 }  // namespace mozilla
 163
 164 #endif