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