modules/stream_filter/prefetch.c

   1 /*****************************************************************************
   2  * prefetch.c: prefetchinging module for VLC
   3  *****************************************************************************
   4  * Copyright © 2015 Rémi Denis-Courmont
   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 by
   8  * 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 <assert.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28
  29 #include <sys/types.h>
  30 #include <unistd.h>
  31
  32 #include <vlc_common.h>
  33 #include <vlc_plugin.h>
  34 #include <vlc_stream.h>
  35 #include <vlc_fs.h>
  36 #include <vlc_interrupt.h>
  37
  38 struct stream_ctrl
  39 {
  40     struct stream_ctrl *next;
  41     int query;
  42     union
  43     {
  44         struct
  45         {
  46             int id;
  47             bool state;
  48         } id_state;
  49     };
  50 };
  51
  52 typedef struct
  53 {
  54     vlc_mutex_t  lock;
  55     vlc_cond_t   wait_data;
  56     vlc_cond_t   wait_space;
  57     vlc_thread_t thread;
  58     vlc_interrupt_t *interrupt;
  59
  60     bool         eof;
  61     bool         error;
  62     bool         paused;
  63
  64     bool         can_seek;
  65     bool         can_pace;
  66     bool         can_pause;
  67     uint64_t     size;
  68     vlc_tick_t   pts_delay;
  69     char        *content_type;
  70
  71     uint64_t     buffer_offset;
  72     uint64_t     stream_offset;
  73     size_t       buffer_length;
  74     size_t       buffer_size;
  75     char        *buffer;
  76     size_t       seek_threshold;
  77
  78     struct stream_ctrl *controls;
  79 } stream_sys_t;
  80
  81 static ssize_t ThreadRead(stream_t *stream, void *buf, size_t length)
  82 {
  83     stream_sys_t *sys = stream->p_sys;
  84
  85     vlc_mutex_unlock(&sys->lock);
  86     assert(length > 0);
  87
  88     ssize_t val = vlc_stream_ReadPartial(stream->s, buf, length);
  89
  90     vlc_mutex_lock(&sys->lock);
  91     return val;
  92 }
  93
  94 static int ThreadSeek(stream_t *stream, uint64_t seek_offset)
  95 {
  96     stream_sys_t *sys = stream->p_sys;
  97
  98     vlc_mutex_unlock(&sys->lock);
  99
 100     int val = vlc_stream_Seek(stream->s, seek_offset);
 101     if (val != VLC_SUCCESS)
 102         msg_Err(stream, "cannot seek (to offset %"PRIu64")", seek_offset);
 103
 104     vlc_mutex_lock(&sys->lock);
 105
 106     return (val == VLC_SUCCESS) ? 0 : -1;
 107 }
 108
 109 static int ThreadControl(stream_t *stream, int query, ...)
 110 {
 111     stream_sys_t *sys = stream->p_sys;
 112
 113     vlc_mutex_unlock(&sys->lock);
 114
 115     va_list ap;
 116     int ret;
 117
 118     va_start(ap, query);
 119     ret = vlc_stream_vaControl(stream->s, query, ap);
 120     va_end(ap);
 121
 122     vlc_mutex_lock(&sys->lock);
 123     return ret;
 124 }
 125
 126 static void *Thread(void *data)
 127 {
 128     stream_t *stream = data;
 129     stream_sys_t *sys = stream->p_sys;
 130     bool paused = false;
 131
 132     vlc_interrupt_set(sys->interrupt);
 133
 134     vlc_mutex_lock(&sys->lock);
 135     while (!vlc_killed())
 136     {
 137         struct stream_ctrl *ctrl = sys->controls;
 138
 139         if (unlikely(ctrl != NULL))
 140         {
 141             sys->controls = ctrl->next;
 142             ThreadControl(stream, ctrl->query, ctrl->id_state.id,
 143                           ctrl->id_state.state);
 144             free(ctrl);
 145             continue;
 146         }
 147
 148         if (sys->paused != paused)
 149         {   /* Update pause state */
 150             msg_Dbg(stream, paused ? "resuming" : "pausing");
 151             paused = sys->paused;
 152             ThreadControl(stream, STREAM_SET_PAUSE_STATE, paused);
 153             continue;
 154         }
 155
 156         if (paused || sys->error)
 157         {   /* Wait for not paused and not failed */
 158             vlc_cond_wait(&sys->wait_space, &sys->lock);
 159             continue;
 160         }
 161
 162         uint_fast64_t stream_offset = sys->stream_offset;
 163
 164         if (stream_offset < sys->buffer_offset)
 165         {   /* Need to seek backward */
 166             if (ThreadSeek(stream, stream_offset) == 0)
 167             {
 168                 sys->buffer_offset = stream_offset;
 169                 sys->buffer_length = 0;
 170                 assert(!sys->error);
 171                 sys->eof = false;
 172             }
 173             else
 174             {
 175                 sys->error = true;
 176                 vlc_cond_signal(&sys->wait_data);
 177             }
 178             continue;
 179         }
 180
 181         if (sys->eof)
 182         {   /* Do not attempt to read at EOF - would busy loop */
 183             vlc_cond_wait(&sys->wait_space, &sys->lock);
 184             continue;
 185         }
 186
 187         assert(stream_offset >= sys->buffer_offset);
 188
 189         /* As long as there is space, the buffer will retain already read
 190          * ("historical") data. The data can be used if/when seeking backward.
 191          * Unread data is however given precedence if the buffer is full. */
 192         uint64_t history = stream_offset - sys->buffer_offset;
 193
 194         /* If upstream supports seeking and if the downstream offset is far
 195          * beyond the upstream offset, then attempt to skip forward.
 196          * If it fails, assume upstream is well-behaved such that the failed
 197          * seek is a no-op, and continue as if seeking was not supported.
 198          * WARNING: Except problems with misbehaving access plug-ins. */
 199         if (sys->can_seek
 200          && history >= (sys->buffer_length + sys->seek_threshold))
 201         {
 202             if (ThreadSeek(stream, stream_offset) == 0)
 203             {
 204                 sys->buffer_offset = stream_offset;
 205                 sys->buffer_length = 0;
 206                 assert(!sys->error);
 207                 assert(!sys->eof);
 208             }
 209             else
 210             {   /* Seek failure is not necessarily fatal here. We could read
 211                  * data instead until the desired seek offset. But in practice,
 212                  * not all upstream accesses handle reads after failed seek
 213                  * correctly. Furthermore, sys->stream_offset and/or
 214                  * sys->paused might have changed in the mean time. */
 215                 sys->error = true;
 216                 vlc_cond_signal(&sys->wait_data);
 217             }
 218             continue;
 219         }
 220
 221         assert(sys->buffer_size >= sys->buffer_length);
 222
 223         size_t len = sys->buffer_size - sys->buffer_length;
 224         if (len == 0)
 225         {   /* Buffer is full */
 226             if (history == 0)
 227             {   /* Wait for data to be read */
 228                 vlc_cond_wait(&sys->wait_space, &sys->lock);
 229                 continue;
 230             }
 231
 232             /* Discard some historical data to make room. */
 233             len = history > sys->buffer_length ? sys->buffer_length : history;
 234
 235             sys->buffer_offset += len;
 236             sys->buffer_length -= len;
 237         }
 238
 239         size_t offset = (sys->buffer_offset + sys->buffer_length)
 240                         % sys->buffer_size;
 241          /* Do not step past the sharp edge of the circular buffer */
 242         if (offset + len > sys->buffer_size)
 243             len = sys->buffer_size - offset;
 244
 245         ssize_t val = ThreadRead(stream, sys->buffer + offset, len);
 246         if (val < 0)
 247             continue;
 248         if (val == 0)
 249         {
 250             assert(len > 0);
 251             msg_Dbg(stream, "end of stream");
 252             sys->eof = true;
 253         }
 254
 255         assert((size_t)val <= len);
 256         sys->buffer_length += val;
 257         assert(sys->buffer_length <= sys->buffer_size);
 258         //msg_Dbg(stream, "buffer: %zu/%zu", sys->buffer_length,
 259         //        sys->buffer_size);
 260         vlc_cond_signal(&sys->wait_data);
 261     }
 262
 263     sys->error = true;
 264     vlc_cond_signal(&sys->wait_data);
 265     vlc_mutex_unlock(&sys->lock);
 266     return NULL;
 267 }
 268
 269 static int Seek(stream_t *stream, uint64_t offset)
 270 {
 271     stream_sys_t *sys = stream->p_sys;
 272
 273     vlc_mutex_lock(&sys->lock);
 274     sys->stream_offset = offset;
 275     sys->error = false;
 276     vlc_cond_signal(&sys->wait_space);
 277     vlc_mutex_unlock(&sys->lock);
 278     return 0;
 279 }
 280
 281 static size_t BufferLevel(const stream_t *stream, bool *eof)
 282 {
 283     stream_sys_t *sys = stream->p_sys;
 284
 285     *eof = false;
 286
 287     if (sys->stream_offset < sys->buffer_offset)
 288         return 0;
 289     if ((sys->stream_offset - sys->buffer_offset) >= sys->buffer_length)
 290     {
 291         *eof = sys->eof;
 292         return 0;
 293     }
 294     return sys->buffer_offset + sys->buffer_length - sys->stream_offset;
 295 }
 296
 297 static ssize_t Read(stream_t *stream, void *buf, size_t buflen)
 298 {
 299     stream_sys_t *sys = stream->p_sys;
 300     size_t copy, offset;
 301     bool eof;
 302
 303     if (buflen == 0)
 304         return buflen;
 305
 306     vlc_mutex_lock(&sys->lock);
 307     if (sys->paused)
 308     {
 309         msg_Err(stream, "reading while paused (buggy demux?)");
 310         sys->paused = false;
 311         vlc_cond_signal(&sys->wait_space);
 312     }
 313
 314     while ((copy = BufferLevel(stream, &eof)) == 0 && !eof)
 315     {
 316         void *data[2];
 317
 318         if (sys->error)
 319         {
 320             vlc_mutex_unlock(&sys->lock);
 321             return 0;
 322         }
 323
 324         vlc_interrupt_forward_start(sys->interrupt, data);
 325         vlc_cond_wait(&sys->wait_data, &sys->lock);
 326         vlc_interrupt_forward_stop(data);
 327     }
 328
 329     offset = sys->stream_offset % sys->buffer_size;
 330     if (copy > buflen)
 331         copy = buflen;
 332     /* Do not step past the sharp edge of the circular buffer */
 333     if (offset + copy > sys->buffer_size)
 334         copy = sys->buffer_size - offset;
 335
 336     memcpy(buf, sys->buffer + offset, copy);
 337     sys->stream_offset += copy;
 338     vlc_cond_signal(&sys->wait_space);
 339     vlc_mutex_unlock(&sys->lock);
 340     return copy;
 341 }
 342
 343 static int Control(stream_t *stream, int query, va_list args)
 344 {
 345     stream_sys_t *sys = stream->p_sys;
 346
 347     switch (query)
 348     {
 349         case STREAM_CAN_SEEK:
 350             *va_arg(args, bool *) = sys->can_seek;
 351             break;
 352         case STREAM_CAN_FASTSEEK:
 353             *va_arg(args, bool *) = false;
 354             break;
 355         case STREAM_CAN_PAUSE:
 356              *va_arg(args, bool *) = sys->can_pause;
 357             break;
 358         case STREAM_CAN_CONTROL_PACE:
 359             *va_arg (args, bool *) = sys->can_pace;
 360             break;
 361         case STREAM_GET_SIZE:
 362             if (sys->size == (uint64_t)-1)
 363                 return VLC_EGENERIC;
 364             *va_arg(args, uint64_t *) = sys->size;
 365             break;
 366         case STREAM_GET_PTS_DELAY:
 367             *va_arg(args, vlc_tick_t *) = sys->pts_delay;
 368             break;
 369         case STREAM_GET_TITLE_INFO:
 370         case STREAM_GET_TITLE:
 371         case STREAM_GET_SEEKPOINT:
 372         case STREAM_GET_META:
 373             return VLC_EGENERIC;
 374         case STREAM_GET_CONTENT_TYPE:
 375             if (sys->content_type == NULL)
 376                 return VLC_EGENERIC;
 377             *va_arg(args, char **) = strdup(sys->content_type);
 378             return VLC_SUCCESS;
 379         case STREAM_GET_SIGNAL:
 380         case STREAM_GET_TAGS:
 381         case STREAM_GET_TYPE:
 382             return VLC_EGENERIC;
 383         case STREAM_SET_PAUSE_STATE:
 384         {
 385             bool paused = va_arg(args, unsigned);
 386
 387             vlc_mutex_lock(&sys->lock);
 388             sys->paused = paused;
 389             vlc_cond_signal(&sys->wait_space);
 390             vlc_mutex_unlock (&sys->lock);
 391             break;
 392         }
 393         case STREAM_SET_TITLE:
 394         case STREAM_SET_SEEKPOINT:
 395             return VLC_EGENERIC;
 396         case STREAM_SET_PRIVATE_ID_STATE:
 397         {
 398             struct stream_ctrl *ctrl = malloc(sizeof (*ctrl)), **pp;
 399             if (unlikely(ctrl == NULL))
 400                 return VLC_ENOMEM;
 401
 402             ctrl->next = NULL;
 403             ctrl->query = query;
 404             ctrl->id_state.id = va_arg(args, int);
 405             ctrl->id_state.state = va_arg(args, int);
 406             vlc_mutex_lock(&sys->lock);
 407             for (pp = &sys->controls; *pp != NULL; pp = &((*pp)->next));
 408             *pp = ctrl;
 409             vlc_cond_signal(&sys->wait_space);
 410             vlc_mutex_unlock(&sys->lock);
 411             break;
 412         }
 413         case STREAM_SET_PRIVATE_ID_CA:
 414         case STREAM_GET_PRIVATE_ID_STATE:
 415             return VLC_EGENERIC;
 416         default:
 417             msg_Err(stream, "unimplemented query (%d) in control", query);
 418             return VLC_EGENERIC;
 419     }
 420     return VLC_SUCCESS;
 421 }
 422
 423 static int Open(vlc_object_t *obj)
 424 {
 425     stream_t *stream = (stream_t *)obj;
 426
 427     bool fast_seek;
 428
 429     if (vlc_stream_Control(stream->s, STREAM_CAN_FASTSEEK, &fast_seek))
 430         return VLC_EGENERIC; /* not a byte stream */
 431     /* For local files, the operating system is likely to do a better work at
 432      * caching/prefetching. Also, prefetching with this module could cause
 433      * undesirable high load at start-up. Lastly, local files may require
 434      * support for title/seekpoint and meta control requests. */
 435     if (fast_seek)
 436         return VLC_EGENERIC;
 437
 438     /* PID-filtered streams are not suitable for prefetching, as they would
 439      * suffer excessive latency to enable a PID. DVB would also require support
 440      * for the signal level and Conditional Access controls.
 441      * TODO? For seekable streams, a forced could work around the problem. */
 442     if (vlc_stream_Control(stream->s, STREAM_GET_PRIVATE_ID_STATE, 0,
 443                            &(bool){ false }) == VLC_SUCCESS)
 444         return VLC_EGENERIC;
 445
 446     stream_sys_t *sys = malloc(sizeof (*sys));
 447     if (unlikely(sys == NULL))
 448         return VLC_ENOMEM;
 449
 450     vlc_stream_Control(stream->s, STREAM_CAN_SEEK, &sys->can_seek);
 451     vlc_stream_Control(stream->s, STREAM_CAN_PAUSE, &sys->can_pause);
 452     vlc_stream_Control(stream->s, STREAM_CAN_CONTROL_PACE, &sys->can_pace);
 453     if (vlc_stream_Control(stream->s, STREAM_GET_SIZE, &sys->size))
 454         sys->size = -1;
 455     vlc_stream_Control(stream->s, STREAM_GET_PTS_DELAY, &sys->pts_delay);
 456     if (vlc_stream_Control(stream->s, STREAM_GET_CONTENT_TYPE,
 457                            &sys->content_type))
 458         sys->content_type = NULL;
 459
 460     sys->eof = false;
 461     sys->error = false;
 462     sys->paused = false;
 463     sys->buffer_offset = 0;
 464     sys->stream_offset = 0;
 465     sys->buffer_length = 0;
 466     sys->buffer_size = var_InheritInteger(obj, "prefetch-buffer-size") << 10u;
 467     sys->seek_threshold = var_InheritInteger(obj, "prefetch-seek-threshold");
 468     sys->controls = NULL;
 469
 470     uint64_t size = stream_Size(stream->s);
 471     if (size > 0)
 472     {   /* No point allocating a buffer larger than the source stream */
 473         if (sys->buffer_size > size)
 474             sys->buffer_size = size;
 475     }
 476
 477     sys->buffer = malloc(sys->buffer_size);
 478     if (sys->buffer == NULL)
 479         goto error;
 480
 481     sys->interrupt = vlc_interrupt_create();
 482     if (unlikely(sys->interrupt == NULL))
 483         goto error;
 484
 485     vlc_mutex_init(&sys->lock);
 486     vlc_cond_init(&sys->wait_data);
 487     vlc_cond_init(&sys->wait_space);
 488
 489     stream->p_sys = sys;
 490
 491     if (vlc_clone(&sys->thread, Thread, stream, VLC_THREAD_PRIORITY_LOW))
 492     {
 493         vlc_interrupt_destroy(sys->interrupt);
 494         goto error;
 495     }
 496
 497     msg_Dbg(stream, "using %zu bytes buffer", sys->buffer_size);
 498     stream->pf_read = Read;
 499     stream->pf_seek = Seek;
 500     stream->pf_control = Control;
 501     return VLC_SUCCESS;
 502
 503 error:
 504     free(sys->buffer);
 505     free(sys->content_type);
 506     free(sys);
 507     return VLC_ENOMEM;
 508 }
 509
 510
 511 /**
 512  * Releases allocate resources.
 513  */
 514 static void Close (vlc_object_t *obj)
 515 {
 516     stream_t *stream = (stream_t *)obj;
 517     stream_sys_t *sys = stream->p_sys;
 518
 519     vlc_mutex_lock(&sys->lock);
 520     vlc_interrupt_kill(sys->interrupt);
 521     vlc_cond_signal(&sys->wait_space);
 522     vlc_mutex_unlock(&sys->lock);
 523
 524     vlc_join(sys->thread, NULL);
 525     vlc_interrupt_destroy(sys->interrupt);
 526
 527     while(sys->controls)
 528     {
 529         struct stream_ctrl *ctrl = sys->controls;
 530         sys->controls = ctrl->next;
 531         free(ctrl);
 532     }
 533     free(sys->buffer);
 534     free(sys->content_type);
 535     free(sys);
 536 }
 537
 538 vlc_module_begin()
 539     set_category(CAT_INPUT)
 540     set_subcategory(SUBCAT_INPUT_STREAM_FILTER)
 541     set_capability("stream_filter", 0)
 542
 543     set_description(N_("Stream prefetch filter"))
 544     set_callbacks(Open, Close)
 545
 546     add_integer("prefetch-buffer-size", 1 << 14, N_("Buffer size"),
 547                 N_("Prefetch buffer size (KiB)"), false)
 548         change_integer_range(4, 1 << 20)
 549     add_obsolete_integer("prefetch-read-size") /* since 4.0.0 */
 550     add_integer("prefetch-seek-threshold", 1 << 14, N_("Seek threshold"),
 551                 N_("Prefetch forward seek threshold (bytes)"), true)
 552         change_integer_range(0, UINT64_C(1) << 60)
 553 vlc_module_end()