modules/stream_filter/cache_read.c

   1 /*****************************************************************************
   2  * cache_read.c
   3  *****************************************************************************
   4  * Copyright (C) 1999-2004 VLC authors and VideoLAN
   5  * $Id$
   6  *
   7  * Authors: Laurent Aimar <fenrir@via.ecp.fr>
   8  *
   9  * This program is free software; you can redistribute it and/or modify it
  10  * under the terms of the GNU Lesser General Public License as published by
  11  * the Free Software Foundation; either version 2.1 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17  * GNU Lesser General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Lesser General Public License
  20  * along with this program; if not, write to the Free Software Foundation,
  21  * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
  22  *****************************************************************************/
  23
  24 #ifdef HAVE_CONFIG_H
  25 # include "config.h"
  26 #endif
  27
  28 #include <assert.h>
  29 #include <stdlib.h>
  30 #include <string.h>
  31
  32 #include <vlc_common.h>
  33 #include <vlc_plugin.h>
  34 #include <vlc_stream.h>
  35 #include <vlc_interrupt.h>
  36
  37 // #define STREAM_DEBUG 1
  38
  39 /*
  40  * Complex scheme using mutliple track to avoid seeking
  41  */
  42
  43 /* How many tracks we have, currently only used for stream mode */
  44 #ifdef OPTIMIZE_MEMORY
  45 #   define STREAM_CACHE_TRACK 1
  46     /* Max size of our cache 128Ko per track */
  47 #   define STREAM_CACHE_SIZE  (STREAM_CACHE_TRACK*1024*128)
  48 #else
  49 #   define STREAM_CACHE_TRACK 3
  50     /* Max size of our cache 4Mo per track */
  51 #   define STREAM_CACHE_SIZE  (4*STREAM_CACHE_TRACK*1024*1024)
  52 #endif
  53
  54 /* How many data we try to prebuffer
  55  * XXX it should be small to avoid useless latency but big enough for
  56  * efficient demux probing */
  57 #define STREAM_CACHE_PREBUFFER_SIZE (128)
  58
  59 /* Method:
  60  *  - We use ring buffers, only one if unseekable, all if seekable
  61  *  - Upon seek date current ring, then search if one ring match the pos,
  62  *      yes: switch to it, seek the access to match the end of the ring
  63  *      no: search the ring with i_end the closer to i_pos,
  64  *          if close enough, read data and use this ring
  65  *          else use the oldest ring, seek and use it.
  66  *
  67  *  TODO: - with access non seekable: use all space available for only one ring, but
  68  *          we have to support seekable/non-seekable switch on the fly.
  69  *        - compute a good value for i_read_size
  70  *        - ?
  71  */
  72 #define STREAM_READ_ATONCE 1024
  73 #define STREAM_CACHE_TRACK_SIZE (STREAM_CACHE_SIZE/STREAM_CACHE_TRACK)
  74
  75 typedef struct
  76 {
  77     mtime_t date;
  78
  79     uint64_t i_start;
  80     uint64_t i_end;
  81
  82     uint8_t *p_buffer;
  83
  84 } stream_track_t;
  85
  86 struct stream_sys_t
  87 {
  88     uint64_t     i_pos;      /* Current reading offset */
  89
  90     unsigned     i_offset;   /* Buffer offset in the current track */
  91     int          i_tk;       /* Current track */
  92     stream_track_t tk[STREAM_CACHE_TRACK];
  93
  94     /* Global buffer */
  95     uint8_t     *p_buffer;
  96
  97     /* */
  98     unsigned     i_used; /* Used since last read */
  99     unsigned     i_read_size;
 100
 101     struct
 102     {
 103         /* Stat about reading data */
 104         uint64_t i_read_count;
 105         uint64_t i_bytes;
 106         uint64_t i_read_time;
 107     } stat;
 108 };
 109
 110 static int AStreamRefillStream(stream_t *s)
 111 {
 112     stream_sys_t *sys = s->p_sys;
 113     stream_track_t *tk = &sys->tk[sys->i_tk];
 114
 115     /* We read but won't increase i_start after initial start + offset */
 116     int i_toread =
 117         __MIN(sys->i_used, STREAM_CACHE_TRACK_SIZE -
 118                (tk->i_end - tk->i_start - sys->i_offset));
 119
 120     if (i_toread <= 0) return VLC_SUCCESS; /* EOF */
 121
 122 #ifdef STREAM_DEBUG
 123     msg_Dbg(s, "AStreamRefillStream: used=%d toread=%d",
 124                  sys->i_used, i_toread);
 125 #endif
 126
 127     mtime_t start = mdate();
 128     while (i_toread > 0)
 129     {
 130         int i_off = tk->i_end % STREAM_CACHE_TRACK_SIZE;
 131         int i_read;
 132
 133         if (vlc_killed())
 134             return VLC_EGENERIC;
 135
 136         i_read = __MIN(i_toread, STREAM_CACHE_TRACK_SIZE - i_off);
 137         i_read = vlc_stream_Read(s->p_source, &tk->p_buffer[i_off], i_read);
 138
 139         /* msg_Dbg(s, "AStreamRefillStream: read=%d", i_read); */
 140         if (i_read <  0)
 141         {
 142             continue;
 143         }
 144         else if (i_read == 0)
 145             return VLC_SUCCESS;
 146
 147         /* Update end */
 148         tk->i_end += i_read;
 149
 150         /* Windows of STREAM_CACHE_TRACK_SIZE */
 151         if (tk->i_start + STREAM_CACHE_TRACK_SIZE < tk->i_end)
 152         {
 153             unsigned i_invalid = tk->i_end - tk->i_start - STREAM_CACHE_TRACK_SIZE;
 154
 155             tk->i_start += i_invalid;
 156             sys->i_offset -= i_invalid;
 157         }
 158
 159         i_toread -= i_read;
 160         sys->i_used -= i_read;
 161
 162         sys->stat.i_bytes += i_read;
 163         sys->stat.i_read_count++;
 164     }
 165
 166     sys->stat.i_read_time += mdate() - start;
 167     return VLC_SUCCESS;
 168 }
 169
 170 static void AStreamPrebufferStream(stream_t *s)
 171 {
 172     stream_sys_t *sys = s->p_sys;
 173     mtime_t start = mdate();
 174     bool first = true;
 175
 176     msg_Dbg(s, "starting pre-buffering");
 177     for (;;)
 178     {
 179         stream_track_t *tk = &sys->tk[sys->i_tk];
 180         mtime_t now = mdate();
 181
 182         int i_read;
 183         int i_buffered = tk->i_end - tk->i_start;
 184
 185         if (vlc_killed() || i_buffered >= STREAM_CACHE_PREBUFFER_SIZE)
 186         {
 187             int64_t i_byterate;
 188
 189             /* Update stat */
 190             sys->stat.i_bytes = i_buffered;
 191             sys->stat.i_read_time = now - start;
 192             i_byterate = (CLOCK_FREQ * sys->stat.i_bytes) /
 193                          (sys->stat.i_read_time+1);
 194
 195             msg_Dbg(s, "pre-buffering done %"PRId64" bytes in %"PRId64"s - "
 196                     "%"PRId64" KiB/s", sys->stat.i_bytes,
 197                     sys->stat.i_read_time / CLOCK_FREQ, i_byterate / 1024);
 198             break;
 199         }
 200
 201         i_read = STREAM_CACHE_TRACK_SIZE - i_buffered;
 202         i_read = __MIN((int)sys->i_read_size, i_read);
 203         i_read = vlc_stream_Read(s->p_source, &tk->p_buffer[i_buffered],
 204                                  i_read);
 205         if (i_read <  0)
 206             continue;
 207         else if (i_read == 0)
 208             break;  /* EOF */
 209
 210         if (first)
 211         {
 212             msg_Dbg(s, "received first data after %"PRId64" ms",
 213                     (mdate() - start) / 1000);
 214             first = false;
 215         }
 216
 217         tk->i_end += i_read;
 218         sys->stat.i_read_count++;
 219     }
 220 }
 221
 222 /****************************************************************************
 223  * AStreamControlReset:
 224  ****************************************************************************/
 225 static void AStreamControlReset(stream_t *s)
 226 {
 227     stream_sys_t *sys = s->p_sys;
 228
 229     sys->i_pos = 0;
 230
 231     /* Setup our tracks */
 232     sys->i_offset = 0;
 233     sys->i_tk     = 0;
 234     sys->i_used   = 0;
 235
 236     for (unsigned i = 0; i < STREAM_CACHE_TRACK; i++)
 237     {
 238         sys->tk[i].date  = 0;
 239         sys->tk[i].i_start = sys->i_pos;
 240         sys->tk[i].i_end   = sys->i_pos;
 241     }
 242
 243     /* Do the prebuffering */
 244     AStreamPrebufferStream(s);
 245 }
 246
 247 static ssize_t AStreamReadStream(stream_t *s, void *buf, size_t len)
 248 {
 249     stream_sys_t *sys = s->p_sys;
 250     stream_track_t *tk = &sys->tk[sys->i_tk];
 251
 252     if (tk->i_start >= tk->i_end)
 253         return 0; /* EOF */
 254
 255 #ifdef STREAM_DEBUG
 256     msg_Dbg(s, "AStreamReadStream: %zd pos=%"PRId64" tk=%d start=%"PRId64
 257             " offset=%d end=%"PRId64, len, sys->i_pos, sys->i_tk,
 258             tk->i_start, sys->i_offset, tk->i_end);
 259 #endif
 260
 261     unsigned i_off = (tk->i_start + sys->i_offset) % STREAM_CACHE_TRACK_SIZE;
 262     size_t i_current = __MIN(tk->i_end - tk->i_start - sys->i_offset,
 263                              STREAM_CACHE_TRACK_SIZE - i_off);
 264     ssize_t i_copy = __MIN(i_current, len);
 265     if (i_copy <= 0)
 266         return 0; /* EOF */
 267
 268     /* Copy data */
 269     /* msg_Dbg(s, "AStreamReadStream: copy %zd", i_copy); */
 270     if (buf != NULL)
 271         memcpy(buf, &tk->p_buffer[i_off], i_copy);
 272     sys->i_offset += i_copy;
 273
 274     /* Update pos now */
 275     sys->i_pos += i_copy;
 276
 277     /* */
 278     sys->i_used += i_copy;
 279
 280     if (tk->i_end + i_copy <= tk->i_start + sys->i_offset + len)
 281     {
 282         const size_t i_read_requested = VLC_CLIP(len - i_copy,
 283                                                  STREAM_READ_ATONCE / 2,
 284                                                  STREAM_READ_ATONCE * 10);
 285         if (sys->i_used < i_read_requested)
 286             sys->i_used = i_read_requested;
 287
 288         AStreamRefillStream(s);
 289     }
 290
 291     return i_copy;
 292 }
 293
 294 static int AStreamSeekStream(stream_t *s, uint64_t i_pos)
 295 {
 296     stream_sys_t *sys = s->p_sys;
 297     stream_track_t *p_current = &sys->tk[sys->i_tk];
 298
 299     if (p_current->i_start >= p_current->i_end  && i_pos >= p_current->i_end)
 300         return 0; /* EOF */
 301
 302 #ifdef STREAM_DEBUG
 303     msg_Dbg(s, "AStreamSeekStream: to %"PRId64" pos=%"PRId64
 304              " tk=%d start=%"PRId64" offset=%d end=%"PRId64,
 305              i_pos, sys->i_pos, sys->i_tk, p_current->i_start,
 306              sys->i_offset, p_current->i_end);
 307 #endif
 308
 309     bool   b_aseek;
 310     vlc_stream_Control(s->p_source, STREAM_CAN_SEEK, &b_aseek);
 311     if (!b_aseek && i_pos < p_current->i_start)
 312     {
 313         msg_Warn(s, "AStreamSeekStream: can't seek");
 314         return VLC_EGENERIC;
 315     }
 316
 317     bool   b_afastseek;
 318     vlc_stream_Control(s->p_source, STREAM_CAN_FASTSEEK, &b_afastseek);
 319
 320     /* FIXME compute seek cost (instead of static 'stupid' value) */
 321     uint64_t i_skip_threshold;
 322     if (b_aseek)
 323         i_skip_threshold = b_afastseek ? 128 : 3 * sys->i_read_size;
 324     else
 325         i_skip_threshold = INT64_MAX;
 326
 327     /* Date the current track */
 328     p_current->date = mdate();
 329
 330     /* Search a new track slot */
 331     stream_track_t *tk = NULL;
 332     int i_tk_idx = -1;
 333
 334     /* Prefer the current track */
 335     if (p_current->i_start <= i_pos && i_pos <= p_current->i_end + i_skip_threshold)
 336     {
 337         tk = p_current;
 338         i_tk_idx = sys->i_tk;
 339     }
 340     if (!tk)
 341     {
 342         /* Try to maximize already read data */
 343         for (int i = 0; i < STREAM_CACHE_TRACK; i++)
 344         {
 345             stream_track_t *t = &sys->tk[i];
 346
 347             if (t->i_start > i_pos || i_pos > t->i_end)
 348                 continue;
 349
 350             if (!tk || tk->i_end < t->i_end)
 351             {
 352                 tk = t;
 353                 i_tk_idx = i;
 354             }
 355         }
 356     }
 357     if (!tk)
 358     {
 359         /* Use the oldest unused */
 360         for (int i = 0; i < STREAM_CACHE_TRACK; i++)
 361         {
 362             stream_track_t *t = &sys->tk[i];
 363
 364             if (!tk || tk->date > t->date)
 365             {
 366                 tk = t;
 367                 i_tk_idx = i;
 368             }
 369         }
 370     }
 371     assert(i_tk_idx >= 0 && i_tk_idx < STREAM_CACHE_TRACK);
 372
 373     if (tk != p_current)
 374         i_skip_threshold = 0;
 375     if (tk->i_start <= i_pos && i_pos <= tk->i_end + i_skip_threshold)
 376     {
 377 #ifdef STREAM_DEBUG
 378         msg_Err(s, "AStreamSeekStream: reusing %d start=%"PRId64
 379                  " end=%"PRId64"(%s)",
 380                  i_tk_idx, tk->i_start, tk->i_end,
 381                  tk != p_current ? "seek" : i_pos > tk->i_end ? "skip" : "noseek");
 382 #endif
 383         if (tk != p_current)
 384         {
 385             assert(b_aseek);
 386
 387             /* Seek at the end of the buffer
 388              * TODO it is stupid to seek now, it would be better to delay it
 389              */
 390             if (vlc_stream_Seek(s->p_source, tk->i_end))
 391             {
 392                 msg_Err(s, "AStreamSeekStream: hard seek failed");
 393                 return VLC_EGENERIC;
 394             }
 395         }
 396         else if (i_pos > tk->i_end)
 397         {
 398             uint64_t i_skip = i_pos - tk->i_end;
 399             while (i_skip > 0)
 400             {
 401                 const int i_read_max = __MIN(10 * STREAM_READ_ATONCE, i_skip);
 402                 int i_read = 0;
 403                 if ((i_read = AStreamReadStream(s, NULL, i_read_max)) < 0)
 404                 {
 405                     msg_Err(s, "AStreamSeekStream: skip failed");
 406                     return VLC_EGENERIC;
 407                 } else if (i_read == 0)
 408                     return VLC_SUCCESS; /* EOF */
 409                 i_skip -= i_read_max;
 410             }
 411         }
 412     }
 413     else
 414     {
 415 #ifdef STREAM_DEBUG
 416         msg_Err(s, "AStreamSeekStream: hard seek");
 417 #endif
 418         /* Nothing good, seek and choose oldest segment */
 419         if (vlc_stream_Seek(s->p_source, i_pos))
 420         {
 421             msg_Err(s, "AStreamSeekStream: hard seek failed");
 422             return VLC_EGENERIC;
 423         }
 424
 425         tk->i_start = i_pos;
 426         tk->i_end   = i_pos;
 427     }
 428     sys->i_offset = i_pos - tk->i_start;
 429     sys->i_tk = i_tk_idx;
 430     sys->i_pos = i_pos;
 431
 432     /* If there is not enough data left in the track, refill  */
 433     /* TODO How to get a correct value for
 434      *    - refilling threshold
 435      *    - how much to refill
 436      */
 437     if (tk->i_end < tk->i_start + sys->i_offset + sys->i_read_size)
 438     {
 439         if (sys->i_used < STREAM_READ_ATONCE / 2)
 440             sys->i_used = STREAM_READ_ATONCE / 2;
 441
 442         if (AStreamRefillStream(s))
 443             return VLC_EGENERIC;
 444     }
 445     return VLC_SUCCESS;
 446 }
 447
 448 /****************************************************************************
 449  * AStreamControl:
 450  ****************************************************************************/
 451 static int AStreamControl(stream_t *s, int i_query, va_list args)
 452 {
 453     switch(i_query)
 454     {
 455         case STREAM_CAN_SEEK:
 456         case STREAM_CAN_FASTSEEK:
 457         case STREAM_CAN_PAUSE:
 458         case STREAM_CAN_CONTROL_PACE:
 459         case STREAM_IS_DIRECTORY:
 460         case STREAM_GET_SIZE:
 461         case STREAM_GET_PTS_DELAY:
 462         case STREAM_GET_TITLE_INFO:
 463         case STREAM_GET_TITLE:
 464         case STREAM_GET_SEEKPOINT:
 465         case STREAM_GET_META:
 466         case STREAM_GET_CONTENT_TYPE:
 467         case STREAM_GET_SIGNAL:
 468         case STREAM_GET_TAGS:
 469         case STREAM_SET_PAUSE_STATE:
 470         case STREAM_SET_PRIVATE_ID_STATE:
 471         case STREAM_SET_PRIVATE_ID_CA:
 472         case STREAM_GET_PRIVATE_ID_STATE:
 473             return vlc_stream_vaControl(s->p_source, i_query, args);
 474
 475         case STREAM_SET_TITLE:
 476         case STREAM_SET_SEEKPOINT:
 477         {
 478             int ret = vlc_stream_vaControl(s->p_source, i_query, args);
 479             if (ret == VLC_SUCCESS)
 480                 AStreamControlReset(s);
 481             return ret;
 482         }
 483
 484         case STREAM_SET_RECORD_STATE:
 485         default:
 486             msg_Err(s, "invalid vlc_stream_vaControl query=0x%x", i_query);
 487             return VLC_EGENERIC;
 488     }
 489     return VLC_SUCCESS;
 490 }
 491
 492 static int Open(vlc_object_t *obj)
 493 {
 494     stream_t *s = (stream_t *)obj;
 495
 496     stream_sys_t *sys = malloc(sizeof (*sys));
 497     if (unlikely(sys == NULL))
 498         return VLC_ENOMEM;
 499
 500     /* Common field */
 501     sys->i_pos = 0;
 502
 503     /* Stats */
 504     sys->stat.i_bytes = 0;
 505     sys->stat.i_read_time = 0;
 506     sys->stat.i_read_count = 0;
 507
 508     msg_Dbg(s, "Using stream method for AStream*");
 509
 510     /* Allocate/Setup our tracks */
 511     sys->i_offset = 0;
 512     sys->i_tk     = 0;
 513     sys->p_buffer = malloc(STREAM_CACHE_SIZE);
 514     if (sys->p_buffer == NULL)
 515     {
 516         free(sys);
 517         return VLC_ENOMEM;
 518     }
 519
 520     sys->i_used   = 0;
 521     sys->i_read_size = STREAM_READ_ATONCE;
 522 #if STREAM_READ_ATONCE < 256
 523 #   error "Invalid STREAM_READ_ATONCE value"
 524 #endif
 525
 526     for (unsigned i = 0; i < STREAM_CACHE_TRACK; i++)
 527     {
 528         sys->tk[i].date  = 0;
 529         sys->tk[i].i_start = sys->i_pos;
 530         sys->tk[i].i_end   = sys->i_pos;
 531         sys->tk[i].p_buffer = &sys->p_buffer[i * STREAM_CACHE_TRACK_SIZE];
 532     }
 533
 534     s->p_sys = sys;
 535
 536     /* Do the prebuffering */
 537     AStreamPrebufferStream(s);
 538
 539     if (sys->tk[sys->i_tk].i_end <= 0)
 540     {
 541         msg_Err(s, "cannot pre fill buffer");
 542         free(sys->p_buffer);
 543         free(sys);
 544         return VLC_EGENERIC;
 545     }
 546
 547     s->pf_read = AStreamReadStream;
 548     s->pf_seek = AStreamSeekStream;
 549     s->pf_control = AStreamControl;
 550     return VLC_SUCCESS;
 551 }
 552
 553 /****************************************************************************
 554  * AStreamDestroy:
 555  ****************************************************************************/
 556 static void Close(vlc_object_t *obj)
 557 {
 558     stream_t *s = (stream_t *)obj;
 559     stream_sys_t *sys = s->p_sys;
 560
 561     free(sys->p_buffer);
 562     free(sys);
 563 }
 564
 565 vlc_module_begin()
 566     set_category(CAT_INPUT)
 567     set_subcategory(SUBCAT_INPUT_STREAM_FILTER)
 568     set_capability("stream_filter", 0)
 569
 570     set_description(N_("Byte stream cache"))
 571     set_callbacks(Open, Close)
 572 vlc_module_end()