vorbis-tools/ogg123/buffer.c

   1 /********************************************************************
   2  *                                                                  *
   3  * THIS FILE IS PART OF THE OggVorbis SOFTWARE CODEC SOURCE CODE.   *
   4  * USE, DISTRIBUTION AND REPRODUCTION OF THIS SOURCE IS GOVERNED BY *
   5  * THE GNU PUBLIC LICENSE 2, WHICH IS INCLUDED WITH THIS SOURCE.    *
   6  * PLEASE READ THESE TERMS BEFORE DISTRIBUTING.                     *
   7  *                                                                  *
   8  * THE Ogg123 SOURCE CODE IS (C) COPYRIGHT 2000-2001                *
   9  * by Stan Seibert <volsung@xiph.org> AND OTHER CONTRIBUTORS        *
  10  * http://www.xiph.org/                                             *
  11  *                                                                  *
  12  ********************************************************************
  13
  14  last mod: $Id$
  15
  16  ********************************************************************/
  17
  18 #ifdef HAVE_CONFIG_H
  19 #include <config.h>
  20 #endif
  21
  22 #include <sys/types.h>
  23 #include <sys/wait.h>
  24 #include <sys/time.h>
  25 #include <string.h>
  26 #include <fcntl.h>
  27 #include <signal.h>
  28 #include <unistd.h>
  29 #include <stdio.h>
  30
  31 #include "compat.h"
  32 #include "buffer.h"
  33 #include "i18n.h"
  34 #include "ogg123.h"
  35
  36 #define MIN(x,y)       ( (x) < (y) ? (x) : (y) )
  37 #define MIN3(x,y,z)    MIN(x,MIN(y,z))
  38 #define MIN4(w,x,y,z)  MIN( MIN(w,x), MIN(y,z) )
  39
  40 #ifdef DEBUG_BUFFER
  41 FILE *debugfile;
  42 #define DEBUG(x) { fprintf (debugfile, "%d: " x "\n", getpid()); }
  43 #define DEBUG1(x, y) { fprintf (debugfile, "%d: " x "\n", getpid(), y); }
  44 #define DEBUG2(x, y, z) { fprintf (debugfile, "%d: " x "\n", getpid(), y, z); }
  45 #else
  46 #define DEBUG(x)
  47 #define DEBUG1(x, y)
  48 #define DEBUG2(x, y, z)
  49 #endif
  50
  51 /* Macros that support debugging of threading structures */
  52
  53 #define LOCK_MUTEX(mutex) { DEBUG1("Locking mutex %s.", #mutex); pthread_mutex_lock (&(mutex)); }
  54 #define UNLOCK_MUTEX(mutex) { DEBUG1("Unlocking mutex %s", #mutex); pthread_mutex_unlock(&(mutex)); }
  55 #define COND_WAIT(cond, mutex) { DEBUG2("Unlocking %s and waiting on %s", #mutex, #cond); pthread_cond_wait(&(cond), &(mutex)); }
  56 #define COND_SIGNAL(cond) { DEBUG1("Signalling %s", #cond); pthread_cond_signal(&(cond)); }
  57
  58 extern signal_request_t sig_request;  /* Need access to global cancel flag */
  59
  60
  61 /* -------------------- Private Functions ------------------ */
  62
  63 void buffer_init_vars (buf_t *buf)
  64 {
  65   /* Initialize buffer flags */
  66   buf->prebuffering = buf->prebuffer_size > 0;
  67   buf->paused = 0;
  68   buf->eos = 0;
  69   buf->abort_write = 0;
  70   buf->cancel_flag = 0;
  71
  72   buf->curfill = 0;
  73   buf->start = 0;
  74   buf->position = 0;
  75   buf->position_end = 0;
  76 }
  77
  78 void buffer_thread_init (buf_t *buf)
  79 {
  80   sigset_t set;
  81
  82   DEBUG("Enter buffer_thread_init");
  83
  84   /* Block signals to this thread */
  85   sigfillset(&set);
  86   sigaddset(&set, SIGINT);
  87   sigaddset(&set, SIGTSTP);
  88   sigaddset(&set, SIGCONT);
  89   if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0)
  90     DEBUG("pthread_sigmask failed");
  91 }
  92
  93
  94 void buffer_thread_cleanup (void *arg)
  95 {
  96   buf_t *buf = (buf_t *)arg;
  97   (void)buf;
  98
  99   DEBUG("Enter buffer_thread_cleanup");
 100 }
 101
 102
 103 void buffer_mutex_unlock (void *arg)
 104 {
 105   buf_t *buf = (buf_t *)arg;
 106
 107   UNLOCK_MUTEX(buf->mutex);
 108 }
 109
 110
 111 action_t *malloc_action (action_func_t action_func, void *action_arg)
 112 {
 113   action_t *action;
 114
 115   action = malloc(sizeof(action_t));
 116
 117   if (action == NULL) {
 118     fprintf(stderr, _("ERROR: Out of memory in malloc_action().\n"));
 119     exit(1);
 120   }
 121
 122   action->position = 0;
 123   action->action_func = action_func;
 124   action->arg = action_arg;
 125   action->next = NULL;
 126
 127   return action;
 128 }
 129
 130
 131 /* insert = 1:  Make this action the first action associated with this position
 132    insert = 0:  Make this action the last action associated with this position
 133 */
 134 #define INSERT 1
 135 #define APPEND 0
 136 void in_order_add_action (action_t **action_list, action_t *action, int insert)
 137 {
 138   insert = insert > 0 ? 1 : 0;  /* Clamp in case caller messed up */
 139
 140   while (*action_list != NULL &&
 141          (*action_list)->position <= (action->position + insert))
 142     action_list = &((*action_list)->next);
 143
 144   action->next = *action_list;
 145   *action_list = action;
 146 }
 147
 148
 149 void execute_actions (buf_t *buf, action_t **action_list, ogg_int64_t position)
 150 {
 151   action_t *action;
 152
 153   while (*action_list != NULL && (*action_list)->position <= position) {
 154     action = *action_list;
 155     action->action_func(buf, action->arg);
 156
 157     *action_list = (*action_list)->next;
 158     free(action);
 159   }
 160 }
 161
 162
 163 void free_action (action_t *action)
 164 {
 165   free(action);
 166 }
 167
 168
 169
 170 int compute_dequeue_size (buf_t *buf, int request_size)
 171 {
 172   ogg_int64_t next_action_pos;
 173
 174   /*
 175      For simplicity, the number of bytes played must satisfy the following
 176      requirements:
 177      1. Do not extract more bytes than are stored in the buffer.
 178      2. Do not extract more bytes than the requested number of bytes.
 179      3. Do not run off the end of the buffer.
 180      4. Do not go past the next action.
 181   */
 182
 183   if (buf->actions != NULL) {
 184
 185     next_action_pos = buf->actions->position;
 186
 187     return MIN4((ogg_int64_t)buf->curfill, (ogg_int64_t)request_size,
 188                (ogg_int64_t)(buf->size - buf->start),
 189                next_action_pos - buf->position);
 190   } else
 191     return MIN3(buf->curfill, (long)request_size, buf->size - buf->start);
 192
 193 }
 194
 195
 196 void *buffer_thread_func (void *arg)
 197 {
 198   buf_t *buf = (buf_t*) arg;
 199   size_t write_amount;
 200
 201   DEBUG("Enter buffer_thread_func");
 202
 203   buffer_thread_init(buf);
 204
 205   pthread_cleanup_push(buffer_thread_cleanup, buf);
 206
 207   DEBUG("Start main play loop");
 208
 209   /* This test is safe since curfill will never decrease and eos will
 210      never be unset. */
 211   while ( !(buf->eos && buf->curfill == 0) && !buf->abort_write) {
 212
 213     if (buf->cancel_flag || sig_request.cancel)
 214       break;
 215
 216     DEBUG("Check for something to play");
 217     /* Block until we can play something */
 218     LOCK_MUTEX (buf->mutex);
 219     if (buf->prebuffering ||
 220         buf->paused ||
 221         (buf->curfill < buf->audio_chunk_size && !buf->eos)) {
 222
 223       DEBUG("Waiting for more data to play.");
 224       COND_WAIT(buf->playback_cond, buf->mutex);
 225     }
 226
 227     DEBUG("Ready to play");
 228
 229     UNLOCK_MUTEX(buf->mutex);
 230
 231     if (buf->cancel_flag || sig_request.cancel)
 232       break;
 233
 234     /* Don't need to lock buffer while running actions since position
 235        won't change.  We clear out any actions before we compute the
 236        dequeue size so we don't consider actions that need to
 237        run right now.  */
 238     execute_actions(buf, &buf->actions, buf->position);
 239
 240     LOCK_MUTEX(buf->mutex);
 241
 242     /* Need to be locked while we check things. */
 243     write_amount = compute_dequeue_size(buf, buf->audio_chunk_size);
 244
 245     UNLOCK_MUTEX(buf->mutex);
 246
 247     if(write_amount){ /* we might have been woken spuriously */
 248       /* No need to lock mutex here because the other thread will
 249          NEVER reduce the number of bytes stored in the buffer */
 250       DEBUG1("Sending %d bytes to the audio device", write_amount);
 251       write_amount = buf->write_func(buf->buffer + buf->start, write_amount,
 252                                      /* Only set EOS if this is the last chunk */
 253                                      write_amount == buf->curfill ? buf->eos : 0,
 254                                      buf->write_arg);
 255
 256       if (!write_amount) {
 257         DEBUG("Error writing to the audio device. Aborting.");
 258         buffer_abort_write(buf);
 259       }
 260
 261       LOCK_MUTEX(buf->mutex);
 262
 263       buf->curfill -= write_amount;
 264       buf->position += write_amount;
 265       buf->start = (buf->start + write_amount) % buf->size;
 266       DEBUG1("Updated buffer fill, curfill = %ld", buf->curfill);
 267
 268       /* If we've essentially emptied the buffer and prebuffering is enabled,
 269          we need to do another prebuffering session */
 270       if (!buf->eos && (buf->curfill < buf->audio_chunk_size))
 271         buf->prebuffering = buf->prebuffer_size > 0;
 272     }else{
 273       DEBUG("Woken spuriously");
 274     }
 275
 276     /* Signal a waiting decoder thread that they can put more audio into the
 277        buffer */
 278     DEBUG("Signal decoder thread that buffer space is available");
 279     COND_SIGNAL(buf->write_cond);
 280
 281     UNLOCK_MUTEX(buf->mutex);
 282   }
 283
 284   pthread_cleanup_pop(1);
 285   DEBUG("exiting buffer_thread_func");
 286
 287   return 0;
 288 }
 289
 290
 291 int submit_data_chunk (buf_t *buf, char *data, size_t size)
 292 {
 293   long   buf_write_pos; /* offset of first available write location */
 294   size_t write_size;
 295
 296   DEBUG1("Enter submit_data_chunk, size %d", size);
 297
 298   pthread_cleanup_push(buffer_mutex_unlock, buf);
 299
 300   /* Put the data into the buffer as space is made available */
 301   while (size > 0 && !buf->abort_write) {
 302
 303     /* Section 1: Write a chunk of data */
 304     DEBUG("Obtaining lock on buffer");
 305     LOCK_MUTEX(buf->mutex);
 306     if (buf->size - buf->curfill > 0) {
 307
 308       /* Figure how much we can write into the buffer.  Requirements:
 309          1. Don't write more data than we have.
 310          2. Don't write more data than we have room for.
 311          3. Don't write past the end of the buffer. */
 312       buf_write_pos = (buf->start + buf->curfill) % buf->size;
 313       write_size = MIN3(size, buf->size - buf->curfill,
 314                         buf->size - buf_write_pos);
 315
 316       memcpy(buf->buffer + buf_write_pos, data, write_size);
 317       buf->curfill += write_size;
 318       data += write_size;
 319       size -= write_size;
 320       buf->position_end += write_size;
 321       DEBUG1("writing chunk into buffer, curfill = %ld", buf->curfill);
 322     }
 323     else {
 324
 325       if (buf->cancel_flag || sig_request.cancel) {
 326         UNLOCK_MUTEX(buf->mutex);
 327         break;
 328       }
 329       /* No room for more data, wait until there is */
 330       DEBUG("No room for data in buffer.  Waiting.");
 331       COND_WAIT(buf->write_cond, buf->mutex);
 332     }
 333     /* Section 2: signal if we are not prebuffering, done
 334        prebuffering, or paused */
 335     if (buf->prebuffering && (buf->prebuffer_size <= buf->curfill)) {
 336
 337       DEBUG("prebuffering done")
 338         buf->prebuffering = 0; /* done prebuffering */
 339     }
 340
 341     if (!buf->prebuffering && !buf->paused) {
 342
 343       DEBUG("Signalling playback thread that more data is available.");
 344       COND_SIGNAL(buf->playback_cond);
 345     } else
 346       DEBUG("Not signalling playback thread since prebuffering or paused.");
 347
 348     UNLOCK_MUTEX(buf->mutex);
 349   }
 350
 351   pthread_cleanup_pop(0);
 352
 353   DEBUG("Exit submit_data_chunk");
 354   return !buf->abort_write;
 355 }
 356
 357
 358 buffer_stats_t *malloc_buffer_stats ()
 359 {
 360   buffer_stats_t *new_stats;
 361
 362   new_stats = malloc(sizeof(buffer_stats_t));
 363
 364   if (new_stats == NULL) {
 365     fprintf(stderr, _("ERROR: Could not allocate memory in malloc_buffer_stats()\n"));
 366     exit(1);
 367   }
 368
 369   return new_stats;
 370 }
 371
 372
 373 /* ------------------ Begin public interface ------------------ */
 374
 375 /* --- Buffer allocation --- */
 376
 377 buf_t *buffer_create (long size, long prebuffer,
 378                       buffer_write_func_t write_func, void *arg,
 379                       int audio_chunk_size)
 380 {
 381   buf_t *buf = malloc (sizeof(buf_t) + sizeof (char) * (size - 1));
 382
 383   if (buf == NULL) {
 384       perror ("malloc");
 385       exit(1);
 386   }
 387
 388 #ifdef DEBUG_BUFFER
 389   debugfile = fopen ("/tmp/bufferdebug", "w");
 390   setvbuf (debugfile, NULL, _IONBF, 0);
 391 #endif
 392
 393   /* Initialize the buffer structure. */
 394   DEBUG1("buffer_create, size = %ld", size);
 395
 396   memset (buf, 0, sizeof(*buf));
 397
 398   buf->write_func = write_func;
 399   buf->write_arg = arg;
 400
 401   /* Setup pthread variables */
 402   pthread_mutex_init(&buf->mutex, NULL);
 403   pthread_cond_init(&buf->write_cond, NULL);
 404   pthread_cond_init(&buf->playback_cond, NULL);
 405
 406   /* Correct for impossible prebuffer and chunk sizes */
 407   if (audio_chunk_size > size || audio_chunk_size == 0)
 408     audio_chunk_size = size / 2;
 409
 410   if (prebuffer > size)
 411     prebuffer = prebuffer / 2;
 412
 413   buf->audio_chunk_size = audio_chunk_size;
 414
 415   buf->prebuffer_size = prebuffer;
 416   buf->size = size;
 417
 418   buf->actions = 0;
 419
 420   /* Initialize flags */
 421   buffer_init_vars(buf);
 422
 423   return buf;
 424 }
 425
 426
 427 void buffer_reset (buf_t *buf)
 428 {
 429   action_t *action;
 430
 431   /* Cleanup pthread variables */
 432   pthread_mutex_destroy(&buf->mutex);
 433   pthread_cond_destroy(&buf->write_cond);
 434   pthread_cond_destroy(&buf->playback_cond);
 435
 436   /* Reinit pthread variables */
 437   pthread_mutex_init(&buf->mutex, NULL);
 438   pthread_cond_init(&buf->write_cond, NULL);
 439   pthread_cond_init(&buf->playback_cond, NULL);
 440
 441   /* Clear old actions */
 442   while (buf->actions != NULL) {
 443     action = buf->actions;
 444     buf->actions = buf->actions->next;
 445     free(action);
 446   }
 447
 448   buffer_init_vars(buf);
 449 }
 450
 451
 452 void buffer_destroy (buf_t *buf)
 453 {
 454   DEBUG("buffer_destroy");
 455
 456   /* Cleanup pthread variables */
 457   pthread_mutex_destroy(&buf->mutex);
 458   COND_SIGNAL(buf->write_cond);
 459   pthread_cond_destroy(&buf->write_cond);
 460   COND_SIGNAL(buf->playback_cond);
 461   pthread_cond_destroy(&buf->playback_cond);
 462
 463   free(buf);
 464 }
 465
 466
 467 /* --- Buffer thread control --- */
 468
 469 int buffer_thread_start (buf_t *buf)
 470 {
 471   DEBUG("Starting new thread.");
 472
 473   return pthread_create(&buf->thread, NULL, buffer_thread_func, buf);
 474 }
 475
 476
 477 /* WARNING: DO NOT call buffer_submit_data after you pause the
 478    playback thread, or you run the risk of deadlocking.  Call
 479    buffer_thread_unpause first. */
 480 void buffer_thread_pause (buf_t *buf)
 481 {
 482   DEBUG("Pausing playback thread");
 483
 484   pthread_cleanup_push(buffer_mutex_unlock, buf);
 485
 486   LOCK_MUTEX(buf->mutex);
 487   buf->paused = 1;
 488   UNLOCK_MUTEX(buf->mutex);
 489
 490   pthread_cleanup_pop(0);
 491 }
 492
 493
 494 void buffer_thread_unpause (buf_t *buf)
 495 {
 496   DEBUG("Unpausing playback thread");
 497
 498   pthread_cleanup_push(buffer_mutex_unlock, buf);
 499
 500   LOCK_MUTEX(buf->mutex);
 501   buf->paused = 0;
 502   COND_SIGNAL(buf->playback_cond);
 503   UNLOCK_MUTEX(buf->mutex);
 504
 505   pthread_cleanup_pop(0);
 506 }
 507
 508
 509 void buffer_thread_kill (buf_t *buf)
 510 {
 511   DEBUG("Attempting to kill playback thread.");
 512
 513   /* Flag the cancellation */
 514   buf->cancel_flag = 1;
 515
 516   /* Signal the playback condition to wake stuff up */
 517   COND_SIGNAL(buf->playback_cond);
 518
 519   pthread_join(buf->thread, NULL);
 520
 521   buffer_thread_cleanup(buf);
 522
 523   DEBUG("Playback thread killed.");
 524 }
 525
 526
 527 /* --- Data buffering functions --- */
 528
 529 int buffer_submit_data (buf_t *buf, char *data, long nbytes) {
 530   return submit_data_chunk (buf, data, nbytes);
 531 }
 532
 533 size_t buffer_get_data (buf_t *buf, char *data, long nbytes)
 534 {
 535   int write_amount;
 536   int orig_size;
 537
 538   orig_size = nbytes;
 539
 540   DEBUG("Enter buffer_get_data");
 541
 542   pthread_cleanup_push(buffer_mutex_unlock, buf);
 543
 544   LOCK_MUTEX(buf->mutex);
 545
 546   /* Put the data into the buffer as space is made available */
 547   while (nbytes > 0) {
 548
 549     if (buf->abort_write)
 550       break;
 551
 552     DEBUG("Obtaining lock on buffer");
 553     /* Block until we can read something */
 554     if (buf->curfill == 0 && buf->eos)
 555       break; /* No more data to read */
 556
 557     if (buf->curfill == 0 || (buf->prebuffering && !buf->eos)) {
 558       DEBUG("Waiting for more data to copy.");
 559       COND_WAIT(buf->playback_cond, buf->mutex);
 560     }
 561
 562     if (buf->abort_write)
 563       break;
 564
 565     /* Note: Even if curfill is still 0, nothing bad will happen here */
 566
 567     /* For simplicity, the number of bytes played must satisfy
 568        the following three requirements:
 569
 570        1. Do not copy more bytes than are stored in the buffer.
 571        2. Do not copy more bytes than the reqested data size.
 572        3. Do not run off the end of the buffer. */
 573     write_amount = compute_dequeue_size(buf, nbytes);
 574
 575     UNLOCK_MUTEX(buf->mutex);
 576     execute_actions(buf, &buf->actions, buf->position);
 577
 578     /* No need to lock mutex here because the other thread will
 579        NEVER reduce the number of bytes stored in the buffer */
 580     DEBUG1("Copying %d bytes from the buffer", write_amount);
 581     memcpy(data, buf->buffer + buf->start, write_amount);
 582     LOCK_MUTEX(buf->mutex);
 583
 584     buf->curfill -= write_amount;
 585     data += write_amount;
 586     nbytes -= write_amount;
 587     buf->start = (buf->start + write_amount) % buf->size;
 588     DEBUG1("Updated buffer fill, curfill = %ld", buf->curfill);
 589
 590     /* Signal a waiting decoder thread that they can put more
 591        audio into the buffer */
 592     DEBUG("Signal decoder thread that buffer space is available");
 593     COND_SIGNAL(buf->write_cond);
 594   }
 595
 596   UNLOCK_MUTEX(buf->mutex);
 597
 598   pthread_cleanup_pop(0);
 599
 600   pthread_testcancel();
 601
 602   DEBUG("Exit buffer_get_data");
 603
 604   return orig_size - nbytes;
 605 }
 606
 607 void buffer_mark_eos (buf_t *buf)
 608 {
 609   DEBUG("buffer_mark_eos");
 610
 611   pthread_cleanup_push(buffer_mutex_unlock, buf);
 612
 613   LOCK_MUTEX(buf->mutex);
 614   buf->eos = 1;
 615   buf->prebuffering = 0;
 616   COND_SIGNAL(buf->playback_cond);
 617   UNLOCK_MUTEX(buf->mutex);
 618
 619   pthread_cleanup_pop(0);
 620 }
 621
 622 void buffer_abort_write (buf_t *buf)
 623 {
 624   DEBUG("buffer_abort_write");
 625
 626   pthread_cleanup_push(buffer_mutex_unlock, buf);
 627
 628   LOCK_MUTEX(buf->mutex);
 629   buf->abort_write = 1;
 630   COND_SIGNAL(buf->write_cond);
 631   COND_SIGNAL(buf->playback_cond);
 632   UNLOCK_MUTEX(buf->mutex);
 633
 634   pthread_cleanup_pop(0);
 635 }
 636
 637
 638 /* --- Action buffering functions --- */
 639
 640 void buffer_action_now (buf_t *buf, action_func_t action_func,
 641                         void *action_arg)
 642 {
 643   action_t *action;
 644
 645   action = malloc_action(action_func, action_arg);
 646
 647   pthread_cleanup_push(buffer_mutex_unlock, buf);
 648
 649   LOCK_MUTEX(buf->mutex);
 650
 651   action->position = buf->position;
 652
 653   /* Insert this action right at the front */
 654   action->next = buf->actions;
 655   buf->actions = action;
 656
 657   UNLOCK_MUTEX(buf->mutex);
 658
 659   pthread_cleanup_pop(0);
 660 }
 661
 662
 663 void buffer_insert_action_at_end (buf_t *buf, action_func_t action_func,
 664                                   void *action_arg)
 665 {
 666   action_t *action;
 667
 668   action = malloc_action(action_func, action_arg);
 669
 670   pthread_cleanup_push(buffer_mutex_unlock, buf);
 671
 672   LOCK_MUTEX(buf->mutex);
 673
 674   /* Stick after the last item in the buffer */
 675   action->position = buf->position_end;
 676
 677   in_order_add_action(&buf->actions, action, INSERT);
 678
 679   UNLOCK_MUTEX(buf->mutex);
 680
 681   pthread_cleanup_pop(0);
 682 }
 683
 684
 685 void buffer_append_action_at_end (buf_t *buf, action_func_t action_func,
 686                                   void *action_arg)
 687 {
 688   action_t *action;
 689
 690   action = malloc_action(action_func, action_arg);
 691
 692   pthread_cleanup_push(buffer_mutex_unlock, buf);
 693
 694   LOCK_MUTEX(buf->mutex);
 695
 696   /* Stick after the last item in the buffer */
 697   action->position = buf->position_end;
 698
 699   in_order_add_action(&buf->actions, action, APPEND);
 700
 701   UNLOCK_MUTEX(buf->mutex);
 702
 703   pthread_cleanup_pop(0);
 704 }
 705
 706
 707 void buffer_insert_action_at (buf_t *buf, action_func_t action_func,
 708                               void *action_arg, ogg_int64_t position)
 709 {
 710   action_t *action;
 711
 712   action = malloc_action(action_func, action_arg);
 713
 714   pthread_cleanup_push(buffer_mutex_unlock, buf);
 715
 716   LOCK_MUTEX(buf->mutex);
 717
 718   action->position = position;
 719
 720   in_order_add_action(&buf->actions, action, INSERT);
 721
 722   UNLOCK_MUTEX(buf->mutex);
 723
 724   pthread_cleanup_pop(0);
 725 }
 726
 727
 728 void buffer_append_action_at (buf_t *buf, action_func_t action_func,
 729                               void *action_arg, ogg_int64_t position)
 730 {
 731   action_t *action;
 732
 733   action = malloc_action(action_func, action_arg);
 734
 735   pthread_cleanup_push(buffer_mutex_unlock, buf);
 736
 737   LOCK_MUTEX(buf->mutex);
 738
 739   action->position = position;
 740
 741   in_order_add_action(&buf->actions, action, APPEND);
 742
 743   UNLOCK_MUTEX(buf->mutex);
 744
 745   pthread_cleanup_pop(0);
 746 }
 747
 748
 749 /* --- Buffer status functions --- */
 750
 751 void buffer_wait_for_empty (buf_t *buf)
 752 {
 753   int empty = 0;
 754
 755   DEBUG("Enter buffer_wait_for_empty");
 756
 757   pthread_cleanup_push(buffer_mutex_unlock, buf);
 758
 759   LOCK_MUTEX(buf->mutex);
 760   while (!empty && !buf->abort_write) {
 761
 762     if (buf->curfill > 0) {
 763       DEBUG1("Buffer curfill = %ld, going back to sleep.", buf->curfill);
 764       COND_WAIT(buf->write_cond, buf->mutex);
 765     } else
 766       empty = 1;
 767   }
 768   UNLOCK_MUTEX(buf->mutex);
 769
 770   pthread_cleanup_pop(0);
 771
 772   DEBUG("Exit buffer_wait_for_empty");
 773 }
 774
 775
 776 long buffer_full (buf_t *buf)
 777 {
 778   return buf->curfill;
 779 }
 780
 781
 782 buffer_stats_t *buffer_statistics (buf_t *buf)
 783 {
 784   buffer_stats_t *stats;
 785
 786   pthread_cleanup_push(buffer_mutex_unlock, buf);
 787
 788   LOCK_MUTEX(buf->mutex);
 789
 790   stats = malloc_buffer_stats();
 791
 792   stats->size = buf->size;
 793   stats->fill = (double) buf->curfill / (double) buf->size * 100.0;
 794   stats->prebuffer_fill = (double) buf->prebuffer_size / (double) buf->size;
 795   stats->prebuffering = buf->prebuffering;
 796   stats->paused = buf->paused;
 797   stats->eos = buf->eos;
 798
 799   UNLOCK_MUTEX(buf->mutex);
 800
 801   pthread_cleanup_pop(0);
 802
 803   return stats;
 804 }