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