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