modules/demux/adaptive/Streams.cpp

   1 /*
   2  * Streams.cpp
   3  *****************************************************************************
   4  * Copyright (C) 2014 - 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
  21 #ifdef HAVE_CONFIG_H
  22 # include "config.h"
  23 #endif
  24
  25 #include "Streams.hpp"
  26 #include "logic/AbstractAdaptationLogic.h"
  27 #include "http/HTTPConnection.hpp"
  28 #include "http/HTTPConnectionManager.h"
  29 #include "playlist/BaseRepresentation.h"
  30 #include "playlist/SegmentChunk.hpp"
  31 #include "plumbing/SourceStream.hpp"
  32 #include "plumbing/CommandsQueue.hpp"
  33 #include "tools/Debug.hpp"
  34 #include <vlc_demux.h>
  35
  36 #include <algorithm>
  37
  38 using namespace adaptive;
  39 using namespace adaptive::http;
  40
  41 AbstractStream::AbstractStream(demux_t * demux_)
  42 {
  43     p_realdemux = demux_;
  44     format = StreamFormat::UNSUPPORTED;
  45     currentChunk = NULL;
  46     eof = false;
  47     dead = false;
  48     disabled = false;
  49     discontinuity = false;
  50     needrestart = false;
  51     inrestart = false;
  52     segmentTracker = NULL;
  53     demuxersource = NULL;
  54     commandsqueue = NULL;
  55     demuxer = NULL;
  56     fakeesout = NULL;
  57     last_buffer_status = buffering_lessthanmin;
  58     vlc_mutex_init(&lock);
  59 }
  60
  61 bool AbstractStream::init(const StreamFormat &format_, SegmentTracker *tracker, AbstractConnectionManager *conn)
  62 {
  63     /* Don't even try if not supported or already init */
  64     if((unsigned)format_ == StreamFormat::UNSUPPORTED || demuxersource)
  65         return false;
  66
  67     demuxersource = new (std::nothrow) ChunksSourceStream( VLC_OBJECT(p_realdemux), this );
  68     if(demuxersource)
  69     {
  70         CommandsFactory *factory = new (std::nothrow) CommandsFactory();
  71         if(factory)
  72         {
  73             commandsqueue = new (std::nothrow) CommandsQueue(factory);
  74             if(commandsqueue)
  75             {
  76                 fakeesout = new (std::nothrow) FakeESOut(p_realdemux->out, commandsqueue);
  77                 if(fakeesout)
  78                 {
  79                     /* All successfull */
  80                     fakeesout->setExtraInfoProvider( this );
  81                     format = format_;
  82                     segmentTracker = tracker;
  83                     segmentTracker->registerListener(this);
  84                     segmentTracker->notifyBufferingState(true);
  85                     connManager = conn;
  86                     return true;
  87                 }
  88                 delete commandsqueue;
  89                 commandsqueue = NULL;
  90             }
  91             else
  92             {
  93                 delete factory;
  94             }
  95         }
  96         delete demuxersource;
  97     }
  98
  99     return false;
 100 }
 101
 102 AbstractStream::~AbstractStream()
 103 {
 104     delete currentChunk;
 105     if(segmentTracker)
 106         segmentTracker->notifyBufferingState(false);
 107     delete segmentTracker;
 108
 109     delete demuxer;
 110     delete demuxersource;
 111     delete fakeesout;
 112     delete commandsqueue;
 113
 114     vlc_mutex_destroy(&lock);
 115 }
 116
 117 void AbstractStream::prepareRestart(bool b_discontinuity)
 118 {
 119     if(demuxer)
 120     {
 121         /* Enqueue Del Commands for all current ES */
 122         demuxer->drain();
 123         setTimeOffset(true);
 124         /* Enqueue Del Commands for all current ES */
 125         fakeesout->scheduleAllForDeletion();
 126         if(b_discontinuity)
 127             fakeesout->schedulePCRReset();
 128         commandsqueue->Commit();
 129         /* ignoring demuxer's own Del commands */
 130         commandsqueue->setDrop(true);
 131         delete demuxer;
 132         commandsqueue->setDrop(false);
 133         demuxer = NULL;
 134     }
 135 }
 136
 137 void AbstractStream::setLanguage(const std::string &lang)
 138 {
 139     language = lang;
 140 }
 141
 142 void AbstractStream::setDescription(const std::string &desc)
 143 {
 144     description = desc;
 145 }
 146
 147 mtime_t AbstractStream::getPCR() const
 148 {
 149     vlc_mutex_lock(const_cast<vlc_mutex_t *>(&lock));
 150     mtime_t pcr = isDisabled() ? VLC_TS_INVALID : commandsqueue->getPCR();
 151     vlc_mutex_unlock(const_cast<vlc_mutex_t *>(&lock));
 152     return pcr;
 153 }
 154
 155 mtime_t AbstractStream::getMinAheadTime() const
 156 {
 157     if(!segmentTracker)
 158         return 0;
 159     return segmentTracker->getMinAheadTime();
 160 }
 161
 162 mtime_t AbstractStream::getFirstDTS() const
 163 {
 164     mtime_t dts;
 165     vlc_mutex_lock(const_cast<vlc_mutex_t *>(&lock));
 166     if(isDisabled())
 167     {
 168         dts = VLC_TS_INVALID;
 169     }
 170     else
 171     {
 172         dts = commandsqueue->getFirstDTS();
 173         if(dts == VLC_TS_INVALID)
 174             dts = commandsqueue->getPCR();
 175     }
 176     vlc_mutex_unlock(const_cast<vlc_mutex_t *>(&lock));
 177     return dts;
 178 }
 179
 180 int AbstractStream::esCount() const
 181 {
 182     return fakeesout->esCount();
 183 }
 184
 185 bool AbstractStream::seekAble() const
 186 {
 187     return (demuxer &&
 188             !fakeesout->restarting() &&
 189             !discontinuity &&
 190             !commandsqueue->isDraining() );
 191 }
 192
 193 bool AbstractStream::isSelected() const
 194 {
 195     return fakeesout->hasSelectedEs();
 196 }
 197
 198 bool AbstractStream::reactivate(mtime_t basetime)
 199 {
 200     if(setPosition(basetime, false))
 201     {
 202         setDisabled(false);
 203         return true;
 204     }
 205     else
 206     {
 207         eof = true; /* can't reactivate */
 208         return false;
 209     }
 210 }
 211
 212 bool AbstractStream::startDemux()
 213 {
 214     if(demuxer)
 215         return false;
 216
 217     demuxersource->Reset();
 218     demuxer = createDemux(format);
 219     if(!demuxer && format != StreamFormat())
 220         msg_Err(p_realdemux, "Failed to create demuxer %p %s", (void *)demuxer,
 221                 format.str().c_str());
 222
 223     return !!demuxer;
 224 }
 225
 226 bool AbstractStream::restartDemux()
 227 {
 228     bool b_ret = true;
 229     if(!demuxer)
 230     {
 231         b_ret = startDemux();
 232     }
 233     else if(demuxer->needsRestartOnSeek())
 234     {
 235         inrestart = true;
 236         /* Push all ES as recycling candidates */
 237         fakeesout->recycleAll();
 238         /* Restart with ignoring es_Del pushes to queue when terminating demux */
 239         commandsqueue->setDrop(true);
 240         demuxer->destroy();
 241         commandsqueue->setDrop(false);
 242         b_ret = demuxer->create();
 243         inrestart = false;
 244     }
 245     else
 246     {
 247         commandsqueue->Commit();
 248     }
 249     return b_ret;
 250 }
 251
 252 void AbstractStream::setDisabled(bool b)
 253 {
 254     if(disabled != b)
 255         segmentTracker->notifyBufferingState(!b);
 256     disabled = b;
 257 }
 258
 259 bool AbstractStream::isDisabled() const
 260 {
 261     return dead || disabled;
 262 }
 263
 264 bool AbstractStream::canActivate() const
 265 {
 266     return !dead;
 267 }
 268
 269 bool AbstractStream::decodersDrained()
 270 {
 271     return fakeesout->decodersDrained();
 272 }
 273
 274 AbstractStream::buffering_status AbstractStream::getLastBufferStatus() const
 275 {
 276     return last_buffer_status;
 277 }
 278
 279 mtime_t AbstractStream::getDemuxedAmount() const
 280 {
 281     return commandsqueue->getDemuxedAmount();
 282 }
 283
 284 AbstractStream::buffering_status AbstractStream::bufferize(mtime_t nz_deadline,
 285                                                            unsigned i_min_buffering, unsigned i_extra_buffering)
 286 {
 287     last_buffer_status = doBufferize(nz_deadline, i_min_buffering, i_extra_buffering);
 288     return last_buffer_status;
 289 }
 290
 291 AbstractStream::buffering_status AbstractStream::doBufferize(mtime_t nz_deadline,
 292                                                              unsigned i_min_buffering, unsigned i_extra_buffering)
 293 {
 294     vlc_mutex_lock(&lock);
 295
 296     /* Ensure it is configured */
 297     if(!segmentTracker || !connManager || dead)
 298     {
 299         vlc_mutex_unlock(&lock);
 300         return AbstractStream::buffering_end;
 301     }
 302
 303     /* Disable streams that are not selected (alternate streams) */
 304     if(esCount() && !isSelected() && !fakeesout->restarting())
 305     {
 306         setDisabled(true);
 307         segmentTracker->reset();
 308         commandsqueue->Abort(false);
 309         msg_Dbg(p_realdemux, "deactivating stream %s", format.str().c_str());
 310         vlc_mutex_unlock(&lock);
 311         return AbstractStream::buffering_end;
 312     }
 313
 314     if(commandsqueue->isDraining())
 315     {
 316         vlc_mutex_unlock(&lock);
 317         return AbstractStream::buffering_suspended;
 318     }
 319
 320     if(!demuxer)
 321     {
 322         format = segmentTracker->getCurrentFormat();
 323         if(!startDemux())
 324         {
 325             /* If demux fails because of probing failure / wrong format*/
 326             if(discontinuity)
 327             {
 328                 msg_Dbg( p_realdemux, "Draining on format change" );
 329                 prepareRestart();
 330                 discontinuity = false;
 331                 commandsqueue->setDraining();
 332                 vlc_mutex_unlock(&lock);
 333                 return AbstractStream::buffering_ongoing;
 334             }
 335             dead = true; /* Prevent further retries */
 336             commandsqueue->setEOF();
 337             vlc_mutex_unlock(&lock);
 338             return AbstractStream::buffering_end;
 339         }
 340     }
 341
 342     const int64_t i_total_buffering = i_min_buffering + i_extra_buffering;
 343
 344     mtime_t i_demuxed = commandsqueue->getDemuxedAmount();
 345     segmentTracker->notifyBufferingLevel(i_min_buffering, i_demuxed, i_total_buffering);
 346     if(i_demuxed < i_total_buffering) /* not already demuxed */
 347     {
 348         if(!segmentTracker->segmentsListReady()) /* Live Streams */
 349         {
 350             vlc_mutex_unlock(&lock);
 351             return AbstractStream::buffering_suspended;
 352         }
 353
 354         mtime_t nz_extdeadline = commandsqueue->getBufferingLevel() +
 355                                 (i_total_buffering - commandsqueue->getDemuxedAmount()) / (CLOCK_FREQ/4);
 356         nz_deadline = std::max(nz_deadline, nz_extdeadline);
 357
 358         /* need to read, demuxer still buffering, ... */
 359         vlc_mutex_unlock(&lock);
 360         int i_ret = demuxer->demux(nz_deadline);
 361         vlc_mutex_lock(&lock);
 362         if(i_ret != VLC_DEMUXER_SUCCESS)
 363         {
 364             if(discontinuity || needrestart)
 365             {
 366                 msg_Dbg(p_realdemux, "Restarting demuxer");
 367                 prepareRestart(discontinuity);
 368                 if(discontinuity)
 369                 {
 370                     msg_Dbg(p_realdemux, "Draining on discontinuity");
 371                     commandsqueue->setDraining();
 372                     discontinuity = false;
 373                 }
 374                 needrestart = false;
 375                 vlc_mutex_unlock(&lock);
 376                 return AbstractStream::buffering_ongoing;
 377             }
 378             commandsqueue->setEOF();
 379             vlc_mutex_unlock(&lock);
 380             return AbstractStream::buffering_end;
 381         }
 382         i_demuxed = commandsqueue->getDemuxedAmount();
 383         segmentTracker->notifyBufferingLevel(i_min_buffering, i_demuxed, i_total_buffering);
 384     }
 385     vlc_mutex_unlock(&lock);
 386
 387     if(i_demuxed < i_total_buffering) /* need to read more */
 388     {
 389         if(i_demuxed < i_min_buffering)
 390             return AbstractStream::buffering_lessthanmin; /* high prio */
 391         return AbstractStream::buffering_ongoing;
 392     }
 393     return AbstractStream::buffering_full;
 394 }
 395
 396 AbstractStream::status AbstractStream::dequeue(mtime_t nz_deadline, mtime_t *pi_pcr)
 397 {
 398     vlc_mutex_locker locker(&lock);
 399
 400     *pi_pcr = nz_deadline;
 401
 402     if(commandsqueue->isDraining())
 403     {
 404         AdvDebug(msg_Dbg(p_realdemux, "Stream %s pcr %" PRId64 " dts %" PRId64 " deadline %" PRId64 " [DRAINING]",
 405                          description.c_str(), commandsqueue->getPCR(), commandsqueue->getFirstDTS(),
 406                          nz_deadline));
 407
 408         *pi_pcr = commandsqueue->Process(p_realdemux->out, VLC_TS_0 + nz_deadline);
 409         if(!commandsqueue->isEmpty())
 410             return AbstractStream::status_demuxed;
 411
 412         if(!commandsqueue->isEOF())
 413         {
 414             commandsqueue->Abort(true); /* reset buffering level and flags */
 415             return AbstractStream::status_discontinuity;
 416         }
 417     }
 418
 419     if(isDisabled() || commandsqueue->isEOF())
 420     {
 421         *pi_pcr = nz_deadline;
 422         return AbstractStream::status_eof;
 423     }
 424
 425     AdvDebug(msg_Dbg(p_realdemux, "Stream %s pcr %" PRId64 " dts %" PRId64 " deadline %" PRId64 " buflevel %" PRId64,
 426                      description.c_str(), commandsqueue->getPCR(), commandsqueue->getFirstDTS(),
 427                      nz_deadline, commandsqueue->getBufferingLevel()));
 428
 429     if(nz_deadline + VLC_TS_0 <= commandsqueue->getBufferingLevel()) /* demuxed */
 430     {
 431         *pi_pcr = commandsqueue->Process( p_realdemux->out, VLC_TS_0 + nz_deadline );
 432         return AbstractStream::status_demuxed;
 433     }
 434
 435     return AbstractStream::status_buffering;
 436 }
 437
 438 block_t * AbstractStream::readNextBlock()
 439 {
 440     if (currentChunk == NULL && !eof)
 441         currentChunk = segmentTracker->getNextChunk(!fakeesout->restarting(), connManager);
 442
 443     if(discontinuity || needrestart)
 444     {
 445         msg_Info(p_realdemux, "Encountered discontinuity");
 446         /* Force stream/demuxer to end for this call */
 447         return NULL;
 448     }
 449
 450     if(currentChunk == NULL)
 451     {
 452         eof = true;
 453         return NULL;
 454     }
 455
 456     const bool b_segment_head_chunk = (currentChunk->getBytesRead() == 0);
 457
 458     block_t *block = currentChunk->readBlock();
 459     if(block == NULL)
 460     {
 461         delete currentChunk;
 462         currentChunk = NULL;
 463         return NULL;
 464     }
 465
 466     if (currentChunk->isEmpty())
 467     {
 468         delete currentChunk;
 469         currentChunk = NULL;
 470     }
 471
 472     block = checkBlock(block, b_segment_head_chunk);
 473
 474     return block;
 475 }
 476
 477 bool AbstractStream::setPosition(mtime_t time, bool tryonly)
 478 {
 479     if(!seekAble())
 480         return false;
 481
 482     bool ret = segmentTracker->setPositionByTime(time, demuxer->needsRestartOnSeek(), tryonly);
 483     if(!tryonly && ret)
 484     {
 485         if(demuxer->needsRestartOnSeek())
 486         {
 487             if(currentChunk)
 488                 delete currentChunk;
 489             currentChunk = NULL;
 490             needrestart = false;
 491
 492             setTimeOffset(-1);
 493             setTimeOffset(segmentTracker->getPlaybackTime());
 494
 495             if( !restartDemux() )
 496                 dead = true;
 497         }
 498         else commandsqueue->Abort( true );
 499
 500         es_out_Control(p_realdemux->out, ES_OUT_SET_NEXT_DISPLAY_TIME,
 501                        VLC_TS_0 + time);
 502     }
 503     return ret;
 504 }
 505
 506 mtime_t AbstractStream::getPlaybackTime() const
 507 {
 508     return segmentTracker->getPlaybackTime();
 509 }
 510
 511 void AbstractStream::runUpdates()
 512 {
 513     if(!isDisabled())
 514         segmentTracker->updateSelected();
 515 }
 516
 517 void AbstractStream::fillExtraFMTInfo( es_format_t *p_fmt ) const
 518 {
 519     if(!p_fmt->psz_language && !language.empty())
 520         p_fmt->psz_language = strdup(language.c_str());
 521     if(!p_fmt->psz_description && !description.empty())
 522         p_fmt->psz_description = strdup(description.c_str());
 523 }
 524
 525 void AbstractStream::setTimeOffset(mtime_t i_offset)
 526 {
 527     /* Check if we need to set an offset as the demuxer
 528      * will start from zero from seek point */
 529     if(i_offset < 0) /* reset */
 530     {
 531         fakeesout->setExpectedTimestampOffset(0);
 532     }
 533     else if(demuxer)
 534     {
 535         fakeesout->setExpectedTimestampOffset(i_offset);
 536     }
 537 }
 538
 539 void AbstractStream::trackerEvent(const SegmentTrackerEvent &event)
 540 {
 541     switch(event.type)
 542     {
 543         case SegmentTrackerEvent::DISCONTINUITY:
 544             discontinuity = true;
 545             break;
 546
 547         case SegmentTrackerEvent::FORMATCHANGE:
 548             /* Check if our current demux is still valid */
 549             if(*event.u.format.f != format)
 550             {
 551                 /* Format has changed between segments, we need to drain and change demux */
 552                 msg_Info(p_realdemux, "Changing stream format %s -> %s",
 553                          format.str().c_str(), event.u.format.f->str().c_str());
 554                 format = *event.u.format.f;
 555
 556                 /* This is an implict discontinuity */
 557                 discontinuity = true;
 558             }
 559             break;
 560
 561         case SegmentTrackerEvent::SWITCHING:
 562             if(demuxer && demuxer->needsRestartOnSwitch() && !inrestart)
 563             {
 564                 needrestart = true;
 565             }
 566         default:
 567             break;
 568     }
 569 }