modules/demux/adaptive/plumbing/CommandsQueue.cpp

   1 /*
   2  * CommandsQueue.cpp
   3  *****************************************************************************
   4  * Copyright © 2015 VideoLAN and VLC Authors
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU Lesser General Public License as published
   8  * by the Free Software Foundation; either version 2.1 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14  * GNU Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General Public License
  17  * along with this program; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
  19  *****************************************************************************/
  20 #ifdef HAVE_CONFIG_H
  21 # include "config.h"
  22 #endif
  23
  24 #include "CommandsQueue.hpp"
  25 #include "FakeESOutID.hpp"
  26 #include "FakeESOut.hpp"
  27 #include <vlc_es_out.h>
  28 #include <vlc_block.h>
  29 #include <vlc_meta.h>
  30 #include <algorithm>
  31 #include <set>
  32
  33 using namespace adaptive;
  34
  35 enum
  36 {
  37     ES_OUT_PRIVATE_COMMAND_ADD = ES_OUT_PRIVATE_START,
  38     ES_OUT_PRIVATE_COMMAND_DEL,
  39     ES_OUT_PRIVATE_COMMAND_DESTROY,
  40     ES_OUT_PRIVATE_COMMAND_SEND,
  41     ES_OUT_PRIVATE_COMMAND_DISCONTINUITY
  42 };
  43
  44 AbstractCommand::AbstractCommand( int type_ )
  45 {
  46     type = type_;
  47 }
  48
  49 AbstractCommand::~AbstractCommand()
  50 {
  51
  52 }
  53
  54 vlc_tick_t AbstractCommand::getTime() const
  55 {
  56     return VLC_TICK_INVALID;
  57 }
  58
  59 int AbstractCommand::getType() const
  60 {
  61     return type;
  62 }
  63
  64 AbstractFakeEsCommand::AbstractFakeEsCommand( int type, AbstractFakeESOutID *p_es ) :
  65     AbstractCommand( type )
  66 {
  67     p_fakeid = p_es;
  68 }
  69
  70 EsOutSendCommand::EsOutSendCommand( AbstractFakeESOutID *p_es, block_t *p_block_ ) :
  71     AbstractFakeEsCommand( ES_OUT_PRIVATE_COMMAND_SEND, p_es )
  72 {
  73     p_block = p_block_;
  74 }
  75
  76 EsOutSendCommand::~EsOutSendCommand()
  77 {
  78     if( p_block )
  79         block_Release( p_block );
  80 }
  81
  82 void EsOutSendCommand::Execute()
  83 {
  84     p_fakeid->sendData( p_block );
  85     p_block = nullptr;
  86 }
  87
  88 vlc_tick_t EsOutSendCommand::getTime() const
  89 {
  90     if( likely(p_block) )
  91         return p_block->i_dts;
  92     else
  93         return AbstractCommand::getTime();
  94 }
  95
  96 const void * EsOutSendCommand::esIdentifier() const
  97 {
  98     return static_cast<const void *>(p_fakeid);
  99 }
 100
 101 EsOutDelCommand::EsOutDelCommand( AbstractFakeESOutID *p_es ) :
 102     AbstractFakeEsCommand( ES_OUT_PRIVATE_COMMAND_DEL, p_es )
 103 {
 104 }
 105
 106 void EsOutDelCommand::Execute( )
 107 {
 108     p_fakeid->release();
 109 }
 110
 111 EsOutAddCommand::EsOutAddCommand( AbstractFakeESOutID *p_es ) :
 112     AbstractFakeEsCommand( ES_OUT_PRIVATE_COMMAND_ADD, p_es )
 113 {
 114 }
 115
 116 EsOutAddCommand::~EsOutAddCommand()
 117 {
 118 }
 119
 120 void EsOutAddCommand::Execute( )
 121 {
 122     /* Create the real ES on the adaptive demux */
 123     p_fakeid->create();
 124 }
 125
 126 EsOutControlPCRCommand::EsOutControlPCRCommand( int group_, vlc_tick_t pcr_ ) :
 127     AbstractCommand( ES_OUT_SET_GROUP_PCR )
 128 {
 129     group = group_;
 130     pcr = pcr_;
 131     type = ES_OUT_SET_GROUP_PCR;
 132 }
 133
 134 void EsOutControlPCRCommand::Execute( )
 135 {
 136     // do nothing here
 137 }
 138
 139 vlc_tick_t EsOutControlPCRCommand::getTime() const
 140 {
 141     return pcr;
 142 }
 143
 144 EsOutDestroyCommand::EsOutDestroyCommand() :
 145     AbstractCommand( ES_OUT_PRIVATE_COMMAND_DESTROY )
 146 {
 147 }
 148
 149 void EsOutDestroyCommand::Execute( )
 150 {
 151 }
 152
 153 EsOutControlResetPCRCommand::EsOutControlResetPCRCommand() :
 154     AbstractCommand( ES_OUT_PRIVATE_COMMAND_DISCONTINUITY )
 155 {
 156 }
 157
 158 void EsOutControlResetPCRCommand::Execute( )
 159 {
 160 }
 161
 162 EsOutMetaCommand::EsOutMetaCommand( AbstractFakeEsOut *out,
 163                                     int i_group, vlc_meta_t *p_meta ) :
 164     AbstractCommand( ES_OUT_SET_GROUP_META )
 165 {
 166     group = i_group;
 167     this->out = out;
 168     this->p_meta = p_meta;
 169 }
 170
 171 EsOutMetaCommand::~EsOutMetaCommand()
 172 {
 173     if( p_meta )
 174         vlc_meta_Delete( p_meta );
 175 }
 176
 177 void EsOutMetaCommand::Execute()
 178 {
 179     out->sendMeta( group, p_meta );
 180 }
 181
 182 /*
 183  * Commands Default Factory
 184  */
 185
 186 EsOutSendCommand * CommandsFactory::createEsOutSendCommand( AbstractFakeESOutID *id, block_t *p_block ) const
 187 {
 188     return new (std::nothrow) EsOutSendCommand( id, p_block );
 189 }
 190
 191 EsOutDelCommand * CommandsFactory::createEsOutDelCommand( AbstractFakeESOutID *id ) const
 192 {
 193     return new (std::nothrow) EsOutDelCommand( id );
 194 }
 195
 196 EsOutAddCommand * CommandsFactory::createEsOutAddCommand( AbstractFakeESOutID *id ) const
 197 {
 198     return new (std::nothrow) EsOutAddCommand( id );
 199 }
 200
 201 EsOutControlPCRCommand * CommandsFactory::createEsOutControlPCRCommand( int group, vlc_tick_t pcr ) const
 202 {
 203     return new (std::nothrow) EsOutControlPCRCommand( group, pcr );
 204 }
 205
 206 EsOutDestroyCommand * CommandsFactory::createEsOutDestroyCommand() const
 207 {
 208     return new (std::nothrow) EsOutDestroyCommand();
 209 }
 210
 211 EsOutControlResetPCRCommand * CommandsFactory::creatEsOutControlResetPCRCommand() const
 212 {
 213     return new (std::nothrow) EsOutControlResetPCRCommand();
 214 }
 215
 216 EsOutMetaCommand * CommandsFactory::createEsOutMetaCommand( AbstractFakeEsOut *out, int group,
 217                                                             const vlc_meta_t *p_meta ) const
 218 {
 219     vlc_meta_t *p_dup = vlc_meta_New();
 220     if( p_dup )
 221     {
 222         vlc_meta_Merge( p_dup, p_meta );
 223         return new (std::nothrow) EsOutMetaCommand( out, group, p_dup );
 224     }
 225     return nullptr;
 226 }
 227
 228 /*
 229  * Commands Queue management
 230  */
 231 #if 0
 232 /* For queue printing/debugging */
 233 std::ostream& operator<<(std::ostream& ostr, const std::list<AbstractCommand *>& list)
 234 {
 235     for (auto &i : list) {
 236         ostr << "[" << i->getType() << "]" << SEC_FROM_VLC_TICK(i->getTime()) << " ";
 237     }
 238     return ostr;
 239 }
 240 #endif
 241
 242 CommandsQueue::CommandsQueue( CommandsFactory *f )
 243 {
 244     bufferinglevel = VLC_TICK_INVALID;
 245     pcr = VLC_TICK_INVALID;
 246     b_drop = false;
 247     b_draining = false;
 248     b_eof = false;
 249     commandsFactory = f;
 250     nextsequence = 0;
 251 }
 252
 253 CommandsQueue::~CommandsQueue()
 254 {
 255     Abort( false );
 256     delete commandsFactory;
 257 }
 258
 259 static bool compareCommands( const Queueentry &a, const Queueentry &b )
 260 {
 261     if(a.second->getTime() == b.second->getTime())
 262     {
 263         /* Reorder the initial clock PCR setting PCR0 DTS0 PCR0 DTS1 PCR1
 264            so it appears after the block, avoiding it not being output */
 265         if(a.second->getType() == ES_OUT_SET_GROUP_PCR &&
 266            b.second->getType() == ES_OUT_PRIVATE_COMMAND_SEND &&
 267            a.first < b.first)
 268             return false;
 269
 270         return a.first < b.first;
 271     }
 272     else if (a.second->getTime() == VLC_TICK_INVALID || b.second->getTime() == VLC_TICK_INVALID)
 273     {
 274         return a.first < b.first;
 275     }
 276     else
 277     {
 278         return a.second->getTime() < b.second->getTime();
 279     }
 280 }
 281
 282 void CommandsQueue::Schedule( AbstractCommand *command )
 283 {
 284     if( b_drop )
 285     {
 286         delete command;
 287     }
 288     else if( command->getType() == ES_OUT_SET_GROUP_PCR )
 289     {
 290         bufferinglevel = command->getTime();
 291         LockedCommit();
 292         commands.push_back( Queueentry(nextsequence++, command) );
 293     }
 294     else
 295     {
 296         incoming.push_back( Queueentry(nextsequence++, command) );
 297     }
 298 }
 299
 300 const CommandsFactory * CommandsQueue::factory() const
 301 {
 302     return commandsFactory;
 303 }
 304
 305 vlc_tick_t CommandsQueue::Process( vlc_tick_t barrier )
 306 {
 307     vlc_tick_t lastdts = barrier;
 308     std::set<const void *> disabled_esids;
 309     bool b_datasent = false;
 310
 311     /* We need to filter the current commands list
 312        We need to return on discontinuity or reset events if data was sent
 313        We must lookup every packet until end or PCR matching barrier,
 314        because packets of multiple stream can arrive delayed (with intermidiate pcr)
 315        ex: for a target time of 2, you must dequeue <= 2 until >= PCR2
 316        A0,A1,A2,B0,PCR0,B1,B2,PCR2,B3,A3,PCR3
 317     */
 318     std::list<Queueentry> output;
 319     std::list<Queueentry> in;
 320
 321
 322     in.splice( in.end(), commands );
 323
 324     while( !in.empty() )
 325     {
 326         Queueentry entry = in.front();
 327         AbstractCommand *command = entry.second;
 328
 329         if( command->getType() == ES_OUT_PRIVATE_COMMAND_DEL && b_datasent )
 330             break;
 331
 332         if( command->getType() == ES_OUT_PRIVATE_COMMAND_DISCONTINUITY && b_datasent )
 333             break;
 334
 335         if(command->getType() == ES_OUT_SET_GROUP_PCR && command->getTime() > barrier )
 336             break;
 337
 338         in.pop_front();
 339         b_datasent = true;
 340
 341         if( command->getType() == ES_OUT_PRIVATE_COMMAND_SEND )
 342         {
 343             EsOutSendCommand *sendcommand = dynamic_cast<EsOutSendCommand *>(command);
 344             /* We need a stream identifier to send NON DATED data following DATA for the same ES */
 345             const void *id = (sendcommand) ? sendcommand->esIdentifier() : 0;
 346
 347             /* Not for now */
 348             if( command->getTime() > barrier ) /* Not for now */
 349             {
 350                 /* ensure no more non dated for that ES is sent
 351                  * since we're sure that data is above barrier */
 352                 disabled_esids.insert( id );
 353                 commands.push_back( entry );
 354             }
 355             else if( command->getTime() == VLC_TICK_INVALID )
 356             {
 357                 if( disabled_esids.find( id ) == disabled_esids.end() )
 358                     output.push_back( entry );
 359                 else
 360                     commands.push_back( entry );
 361             }
 362             else /* Falls below barrier, send */
 363             {
 364                 output.push_back( entry );
 365             }
 366         }
 367         else output.push_back( entry ); /* will discard below */
 368     }
 369
 370     /* push remaining ones if broke above */
 371     commands.splice( commands.end(), in );
 372
 373     if(commands.empty() && b_draining)
 374         b_draining = false;
 375
 376     /* Now execute our selected commands */
 377     while( !output.empty() )
 378     {
 379         AbstractCommand *command = output.front().second;
 380         output.pop_front();
 381
 382         if( command->getType() == ES_OUT_PRIVATE_COMMAND_SEND )
 383         {
 384             vlc_tick_t dts = command->getTime();
 385             if( dts != VLC_TICK_INVALID )
 386                 lastdts = dts;
 387         }
 388
 389         command->Execute();
 390         delete command;
 391     }
 392     pcr = lastdts; /* Warn! no PCR update/lock release until execution */
 393
 394
 395     return lastdts;
 396 }
 397
 398 void CommandsQueue::LockedCommit()
 399 {
 400     /* reorder all blocks by time between 2 PCR and merge with main list */
 401     incoming.sort( compareCommands );
 402     commands.splice( commands.end(), incoming );
 403 }
 404
 405 void CommandsQueue::Commit()
 406 {
 407     LockedCommit();
 408 }
 409
 410 void CommandsQueue::Abort( bool b_reset )
 411 {
 412     commands.splice( commands.end(), incoming );
 413     while( !commands.empty() )
 414     {
 415         delete commands.front().second;
 416         commands.pop_front();
 417     }
 418
 419     if( b_reset )
 420     {
 421         bufferinglevel = VLC_TICK_INVALID;
 422         pcr = VLC_TICK_INVALID;
 423         b_draining = false;
 424         b_eof = false;
 425     }
 426 }
 427
 428 bool CommandsQueue::isEmpty() const
 429 {
 430     return commands.empty() && incoming.empty();
 431 }
 432
 433 void CommandsQueue::setDrop( bool b )
 434 {
 435     b_drop = b;
 436 }
 437
 438 void CommandsQueue::setDraining()
 439 {
 440     LockedSetDraining();
 441 }
 442
 443 bool CommandsQueue::isDraining() const
 444 {
 445     return b_draining;
 446 }
 447
 448 void CommandsQueue::setEOF( bool b )
 449 {
 450     b_eof = b;
 451     if( b_eof )
 452         LockedSetDraining();
 453     else
 454         b_draining = false;
 455 }
 456
 457 bool CommandsQueue::isEOF() const
 458 {
 459     return b_eof;
 460 }
 461
 462 vlc_tick_t CommandsQueue::getDemuxedAmount(vlc_tick_t from) const
 463 {
 464     if( bufferinglevel == VLC_TICK_INVALID || from > bufferinglevel )
 465         return 0;
 466     if( from > getFirstDTS() )
 467         return bufferinglevel - from;
 468     else
 469         return bufferinglevel - getFirstDTS();
 470 }
 471
 472 vlc_tick_t CommandsQueue::getBufferingLevel() const
 473 {
 474     return bufferinglevel;
 475 }
 476
 477 vlc_tick_t CommandsQueue::getFirstDTS() const
 478 {
 479     std::list<Queueentry>::const_iterator it;
 480     vlc_tick_t i_firstdts = pcr;
 481     for( it = commands.begin(); it != commands.end(); ++it )
 482     {
 483         const vlc_tick_t i_dts = (*it).second->getTime();
 484         if( i_dts != VLC_TICK_INVALID )
 485         {
 486             if( i_dts < i_firstdts || i_firstdts == VLC_TICK_INVALID )
 487                 i_firstdts = i_dts;
 488             break;
 489         }
 490     }
 491     return i_firstdts;
 492 }
 493
 494 void CommandsQueue::LockedSetDraining()
 495 {
 496     LockedCommit();
 497     b_draining = !commands.empty();
 498 }
 499
 500 vlc_tick_t CommandsQueue::getPCR() const
 501 {
 502     return pcr;
 503 }