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     bool restarting = fakeesout->restarting();
 188     bool draining = commandsqueue->isDraining();
 189     bool eof = commandsqueue->isEOF();
 190
 191     msg_Dbg(p_realdemux, "demuxer %p, fakeesout restarting %d, "
 192             "discontinuity %d, commandsqueue draining %d, commandsqueue eof %d",
 193             static_cast<void *>(demuxer), restarting, discontinuity, draining, eof);
 194
 195     if(!demuxer || restarting || discontinuity || (!eof && draining))
 196     {
 197         msg_Warn(p_realdemux, "not seekable");
 198         return false;
 199     }
 200     else
 201     {
 202         return true;
 203     }
 204 }
 205
 206 bool AbstractStream::isSelected() const
 207 {
 208     return fakeesout->hasSelectedEs();
 209 }
 210
 211 bool AbstractStream::reactivate(mtime_t basetime)
 212 {
 213     if(setPosition(basetime, false))
 214     {
 215         setDisabled(false);
 216         return true;
 217     }
 218     else
 219     {
 220         eof = true; /* can't reactivate */
 221         return false;
 222     }
 223 }
 224
 225 bool AbstractStream::startDemux()
 226 {
 227     if(demuxer)
 228         return false;
 229
 230     demuxersource->Reset();
 231     demuxer = createDemux(format);
 232     if(!demuxer && format != StreamFormat())
 233         msg_Err(p_realdemux, "Failed to create demuxer %p %s", (void *)demuxer,
 234                 format.str().c_str());
 235
 236     return !!demuxer;
 237 }
 238
 239 bool AbstractStream::restartDemux()
 240 {
 241     bool b_ret = true;
 242     if(!demuxer)
 243     {
 244         b_ret = startDemux();
 245     }
 246     else if(demuxer->needsRestartOnSeek())
 247     {
 248         inrestart = true;
 249         /* Push all ES as recycling candidates */
 250         fakeesout->recycleAll();
 251         /* Restart with ignoring es_Del pushes to queue when terminating demux */
 252         commandsqueue->setDrop(true);
 253         demuxer->destroy();
 254         commandsqueue->setDrop(false);
 255         b_ret = demuxer->create();
 256         inrestart = false;
 257     }
 258     else
 259     {
 260         commandsqueue->Commit();
 261     }
 262     return b_ret;
 263 }
 264
 265 void AbstractStream::setDisabled(bool b)
 266 {
 267     if(disabled != b)
 268         segmentTracker->notifyBufferingState(!b);
 269     disabled = b;
 270 }
 271
 272 bool AbstractStream::isDisabled() const
 273 {
 274     return dead || disabled;
 275 }
 276
 277 bool AbstractStream::canActivate() const
 278 {
 279     return !dead;
 280 }
 281
 282 bool AbstractStream::decodersDrained()
 283 {
 284     return fakeesout->decodersDrained();
 285 }
 286
 287 AbstractStream::buffering_status AbstractStream::getLastBufferStatus() const
 288 {
 289     return last_buffer_status;
 290 }
 291
 292 mtime_t AbstractStream::getDemuxedAmount() const
 293 {
 294     return commandsqueue->getDemuxedAmount();
 295 }
 296
 297 AbstractStream::buffering_status AbstractStream::bufferize(mtime_t nz_deadline,
 298                                                            unsigned i_min_buffering, unsigned i_extra_buffering)
 299 {
 300     last_buffer_status = doBufferize(nz_deadline, i_min_buffering, i_extra_buffering);
 301     return last_buffer_status;
 302 }
 303
 304 AbstractStream::buffering_status AbstractStream::doBufferize(mtime_t nz_deadline,
 305                                                              unsigned i_min_buffering, unsigned i_extra_buffering)
 306 {
 307     vlc_mutex_lock(&lock);
 308
 309     /* Ensure it is configured */
 310     if(!segmentTracker || !connManager || dead)
 311     {
 312         vlc_mutex_unlock(&lock);
 313         return AbstractStream::buffering_end;
 314     }
 315
 316     /* Disable streams that are not selected (alternate streams) */
 317     if(esCount() && !isSelected() && !fakeesout->restarting())
 318     {
 319         setDisabled(true);
 320         segmentTracker->reset();
 321         commandsqueue->Abort(false);
 322         msg_Dbg(p_realdemux, "deactivating %s stream %s",
 323                 format.str().c_str(), description.c_str());
 324         vlc_mutex_unlock(&lock);
 325         return AbstractStream::buffering_end;
 326     }
 327
 328     if(commandsqueue->isDraining())
 329     {
 330         vlc_mutex_unlock(&lock);
 331         return AbstractStream::buffering_suspended;
 332     }
 333
 334     if(!demuxer)
 335     {
 336         format = segmentTracker->getCurrentFormat();
 337         if(!startDemux())
 338         {
 339             /* If demux fails because of probing failure / wrong format*/
 340             if(discontinuity)
 341             {
 342                 msg_Dbg( p_realdemux, "Draining on format change" );
 343                 prepareRestart();
 344                 discontinuity = false;
 345                 commandsqueue->setDraining();
 346                 vlc_mutex_unlock(&lock);
 347                 return AbstractStream::buffering_ongoing;
 348             }
 349             dead = true; /* Prevent further retries */
 350             commandsqueue->setEOF(true);
 351             vlc_mutex_unlock(&lock);
 352             return AbstractStream::buffering_end;
 353         }
 354     }
 355
 356     const int64_t i_total_buffering = i_min_buffering + i_extra_buffering;
 357
 358     mtime_t i_demuxed = commandsqueue->getDemuxedAmount();
 359     segmentTracker->notifyBufferingLevel(i_min_buffering, i_demuxed, i_total_buffering);
 360     if(i_demuxed < i_total_buffering) /* not already demuxed */
 361     {
 362         if(!segmentTracker->segmentsListReady()) /* Live Streams */
 363         {
 364             vlc_mutex_unlock(&lock);
 365             return AbstractStream::buffering_suspended;
 366         }
 367
 368         mtime_t nz_extdeadline = commandsqueue->getBufferingLevel() +
 369                                 (i_total_buffering - commandsqueue->getDemuxedAmount()) / (CLOCK_FREQ/4);
 370         nz_deadline = std::max(nz_deadline, nz_extdeadline);
 371
 372         /* need to read, demuxer still buffering, ... */
 373         vlc_mutex_unlock(&lock);
 374         int i_ret = demuxer->demux(nz_deadline);
 375         vlc_mutex_lock(&lock);
 376         if(i_ret != VLC_DEMUXER_SUCCESS)
 377         {
 378             if(discontinuity || needrestart)
 379             {
 380                 msg_Dbg(p_realdemux, "Restarting demuxer");
 381                 prepareRestart(discontinuity);
 382                 if(discontinuity)
 383                 {
 384                     msg_Dbg(p_realdemux, "Draining on discontinuity");
 385                     commandsqueue->setDraining();
 386                     discontinuity = false;
 387                 }
 388                 needrestart = false;
 389                 vlc_mutex_unlock(&lock);
 390                 return AbstractStream::buffering_ongoing;
 391             }
 392             commandsqueue->setEOF(true);
 393             vlc_mutex_unlock(&lock);
 394             return AbstractStream::buffering_end;
 395         }
 396         i_demuxed = commandsqueue->getDemuxedAmount();
 397         segmentTracker->notifyBufferingLevel(i_min_buffering, i_demuxed, i_total_buffering);
 398     }
 399     vlc_mutex_unlock(&lock);
 400
 401     if(i_demuxed < i_total_buffering) /* need to read more */
 402     {
 403         if(i_demuxed < i_min_buffering)
 404             return AbstractStream::buffering_lessthanmin; /* high prio */
 405         return AbstractStream::buffering_ongoing;
 406     }
 407     return AbstractStream::buffering_full;
 408 }
 409
 410 AbstractStream::status AbstractStream::dequeue(mtime_t nz_deadline, mtime_t *pi_pcr)
 411 {
 412     vlc_mutex_locker locker(&lock);
 413
 414     *pi_pcr = nz_deadline;
 415
 416     if(commandsqueue->isDraining())
 417     {
 418         AdvDebug(msg_Dbg(p_realdemux, "Stream %s pcr %" PRId64 " dts %" PRId64 " deadline %" PRId64 " [DRAINING]",
 419                          description.c_str(), commandsqueue->getPCR(), commandsqueue->getFirstDTS(),
 420                          nz_deadline));
 421
 422         *pi_pcr = commandsqueue->Process(p_realdemux->out, VLC_TS_0 + nz_deadline);
 423         if(!commandsqueue->isEmpty())
 424             return AbstractStream::status_demuxed;
 425
 426         if(!commandsqueue->isEOF())
 427         {
 428             commandsqueue->Abort(true); /* reset buffering level and flags */
 429             return AbstractStream::status_discontinuity;
 430         }
 431     }
 432
 433     if(isDisabled() || commandsqueue->isEOF())
 434     {
 435         *pi_pcr = nz_deadline;
 436         return AbstractStream::status_eof;
 437     }
 438
 439     AdvDebug(msg_Dbg(p_realdemux, "Stream %s pcr %" PRId64 " dts %" PRId64 " deadline %" PRId64 " buflevel %" PRId64,
 440                      description.c_str(), commandsqueue->getPCR(), commandsqueue->getFirstDTS(),
 441                      nz_deadline, commandsqueue->getBufferingLevel()));
 442
 443     if(nz_deadline + VLC_TS_0 <= commandsqueue->getBufferingLevel()) /* demuxed */
 444     {
 445         *pi_pcr = commandsqueue->Process( p_realdemux->out, VLC_TS_0 + nz_deadline );
 446         return AbstractStream::status_demuxed;
 447     }
 448
 449     return AbstractStream::status_buffering;
 450 }
 451
 452 block_t * AbstractStream::readNextBlock()
 453 {
 454     if (currentChunk == NULL && !eof)
 455         currentChunk = segmentTracker->getNextChunk(!fakeesout->restarting(), connManager);
 456
 457     if(discontinuity || needrestart)
 458     {
 459         msg_Info(p_realdemux, "Encountered discontinuity");
 460         /* Force stream/demuxer to end for this call */
 461         return NULL;
 462     }
 463
 464     if(currentChunk == NULL)
 465     {
 466         eof = true;
 467         return NULL;
 468     }
 469
 470     const bool b_segment_head_chunk = (currentChunk->getBytesRead() == 0);
 471
 472     block_t *block = currentChunk->readBlock();
 473     if(block == NULL)
 474     {
 475         delete currentChunk;
 476         currentChunk = NULL;
 477         return NULL;
 478     }
 479
 480     if (currentChunk->isEmpty())
 481     {
 482         delete currentChunk;
 483         currentChunk = NULL;
 484     }
 485
 486     block = checkBlock(block, b_segment_head_chunk);
 487
 488     return block;
 489 }
 490
 491 bool AbstractStream::setPosition(mtime_t time, bool tryonly)
 492 {
 493     if(!seekAble())
 494         return false;
 495
 496     bool ret = segmentTracker->setPositionByTime(time, demuxer->needsRestartOnSeek(), tryonly);
 497     if(!tryonly && ret)
 498     {
 499         // clear eof flag before restartDemux() to prevent readNextBlock() fail
 500         eof = false;
 501         if(demuxer->needsRestartOnSeek())
 502         {
 503             if(currentChunk)
 504                 delete currentChunk;
 505             currentChunk = NULL;
 506             needrestart = false;
 507
 508             setTimeOffset(-1);
 509             setTimeOffset(segmentTracker->getPlaybackTime());
 510
 511             if( !restartDemux() )
 512             {
 513                 msg_Info(p_realdemux, "Restart demux failed");
 514                 eof = true;
 515                 dead = true;
 516                 ret = false;
 517             }
 518             else
 519             {
 520                 commandsqueue->setEOF(false);
 521             }
 522         }
 523         else commandsqueue->Abort( true );
 524
 525         es_out_Control(p_realdemux->out, ES_OUT_SET_NEXT_DISPLAY_TIME,
 526                        VLC_TS_0 + time);
 527     }
 528     return ret;
 529 }
 530
 531 mtime_t AbstractStream::getPlaybackTime() const
 532 {
 533     return segmentTracker->getPlaybackTime();
 534 }
 535
 536 void AbstractStream::runUpdates()
 537 {
 538     if(!isDisabled())
 539         segmentTracker->updateSelected();
 540 }
 541
 542 void AbstractStream::fillExtraFMTInfo( es_format_t *p_fmt ) const
 543 {
 544     if(!p_fmt->psz_language && !language.empty())
 545         p_fmt->psz_language = strdup(language.c_str());
 546     if(!p_fmt->psz_description && !description.empty())
 547         p_fmt->psz_description = strdup(description.c_str());
 548 }
 549
 550 void AbstractStream::setTimeOffset(mtime_t i_offset)
 551 {
 552     /* Check if we need to set an offset as the demuxer
 553      * will start from zero from seek point */
 554     if(i_offset < 0) /* reset */
 555     {
 556         fakeesout->setExpectedTimestampOffset(0);
 557     }
 558     else if(demuxer)
 559     {
 560         fakeesout->setExpectedTimestampOffset(i_offset);
 561     }
 562 }
 563
 564 void AbstractStream::trackerEvent(const SegmentTrackerEvent &event)
 565 {
 566     switch(event.type)
 567     {
 568         case SegmentTrackerEvent::DISCONTINUITY:
 569             discontinuity = true;
 570             break;
 571
 572         case SegmentTrackerEvent::FORMATCHANGE:
 573             /* Check if our current demux is still valid */
 574             if(*event.u.format.f != format)
 575             {
 576                 /* Format has changed between segments, we need to drain and change demux */
 577                 msg_Info(p_realdemux, "Changing stream format %s -> %s",
 578                          format.str().c_str(), event.u.format.f->str().c_str());
 579                 format = *event.u.format.f;
 580
 581                 /* This is an implict discontinuity */
 582                 discontinuity = true;
 583             }
 584             break;
 585
 586         case SegmentTrackerEvent::SWITCHING:
 587             if(demuxer && demuxer->needsRestartOnSwitch() && !inrestart)
 588             {
 589                 needrestart = true;
 590             }
 591             break;
 592
 593         case SegmentTrackerEvent::SEGMENT_CHANGE:
 594             if(demuxer && demuxer->needsRestartOnEachSegment() && !inrestart)
 595             {
 596                 needrestart = true;
 597             }
 598             break;
 599
 600         default:
 601             break;
 602     }
 603 }