- for(;;) {
- result = write(fd, bufc, count);
- if(result <= 0) {
- return result;
- }
- count -= result;
- bufc += result;
-
- if(count == 0) {
- break;
- }
- }
-
- return bufc-(const char *)buf;
-}
-
-void decrement_active_buffers(void *arg)
-{
- pthread_mutex_lock(&active_buffers_mutex);
- active_buffers--;
- pthread_mutex_unlock(&active_buffers_mutex);
-}
-
-void *consumer_thread(void *arg)
-{
- struct buffer_info *buf = (struct buffer_info *) arg;
- int result;
-
- pthread_cleanup_push(decrement_active_buffers, NULL);
-
- for(;;) {
- /* get the subbuffer */
- result = get_subbuffer(buf);
- if(result == -1) {
- ERR("error getting subbuffer");
- continue;
- }
- else if(result == GET_SUBBUF_DONE) {
- /* this is done */
- break;
- }
- else if(result == GET_SUBBUF_DIED) {
- finish_consuming_dead_subbuffer(buf);
- break;
- }
-
- /* write data to file */
- result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size);
- if(result == -1) {
- PERROR("write");
- /* FIXME: maybe drop this trace */
- }
-
- /* put the subbuffer */
- result = put_subbuffer(buf);
- if(result == -1) {
- ERR("unknown error putting subbuffer (channel=%s)", buf->name);
- break;
- }
- else if(result == PUT_SUBBUF_PUSHED) {
- ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name);
- break;
- }
- else if(result == PUT_SUBBUF_DIED) {
- WARN("application died while putting subbuffer");
- /* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */
- finish_consuming_dead_subbuffer(buf);
- }
- else if(result == PUT_SUBBUF_OK) {
- }
- }
-
- DBG("thread for buffer %s is stopping", buf->name);
-
- /* FIXME: destroy, unalloc... */
-
- pthread_cleanup_pop(1);
-
- return NULL;