Index: apps/buffering.c =================================================================== --- apps/buffering.c (revision 29263) +++ apps/buffering.c (working copy) @@ -109,6 +109,13 @@ }; /* invariant: filesize == offset + available + filerem */ + +struct buf_message_data +{ + int handle_id; + intptr_t data; +}; + static char *buffer; static char *guard_buffer; @@ -133,14 +140,15 @@ static int base_handle_id; -static struct mutex llist_mutex; -static struct mutex llist_mod_mutex; +/* Lock this during off-buffering-thread operations that may yield and + having a handle get moved would be a bad thing. */ +static struct mutex llist_mutex SHAREDBSS_ATTR; /* Handle cache (makes find_handle faster). This is global so that move_handle and rm_handle can invalidate it. */ static struct memory_handle *cached_handle = NULL; -static struct { +static struct data_counters { size_t remaining; /* Amount of data needing to be buffered */ size_t wasted; /* Amount of space available for freeing */ size_t buffered; /* Amount of data currently in the buffer */ @@ -152,8 +160,8 @@ enum { Q_BUFFER_HANDLE = 1, /* Request buffering of a handle, this should not be used in a low buffer situation. */ - Q_RESET_HANDLE, /* (internal) Request resetting of a handle to its - offset (the offset has to be set beforehand) */ + Q_REBUFFER_HANDLE, /* Request reset and rebuffering of a handle at a new + file starting position. */ Q_CLOSE_HANDLE, /* Request closing a handle */ Q_BASE_HANDLE, /* Set the reference handle for buf_useful_data */ @@ -260,9 +268,6 @@ if (num_handles >= BUF_MAX_HANDLES) return NULL; - mutex_lock(&llist_mutex); - mutex_lock(&llist_mod_mutex); - if (cur_handle && cur_handle->filerem > 0) { /* the current handle hasn't finished buffering. We can only add a new one if there is already enough free space to finish @@ -270,8 +275,6 @@ size_t req = cur_handle->filerem + sizeof(struct memory_handle); if (ringbuf_add_cross(cur_handle->widx, req, buf_ridx) >= 0) { /* Not enough space */ - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return NULL; } else { /* Allocate the remainder of the space for the current handle */ @@ -299,8 +302,6 @@ overlap = ringbuf_add_cross(buf_widx, shift + len, buf_ridx); if (overlap >= 0 && (alloc_all || (size_t)overlap >= data_size)) { /* Not enough space for required allocations */ - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return NULL; } @@ -329,8 +330,6 @@ cur_handle = new_handle; - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return new_handle; } @@ -341,9 +340,6 @@ if (h == NULL) return true; - mutex_lock(&llist_mutex); - mutex_lock(&llist_mod_mutex); - if (h == first_handle) { first_handle = h->next; if (h == cur_handle) { @@ -367,8 +363,6 @@ buf_widx = cur_handle->widx; } } else { - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return false; } } @@ -378,9 +372,6 @@ cached_handle = NULL; num_handles--; - - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return true; } @@ -391,19 +382,15 @@ if (handle_id < 0) return NULL; - mutex_lock(&llist_mutex); - /* simple caching because most of the time the requested handle will either be the same as the last, or the one after the last */ if (cached_handle) { if (cached_handle->id == handle_id) { - mutex_unlock(&llist_mutex); return cached_handle; } else if (cached_handle->next && (cached_handle->next->id == handle_id)) { cached_handle = cached_handle->next; - mutex_unlock(&llist_mutex); return cached_handle; } } @@ -416,7 +403,6 @@ if (m) cached_handle = m; - mutex_unlock(&llist_mutex); return m; } @@ -454,9 +440,6 @@ return false; } - mutex_lock(&llist_mutex); - mutex_lock(&llist_mod_mutex); - oldpos = ringbuf_offset(src); newpos = ringbuf_add(oldpos, final_delta); overlap = ringbuf_add_cross(newpos, size_to_move, buffer_len - 1); @@ -481,8 +464,6 @@ correction = (correction + 3) & ~3; if (final_delta < correction + sizeof(struct memory_handle)) { /* Delta cannot end up less than the size of the struct */ - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return false; } newpos -= correction; @@ -504,8 +485,6 @@ if (m && m->next == src) { m->next = dest; } else { - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return false; } } @@ -550,8 +529,6 @@ /* Update the caller with the new location of h and the distance moved */ *h = dest; *delta = final_delta; - mutex_unlock(&llist_mod_mutex); - mutex_unlock(&llist_mutex); return true; } @@ -571,18 +548,24 @@ These functions are used by the buffering thread to manage buffer space. */ -static void update_data_counters(void) +static void update_data_counters(struct data_counters *dc) { - struct memory_handle *m = find_handle(base_handle_id); - bool is_useful = m==NULL; - size_t buffered = 0; size_t wasted = 0; size_t remaining = 0; size_t useful = 0; + struct memory_handle *m; + bool is_useful; + + if (dc == NULL) + dc = &data_counters; + mutex_lock(&llist_mutex); + m = find_handle(base_handle_id); + is_useful = m == NULL; + m = first_handle; while (m) { buffered += m->available; @@ -600,19 +583,19 @@ mutex_unlock(&llist_mutex); - data_counters.buffered = buffered; - data_counters.wasted = wasted; - data_counters.remaining = remaining; - data_counters.useful = useful; + dc->buffered = buffered; + dc->wasted = wasted; + dc->remaining = remaining; + dc->useful = useful; } static inline bool buffer_is_low(void) { - update_data_counters(); + update_data_counters(NULL); return data_counters.useful < (conf_watermark / 2); } -/* Buffer data for the given handle. +/* Q_BUFFER_HANDLE event and buffer data for the given handle. Return whether or not the buffering should continue explicitly. */ static bool buffer_handle(int handle_id) { @@ -659,7 +642,7 @@ h->filerem = 0; h->available = sizeof(struct mp3entry); h->widx += sizeof(struct mp3entry); - send_event(BUFFER_EVENT_FINISHED, &h->id); + send_event(BUFFER_EVENT_FINISHED, &handle_id); return true; } @@ -676,9 +659,11 @@ if (ringbuf_add_cross(h->widx, copy_n, buf_ridx) >= 0) return false; - /* FIXME: This would overwrite the next handle + /* This would overwrite the next handle? * If this is true, then there's a handle even though we have still - * data to buffer. This should NEVER EVER happen! (but it does :( ) */ + * data to buffer. This may happen if a rebuffer is required and the + * thread responsible for clearing the buffer after the handle being + * rebuffered has not yet completed the cleanup. */ if (h->next && (overlap = ringbuf_add_cross(h->widx, copy_n, next_handle)) > 0) { @@ -735,90 +720,36 @@ /* finished buffering the file */ close(h->fd); h->fd = -1; - send_event(BUFFER_EVENT_FINISHED, &h->id); + send_event(BUFFER_EVENT_FINISHED, &handle_id); } return !stop; } -/* Reset writing position and data buffer of a handle to its current offset. - Use this after having set the new offset to use. */ -static void reset_handle(int handle_id) +/* Close the specified handle id and free its allocation. */ +static bool close_handle(int handle_id) { - size_t alignment_pad; + bool retval = false; - logf("reset_handle(%d)", handle_id); + struct memory_handle *h; - struct memory_handle *h = find_handle(handle_id); - if (!h) - return; + mutex_lock(&llist_mutex); + h = find_handle(handle_id); - /* Align to desired storage alignment */ - alignment_pad = STORAGE_OVERLAP(h->offset - (size_t)(&buffer[h->start])); - h->ridx = h->widx = h->data = ringbuf_add(h->start, alignment_pad); - - if (h == cur_handle) - buf_widx = h->widx; - h->available = 0; - h->filerem = h->filesize - h->offset; - - if (h->fd >= 0) { - lseek(h->fd, h->offset, SEEK_SET); - } -} - -/* Seek to a nonbuffered part of a handle by rebuffering the data. */ -static void rebuffer_handle(int handle_id, size_t newpos) -{ - struct memory_handle *h = find_handle(handle_id); - if (!h) - return; - - /* When seeking foward off of the buffer, if it is a short seek don't - rebuffer the whole track, just read enough to satisfy */ - if (newpos > h->offset && newpos - h->offset < BUFFERING_DEFAULT_FILECHUNK) - { - LOGFQUEUE("buffering >| Q_BUFFER_HANDLE %d", handle_id); - queue_send(&buffering_queue, Q_BUFFER_HANDLE, handle_id); - h->ridx = ringbuf_add(h->data, newpos - h->offset); - return; - } - - h->offset = newpos; - - /* Reset the handle to its new offset */ - LOGFQUEUE("buffering >| Q_RESET_HANDLE %d", handle_id); - queue_send(&buffering_queue, Q_RESET_HANDLE, handle_id); - - uintptr_t next = ringbuf_offset(h->next); - if (ringbuf_sub(next, h->data) < h->filesize - newpos) - { - /* There isn't enough space to rebuffer all of the track from its new - offset, so we ask the user to free some */ - DEBUGF("%s(): space is needed\n", __func__); - send_event(BUFFER_EVENT_REBUFFER, &handle_id); - } - - /* Now we ask for a rebuffer */ - LOGFQUEUE("buffering >| Q_BUFFER_HANDLE %d", handle_id); - queue_send(&buffering_queue, Q_BUFFER_HANDLE, handle_id); -} - -static bool close_handle(int handle_id) -{ - struct memory_handle *h = find_handle(handle_id); - /* If the handle is not found, it is closed */ - if (!h) - return true; + if (h) { + if (h->fd >= 0) { + close(h->fd); + h->fd = -1; + } - if (h->fd >= 0) { - close(h->fd); - h->fd = -1; + /* rm_handle returns true unless the handle somehow persists after + exit */ + retval = rm_handle(h); } - /* rm_handle returns true unless the handle somehow persists after exit */ - return rm_handle(h); + mutex_unlock(&llist_mutex); + return retval; } /* Free buffer space by moving the handle struct right before the useful @@ -881,7 +812,11 @@ { logf("fill_buffer()"); struct memory_handle *m; + + mutex_lock(&llist_mutex); shrink_handle(first_handle); + mutex_unlock(&llist_mutex); + m = first_handle; while (queue_empty(&buffering_queue) && m) { if (m->filerem > 0) { @@ -970,7 +905,7 @@ /* Reserve space in the buffer for a file. filename: name of the file to open offset: offset at which to start buffering the file, useful when the first - (offset-1) bytes of the file aren't needed. + offset bytes of the file aren't needed. type: one of the data types supported (audio, image, cuesheet, others user_data: user data passed possibly passed in subcalls specific to a data_type (only used for image (albumart) buffering so far ) @@ -984,6 +919,8 @@ /* currently only used for aa loading */ (void)user_data; #endif + int handle_id; + if (type == TYPE_ID3) { /* ID3 case: allocate space, init the handle and return. */ @@ -992,6 +929,7 @@ if (!h) return ERR_BUFFER_FULL; + handle_id = h->id; h->fd = -1; h->filesize = sizeof(struct mp3entry); h->filerem = sizeof(struct mp3entry); @@ -1007,10 +945,10 @@ can't wrap */ /* Inform the buffering thread that we added a handle */ - LOGFQUEUE("buffering > Q_HANDLE_ADDED %d", h->id); - queue_post(&buffering_queue, Q_HANDLE_ADDED, h->id); + LOGFQUEUE("buffering > Q_HANDLE_ADDED %d", handle_id); + queue_post(&buffering_queue, Q_HANDLE_ADDED, handle_id); - return h->id; + return handle_id; } #ifdef APPLICATION /* loading code from memory is not supported in application builds */ @@ -1050,6 +988,7 @@ return ERR_BUFFER_FULL; } + handle_id = h->id; strlcpy(h->path, file, MAX_PATH); h->offset = adjusted_offset; @@ -1080,10 +1019,11 @@ { /* Bitmap file: we load the data instead of the file */ int rc; - mutex_lock(&llist_mod_mutex); /* Lock because load_bitmap yields */ + mutex_lock(&llist_mutex); /* Lock because load_bitmap yields */ rc = load_image(fd, file, (struct bufopen_bitmap_data*)user_data); - mutex_unlock(&llist_mod_mutex); - if (rc <= 0) + mutex_unlock(&llist_mutex); + h = find_handle(handle_id); /* Mutex unlock yields */ + if (!h || rc <= 0) { rm_handle(h); close(fd); @@ -1107,20 +1047,20 @@ if (type == TYPE_CUESHEET) { h->fd = fd; /* Immediately start buffering those */ - LOGFQUEUE("buffering >| Q_BUFFER_HANDLE %d", h->id); - queue_send(&buffering_queue, Q_BUFFER_HANDLE, h->id); + LOGFQUEUE("buffering >| Q_BUFFER_HANDLE %d", handle_id); + queue_send(&buffering_queue, Q_BUFFER_HANDLE, handle_id); } else { /* Other types will get buffered in the course of normal operations */ h->fd = -1; close(fd); /* Inform the buffering thread that we added a handle */ - LOGFQUEUE("buffering > Q_HANDLE_ADDED %d", h->id); - queue_post(&buffering_queue, Q_HANDLE_ADDED, h->id); + LOGFQUEUE("buffering > Q_HANDLE_ADDED %d", handle_id); + queue_post(&buffering_queue, Q_HANDLE_ADDED, handle_id); } - logf("bufopen: new hdl %d", h->id); - return h->id; + logf("bufopen: new hdl %d", handle_id); + return handle_id; } /* Open a new handle from data that needs to be copied from memory. @@ -1173,42 +1113,105 @@ return queue_send(&buffering_queue, Q_CLOSE_HANDLE, handle_id); } -/* Set reading index in handle (relatively to the start of the file). - Access before the available data will trigger a rebuffer. - Return 0 for success and < 0 for failure: - -1 if the handle wasn't found - -2 if the new requested position was beyond the end of the file -*/ -int bufseek(int handle_id, size_t newpos) +/* Backend to bufseek and bufadvance */ +static void rebuffer_handle(int handle_id, size_t newpos) { + /* When seeking foward off of the buffer, if it is a short seek don't + rebuffer the whole track, just read enough to satisfy */ struct memory_handle *h = find_handle(handle_id); if (!h) - return ERR_HANDLE_NOT_FOUND; + { + queue_reply(&buffering_queue, ERR_HANDLE_NOT_FOUND); + return; + } + if (newpos > h->offset && newpos - h->offset < BUFFERING_DEFAULT_FILECHUNK) + { + h->ridx = ringbuf_add(h->data, newpos - h->offset); + queue_reply(&buffering_queue, 0); + buffer_handle(handle_id); + return; + } + + h->offset = newpos; + + /* Reset the handle to its new offset */ + + /* Align to desired storage alignment */ + size_t alignment_pad = STORAGE_OVERLAP(h->offset - (size_t)(&buffer[h->start])); + h->ridx = h->widx = h->data = ringbuf_add(h->start, alignment_pad); + + if (h == cur_handle) + buf_widx = h->widx; + + h->available = 0; + h->filerem = h->filesize - h->offset; + + if (h->fd >= 0) + lseek(h->fd, h->offset, SEEK_SET); + + uintptr_t next = ringbuf_offset(h->next); + if (ringbuf_sub(next, h->data) < h->filesize - newpos) + { + /* There isn't enough space to rebuffer all of the track from its new + offset, so we ask the user to free some */ + DEBUGF("%s(): space is needed\n", __func__); + int hid = handle_id; + send_event(BUFFER_EVENT_REBUFFER, &hid); + } + + /* Now we do the rebuffer */ + queue_reply(&buffering_queue, 0); + buffer_handle(handle_id); +} + +/* Backend to bufseek and bufadvance */ +static int seek_handle(struct memory_handle *h, size_t newpos) +{ if (newpos > h->filesize) { /* access beyond the end of the file */ return ERR_INVALID_VALUE; } - else if (newpos < h->offset || h->offset + h->available < newpos) { - /* access before or after buffered data. A rebuffer is needed. */ - rebuffer_handle(handle_id, newpos); + else if ((newpos < h->offset || h->offset + h->available <= newpos) && + (newpos < h->filesize || h->filerem > 0)) { + /* access before or after buffered data and not to end of file or file + is not buffered to the end-- a rebuffer is needed. */ + struct buf_message_data parm = { h->id, newpos }; + return queue_send(&buffering_queue, Q_REBUFFER_HANDLE, + (intptr_t)&parm); } else { h->ridx = ringbuf_add(h->data, newpos - h->offset); } + return 0; } +/* Set reading index in handle (relatively to the start of the file). + Access before the available data will trigger a rebuffer. + Return 0 for success and < 0 for failure: + -1 if the handle wasn't found + -2 if the new requested position was beyond the end of the file +*/ +int bufseek(int handle_id, size_t newpos) +{ + struct memory_handle *h = find_handle(handle_id); + if (!h) + return ERR_HANDLE_NOT_FOUND; + + return seek_handle(h, newpos); +} + /* Advance the reading index in a handle (relatively to its current position). Return 0 for success and < 0 for failure */ int bufadvance(int handle_id, off_t offset) { - const struct memory_handle *h = find_handle(handle_id); + struct memory_handle *h = find_handle(handle_id); if (!h) return ERR_HANDLE_NOT_FOUND; size_t newpos = h->offset + ringbuf_sub(h->ridx, h->data) + offset; - return bufseek(handle_id, newpos); + return seek_handle(h, newpos); } /* Used by bufread and bufgetdata to prepare the buffer and retrieve the @@ -1250,7 +1253,7 @@ /* Wait for the data to be ready */ do { - sleep(1); + sleep(0); /* it is not safe for a non-buffering thread to sleep while * holding a handle */ h = find_handle(handle_id); @@ -1265,6 +1268,15 @@ return h; } + +/* Note: It is safe for the thread responsible for handling the rebuffer + * cleanup request to call bufread or bufgetdata only when the data will + * be available-- not if it could be blocked waiting for it in prep_bufdata. + * It should be apparent that if said thread is being forced to wait for + * buffering but has not yet responded to the cleanup request, the space + * can never be cleared to allow further reading of the file because it is + * not listening to callbacks any longer. */ + /* Copy data from the given handle to the dest buffer. Return the number of bytes copied or < 0 for failure (handle not found). The caller is blocked until the requested amount of data is available. @@ -1454,13 +1466,16 @@ static void shrink_buffer(void) { logf("shrink_buffer()"); + mutex_lock(&llist_mutex); shrink_buffer_inner(first_handle); + mutex_unlock(&llist_mutex); } void buffering_thread(void) { bool filling = false; struct queue_event ev; + struct buf_message_data *parm; while (true) { @@ -1488,10 +1503,11 @@ buffer_handle((int)ev.data); break; - case Q_RESET_HANDLE: - LOGFQUEUE("buffering < Q_RESET_HANDLE %d", (int)ev.data); - queue_reply(&buffering_queue, 1); - reset_handle((int)ev.data); + case Q_REBUFFER_HANDLE: + parm = (struct buf_message_data *)ev.data; + LOGFQUEUE("buffering < Q_REBUFFER_HANDLE %d %ld", + parm->handle_id, parm->data); + rebuffer_handle(parm->handle_id, parm->data); break; case Q_CLOSE_HANDLE: @@ -1523,7 +1539,7 @@ break; } - update_data_counters(); + update_data_counters(NULL); /* If the buffer is low, call the callbacks to get new data */ if (num_handles > 0 && data_counters.useful <= conf_watermark) @@ -1545,7 +1561,7 @@ if (!filling) shrink_buffer(); filling = fill_buffer(); - update_data_counters(); + update_data_counters(NULL); } } #endif @@ -1573,12 +1589,6 @@ void buffering_init(void) { mutex_init(&llist_mutex); - mutex_init(&llist_mod_mutex); -#ifdef HAVE_PRIORITY_SCHEDULING - /* This behavior not safe atm */ - mutex_set_preempt(&llist_mutex, false); - mutex_set_preempt(&llist_mod_mutex, false); -#endif conf_watermark = BUFFERING_DEFAULT_WATERMARK; @@ -1628,11 +1638,12 @@ void buffering_get_debugdata(struct buffering_debug *dbgdata) { - update_data_counters(); + struct data_counters dc; + update_data_counters(&dc); dbgdata->num_handles = num_handles; - dbgdata->data_rem = data_counters.remaining; - dbgdata->wasted_space = data_counters.wasted; - dbgdata->buffered_data = data_counters.buffered; - dbgdata->useful_data = data_counters.useful; + dbgdata->data_rem = dc.remaining; + dbgdata->wasted_space = dc.wasted; + dbgdata->buffered_data = dc.buffered; + dbgdata->useful_data = dc.useful; dbgdata->watermark = conf_watermark; }