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