apps/plugins/mpegplayer/disk_buf.c

   1 /***************************************************************************
   2  *             __________               __   ___.
   3  *   Open      \______   \ ____   ____ |  | _\_ |__   _______  ___
   4  *   Source     |       _//  _ \_/ ___\|  |/ /| __ \ /  _ \  \/  /
   5  *   Jukebox    |    |   (  <_> )  \___|    < | \_\ (  <_> > <  <
   6  *   Firmware   |____|_  /\____/ \___  >__|_ \|___  /\____/__/\_ \
   7  *                     \/            \/     \/    \/            \/
   8  * $Id$
   9  *
  10  * mpegplayer buffering routines
  11  *
  12  * Copyright (c) 2007 Michael Sevakis
  13  *
  14  * This program is free software; you can redistribute it and/or
  15  * modify it under the terms of the GNU General Public License
  16  * as published by the Free Software Foundation; either version 2
  17  * of the License, or (at your option) any later version.
  18  *
  19  * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
  20  * KIND, either express or implied.
  21  *
  22  ****************************************************************************/
  23 #include "plugin.h"
  24 #include "mpegplayer.h"
  25 #include <system.h>
  26
  27 static struct mutex disk_buf_mtx SHAREDBSS_ATTR;
  28 static struct event_queue disk_buf_queue SHAREDBSS_ATTR;
  29 static struct queue_sender_list disk_buf_queue_send SHAREDBSS_ATTR;
  30 static uint32_t disk_buf_stack[DEFAULT_STACK_SIZE*2/sizeof(uint32_t)];
  31
  32 struct disk_buf disk_buf SHAREDBSS_ATTR;
  33 static struct list_item nf_list;
  34
  35 static inline void disk_buf_lock(void)
  36 {
  37     rb->mutex_lock(&disk_buf_mtx);
  38 }
  39
  40 static inline void disk_buf_unlock(void)
  41 {
  42     rb->mutex_unlock(&disk_buf_mtx);
  43 }
  44
  45 static inline void disk_buf_on_clear_data_notify(struct stream_hdr *sh)
  46 {
  47     DEBUGF("DISK_BUF_CLEAR_DATA_NOTIFY: 0x%02X (cleared)\n",
  48             STR_FROM_HEADER(sh)->id);
  49     list_remove_item(&sh->nf);
  50 }
  51
  52 static int disk_buf_on_data_notify(struct stream_hdr *sh)
  53 {
  54     DEBUGF("DISK_BUF_DATA_NOTIFY: 0x%02X ", STR_FROM_HEADER(sh)->id);
  55
  56     if (sh->win_left <= sh->win_right)
  57     {
  58         /* Check if the data is already ready already */
  59         if (disk_buf_is_data_ready(sh, 0))
  60         {
  61             /* It was - don't register */
  62             DEBUGF("(was ready)\n"
  63                    "  swl:%lu swr:%lu\n"
  64                    "  dwl:%lu dwr:%lu\n",
  65                    sh->win_left, sh->win_right,
  66                    disk_buf.win_left, disk_buf.win_right);
  67             /* Be sure it's not listed though if multiple requests were made */
  68             list_remove_item(&sh->nf);
  69             return DISK_BUF_NOTIFY_OK;
  70         }
  71
  72         switch (disk_buf.state)
  73         {
  74         case TSTATE_DATA:
  75         case TSTATE_BUFFERING:
  76         case TSTATE_INIT:
  77             disk_buf.state = TSTATE_BUFFERING;
  78             list_add_item(&nf_list, &sh->nf);
  79             DEBUGF("(registered)\n"
  80                    "  swl:%lu swr:%lu\n"
  81                    "  dwl:%lu dwr:%lu\n",
  82                    sh->win_left, sh->win_right,
  83                    disk_buf.win_left, disk_buf.win_right);
  84             return DISK_BUF_NOTIFY_REGISTERED;
  85         }
  86     }
  87
  88     DEBUGF("(error)\n");
  89     return DISK_BUF_NOTIFY_ERROR;
  90 }
  91
  92 static bool check_data_notifies_callback(struct list_item *item,
  93                                          intptr_t data)
  94 {
  95     struct stream_hdr *sh = TYPE_FROM_MEMBER(struct stream_hdr, item, nf);
  96
  97     if (disk_buf_is_data_ready(sh, 0))
  98     {
  99         /* Remove from list then post notification - post because send
 100          * could result in a wait for each thread to finish resulting
 101          * in deadlock */
 102         list_remove_item(item);
 103         str_post_msg(STR_FROM_HEADER(sh), DISK_BUF_DATA_NOTIFY, 0);
 104         DEBUGF("DISK_BUF_DATA_NOTIFY: 0x%02X (notified)\n",
 105                STR_FROM_HEADER(sh)->id);
 106     }
 107
 108     return true;
 109     (void)data;
 110 }
 111
 112 /* Check registered streams and notify them if their data is available */
 113 static void check_data_notifies(void)
 114 {
 115     list_enum_items(&nf_list, check_data_notifies_callback, 0);
 116 }
 117
 118 /* Clear all registered notifications - do not post them */
 119 static inline void clear_data_notifies(void)
 120 {
 121     list_clear_all(&nf_list);
 122 }
 123
 124 /* Background buffering when streaming */
 125 static inline void disk_buf_buffer(void)
 126 {
 127     struct stream_window sw;
 128
 129     switch (disk_buf.state)
 130     {
 131     default:
 132     {
 133         size_t wm;
 134         uint32_t time;
 135
 136         /* Get remaining minimum data based upon the stream closest to the
 137          * right edge of the window */
 138         if (!stream_get_window(&sw))
 139             break;
 140
 141         time = stream_get_ticks(NULL);
 142         wm = muldiv_uint32(5*CLOCK_RATE, sw.right - disk_buf.pos_last,
 143                            time - disk_buf.time_last);
 144         wm = MIN(wm, (size_t)disk_buf.size);
 145         wm = MAX(wm, DISK_BUF_LOW_WATERMARK);
 146
 147         disk_buf.time_last = time;
 148         disk_buf.pos_last = sw.right;
 149
 150         /* Fast attack, slow decay */
 151         disk_buf.low_wm = (wm > (size_t)disk_buf.low_wm) ?
 152             wm : AVERAGE(disk_buf.low_wm, wm, 16);
 153
 154 #if 0
 155         rb->splashf(0, "*%10ld %10ld", disk_buf.low_wm,
 156                    disk_buf.win_right - sw.right);
 157 #endif
 158
 159         if (disk_buf.win_right - sw.right > disk_buf.low_wm)
 160             break;
 161
 162         disk_buf.state = TSTATE_BUFFERING;
 163         } /* default: */
 164
 165         /* Fall-through */
 166     case TSTATE_BUFFERING:
 167     {
 168         ssize_t len, n;
 169         uint32_t tag, *tag_p;
 170
 171         /* Limit buffering up to the stream with the least progress */
 172         if (!stream_get_window(&sw))
 173         {
 174             disk_buf.state = TSTATE_DATA;
 175             rb->storage_sleep();
 176             break;
 177         }
 178
 179         /* Wrap pointer */
 180         if (disk_buf.tail >= disk_buf.end)
 181             disk_buf.tail = disk_buf.start;
 182
 183         len = disk_buf.size - disk_buf.win_right + sw.left;
 184
 185         if (len < DISK_BUF_PAGE_SIZE)
 186         {
 187             /* Free space is less than one page */
 188             disk_buf.state = TSTATE_DATA;
 189             disk_buf.low_wm = DISK_BUF_LOW_WATERMARK;
 190             rb->storage_sleep();
 191             break;
 192         }
 193
 194         len = disk_buf.tail - disk_buf.start;
 195         tag = MAP_OFFSET_TO_TAG(disk_buf.win_right);
 196         tag_p = &disk_buf.cache[len >> DISK_BUF_PAGE_SHIFT];
 197
 198         if (*tag_p != tag)
 199         {
 200             if (disk_buf.need_seek)
 201             {
 202                 rb->lseek(disk_buf.in_file, disk_buf.win_right, SEEK_SET);
 203                 disk_buf.need_seek = false;
 204             }
 205
 206             n = rb->read(disk_buf.in_file, disk_buf.tail, DISK_BUF_PAGE_SIZE);
 207
 208             if (n <= 0)
 209             {
 210                 /* Error or end of stream */
 211                 disk_buf.state = TSTATE_EOS;
 212                 rb->storage_sleep();
 213                 break;
 214             }
 215
 216             if (len < DISK_GUARDBUF_SIZE)
 217             {
 218                 /* Autoguard guard-o-rama - maintain guardbuffer coherency */
 219                 rb->memcpy(disk_buf.end + len, disk_buf.tail,
 220                            MIN(DISK_GUARDBUF_SIZE - len, n));
 221             }
 222
 223             /* Update the cache entry for this page */
 224             *tag_p = tag;
 225         }
 226         else
 227         {
 228             /* Skipping a read */
 229             n = MIN(DISK_BUF_PAGE_SIZE,
 230                     disk_buf.filesize - disk_buf.win_right);
 231             disk_buf.need_seek = true;
 232         }
 233
 234         disk_buf.tail += DISK_BUF_PAGE_SIZE;
 235
 236         /* Keep left edge moving forward */
 237
 238         /* Advance right edge in temp variable first, then move
 239          * left edge if overflow would occur. This avoids a stream
 240          * thinking its data might be available when it actually
 241          * may not end up that way after a slide of the window. */
 242         len = disk_buf.win_right + n;
 243
 244         if (len - disk_buf.win_left > disk_buf.size)
 245             disk_buf.win_left += n;
 246
 247         disk_buf.win_right = len;
 248
 249         /* Continue buffering until filled or file end */
 250         rb->yield();
 251         } /* TSTATE_BUFFERING: */
 252
 253     case TSTATE_EOS:
 254         break;
 255     } /* end switch */
 256 }
 257
 258 static void disk_buf_on_reset(ssize_t pos)
 259 {
 260     int page;
 261     uint32_t tag;
 262     off_t anchor;
 263
 264     disk_buf.state = TSTATE_INIT;
 265     disk_buf.status = STREAM_STOPPED;
 266     clear_data_notifies();
 267
 268     if (pos >= disk_buf.filesize)
 269     {
 270         /* Anchor on page immediately following the one containing final
 271          * data */
 272         anchor = disk_buf.file_pages * DISK_BUF_PAGE_SIZE;
 273         disk_buf.win_left = disk_buf.filesize;
 274     }
 275     else
 276     {
 277         anchor = pos & ~DISK_BUF_PAGE_MASK;
 278         disk_buf.win_left = anchor;
 279     }
 280
 281     /* Collect all valid data already buffered that is contiguous with the
 282      * current position - probe to left, then to right */
 283     if (anchor > 0)
 284     {
 285         page = MAP_OFFSET_TO_PAGE(anchor);
 286         tag = MAP_OFFSET_TO_TAG(anchor);
 287
 288         do
 289         {
 290             if (--tag, --page < 0)
 291                 page = disk_buf.pgcount - 1;
 292
 293             if (disk_buf.cache[page] != tag)
 294                 break;
 295
 296             disk_buf.win_left = tag << DISK_BUF_PAGE_SHIFT;
 297         }
 298         while (tag > 0);
 299     }
 300
 301     if (anchor < disk_buf.filesize)
 302     {
 303         page = MAP_OFFSET_TO_PAGE(anchor);
 304         tag = MAP_OFFSET_TO_TAG(anchor);
 305
 306         do
 307         {
 308             if (disk_buf.cache[page] != tag)
 309                 break;
 310
 311             if (++tag, ++page >= disk_buf.pgcount)
 312                 page = 0;
 313
 314             anchor += DISK_BUF_PAGE_SIZE;
 315         }
 316         while (anchor < disk_buf.filesize);
 317     }
 318
 319     if (anchor >= disk_buf.filesize)
 320     {
 321         disk_buf.win_right = disk_buf.filesize;
 322         disk_buf.state = TSTATE_EOS;
 323     }
 324     else
 325     {
 326         disk_buf.win_right = anchor;
 327     }
 328
 329     disk_buf.tail = disk_buf.start + MAP_OFFSET_TO_BUFFER(anchor);
 330
 331     DEBUGF("disk buf reset\n"
 332            "  dwl:%ld dwr:%ld\n",
 333            disk_buf.win_left, disk_buf.win_right);
 334
 335     /* Next read position is at right edge */
 336     rb->lseek(disk_buf.in_file, disk_buf.win_right, SEEK_SET);
 337     disk_buf.need_seek = false;
 338
 339     disk_buf_reply_msg(disk_buf.win_right - disk_buf.win_left);
 340 }
 341
 342 static void disk_buf_on_stop(void)
 343 {
 344     bool was_buffering = disk_buf.state == TSTATE_BUFFERING;
 345
 346     disk_buf.state = TSTATE_EOS;
 347     disk_buf.status = STREAM_STOPPED;
 348     clear_data_notifies();
 349
 350     disk_buf_reply_msg(was_buffering);
 351 }
 352
 353 static void disk_buf_on_play_pause(bool play, bool forcefill)
 354 {
 355     struct stream_window sw;
 356
 357     if (disk_buf.state != TSTATE_EOS)
 358     {
 359         if (forcefill)
 360         {
 361             /* Force buffer filling to top */
 362             disk_buf.state = TSTATE_BUFFERING;
 363         }
 364         else if (disk_buf.state != TSTATE_BUFFERING)
 365         {
 366             /* If not filling already, simply monitor */
 367             disk_buf.state = TSTATE_DATA;
 368         }
 369     }
 370     /* else end of stream - no buffering to do */
 371
 372     disk_buf.pos_last = stream_get_window(&sw) ? sw.right : 0;
 373     disk_buf.time_last = stream_get_ticks(NULL);
 374
 375     disk_buf.status = play ? STREAM_PLAYING : STREAM_PAUSED;
 376 }
 377
 378 static int disk_buf_on_load_range(struct dbuf_range *rng)
 379 {
 380     uint32_t tag = rng->tag_start;
 381     uint32_t tag_end = rng->tag_end;
 382     int page = rng->pg_start;
 383
 384     /* Check if a seek is required */
 385     bool need_seek = rb->lseek(disk_buf.in_file, 0, SEEK_CUR)
 386                         != (off_t)(tag << DISK_BUF_PAGE_SHIFT);
 387
 388     do
 389     {
 390         uint32_t *tag_p = &disk_buf.cache[page];
 391
 392         if (*tag_p != tag)
 393         {
 394             /* Page not cached - load it */
 395             ssize_t o, n;
 396
 397             if (need_seek)
 398             {
 399                 rb->lseek(disk_buf.in_file, tag << DISK_BUF_PAGE_SHIFT,
 400                           SEEK_SET);
 401                 need_seek = false;
 402             }
 403
 404             o = page << DISK_BUF_PAGE_SHIFT;
 405             n = rb->read(disk_buf.in_file, disk_buf.start + o,
 406                          DISK_BUF_PAGE_SIZE);
 407
 408             if (n < 0)
 409             {
 410                 /* Read error */
 411                 return DISK_BUF_NOTIFY_ERROR;
 412             }
 413
 414             if (n == 0)
 415             {
 416                 /* End of file */
 417                 break;
 418             }
 419
 420             if (o < DISK_GUARDBUF_SIZE)
 421             {
 422                 /* Autoguard guard-o-rama - maintain guardbuffer coherency */
 423                 rb->memcpy(disk_buf.end + o, disk_buf.start + o,
 424                            MIN(DISK_GUARDBUF_SIZE - o, n));
 425             }
 426
 427             /* Update the cache entry */
 428             *tag_p = tag;
 429         }
 430         else
 431         {
 432             /* Skipping a disk read - must seek on next one */
 433             need_seek = true;
 434         }
 435
 436         if (++page >= disk_buf.pgcount)
 437             page = 0;
 438     }
 439     while (++tag <= tag_end);
 440
 441     return DISK_BUF_NOTIFY_OK;
 442 }
 443
 444 static void disk_buf_thread(void)
 445 {
 446     struct queue_event ev;
 447
 448     disk_buf.state = TSTATE_EOS;
 449     disk_buf.status = STREAM_STOPPED;
 450
 451     while (1)
 452     {
 453         if (disk_buf.state != TSTATE_EOS)
 454         {
 455             /* Poll buffer status and messages */
 456             rb->queue_wait_w_tmo(disk_buf.q, &ev,
 457                                  disk_buf.state == TSTATE_BUFFERING ?
 458                                     0 : HZ/5);
 459         }
 460         else
 461         {
 462             /* Sit idle and wait for commands */
 463             rb->queue_wait(disk_buf.q, &ev);
 464         }
 465
 466         switch (ev.id)
 467         {
 468         case SYS_TIMEOUT:
 469             if (disk_buf.state == TSTATE_EOS)
 470                 break;
 471
 472             disk_buf_buffer();
 473
 474             /* Check for any due notifications if any are pending */
 475             if (nf_list.next != NULL)
 476                 check_data_notifies();
 477
 478             /* Still more data left? */
 479             if (disk_buf.state != TSTATE_EOS)
 480                 continue;
 481
 482             /* Nope - end of stream */
 483             break;
 484
 485         case DISK_BUF_CACHE_RANGE:
 486             disk_buf_reply_msg(disk_buf_on_load_range(
 487                                 (struct dbuf_range *)ev.data));
 488             break;
 489
 490         case STREAM_RESET:
 491             disk_buf_on_reset(ev.data);
 492             break;
 493
 494         case STREAM_STOP:
 495             disk_buf_on_stop();
 496             break;
 497
 498         case STREAM_PAUSE:
 499         case STREAM_PLAY:
 500             disk_buf_on_play_pause(ev.id == STREAM_PLAY, ev.data != 0);
 501             disk_buf_reply_msg(1);
 502             break;
 503
 504         case STREAM_QUIT:
 505             disk_buf.state = TSTATE_EOS;
 506             return;
 507
 508         case DISK_BUF_DATA_NOTIFY:
 509             disk_buf_reply_msg(disk_buf_on_data_notify(
 510                                 (struct stream_hdr *)ev.data));
 511             break;
 512
 513         case DISK_BUF_CLEAR_DATA_NOTIFY:
 514             disk_buf_on_clear_data_notify((struct stream_hdr *)ev.data);
 515             disk_buf_reply_msg(1);
 516             break;
 517         }
 518     }
 519 }
 520
 521 /* Caches some data from the current file */
 522 static int disk_buf_probe(off_t start, size_t length,
 523                           void **p, size_t *outlen)
 524 {
 525     off_t end;
 526     uint32_t tag, tag_end;
 527     int page;
 528
 529     /* Can't read past end of file */
 530     if (length > (size_t)(disk_buf.filesize - disk_buf.offset))
 531     {
 532         length = disk_buf.filesize - disk_buf.offset;
 533     }
 534
 535     /* Can't cache more than the whole buffer size */
 536     if (length > (size_t)disk_buf.size)
 537     {
 538         length = disk_buf.size;
 539     }
 540     /* Zero-length probes permitted */
 541
 542     end = start + length;
 543
 544     /* Prepare the range probe */
 545     tag = MAP_OFFSET_TO_TAG(start);
 546     tag_end = MAP_OFFSET_TO_TAG(end);
 547     page = MAP_OFFSET_TO_PAGE(start);
 548
 549     /* If the end is on a page boundary, check one less or an extra
 550      * one will be probed */
 551     if (tag_end > tag && (end & DISK_BUF_PAGE_MASK) == 0)
 552     {
 553         tag_end--;
 554     }
 555
 556     if (p != NULL)
 557     {
 558         *p = disk_buf.start + (page << DISK_BUF_PAGE_SHIFT)
 559                 + (start & DISK_BUF_PAGE_MASK);
 560     }
 561
 562     if (outlen != NULL)
 563     {
 564         *outlen = length;
 565     }
 566
 567     /* Obtain initial load point. If all data was cached, no message is sent
 568      * otherwise begin on the first page that is not cached. Since we have to
 569      * send the message anyway, the buffering thread will determine what else
 570      * requires loading on its end in order to cache the specified range. */
 571     do
 572     {
 573         if (disk_buf.cache[page] != tag)
 574         {
 575             static struct dbuf_range rng IBSS_ATTR;
 576             DEBUGF("disk_buf: cache miss\n");
 577             rng.tag_start = tag;
 578             rng.tag_end = tag_end;
 579             rng.pg_start = page;
 580             return rb->queue_send(disk_buf.q, DISK_BUF_CACHE_RANGE,
 581                                   (intptr_t)&rng);
 582         }
 583
 584         if (++page >= disk_buf.pgcount)
 585             page = 0;
 586     }
 587     while (++tag <= tag_end);
 588
 589     return DISK_BUF_NOTIFY_OK;
 590 }
 591
 592 /* Attempt to get a pointer to size bytes on the buffer. Returns real amount of
 593  * data available as well as the size of non-wrapped data after *p. */
 594 ssize_t _disk_buf_getbuffer(size_t size, void **pp, void **pwrap, size_t *sizewrap)
 595 {
 596     disk_buf_lock();
 597
 598     if (disk_buf_probe(disk_buf.offset, size, pp, &size) == DISK_BUF_NOTIFY_OK)
 599     {
 600         if (pwrap && sizewrap)
 601         {
 602             uint8_t *p = (uint8_t *)*pp;
 603
 604             if (p + size > disk_buf.end + DISK_GUARDBUF_SIZE)
 605             {
 606                 /* Return pointer to wraparound and the size of same */
 607                 size_t nowrap = (disk_buf.end + DISK_GUARDBUF_SIZE) - p;
 608                 *pwrap = disk_buf.start + DISK_GUARDBUF_SIZE;
 609                 *sizewrap = size - nowrap;
 610             }
 611             else
 612             {
 613                 *pwrap = NULL;
 614                 *sizewrap = 0;
 615             }
 616         }
 617     }
 618     else
 619     {
 620         size = -1;
 621     }
 622
 623     disk_buf_unlock();
 624
 625     return size;
 626 }
 627
 628 /* Read size bytes of data into a buffer - advances the buffer pointer
 629  * and returns the real size read. */
 630 ssize_t disk_buf_read(void *buffer, size_t size)
 631 {
 632     uint8_t *p;
 633
 634     disk_buf_lock();
 635
 636     if (disk_buf_probe(disk_buf.offset, size, PUN_PTR(void **, &p),
 637                        &size) == DISK_BUF_NOTIFY_OK)
 638     {
 639         if (p + size > disk_buf.end + DISK_GUARDBUF_SIZE)
 640         {
 641             /* Read wraps */
 642             size_t nowrap = (disk_buf.end + DISK_GUARDBUF_SIZE) - p;
 643             rb->memcpy(buffer, p, nowrap);
 644             rb->memcpy(buffer + nowrap, disk_buf.start + DISK_GUARDBUF_SIZE,
 645                        size - nowrap);
 646         }
 647         else
 648         {
 649             /* Read wasn't wrapped or guardbuffer holds it */
 650             rb->memcpy(buffer, p, size);
 651         }
 652
 653         disk_buf.offset += size;
 654     }
 655     else
 656     {
 657         size = -1;
 658     }
 659
 660     disk_buf_unlock();
 661
 662     return size;
 663 }
 664
 665 off_t disk_buf_lseek(off_t offset, int whence)
 666 {
 667     disk_buf_lock();
 668
 669     /* The offset returned is the result of the current thread's action and
 670      * may be invalidated so a local result is returned and not the value
 671      * of disk_buf.offset directly */
 672     switch (whence)
 673     {
 674     case SEEK_SET:
 675         /* offset is just the offset */
 676         break;
 677     case SEEK_CUR:
 678         offset += disk_buf.offset;
 679         break;
 680     case SEEK_END:
 681         offset = disk_buf.filesize + offset;
 682         break;
 683     default:
 684         disk_buf_unlock();
 685         return -2; /* Invalid request */
 686     }
 687
 688     if (offset < 0 || offset > disk_buf.filesize)
 689     {
 690         offset = -3;
 691     }
 692     else
 693     {
 694         disk_buf.offset = offset;
 695     }
 696
 697     disk_buf_unlock();
 698
 699     return offset;
 700 }
 701
 702 /* Prepare the buffer to enter the streaming state. Evaluates the available
 703  * streaming window. */
 704 ssize_t disk_buf_prepare_streaming(off_t pos, size_t len)
 705 {
 706     disk_buf_lock();
 707
 708     if (pos < 0)
 709         pos = 0;
 710     else if (pos > disk_buf.filesize)
 711         pos = disk_buf.filesize;
 712
 713     DEBUGF("prepare streaming:\n  pos:%ld len:%lu\n", pos, len);
 714
 715     pos = disk_buf_lseek(pos, SEEK_SET);
 716     disk_buf_probe(pos, len, NULL, &len);
 717
 718     DEBUGF("  probe done: pos:%ld len:%lu\n", pos, len);
 719
 720     len = disk_buf_send_msg(STREAM_RESET, pos);
 721
 722     disk_buf_unlock();
 723
 724     return len;
 725 }
 726
 727 /* Set the streaming window to an arbitrary position within the file. Makes no
 728  * probes to validate data. Use after calling another function to cause data
 729  * to be cached and correct values are known. */
 730 ssize_t disk_buf_set_streaming_window(off_t left, off_t right)
 731 {
 732     ssize_t len;
 733
 734     disk_buf_lock();
 735
 736     if (left < 0)
 737         left = 0;
 738     else if (left > disk_buf.filesize)
 739         left = disk_buf.filesize;
 740
 741     if (left > right)
 742         right = left;
 743
 744     if (right > disk_buf.filesize)
 745         right = disk_buf.filesize;
 746
 747     disk_buf.win_left = left;
 748     disk_buf.win_right = right;
 749     disk_buf.tail = disk_buf.start + ((right + DISK_BUF_PAGE_SIZE-1) &
 750                                        ~DISK_BUF_PAGE_MASK) % disk_buf.size;
 751
 752     len = disk_buf.win_right - disk_buf.win_left;
 753
 754     disk_buf_unlock();
 755
 756     return len;
 757 }
 758
 759 void * disk_buf_offset2ptr(off_t offset)
 760 {
 761     if (offset < 0)
 762         offset = 0;
 763     else if (offset > disk_buf.filesize)
 764         offset = disk_buf.filesize;
 765
 766     return disk_buf.start + (offset % disk_buf.size);
 767 }
 768
 769 void disk_buf_close(void)
 770 {
 771     disk_buf_lock();
 772
 773     if (disk_buf.in_file >= 0)
 774     {
 775         rb->close(disk_buf.in_file);
 776         disk_buf.in_file = -1;
 777
 778         /* Invalidate entire cache */
 779         rb->memset(disk_buf.cache, 0xff,
 780                    disk_buf.pgcount*sizeof (*disk_buf.cache));
 781         disk_buf.file_pages = 0;
 782         disk_buf.filesize = 0;
 783         disk_buf.offset = 0;
 784     }
 785
 786     disk_buf_unlock();
 787 }
 788
 789 int disk_buf_open(const char *filename)
 790 {
 791     int fd;
 792
 793     disk_buf_lock();
 794
 795     disk_buf_close();
 796
 797     fd = rb->open(filename, O_RDONLY);
 798
 799     if (fd >= 0)
 800     {
 801         ssize_t filesize = rb->filesize(fd);
 802
 803         if (filesize <= 0)
 804         {
 805             rb->close(disk_buf.in_file);
 806         }
 807         else
 808         {
 809             disk_buf.filesize = filesize;
 810             /* Number of file pages rounded up toward +inf */
 811             disk_buf.file_pages = ((size_t)filesize + DISK_BUF_PAGE_SIZE-1)
 812                                     / DISK_BUF_PAGE_SIZE;
 813             disk_buf.in_file = fd;
 814         }
 815     }
 816
 817     disk_buf_unlock();
 818
 819     return fd;
 820 }
 821
 822 intptr_t disk_buf_send_msg(long id, intptr_t data)
 823 {
 824     return rb->queue_send(disk_buf.q, id, data);
 825 }
 826
 827 void disk_buf_post_msg(long id, intptr_t data)
 828 {
 829     rb->queue_post(disk_buf.q, id, data);
 830 }
 831
 832 void disk_buf_reply_msg(intptr_t retval)
 833 {
 834     rb->queue_reply(disk_buf.q, retval);
 835 }
 836
 837 bool disk_buf_init(void)
 838 {
 839     disk_buf.thread = 0;
 840     list_initialize(&nf_list);
 841
 842     rb->mutex_init(&disk_buf_mtx);
 843
 844     disk_buf.q = &disk_buf_queue;
 845     rb->queue_init(disk_buf.q, false);
 846
 847     disk_buf.state  = TSTATE_EOS;
 848     disk_buf.status = STREAM_STOPPED;
 849
 850     disk_buf.in_file = -1;
 851     disk_buf.filesize = 0;
 852     disk_buf.win_left = 0;
 853     disk_buf.win_right = 0;
 854     disk_buf.time_last = 0;
 855     disk_buf.pos_last = 0;
 856     disk_buf.low_wm = DISK_BUF_LOW_WATERMARK;
 857
 858     disk_buf.start = mpeg_malloc_all(&disk_buf.size, MPEG_ALLOC_DISKBUF);
 859     if (disk_buf.start == NULL)
 860         return false;
 861
 862 #ifdef PROC_NEEDS_CACHEALIGN
 863     CACHEALIGN_BUFFER(disk_buf.start, disk_buf.size);
 864     disk_buf.start = UNCACHED_ADDR(disk_buf.start);
 865 #endif
 866     disk_buf.size -= DISK_GUARDBUF_SIZE;
 867     disk_buf.pgcount = disk_buf.size / DISK_BUF_PAGE_SIZE;
 868
 869     /* Fit it as tightly as possible */
 870     while (disk_buf.pgcount*(sizeof (*disk_buf.cache) + DISK_BUF_PAGE_SIZE)
 871             > (size_t)disk_buf.size)
 872     {
 873         disk_buf.pgcount--;
 874     }
 875
 876     disk_buf.cache = (typeof (disk_buf.cache))disk_buf.start;
 877     disk_buf.start += sizeof (*disk_buf.cache)*disk_buf.pgcount;
 878     disk_buf.size = disk_buf.pgcount*DISK_BUF_PAGE_SIZE;
 879     disk_buf.end = disk_buf.start + disk_buf.size;
 880     disk_buf.tail = disk_buf.start;
 881
 882     DEBUGF("disk_buf info:\n"
 883            "  page count: %d\n"
 884            "  size:       %ld\n",
 885            disk_buf.pgcount, disk_buf.size);
 886
 887     rb->memset(disk_buf.cache, 0xff,
 888                disk_buf.pgcount*sizeof (*disk_buf.cache));
 889
 890     disk_buf.thread = rb->create_thread(
 891         disk_buf_thread, disk_buf_stack, sizeof(disk_buf_stack), 0,
 892         "mpgbuffer" IF_PRIO(, PRIORITY_BUFFERING) IF_COP(, CPU));
 893
 894     rb->queue_enable_queue_send(disk_buf.q, &disk_buf_queue_send,
 895                                 disk_buf.thread);
 896
 897     if (disk_buf.thread == 0)
 898         return false;
 899
 900     /* Wait for thread to initialize */
 901     disk_buf_send_msg(STREAM_NULL, 0);
 902
 903     return true;
 904 }
 905
 906 void disk_buf_exit(void)
 907 {
 908     if (disk_buf.thread != 0)
 909     {
 910         rb->queue_post(disk_buf.q, STREAM_QUIT, 0);
 911         rb->thread_wait(disk_buf.thread);
 912         disk_buf.thread = 0;
 913     }
 914 }