prefetch_pipe.c

   1 /*
   2 Calf Box, an open source musical instrument.
   3 Copyright (C) 2010-2013 Krzysztof Foltman
   4
   5 This program is free software: you can redistribute it and/or modify
   6 it under the terms of the GNU General Public License as published by
   7 the Free Software Foundation, either version 3 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU General Public License for more details.
  14
  15 You should have received a copy of the GNU General Public License
  16 along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 #include "prefetch_pipe.h"
  20 #include "wavebank.h"
  21 #include <assert.h>
  22 #include <malloc.h>
  23 #include <memory.h>
  24 #include <stdlib.h>
  25 #include <unistd.h>
  26
  27 // Don't bother fetching less than 4 (mono) or 8 KB (stereo)
  28 #define PIPE_MIN_PREFETCH_SIZE_FRAMES 2048
  29
  30 void cbox_prefetch_pipe_init(struct cbox_prefetch_pipe *pipe, uint32_t buffer_size)
  31 {
  32     pipe->data = malloc(buffer_size);
  33     pipe->buffer_size = buffer_size;
  34     pipe->sndfile = NULL;
  35     pipe->state = pps_free;
  36 }
  37
  38 gboolean cbox_prefetch_pipe_openfile(struct cbox_prefetch_pipe *pipe)
  39 {
  40     pipe->sndfile = sf_open(pipe->waveform->canonical_name, SFM_READ, &pipe->info);
  41     if (!pipe->sndfile)
  42         return FALSE;
  43     pipe->file_pos_frame = sf_seek(pipe->sndfile, pipe->waveform->preloaded_frames, SEEK_SET);
  44     if (pipe->file_loop_end > pipe->info.frames)
  45         pipe->file_loop_end = pipe->info.frames;
  46     pipe->buffer_loop_end = pipe->buffer_size / (sizeof(int16_t) * pipe->info.channels);
  47     pipe->produced = pipe->file_pos_frame;
  48     pipe->write_ptr = 0;
  49     pipe->state = pps_active;
  50
  51     return TRUE;
  52 }
  53
  54 void cbox_prefetch_pipe_consumed(struct cbox_prefetch_pipe *pipe, uint32_t frames)
  55 {
  56     pipe->consumed += frames;
  57 }
  58
  59 void cbox_prefetch_pipe_fetch(struct cbox_prefetch_pipe *pipe)
  60 {
  61     gboolean retry;
  62     do {
  63         retry = FALSE;
  64         // XXXKF take consumption rate into account
  65
  66         // How many frames left to consume
  67         int32_t supply = pipe->produced - pipe->consumed;
  68         if (supply < 0)
  69         {
  70             // Overrun already occurred. Cut the losses by skipping already missed
  71             // part.
  72             uint32_t overrun = -supply;
  73
  74             // XXXKF This may or may not be stupid. I didn't put much thought into it.
  75             pipe->produced += overrun;
  76             pipe->file_pos_frame = sf_seek(pipe->sndfile, overrun, SEEK_CUR);
  77             pipe->write_ptr += overrun;
  78             if (pipe->write_ptr >= pipe->buffer_loop_end)
  79                 pipe->write_ptr %= pipe->buffer_loop_end;
  80         }
  81         //
  82         if (supply >= pipe->buffer_loop_end)
  83             return;
  84
  85         // How many frames to read to fill the full prefetch size
  86         int32_t readsize = pipe->buffer_loop_end - supply;
  87         //
  88         if (readsize < PIPE_MIN_PREFETCH_SIZE_FRAMES)
  89             return;
  90
  91         if (pipe->write_ptr == pipe->buffer_loop_end)
  92             pipe->write_ptr = 0;
  93
  94         // If reading across buffer boundary, only read the part up to buffer
  95         // end, and then retry from start of the buffer.
  96         if (pipe->write_ptr + readsize > pipe->buffer_loop_end)
  97         {
  98             readsize = pipe->buffer_loop_end - pipe->write_ptr;
  99             retry = TRUE;
 100         }
 101         // If past the file loop end, restart at file loop start
 102         if (pipe->file_pos_frame >= pipe->file_loop_end)
 103         {
 104             if (pipe->file_loop_start == (uint32_t)-1 || (pipe->loop_count && pipe->play_count >= pipe->loop_count - 1))
 105             {
 106                 pipe->finished = TRUE;
 107                 for (int i = 0; i < readsize * pipe->info.channels; i++)
 108                     pipe->data[pipe->write_ptr * pipe->info.channels + i] = rand();
 109                 break;
 110             }
 111             else
 112             {
 113                 pipe->play_count++;
 114                 pipe->file_pos_frame = pipe->file_loop_start;
 115                 sf_seek(pipe->sndfile, pipe->file_loop_start, SEEK_SET);
 116             }
 117         }
 118         // If reading across file loop boundary, read up to loop end and
 119         // retry to restart
 120         if (pipe->file_pos_frame + readsize > pipe->file_loop_end)
 121         {
 122             readsize = pipe->file_loop_end - pipe->file_pos_frame;
 123             retry = TRUE;
 124         }
 125
 126         int32_t actread = sf_readf_short(pipe->sndfile, pipe->data + pipe->write_ptr * pipe->info.channels, readsize);
 127         pipe->produced += actread;
 128         pipe->file_pos_frame += actread;
 129         pipe->write_ptr += actread;
 130     } while(retry);
 131 }
 132
 133 void cbox_prefetch_pipe_closefile(struct cbox_prefetch_pipe *pipe)
 134 {
 135     assert(pipe->state == pps_closing);
 136     assert(pipe->sndfile);
 137     sf_close(pipe->sndfile);
 138     pipe->sndfile = NULL;
 139     pipe->state = pps_free;
 140 }
 141
 142 void cbox_prefetch_pipe_close(struct cbox_prefetch_pipe *pipe)
 143 {
 144     if (pipe->sndfile)
 145         cbox_prefetch_pipe_closefile(pipe);
 146     if (pipe->data)
 147     {
 148         free(pipe->data);
 149         pipe->data = NULL;
 150     }
 151 }
 152
 153 static void *prefetch_thread(void *user_data)
 154 {
 155     struct cbox_prefetch_stack *stack = user_data;
 156
 157     while(!stack->finished)
 158     {
 159         usleep(1000);
 160         for (int i = 0; i < stack->pipe_count; i++)
 161         {
 162             struct cbox_prefetch_pipe *pipe = &stack->pipes[i];
 163             switch(pipe->state)
 164             {
 165             case pps_free:
 166             case pps_finished:
 167             case pps_error:
 168                 break;
 169             case pps_opening:
 170                 if (!cbox_prefetch_pipe_openfile(pipe))
 171                     pipe->state = pps_error;
 172                 assert(pipe->state != pps_opening);
 173                 break;
 174             case pps_active:
 175                 if (pipe->returned)
 176                     pipe->state = pps_closing;
 177                 else
 178                     cbox_prefetch_pipe_fetch(pipe);
 179                 break;
 180             case pps_closing:
 181                 cbox_prefetch_pipe_closefile(pipe);
 182                 break;
 183             default:
 184                 break;
 185             }
 186         }
 187     }
 188     return 0;
 189 }
 190
 191 struct cbox_prefetch_stack *cbox_prefetch_stack_new(int npipes, uint32_t buffer_size)
 192 {
 193     struct cbox_prefetch_stack *stack = calloc(1, sizeof(struct cbox_prefetch_stack));
 194     stack->pipes = calloc(npipes, sizeof(struct cbox_prefetch_pipe));
 195
 196     for (int i = 0; i < npipes; i++)
 197     {
 198         cbox_prefetch_pipe_init(&stack->pipes[i], buffer_size);
 199         stack->pipes[i].next_free_pipe = i - 1;
 200     }
 201     stack->pipe_count = npipes;
 202     stack->last_free_pipe = npipes - 1;
 203     stack->finished = FALSE;
 204
 205     if (pthread_create(&stack->thr_prefetch, NULL, prefetch_thread, stack))
 206     {
 207         // XXXKF set thread priority
 208         g_warning("Cannot create a prefetch thread. Exiting.\n");
 209         return NULL;
 210     }
 211
 212     return stack;
 213 }
 214
 215 struct cbox_prefetch_pipe *cbox_prefetch_stack_pop(struct cbox_prefetch_stack *stack, struct cbox_waveform *waveform, uint32_t file_loop_start, uint32_t file_loop_end, uint32_t loop_count)
 216 {
 217     // The stack may include some pipes that are already returned but not yet
 218     // fully prepared for opening a new file
 219     int pos = stack->last_free_pipe;
 220     while(pos != -1 && stack->pipes[pos].state != pps_free)
 221         pos = stack->pipes[pos].next_free_pipe;
 222     if (pos == -1)
 223         return NULL;
 224
 225     struct cbox_prefetch_pipe *pipe = &stack->pipes[pos];
 226
 227     stack->last_free_pipe = pipe->next_free_pipe;
 228     pipe->next_free_pipe = -1;
 229
 230     pipe->waveform = waveform;
 231     if (file_loop_start == (uint32_t)-1 && loop_count)
 232         file_loop_start = 0;
 233     pipe->file_loop_start = file_loop_start;
 234     pipe->file_loop_end = file_loop_end;
 235     pipe->buffer_loop_end = 0;
 236     pipe->finished = FALSE;
 237     pipe->returned = FALSE;
 238     pipe->produced = waveform->preloaded_frames;
 239     pipe->consumed = 0;
 240     pipe->play_count = 0;
 241     pipe->loop_count = loop_count;
 242
 243     __sync_synchronize();
 244     pipe->state = pps_opening;
 245     return pipe;
 246 }
 247
 248 void cbox_prefetch_stack_push(struct cbox_prefetch_stack *stack, struct cbox_prefetch_pipe *pipe)
 249 {
 250     switch(pipe->state)
 251     {
 252     case pps_free:
 253         assert(0);
 254         break;
 255     case pps_error:
 256     case pps_closed:
 257         pipe->state = pps_free;
 258         break;
 259     case pps_opening:
 260         // Close the file as soon as open operation completes
 261         pipe->returned = TRUE;
 262         break;
 263     default:
 264         pipe->state = pps_closing;
 265         break;
 266     }
 267     __sync_synchronize();
 268
 269     assert(pipe->next_free_pipe == -1);
 270     pipe->next_free_pipe = stack->last_free_pipe;
 271     int pos = pipe - stack->pipes;
 272
 273     __sync_synchronize();
 274     stack->last_free_pipe = pos;
 275 }
 276
 277 int cbox_prefetch_stack_get_active_pipe_count(struct cbox_prefetch_stack *stack)
 278 {
 279     int count = 0;
 280     for (int i = 0; i < stack->pipe_count; i++)
 281     {
 282         if (stack->pipes[i].state != pps_free)
 283             count++;
 284     }
 285     return count;
 286 }
 287
 288 void cbox_prefetch_stack_destroy(struct cbox_prefetch_stack *stack)
 289 {
 290     void *result = NULL;
 291     stack->finished = TRUE;
 292     pthread_join(stack->thr_prefetch, &result);
 293     for (int i = 0; i < stack->pipe_count; i++)
 294         cbox_prefetch_pipe_close(&stack->pipes[i]);
 295     free(stack->pipes);
 296     free(stack);
 297 }