static ssize_t pulse_write(void *data, const void *buf_, size_t size)
{
   pa_t *pa = (pa_t*)data;
   const uint8_t *buf = (const uint8_t*)buf_;
   size_t written = 0;

   pa_threaded_mainloop_lock(pa->mainloop);
   while (size)
   {
      size_t writable = MIN(size, pa_stream_writable_size(pa->stream));

      if (writable)
      {
         pa_stream_write(pa->stream, buf, writable, NULL, 0, PA_SEEK_RELATIVE);
         buf += writable;
         size -= writable;
         written += writable;
      }
      else if (!pa->nonblock)
         pa_threaded_mainloop_wait(pa->mainloop);
      else
         break;
   }

   pa_threaded_mainloop_unlock(pa->mainloop);

   return written;
}

static int outstream_start_pa(struct SoundIoPrivate *si, struct SoundIoOutStreamPrivate *os) {
    struct SoundIoOutStream *outstream = &os->pub;
    struct SoundIoPulseAudio *sipa = &si->backend_data.pulseaudio;
    struct SoundIoOutStreamPulseAudio *ospa = &os->backend_data.pulseaudio;

    pa_threaded_mainloop_lock(sipa->main_loop);

    ospa->write_byte_count = pa_stream_writable_size(ospa->stream);
    int frame_count = ospa->write_byte_count / outstream->bytes_per_frame;
    outstream->write_callback(outstream, 0, frame_count);

    pa_operation *op = pa_stream_cork(ospa->stream, false, NULL, NULL);
    if (!op) {
        pa_threaded_mainloop_unlock(sipa->main_loop);
        return SoundIoErrorStreaming;
    }
    pa_operation_unref(op);
    pa_stream_set_write_callback(ospa->stream, playback_stream_write_callback, os);
    pa_stream_set_underflow_callback(ospa->stream, playback_stream_underflow_callback, outstream);
    pa_stream_set_overflow_callback(ospa->stream, playback_stream_underflow_callback, outstream);

    pa_threaded_mainloop_unlock(sipa->main_loop);

    return 0;
}

static void pulse_write(void* ptr, int length) {
    int writeoffs, remain, writable;

    CHECK_CONNECTED();

    pa_threaded_mainloop_lock(mainloop);
    CHECK_DEAD_GOTO(fail, 1);

    /* break large fragments into smaller fragments. --nenolod */
    for (writeoffs = 0, remain = length;
         writeoffs < length;
         writeoffs += writable, remain -= writable)
    {
         void * pptr = (char *) ptr + writeoffs;

         writable = length - writeoffs;
         size_t fragsize = pa_stream_writable_size(stream);

         /* don't write more than what PA is willing to handle right now. */
         if (writable > fragsize)
             writable = fragsize;

         if (pa_stream_write(stream, pptr, writable, NULL, PA_SEEK_RELATIVE, 0) < 0) {
             AUDDBG("pa_stream_write() failed: %s", pa_strerror(pa_context_errno(context)));
             goto fail;
         }
    }

    do_trigger = 0;
    written += length;

fail:
    pa_threaded_mainloop_unlock(mainloop);
}

int AudioOutputPulseAudio::GetBufferedOnSoundcard(void) const
{
    pa_usec_t latency = 0;
    size_t buffered = 0;

    if (!pcontext || pa_context_get_state(pcontext) != PA_CONTEXT_READY)
        return 0;

    if (!pstream || pa_stream_get_state(pstream) != PA_STREAM_READY)
        return 0;

    const pa_buffer_attr *buf_attr =  pa_stream_get_buffer_attr(pstream);
    size_t bfree = pa_stream_writable_size(pstream);
    buffered = buf_attr->tlength - bfree;

    pa_threaded_mainloop_lock(mainloop);

    while (pa_stream_get_latency(pstream, &latency, NULL) < 0)
    {
        if (pa_context_errno(pcontext) != PA_ERR_NODATA)
        {
            latency = 0;
            break;
        }
        pa_threaded_mainloop_wait(mainloop);
    }

    pa_threaded_mainloop_unlock(mainloop);

    return ((uint64_t)latency * samplerate *
            output_bytes_per_frame / 1000000) + buffered;
}

/** Return number of bytes that may be written to the server without blocking */
static int get_space(void) {
    size_t l;
    pa_threaded_mainloop_lock(mainloop);
    l = pa_stream_writable_size(stream);
    pa_threaded_mainloop_unlock(mainloop);
    return l;
}

unsigned long SoundGetBytesBuffered (void)
{
     int free_space;
     int error_code;
     long latency;
     int playing = 0;

     if ((device.mainloop == NULL) || (device.api == NULL) || ( device.context == NULL) || (device.stream == NULL))
     	  return SOUNDSIZE;

     pa_threaded_mainloop_lock (device.mainloop);
     free_space = pa_stream_writable_size (device.stream);
     pa_threaded_mainloop_unlock (device.mainloop);

     //fprintf (stderr, "Free space: %d\n", free_space);
     //fprintf (stderr, "Used space: %d\n", maxlength - free_space);
     if  (free_space < mixlen * 3)
     {
	  // Don't buffer anymore, just play
	  //fprintf (stderr, "Not buffering.\n");
     	  return SOUNDSIZE;
     }
     else 
     {
	  // Buffer some sound
	  //fprintf (stderr, "Buffering.\n");
     	  return 0;
     }
}

unsigned int CAESinkPULSE::AddPackets(uint8_t **data, unsigned int frames, unsigned int offset)
{
  if (!m_IsAllocated)
    return 0;

  if (m_IsStreamPaused)
  {
    Pause(false);
  }

  pa_threaded_mainloop_lock(m_MainLoop);

  unsigned int available = frames * m_format.m_frameSize;
  unsigned int length = 0;
  void *buffer = data[0]+offset*m_format.m_frameSize;
  // care a bit for fragmentation
  while ((length = pa_stream_writable_size(m_Stream)) < m_periodSize)
    pa_threaded_mainloop_wait(m_MainLoop);

  length =  std::min((unsigned int)length, available);

  int error = pa_stream_write(m_Stream, buffer, length, NULL, 0, PA_SEEK_RELATIVE);
  pa_threaded_mainloop_unlock(m_MainLoop);

  if (error)
  {
    CLog::Log(LOGERROR, "CPulseAudioDirectSound::AddPackets - pa_stream_write failed\n");
    return 0;
  }
  unsigned int res = (unsigned int)(length / m_format.m_frameSize);

  return res;
}

static size_t pulse_write_avail(void *data)
{
    pa_t *pa = (pa_t*)data;
    pa_threaded_mainloop_lock(pa->mainloop);
    size_t length = pa_stream_writable_size(pa->stream);
    pa_threaded_mainloop_unlock(pa->mainloop);
    return length;
}

// Return number of bytes that may be written to the server without blocking
static int get_space(struct ao *ao)
{
    struct priv *priv = ao->priv;
    pa_threaded_mainloop_lock(priv->mainloop);
    size_t space = pa_stream_writable_size(priv->stream);
    pa_threaded_mainloop_unlock(priv->mainloop);
    return space;
}

static int
rdpsnd_pulse_play(rdpsndDevicePlugin * devplugin, char * data, int size)
{
	struct pulse_device_data * pulse_data;
	int len;
	int ret;
	uint8 * decoded_data;
	char * src;
	int decoded_size;

	pulse_data = (struct pulse_device_data *) devplugin->device_data;
	if (!pulse_data->stream)
		return 1;

	if (pulse_data->format == 0x11)
	{
		decoded_data = pulse_data->pDecodeImaAdpcm(&pulse_data->adpcm,
			(uint8 *) data, size, pulse_data->sample_spec.channels, pulse_data->block_size, &decoded_size);
		size = decoded_size;
		src = (char *) decoded_data;
	}
	else
	{
		decoded_data = NULL;
		src = data;
	}

	LLOGLN(10, ("rdpsnd_pulse_play: size %d", size));

	pa_threaded_mainloop_lock(pulse_data->mainloop);
	while (size > 0)
	{
		while ((len = pa_stream_writable_size(pulse_data->stream)) == 0)
		{
			LLOGLN(10, ("rdpsnd_pulse_play: waiting"));
			pa_threaded_mainloop_wait(pulse_data->mainloop);
		}
		if (len < 0)
			break;
		if (len > size)
			len = size;
		ret = pa_stream_write(pulse_data->stream, src, len, NULL, 0LL, PA_SEEK_RELATIVE);
		if (ret < 0)
		{
			LLOGLN(0, ("rdpsnd_pulse_play: pa_stream_write failed (%d)",
				pa_context_errno(pulse_data->context)));
			break;
		}
		src += len;
		size -= len;
	}
	pa_threaded_mainloop_unlock(pulse_data->mainloop);

	if (decoded_data)
		free(decoded_data);

	return 0;
}

void AudioOutputPulseAudio::WriteAudio(uchar *aubuf, int size)
{
    QString fn_log_tag = "WriteAudio, ";
    pa_stream_state_t sstate = pa_stream_get_state(pstream);

    VBAUDIOTS(fn_log_tag + QString("writing %1 bytes").arg(size));

    /* NB This "if" check can be replaced with PA_STREAM_IS_GOOD() in
       PulseAudio API from 0.9.11. As 0.9.10 is still widely used
       we use the more verbose version for now */

    if (sstate == PA_STREAM_CREATING || sstate == PA_STREAM_READY)
    {
        int write_status = PA_ERR_INVALID;
        size_t to_write = size;
        unsigned char *buf_ptr = aubuf;

        pa_threaded_mainloop_lock(mainloop);
        while (to_write > 0)
        {
            write_status = 0;
            size_t writable = pa_stream_writable_size(pstream);
            if (writable > 0)
            {
                size_t write = min(to_write, writable);
                write_status = pa_stream_write(pstream, buf_ptr, write,
                                               NULL, 0, PA_SEEK_RELATIVE);

                if (0 != write_status)
                    break;

                buf_ptr += write;
                to_write -= write;
            }
            else
            {
                pa_threaded_mainloop_wait(mainloop);
            }
        }
        pa_threaded_mainloop_unlock(mainloop);

        if (to_write > 0)
        {
            if (write_status != 0)
                VBERROR(fn_log_tag + QString("stream write failed: %1")
                                     .arg(write_status == PA_ERR_BADSTATE
                                                ? "PA_ERR_BADSTATE"
                                                : "PA_ERR_INVALID"));

            VBERROR(fn_log_tag + QString("short write, %1 of %2")
                                 .arg(size - to_write).arg(size));
        }
    }
    else
        VBERROR(fn_log_tag + QString("stream state not good: %1")
                             .arg(sstate,0,16));
}

static void underrun_update_cb(pa_stream *s, void *data)
{
   pa_t *pa = (pa_t*)data;

   (void)s;

   RARCH_LOG("[PulseAudio]: Underrun (Buffer: %u, Writable size: %u).\n",
         (unsigned)pa->buffer_size,
         (unsigned)pa_stream_writable_size(pa->stream));
}

/** Return number of bytes that may be written to the server without blocking */
static int get_space(void) {
    uint32_t l;
    assert(stream && context && pa_stream_get_state(stream) == PA_STREAM_READY);
    
    keep_alive();

    l = pa_stream_writable_size(stream);
    
    return l;
}

unsigned int CPulseAEStream::GetSpace()
{
  if (!m_Initialized)
    return 0;

  pa_threaded_mainloop_lock(m_MainLoop);
  unsigned int size = pa_stream_writable_size(m_Stream);
  pa_threaded_mainloop_unlock(m_MainLoop);

  return size;
}

static void
stream_write_callback(pa_stream *s, size_t length, void *userdata)
{
  decoder_t *d = userdata;
  media_pipe_t *mp = d->ad.ad_mp;
  int writable = pa_stream_writable_size(d->s);

  if(writable && d->blocked) {
    d->blocked = 0;
    mp_send_cmd(mp, &mp->mp_audio, MB_CTRL_UNBLOCK);
  }
}

static int pulse_free(void) {
  ENTER(__FUNCTION__);
    size_t l = 0;
    pa_operation *o = NULL;

    CHECK_CONNECTED(0);

    SHOW("pulse_free: %s (call)\n", "pa_threaded_main_loop_lock");
    pa_threaded_mainloop_lock(mainloop);
    CHECK_DEAD_GOTO(fail, 1);

    if ((l = pa_stream_writable_size(stream)) == (size_t) -1) {
        SHOW("pa_stream_writable_size() failed: %s", pa_strerror(pa_context_errno(context)));
        l = 0;
        goto fail;
    }

    SHOW("pulse_free: %s (ret=%d)\n", "pa_stream_writable_size", l);

    /* If this function is called twice with no pulse_write() call in
     * between this means we should trigger the playback */
    if (do_trigger) {
        int success = 0;

	SHOW("pulse_free: %s (call)\n", "pa_stream_trigger");
        if (!(o = pa_stream_trigger(stream, stream_success_cb, &success))) {
            SHOW("pa_stream_trigger() failed: %s", pa_strerror(pa_context_errno(context)));
            goto fail;
        }
        
	SHOW("pulse_free: %s (call)\n", "pa_threaded_main_loop");
        while (pa_operation_get_state(o) != PA_OPERATION_DONE) {
            CHECK_DEAD_GOTO(fail, 1);
            pa_threaded_mainloop_wait(mainloop);
        } 
	SHOW("pulse_free: %s (ret)\n", "pa_threaded_main_loop");
       
        if (!success)
            SHOW("pa_stream_trigger() failed: %s", pa_strerror(pa_context_errno(context)));
    }
    
fail:
    SHOW("pulse_free: %s (call)\n", "pa_operation_unref");
    if (o)
        pa_operation_unref(o);
    
    SHOW("pulse_free: %s (call)\n", "pa_threaded_main_loop_unlock");
    pa_threaded_mainloop_unlock(mainloop);

    do_trigger = !!l;
    SHOW("pulse_free: %d (ret)\n", (int)l);
    return (int) l;
}

static size_t pulse_write_avail(void *data)
{
   size_t length;
   pa_t *pa = (pa_t*)data;

   pa_threaded_mainloop_lock(pa->mainloop);
   length = pa_stream_writable_size(pa->stream);

   audio_driver_set_buffer_size(pa->buffer_size); /* Can change spuriously. */
   pa_threaded_mainloop_unlock(pa->mainloop);
   return length;
}

static void rdpsnd_pulse_play(rdpsndDevicePlugin* device, uint8* data, int size)
{
	rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
	int len;
	int ret;
	uint8* decoded_data;
	uint8* src;
	int decoded_size;

	if (!pulse->stream)
		return;

	if (pulse->format == 0x11)
	{
		decoded_data = dsp_decode_ima_adpcm(&pulse->adpcm,
			data, size, pulse->sample_spec.channels, pulse->block_size, &decoded_size);
		size = decoded_size;
		src = decoded_data;
	}
	else
	{
		decoded_data = NULL;
		src = data;
	}

	pa_threaded_mainloop_lock(pulse->mainloop);
	while (size > 0)
	{
		while ((len = pa_stream_writable_size(pulse->stream)) == 0)
		{
			pa_threaded_mainloop_wait(pulse->mainloop);
		}
		if (len < 0)
			break;
		if (len > size)
			len = size;
		ret = pa_stream_write(pulse->stream, src, len, NULL, 0LL, PA_SEEK_RELATIVE);
		if (ret < 0)
		{
			DEBUG_WARN("pa_stream_write failed (%d)",
				pa_context_errno(pulse->context));
			break;
		}
		src += len;
		size -= len;
	}
	pa_threaded_mainloop_unlock(pulse->mainloop);

	if (decoded_data)
		xfree(decoded_data);
}

gboolean xmms_pulse_backend_write (xmms_pulse *p, const char *data,
                                   size_t length, int *rerror)
{
	assert (p);

	if (!data || !length) {
		if (rerror)
			*rerror = PA_ERR_INVALID;
		return FALSE;
	}

	pa_threaded_mainloop_lock (p->mainloop);
	if (!check_pulse_health (p, rerror))
		goto unlock_and_fail;

	while (length > 0) {
		size_t buf_len;
		int ret;

		while (!(buf_len = pa_stream_writable_size (p->stream))) {
			pa_threaded_mainloop_wait (p->mainloop);
			if (!check_pulse_health (p, rerror))
				goto unlock_and_fail;
		}

		if (buf_len == (size_t)-1) {
			if (rerror)
				*rerror = pa_context_errno ((p)->context);
			goto unlock_and_fail;
		}
		if (buf_len > length)
			buf_len = length;

		ret = pa_stream_write (p->stream, data, buf_len, NULL, 0, PA_SEEK_RELATIVE);
		if (ret < 0) {
			if (rerror)
				*rerror = pa_context_errno ((p)->context);
			goto unlock_and_fail;
		}

		data += buf_len;
		length -= buf_len;
	}

	pa_threaded_mainloop_unlock (p->mainloop);
	return TRUE;

 unlock_and_fail:
	pa_threaded_mainloop_unlock (p->mainloop);
	return FALSE;
}

static ssize_t pulse_write(void *data, const void *buf, size_t size)
{
    pa_t *pa = (pa_t*)data;

    pa_threaded_mainloop_lock(pa->mainloop);
    size_t length = pa_stream_writable_size(pa->stream);
    pa_threaded_mainloop_unlock(pa->mainloop);

    while (length < size && !pa->nonblock)
    {
        pa_threaded_mainloop_lock(pa->mainloop);
        pa_threaded_mainloop_wait(pa->mainloop);
        length = pa_stream_writable_size(pa->stream);
        pa_threaded_mainloop_unlock(pa->mainloop);
    }

    size_t write_size = min(length, size);

    pa_threaded_mainloop_lock(pa->mainloop);
    pa_stream_write(pa->stream, buf, write_size, NULL, 0LL, PA_SEEK_RELATIVE);
    pa_threaded_mainloop_unlock(pa->mainloop);
    return write_size;
}