static void data_thread_loop(void *data)
{
   data_runloop_t *runloop = (data_runloop_t*)data;

   RARCH_LOG("[Data Thread]: Initializing data thread.\n");

   slock_lock(runloop->lock);
   while (!runloop->thread_inited)
      scond_wait(runloop->cond, runloop->lock);
   slock_unlock(runloop->lock);

   RARCH_LOG("[Data Thread]: Starting data thread.\n");

   while (runloop->alive)
   {
      slock_lock(runloop->lock);

      if (!runloop->alive)
         break;

      data_runloop_iterate(true);

      while (!rarch_main_data_active())
         scond_wait(runloop->cond, runloop->lock);

      slock_unlock(runloop->lock);

   }

   RARCH_LOG("[Data Thread]: Stopping data thread.\n");
}

static void audio_thread_loop(void *data)
{
   audio_thread_t *thr = (audio_thread_t*)data;

   if (!thr)
      return;

   RARCH_LOG("[Audio Thread]: Initializing audio driver.\n");
   thr->driver_data   = thr->driver->init(thr->device, thr->out_rate, thr->latency);
   slock_lock(thr->lock);
   thr->inited        = thr->driver_data ? 1 : -1;
   if (thr->inited > 0 && thr->driver->use_float)
      thr->use_float  = thr->driver->use_float(thr->driver_data);
   scond_signal(thr->cond);
   slock_unlock(thr->lock);

   if (thr->inited < 0)
      return;

   /* Wait until we start to avoid calling 
    * stop immediately after initialization. */
   slock_lock(thr->lock);
   while (thr->stopped)
      scond_wait(thr->cond, thr->lock);
   slock_unlock(thr->lock);

   RARCH_LOG("[Audio Thread]: Starting audio.\n");

   for (;;)
   {
      global_t *global = global_get_ptr();
      
      slock_lock(thr->lock);

      if (!thr->alive)
      {
         scond_signal(thr->cond);
         slock_unlock(thr->lock);
         break;
      }

      if (thr->stopped)
      {
         thr->driver->stop(thr->driver_data);
         while (thr->stopped)
            scond_wait(thr->cond, thr->lock);
         thr->driver->start(thr->driver_data);
      }

      slock_unlock(thr->lock);
      global->system.audio_callback.callback();
   }

   RARCH_LOG("[Audio Thread]: Tearing down driver.\n");
   thr->driver->free(thr->driver_data);
}

// Returns FALSE if message not read, TRUE if it was read.  Will always return TRUE if "blocking" is set.
// Will throw MDFN_Error if the read message code is CDIF_MSG_FATAL_ERROR
bool CDIF_Queue::Read(CDIF_Message *message, bool blocking)
{
   bool ret = true;

   slock_lock((slock_t*)ze_mutex);

   if(blocking)
   {
      while(ze_queue.size() == 0)	// while, not just if.
         scond_wait((scond_t*)ze_cond, (slock_t*)ze_mutex);
   }

   if(ze_queue.size() == 0)
      ret = false;
   else
   {
      *message = ze_queue.front();
      ze_queue.pop();
   }  

   slock_unlock((slock_t*)ze_mutex);

   if(ret && message->message == CDIF_MSG_FATAL_ERROR)
      throw MDFN_Error(0, "%s", message->str_message.c_str());

   return(ret);
}

static void sunxi_update_main(const void *frame, struct sunxi_video *_dispvars)
{
   slock_lock(_dispvars->pending_mutex);

   if (_dispvars->pageflip_pending)
      scond_wait(_dispvars->vsync_condition, _dispvars->pending_mutex);

   slock_unlock(_dispvars->pending_mutex);

   /* Frame blitting */
   pixman_blit(
      _dispvars->src_width,
      _dispvars->src_height,
      _dispvars->nextPage->address, 
      _dispvars->dst_pixels_per_line,
      (uint16_t*)frame,
      _dispvars->src_pixels_per_line
      );

   /* Issue pageflip. Will flip on next vsync. */
   sunxi_layer_set_rgb_input_buffer(_dispvars->sunxi_disp, _dispvars->sunxi_disp->bits_per_pixel, 
      _dispvars->nextPage->offset,
      _dispvars->src_width, _dispvars->src_height, _dispvars->sunxi_disp->xres);

   slock_lock(_dispvars->pending_mutex);
   _dispvars->pageflip_pending = true;
   slock_unlock(_dispvars->pending_mutex);
}

static void gfx_ctx_vc_swap_buffers(void *data, void *data2)
{
#ifdef HAVE_EGL
   vc_ctx_data_t              *vc = (vc_ctx_data_t*)data;
   video_frame_info_t *video_info = (video_frame_info_t*)data2;

   if (!vc)
      return;

   egl_swap_buffers(&vc->egl);

   /* Wait for vsync immediately if we don't 
    * want egl_swap_buffers to triple-buffer */
   if (video_info->max_swapchain_images <= 2)
   {
      /* We DON'T wait to wait without callback function ready! */
      if (!vc->vsync_callback_set)
      {
         vc_dispmanx_vsync_callback(vc->dispman_display,
               dispmanx_vsync_callback, (void*)vc);
         vc->vsync_callback_set = true;
      }
      slock_lock(vc->vsync_condition_mutex);
      scond_wait(vc->vsync_condition, vc->vsync_condition_mutex);
      slock_unlock(vc->vsync_condition_mutex);
   }
   /* Stop generating vsync callbacks from now on */
   else if (vc->vsync_callback_set)
      vc_dispmanx_vsync_callback(vc->dispman_display, NULL, NULL);
#endif
}

/* If no free page is available when called, wait for a page flip. */
static struct dispmanx_page *dispmanx_get_free_page(void *data, struct dispmanx_surface *surface) {
   unsigned i;
   struct dispmanx_video *_dispvars = data;
   struct dispmanx_page *page = NULL;

   while (!page)
   {
      /* Try to find a free page */
      for (i = 0; i < surface->numpages; ++i) {
         if (!surface->pages[i].used)
         {
            page = (surface->pages) + i;
            break;
         }
      }

      /* If no page is free at the moment,
       * wait until a free page is freed by vsync CB. */
      if (!page) {
	 slock_lock(_dispvars->vsync_cond_mutex);
	 scond_wait(_dispvars->vsync_condition, _dispvars->vsync_cond_mutex);
	 slock_unlock(_dispvars->vsync_cond_mutex);
      }
   }
   
   /* We mark the choosen page as used */
   slock_lock(page->page_used_mutex);
   page->used = true;
   slock_unlock(page->page_used_mutex);

   return page;
}

static void filter_thread_loop(void *data)
{
   struct filter_thread_data *thr = (struct filter_thread_data*)data;

   for (;;)
   {
      bool die;
      slock_lock(thr->lock);
      while (thr->done && !thr->die)
         scond_wait(thr->cond, thr->lock);
      die = thr->die;
      slock_unlock(thr->lock);

      if (die)
         break;

      if (thr->packet && thr->packet->work)
         thr->packet->work(thr->userdata, thr->packet->thread_data);

      slock_lock(thr->lock);
      thr->done = true;
      scond_signal(thr->cond);
      slock_unlock(thr->lock);
   }
}

static void dispmanx_surface_update(void *data, const void *frame, struct dispmanx_surface *surface)
{
   struct dispmanx_video *_dispvars = data;
   struct dispmanx_page *page = NULL;
   
   /* Wait until last issued flip completes to get a free page. Also, 
      dispmanx doesn't support issuing more than one pageflip.*/
   slock_lock(_dispvars->pending_mutex);
   if (_dispvars->pageflip_pending > 0)
   {
      scond_wait(_dispvars->vsync_condition, _dispvars->pending_mutex);
   }
   slock_unlock(_dispvars->pending_mutex);
  
   page = dispmanx_get_free_page(_dispvars, surface);
 
   /* Frame blitting */
   vc_dispmanx_resource_write_data(page->resource, surface->pixformat,
      surface->pitch, (void*)frame, &(surface->bmp_rect));
   
   /* Issue a page flip that will be done at the next vsync. */
   _dispvars->update = vc_dispmanx_update_start(0);

   vc_dispmanx_element_change_source(_dispvars->update, surface->element,
         page->resource);

   vc_dispmanx_update_submit(_dispvars->update, dispmanx_vsync_callback, (void*)page);

   slock_lock(_dispvars->pending_mutex);
   _dispvars->pageflip_pending++;	
   slock_unlock(_dispvars->pending_mutex);
}

static void thread_wait_reply(thread_video_t *thr, enum thread_cmd cmd)
{
   slock_lock(thr->lock);
   while (cmd != thr->reply_cmd)
      scond_wait(thr->cond_cmd, thr->lock);
   slock_unlock(thr->lock);
}

bool ffemu_push_audio(ffemu_t *handle, const struct ffemu_audio_data *data)
{
   for (;;)
   {
      slock_lock(handle->lock);
      unsigned avail = fifo_write_avail(handle->audio_fifo);
      slock_unlock(handle->lock);

      if (!handle->alive)
         return false;

      if (avail >= data->frames * handle->params.channels * sizeof(int16_t))
         break;

      slock_lock(handle->cond_lock);
      if (handle->can_sleep)
      {
         handle->can_sleep = false;
         scond_wait(handle->cond, handle->cond_lock);
         handle->can_sleep = true;
      }
      else
         scond_signal(handle->cond);

      slock_unlock(handle->cond_lock);
   }

   slock_lock(handle->lock);
   fifo_write(handle->audio_fifo, data->data, data->frames * handle->params.channels * sizeof(int16_t));
   slock_unlock(handle->lock);
   scond_signal(handle->cond);

   return true;
}

static void dispmanx_flip(struct dispmanx_page *page, void *data)
{
   struct dispmanx_video *_dispvars = data;

   if (!_dispvars)
      return;

   /* Dispmanx doesn't support issuing more than one pageflip. 
    * If we do, the second CB isn't called. */
   if (_dispvars->pageflip_pending > 0)
   {
      slock_lock(_dispvars->vsync_cond_mutex);
      scond_wait(_dispvars->vsync_condition, _dispvars->vsync_cond_mutex);
      slock_unlock(_dispvars->vsync_cond_mutex);
   }

   /* Issue a page flip at the next vblank interval 
    * (will be done at vsync anyway). */
   _dispvars->update = vc_dispmanx_update_start(0);

   vc_dispmanx_element_change_source(_dispvars->update, _dispvars->element,
         _dispvars->resources[page->numpage]);

   vc_dispmanx_update_submit(_dispvars->update, vsync_callback, (void*)page);

   slock_lock(_dispvars->pending_mutex);
   _dispvars->pageflip_pending++;	
   slock_unlock(_dispvars->pending_mutex);
}

static size_t write_buffer(jack_t *jd, const float *buf, size_t size)
{
   int i;
   size_t j, frames, written = 0;
   jack_default_audio_sample_t out_deinterleaved_buffer[2][AUDIO_CHUNK_SIZE_NONBLOCKING * AUDIO_MAX_RATIO];

   frames = FRAMES(size);

   /* Avoid buffer overflow if a DSP plugin generated a huge number of frames. */
   if (frames > AUDIO_CHUNK_SIZE_NONBLOCKING * AUDIO_MAX_RATIO)
      frames = AUDIO_CHUNK_SIZE_NONBLOCKING * AUDIO_MAX_RATIO;

   for (i = 0; i < 2; i++)
      for (j = 0; j < frames; j++)
         out_deinterleaved_buffer[i][j] = buf[j * 2 + i];

   while (written < frames)
   {
      size_t avail[2], min_avail, write_frames;
      if (jd->shutdown)
         return 0;

      avail[0] = jack_ringbuffer_write_space(jd->buffer[0]);
      avail[1] = jack_ringbuffer_write_space(jd->buffer[1]);

      min_avail = avail[0] < avail[1] ? avail[0] : avail[1];
      min_avail /= sizeof(float);

      write_frames = frames - written > min_avail ? min_avail : frames - written;

      if (write_frames > 0)
      {
         for (i = 0; i < 2; i++)
         {
            jack_ringbuffer_write(jd->buffer[i], (const char*)&out_deinterleaved_buffer[i][written],
                  write_frames * sizeof(jack_default_audio_sample_t));
         }
         written += write_frames;
      }
#ifdef HAVE_THREADS
      else
      {
         slock_lock(jd->cond_lock);
         scond_wait(jd->cond, jd->cond_lock);
         slock_unlock(jd->cond_lock);
      }
#endif

      if (jd->nonblock)
         break;
   }

   return written * sizeof(float) * 2;
}

bool ffemu_push_video(ffemu_t *handle, const struct ffemu_video_data *data)
{
   unsigned y;
   bool drop_frame = handle->video.frame_drop_count++ % handle->video.frame_drop_ratio;
   handle->video.frame_drop_count %= handle->video.frame_drop_ratio;
   if (drop_frame)
      return true;

   for (;;)
   {
      slock_lock(handle->lock);
      unsigned avail = fifo_write_avail(handle->attr_fifo);
      slock_unlock(handle->lock);

      if (!handle->alive)
         return false;

      if (avail >= sizeof(*data))
         break;

      slock_lock(handle->cond_lock);
      if (handle->can_sleep)
      {
         handle->can_sleep = false;
         scond_wait(handle->cond, handle->cond_lock);
         handle->can_sleep = true;
      }
      else
         scond_signal(handle->cond);

      slock_unlock(handle->cond_lock);
   }

   slock_lock(handle->lock);

   // Tightly pack our frame to conserve memory. libretro tends to use a very large pitch.
   struct ffemu_video_data attr_data = *data;

   if (attr_data.is_dupe)
      attr_data.width = attr_data.height = attr_data.pitch = 0;
   else
      attr_data.pitch = attr_data.width * handle->video.pix_size;

   fifo_write(handle->attr_fifo, &attr_data, sizeof(attr_data));

   int offset = 0;
   for (y = 0; y < attr_data.height; y++, offset += data->pitch)
      fifo_write(handle->video_fifo, (const uint8_t*)data->data + offset, attr_data.pitch);

   slock_unlock(handle->lock);
   scond_signal(handle->cond);

   return true;
}

/* user -> thread */
static void thread_wait_reply(thread_video_t *thr, thread_packet_t *pkt)
{
   slock_lock(thr->lock);

   while (pkt->type != thr->reply_cmd)
      scond_wait(thr->cond_cmd, thr->lock);

   *pkt = thr->cmd_data;
   thr->cmd_data.type = CMD_NONE;

   slock_unlock(thr->lock);
}

static void threaded_worker(void *userdata)
{
   (void)userdata;

   RARCH_LOG("Threaded rarch_task started\n");

   for (;;)
   {
      rarch_task_t *queue = NULL;
      rarch_task_t *task  = NULL;
      rarch_task_t *next  = NULL;

      /* pop all into a local queue to avoid trouble with rarch_task_push(),
       * tasks are in the reverse order here. */
      slock_lock(running_lock);

      if (!worker_continue)
         break; /* should we keep running until all tasks finished? */

      while ((task = task_queue_get(&tasks_running)) != NULL)
      {
         task->next = queue;
         queue = task;
      }

      if (queue == NULL) /* no tasks running, lets wait a bit */
      {
         scond_wait(worker_cond, running_lock);
         slock_unlock(running_lock);
         continue;
      }

      slock_unlock(running_lock);

      for (task = queue; task; task = next)
      {
         next = task->next;
         task->handler(task);

         if (task->finished)
         {
            slock_lock(finished_lock);
            task_queue_put(&tasks_finished, task);
            slock_unlock(finished_lock);
         }
         else
            threaded_push_running(task);
      }
   }

   slock_unlock(running_lock);
}

static ssize_t sl_write(void *data, const void *buf_, size_t size)
{
   sl_t *sl = (sl_t*)data;
   size_t written = 0;
   const uint8_t *buf = (const uint8_t*)buf_;

   while (size)
   {
      size_t avail_write;

      if (sl->nonblock)
      {
         if (sl->buffered_blocks == sl->buf_count)
            break;
      }
      else
      {
         slock_lock(sl->lock);
         while (sl->buffered_blocks == sl->buf_count)
            scond_wait(sl->cond, sl->lock);
         slock_unlock(sl->lock);
      }

      avail_write = min(sl->buf_size - sl->buffer_ptr, size);

      if (avail_write)
      {
         memcpy(sl->buffer[sl->buffer_index] + sl->buffer_ptr, buf, avail_write);
         sl->buffer_ptr += avail_write;
         buf            += avail_write;
         size           -= avail_write;
         written        += avail_write;
      }

      if (sl->buffer_ptr >= sl->buf_size)
      {
         SLresult res = (*sl->buffer_queue)->Enqueue(sl->buffer_queue, sl->buffer[sl->buffer_index], sl->buf_size);
         sl->buffer_index = (sl->buffer_index + 1) % sl->buf_count;
         __sync_fetch_and_add(&sl->buffered_blocks, 1);
         sl->buffer_ptr = 0;

         if (res != SL_RESULT_SUCCESS)
         {
            RARCH_ERR("[OpenSL]: Failed to write! (Error: 0x%x)\n", (unsigned)res);
            return -1;
         }
      }
   }

   return written;
}

static ssize_t rs_write(void *data, const void *buf, size_t size)
{
   rsd_t *rsd = (rsd_t*)data;

   if (rsd->has_error)
      return -1;

   if (rsd->nonblock)
   {
      size_t avail, write_amt;

      rsd_callback_lock(rsd->rd);

      avail = fifo_write_avail(rsd->buffer);
      write_amt = avail > size ? size : avail;

      fifo_write(rsd->buffer, buf, write_amt);
      rsd_callback_unlock(rsd->rd);
      return write_amt;
   }
   else
   {
      size_t written = 0;
      while (written < size && !rsd->has_error)
      {
         size_t avail;
         rsd_callback_lock(rsd->rd);

         avail = fifo_write_avail(rsd->buffer);

         if (avail == 0)
         {
            rsd_callback_unlock(rsd->rd);
            if (!rsd->has_error)
            {
               slock_lock(rsd->cond_lock);
               scond_wait(rsd->cond, rsd->cond_lock);
               slock_unlock(rsd->cond_lock);
            }
         }
         else
         {
            size_t write_amt = size - written > avail ? avail : size - written;
            fifo_write(rsd->buffer, (const char*)buf + written, write_amt);
            rsd_callback_unlock(rsd->rd);
            written += write_amt;
         }
      }
      return written;
   }
}

static void threaded_worker(void *userdata)
{
   (void)userdata;

   for (;;)
   {
      retro_task_t *task  = NULL;

      if (!worker_continue)
         break; /* should we keep running until all tasks finished? */

      slock_lock(running_lock);

      /* Get first task to run */
      task = tasks_running.front;
      if (task == NULL)
      {
         scond_wait(worker_cond, running_lock);
         slock_unlock(running_lock);
         continue;
      }

      slock_unlock(running_lock);

      task->handler(task);

      slock_lock(running_lock);
      task_queue_remove(&tasks_running, task);
      slock_unlock(running_lock);

      /* Update queue */
      if (!task->finished)
      {
         /* Re-add task to running queue */
         retro_task_threaded_push_running(task);
      }
      else
      {
         /* Add task to finished queue */
         slock_lock(finished_lock);
         task_queue_put(&tasks_finished, task);
         slock_unlock(finished_lock);
      }

#if 0
      retro_sleep(10);
#endif
   }

   slock_unlock(running_lock);
}

static void android_app_set_input (void *data, AInputQueue* inputQueue)
{
   struct android_app *android_app = (struct android_app*)data;

   if (!android_app)
      return;

   slock_lock(android_app->mutex);
   android_app->pendingInputQueue = inputQueue;
   android_app_write_cmd(android_app, APP_CMD_INPUT_CHANGED);
   while (android_app->inputQueue != android_app->pendingInputQueue)
      scond_wait(android_app->cond, android_app->mutex);
   slock_unlock(android_app->mutex);
}

static ssize_t alsa_thread_write(void *data, const void *buf, size_t size)
{
   alsa_thread_t *alsa = (alsa_thread_t*)data;

   if (alsa->thread_dead)
      return -1;

   if (alsa->nonblock)
   {
      size_t avail;
      size_t write_amt;

      slock_lock(alsa->fifo_lock);
      avail           = fifo_write_avail(alsa->buffer);
      write_amt       = MIN(avail, size);

      fifo_write(alsa->buffer, buf, write_amt);
      slock_unlock(alsa->fifo_lock);

      return write_amt;
   }
   else
   {
      size_t written = 0;
      while (written < size && !alsa->thread_dead)
      {
         size_t avail;
         slock_lock(alsa->fifo_lock);
         avail = fifo_write_avail(alsa->buffer);

         if (avail == 0)
         {
            slock_unlock(alsa->fifo_lock);
            slock_lock(alsa->cond_lock);
            if (!alsa->thread_dead)
               scond_wait(alsa->cond, alsa->cond_lock);
            slock_unlock(alsa->cond_lock);
         }
         else
         {
            size_t write_amt = MIN(size - written, avail);
            fifo_write(alsa->buffer,
                  (const char*)buf + written, write_amt);
            slock_unlock(alsa->fifo_lock);
            written += write_amt;
         }
      }
      return written;
   }
}