xbmc/cores/VideoPlayer/DVDMessageQueue.cpp

   1 /*
   2  *  Copyright (C) 2005-2018 Team Kodi
   3  *  This file is part of Kodi - https://kodi.tv
   4  *
   5  *  SPDX-License-Identifier: GPL-2.0-or-later
   6  *  See LICENSES/README.md for more information.
   7  */
   8
   9 #include "DVDMessageQueue.h"
  10
  11 #include "cores/VideoPlayer/Interface/DemuxPacket.h"
  12 #include "cores/VideoPlayer/Interface/TimingConstants.h"
  13 #include "utils/log.h"
  14
  15 #include <math.h>
  16 #include <mutex>
  17
  18 using namespace std::chrono_literals;
  19
  20 CDVDMessageQueue::CDVDMessageQueue(const std::string &owner) : m_hEvent(true), m_owner(owner)
  21 {
  22   m_iDataSize     = 0;
  23   m_bInitialized = false;
  24
  25   m_TimeBack = DVD_NOPTS_VALUE;
  26   m_TimeFront = DVD_NOPTS_VALUE;
  27   m_TimeSize = 1.0 / 4.0; /* 4 seconds */
  28   m_iMaxDataSize = 0;
  29 }
  30
  31 CDVDMessageQueue::~CDVDMessageQueue()
  32 {
  33   // remove all remaining messages
  34   Flush(CDVDMsg::NONE);
  35 }
  36
  37 void CDVDMessageQueue::Init()
  38 {
  39   m_iDataSize = 0;
  40   m_bAbortRequest = false;
  41   m_bInitialized = true;
  42   m_TimeBack = DVD_NOPTS_VALUE;
  43   m_TimeFront = DVD_NOPTS_VALUE;
  44   m_drain = false;
  45 }
  46
  47 void CDVDMessageQueue::Flush(CDVDMsg::Message type)
  48 {
  49   std::unique_lock<CCriticalSection> lock(m_section);
  50
  51   m_messages.remove_if([type](const DVDMessageListItem &item){
  52     return type == CDVDMsg::NONE || item.message->IsType(type);
  53   });
  54
  55   m_prioMessages.remove_if([type](const DVDMessageListItem &item){
  56     return type == CDVDMsg::NONE || item.message->IsType(type);
  57   });
  58
  59   if (type == CDVDMsg::DEMUXER_PACKET ||  type == CDVDMsg::NONE)
  60   {
  61     m_iDataSize = 0;
  62     m_TimeBack = DVD_NOPTS_VALUE;
  63     m_TimeFront = DVD_NOPTS_VALUE;
  64   }
  65 }
  66
  67 void CDVDMessageQueue::Abort()
  68 {
  69   std::unique_lock<CCriticalSection> lock(m_section);
  70
  71   m_bAbortRequest = true;
  72
  73   // inform waiter for abort action
  74   m_hEvent.Set();
  75 }
  76
  77 void CDVDMessageQueue::End()
  78 {
  79   std::unique_lock<CCriticalSection> lock(m_section);
  80
  81   Flush(CDVDMsg::NONE);
  82
  83   m_bInitialized = false;
  84   m_iDataSize = 0;
  85   m_bAbortRequest = false;
  86 }
  87
  88 MsgQueueReturnCode CDVDMessageQueue::Put(const std::shared_ptr<CDVDMsg>& pMsg, int priority)
  89 {
  90   return Put(pMsg, priority, true);
  91 }
  92
  93 MsgQueueReturnCode CDVDMessageQueue::PutBack(const std::shared_ptr<CDVDMsg>& pMsg, int priority)
  94 {
  95   return Put(pMsg, priority, false);
  96 }
  97
  98 MsgQueueReturnCode CDVDMessageQueue::Put(const std::shared_ptr<CDVDMsg>& pMsg,
  99                                          int priority,
 100                                          bool front)
 101 {
 102   std::unique_lock<CCriticalSection> lock(m_section);
 103
 104   if (!m_bInitialized)
 105   {
 106     CLog::Log(LOGWARNING, "CDVDMessageQueue({})::Put MSGQ_NOT_INITIALIZED", m_owner);
 107     return MSGQ_NOT_INITIALIZED;
 108   }
 109   if (!pMsg)
 110   {
 111     CLog::Log(LOGFATAL, "CDVDMessageQueue({})::Put MSGQ_INVALID_MSG", m_owner);
 112     return MSGQ_INVALID_MSG;
 113   }
 114
 115   if (priority > 0)
 116   {
 117     int prio = priority;
 118     if (!front)
 119       prio++;
 120
 121     auto it = std::find_if(m_prioMessages.begin(), m_prioMessages.end(),
 122                            [prio](const DVDMessageListItem &item){
 123                              return prio <= item.priority;
 124                            });
 125     m_prioMessages.emplace(it, pMsg, priority);
 126   }
 127   else
 128   {
 129     if (m_messages.empty())
 130     {
 131       m_iDataSize = 0;
 132       m_TimeBack = DVD_NOPTS_VALUE;
 133       m_TimeFront = DVD_NOPTS_VALUE;
 134     }
 135
 136     if (front)
 137       m_messages.emplace_front(pMsg, priority);
 138     else
 139       m_messages.emplace_back(pMsg, priority);
 140   }
 141
 142   if (pMsg->IsType(CDVDMsg::DEMUXER_PACKET) && priority == 0)
 143   {
 144     DemuxPacket* packet = static_cast<CDVDMsgDemuxerPacket*>(pMsg.get())->GetPacket();
 145     if (packet)
 146     {
 147       m_iDataSize += packet->iSize;
 148       if (front)
 149         UpdateTimeFront();
 150       else
 151         UpdateTimeBack();
 152     }
 153   }
 154
 155   // inform waiter for new packet
 156   m_hEvent.Set();
 157
 158   return MSGQ_OK;
 159 }
 160
 161 MsgQueueReturnCode CDVDMessageQueue::Get(std::shared_ptr<CDVDMsg>& pMsg,
 162                                          std::chrono::milliseconds timeout,
 163                                          int& priority)
 164 {
 165   std::unique_lock<CCriticalSection> lock(m_section);
 166
 167   int ret = 0;
 168
 169   if (!m_bInitialized)
 170   {
 171     CLog::Log(LOGFATAL, "CDVDMessageQueue({})::Get MSGQ_NOT_INITIALIZED", m_owner);
 172     return MSGQ_NOT_INITIALIZED;
 173   }
 174
 175   while (!m_bAbortRequest)
 176   {
 177     std::list<DVDMessageListItem> &msgs = (priority > 0 || !m_prioMessages.empty()) ? m_prioMessages : m_messages;
 178
 179     if (!msgs.empty() && (msgs.back().priority >= priority || m_drain))
 180     {
 181       DVDMessageListItem& item(msgs.back());
 182       priority = item.priority;
 183
 184       if (item.message->IsType(CDVDMsg::DEMUXER_PACKET) && item.priority == 0)
 185       {
 186         DemuxPacket* packet =
 187             std::static_pointer_cast<CDVDMsgDemuxerPacket>(item.message)->GetPacket();
 188         if (packet)
 189         {
 190           m_iDataSize -= packet->iSize;
 191         }
 192       }
 193
 194       pMsg = std::move(item.message);
 195       msgs.pop_back();
 196       UpdateTimeBack();
 197       ret = MSGQ_OK;
 198       break;
 199     }
 200     else if (timeout == 0ms)
 201     {
 202       ret = MSGQ_TIMEOUT;
 203       break;
 204     }
 205     else
 206     {
 207       m_hEvent.Reset();
 208       lock.unlock();
 209
 210       // wait for a new message
 211       if (!m_hEvent.Wait(timeout))
 212         return MSGQ_TIMEOUT;
 213
 214       lock.lock();
 215     }
 216   }
 217
 218   if (m_bAbortRequest)
 219     return MSGQ_ABORT;
 220
 221   return (MsgQueueReturnCode)ret;
 222 }
 223
 224 void CDVDMessageQueue::UpdateTimeFront()
 225 {
 226   if (!m_messages.empty())
 227   {
 228     auto &item = m_messages.front();
 229     if (item.message->IsType(CDVDMsg::DEMUXER_PACKET))
 230     {
 231       DemuxPacket* packet =
 232           std::static_pointer_cast<CDVDMsgDemuxerPacket>(item.message)->GetPacket();
 233       if (packet)
 234       {
 235         if (packet->dts != DVD_NOPTS_VALUE)
 236           m_TimeFront = packet->dts;
 237         else if (packet->pts != DVD_NOPTS_VALUE)
 238           m_TimeFront = packet->pts;
 239
 240         if (m_TimeBack == DVD_NOPTS_VALUE)
 241           m_TimeBack = m_TimeFront;
 242       }
 243     }
 244   }
 245 }
 246
 247 void CDVDMessageQueue::UpdateTimeBack()
 248 {
 249   if (!m_messages.empty())
 250   {
 251     auto &item = m_messages.back();
 252     if (item.message->IsType(CDVDMsg::DEMUXER_PACKET))
 253     {
 254       DemuxPacket* packet =
 255           std::static_pointer_cast<CDVDMsgDemuxerPacket>(item.message)->GetPacket();
 256       if (packet)
 257       {
 258         if (packet->dts != DVD_NOPTS_VALUE)
 259           m_TimeBack = packet->dts;
 260         else if (packet->pts != DVD_NOPTS_VALUE)
 261           m_TimeBack = packet->pts;
 262
 263         if (m_TimeFront == DVD_NOPTS_VALUE)
 264           m_TimeFront = m_TimeBack;
 265       }
 266     }
 267   }
 268 }
 269
 270 unsigned CDVDMessageQueue::GetPacketCount(CDVDMsg::Message type)
 271 {
 272   std::unique_lock<CCriticalSection> lock(m_section);
 273
 274   if (!m_bInitialized)
 275     return 0;
 276
 277   unsigned count = 0;
 278   for (const auto &item : m_messages)
 279   {
 280     if(item.message->IsType(type))
 281       count++;
 282   }
 283   for (const auto &item : m_prioMessages)
 284   {
 285     if(item.message->IsType(type))
 286       count++;
 287   }
 288
 289   return count;
 290 }
 291
 292 void CDVDMessageQueue::WaitUntilEmpty()
 293 {
 294   {
 295     std::unique_lock<CCriticalSection> lock(m_section);
 296     m_drain = true;
 297   }
 298
 299   CLog::Log(LOGINFO, "CDVDMessageQueue({})::WaitUntilEmpty", m_owner);
 300   auto msg = std::make_shared<CDVDMsgGeneralSynchronize>(40s, SYNCSOURCE_ANY);
 301   Put(msg);
 302   msg->Wait(m_bAbortRequest, 0);
 303
 304   {
 305     std::unique_lock<CCriticalSection> lock(m_section);
 306     m_drain = false;
 307   }
 308 }
 309
 310 int CDVDMessageQueue::GetLevel() const
 311 {
 312   std::unique_lock<CCriticalSection> lock(m_section);
 313
 314   if (m_iDataSize > m_iMaxDataSize)
 315     return 100;
 316   if (m_iDataSize == 0)
 317     return 0;
 318
 319   if (IsDataBased())
 320   {
 321     return std::min(100, 100 * m_iDataSize / m_iMaxDataSize);
 322   }
 323
 324   int level = std::min(100.0, ceil(100.0 * m_TimeSize * (m_TimeFront - m_TimeBack) / DVD_TIME_BASE ));
 325
 326   // if we added lots of packets with NOPTS, make sure that the queue is not signalled empty
 327   if (level == 0 && m_iDataSize != 0)
 328   {
 329     CLog::Log(LOGDEBUG, "CDVDMessageQueue::GetLevel() - can't determine level");
 330     return 1;
 331   }
 332
 333   return level;
 334 }
 335
 336 double CDVDMessageQueue::GetTimeSize() const
 337 {
 338   std::unique_lock<CCriticalSection> lock(m_section);
 339
 340   if (IsDataBased())
 341     return 0.0;
 342   else
 343     return (m_TimeFront - m_TimeBack) / DVD_TIME_BASE;
 344 }
 345
 346 bool CDVDMessageQueue::IsDataBased() const
 347 {
 348   return (m_TimeBack == DVD_NOPTS_VALUE  ||
 349           m_TimeFront == DVD_NOPTS_VALUE ||
 350           m_TimeFront <= m_TimeBack);
 351 }