int eDVBRecordFileThread::asyncWrite(int len)
{
#ifdef SHOW_WRITE_TIME
	struct timeval starttime;
	struct timeval now;
	suseconds_t diff;
	gettimeofday(&starttime, NULL);
#endif

	m_ts_parser.parseData(m_current_offset, m_buffer, len);

#ifdef SHOW_WRITE_TIME
	gettimeofday(&now, NULL);
	diff = (1000000 * (now.tv_sec - starttime.tv_sec)) + now.tv_usec - starttime.tv_usec;
	eDebug("[eFilePushThreadRecorder] m_ts_parser.parseData: %9u us", (unsigned int)diff);
	gettimeofday(&starttime, NULL);
#endif

	int r = m_current_buffer->start(m_fd_dest, m_current_offset, len, m_buffer);
	if (r < 0)
	{
		eDebug("[eDVBRecordFileThread] aio_write failed: %m");
		return r;
	}
	m_current_offset += len;

#ifdef SHOW_WRITE_TIME
	gettimeofday(&now, NULL);
	diff = (1000000 * (now.tv_sec - starttime.tv_sec)) + now.tv_usec - starttime.tv_usec;
	eDebug("[eFilePushThreadRecorder] aio_write: %9u us", (unsigned int)diff);
#endif
	// Count how many buffers are still "busy". Move backwards from current,
	// because they can reasonably be expected to finish in that order.
	AsyncIOvector::iterator i = m_current_buffer;
	r = i->poll();
	int busy_count = 0;
	while (r > 0)
	{
		++busy_count;
		if (i == m_aio.begin())
			i = m_aio.end();
		--i;
		if (i == m_current_buffer)
		{
			eDebug("[eFilePushThreadRecorder] Warning: All write buffers busy");
			break;
		}
		r = i->poll();
		if (r < 0)
			return r;
	}
	++m_buffer_use_histogram[busy_count];

	++m_current_buffer;
	if (m_current_buffer == m_aio.end())
		m_current_buffer = m_aio.begin();
	m_buffer = m_current_buffer->buffer;
	return len;
}

void eDVBRecordFileThread::flush()
{
	eDebug("[eDVBRecordFileThread] waiting for aio to complete");
	for (AsyncIOvector::iterator it = m_aio.begin(); it != m_aio.end(); ++it)
	{
		it->wait();
	}
	int bufferCount = m_aio.size();
	eDebug("[eDVBRecordFileThread] buffer usage histogram (%d buffers of %d kB)", bufferCount, m_buffersize>>10);
	for (int i=0; i <= bufferCount; ++i)
	{
		if (m_buffer_use_histogram[i] != 0) eDebug("     %2d: %6d", i, m_buffer_use_histogram[i]);
	}
	if (m_overflow_count)
	{
		eDebug("[eDVBRecordFileThread] Demux buffer overflows: %d", m_overflow_count);
	}
	if (m_fd_dest >= 0)
	{
		posix_fadvise(m_fd_dest, 0, 0, POSIX_FADV_DONTNEED);
	}
}

int eDVBRecordStreamThread::writeData(int len)
{
	len = asyncWrite(len);
	if (len < 0)
		return len;
	// Cancel aio on this buffer before returning, streams should not be held up. So we CANCEL
	// any request that hasn't finished on the second round.
	int r = m_current_buffer->cancel(m_fd_dest);
	switch (r)
	{
		//case 0: // that's one of these two:
		case AIO_CANCELED:
		case AIO_ALLDONE:
			break;
		case AIO_NOTCANCELED:
			eDebug("[eDVBRecordStreamThread] failed to cancel, killing all waiting IO");
			aio_cancel(m_fd_dest, NULL);
			// Poll all open requests, because they are all in error state now.
			for (AsyncIOvector::iterator it = m_aio.begin(); it != m_aio.end(); ++it)
			{
				it->poll();
			}
			break;
		case -1:
			eDebug("[eDVBRecordStreamThread] failed: %m");
			return r;
	}
	// we want to have a consistent state, so wait for completion, just to be sure
	r = m_current_buffer->wait();
	if (r < 0)
	{
		eDebug("[eDVBRecordStreamThread] wait failed: %m");
		return -1;
	}
	return len;
}

int eDVBRecordFileThread::writeData(int len)
{
	len = asyncWrite(len);
	if (len < 0)
		return len;
	// Wait for previous aio to complete on this buffer before returning
	int r = m_current_buffer->wait();
	if (r < 0)
		return -1;
	return len;
}