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