text: 
  /*
   * machinarium.
   *
   * cooperative multitasking engine.
  */
  
  #include <machinarium.h>
  #include <machinarium_private.h>
  
  void
  mm_readahead_cb(mm_fd_t *handle)
  {
  	mm_io_t *io = handle->on_read_arg;
  	mm_call_t *call = &io->call;
  	if (mm_call_is_aborted(call))
  		return;
  
  	int left = io->readahead_size - io->readahead_pos;
  	int rc;
  	while (left > 0)
  	{
  		rc = mm_socket_read(io->fd, io->readahead_buf.start + io->readahead_pos, left);
  		if (rc == -1) {
  			if (errno == EAGAIN ||
  			    errno == EWOULDBLOCK)
  				break;
  			if (errno == EINTR)
  				continue;
  			io->readahead_status = errno;
  			io->connected = 0;
  
  			if (mm_call_is(call, MM_CALL_READ)) {
  				call->status = errno;
  				mm_scheduler_wakeup(&mm_self->scheduler, call->coroutine);
  			}
  			return;
  		}
  		io->readahead_pos += rc;
  		left = io->readahead_size - io->readahead_pos;
  		assert(left >= 0);
  		if (rc == 0) {
  			/* eof */
  			mm_readahead_stop(io);
  			io->connected = 0;
  			io->read_eof = 1;
  			io->readahead_status = 0;
  			if (mm_call_is(call, MM_CALL_READ))
  				call->status = 0;
  			break;
  		}
  		break;
  	}
  	io->readahead_status = 0;
  
  	if (mm_call_is(call, MM_CALL_READ)) {
  		call->status = 0;
  		int ra_left = io->readahead_pos - io->readahead_pos_read;
  		if (io->read_eof || ra_left >= io->read_size)
  			mm_scheduler_wakeup(&mm_self->scheduler, call->coroutine);
  	}
  }
  
  int mm_readahead_start(mm_io_t *io, mm_fd_callback_t callback, void *arg)
  {
  	mm_machine_t *machine = mm_self;
  	int rc;
  	rc = mm_buf_ensure(&io->readahead_buf, io->readahead_size);
  	if (rc == -1) {
  		mm_errno_set(ENOMEM);
  		return -1;
  	}
  	rc = mm_loop_read(&machine->loop, &io->handle, callback, arg);
  	if (rc == -1) {
  		mm_errno_set(errno);
  		return -1;
  	}
  	return 0;
  }
  
  int mm_readahead_stop(mm_io_t *io)
  {
  	mm_machine_t *machine = mm_self;
  	int rc;
  	rc = mm_loop_read_stop(&machine->loop, &io->handle);
  	if (rc == -1) {
  		mm_errno_set(errno);
  		return -1;
  	}
  	return 0;
  }
  
  static int
  mm_readahead_read(mm_io_t *io, uint32_t time_ms)
  {
  	assert(io->read_size <= io->readahead_size);
  
  	/* try to use readahead first */
  	assert(io->readahead_pos >= io->readahead_pos_read);
  	int ra_left = io->readahead_pos - io->readahead_pos_read;
  	if (ra_left >= io->read_size) {
  		memcpy(io->read_buf, io->readahead_buf.start + io->readahead_pos_read,
  		       io->read_size);
  		io->readahead_pos_read += io->read_size;
  		return 0;
  	}
  	if (io->readahead_status != 0) {
  		mm_errno_set(io->readahead_status);
  		return -1;
  	}
  	if (io->read_eof || !io->connected) {
  		mm_errno_set(ECONNRESET);
  		return -1;
  	}
  
  	/* copy readahead chunk */
  	int copy_pos = 0;
  	if (ra_left > 0) {
  		memcpy(io->read_buf,
  		       io->readahead_buf.start + io->readahead_pos_read,
  		       ra_left);
  		io->readahead_pos_read += ra_left;
  		io->read_size -= ra_left;
  		copy_pos = ra_left;
  	}
  
  	/* reset readahead position */
  	assert(io->readahead_pos_read == io->readahead_pos);
  	io->readahead_pos = 0;
  	io->readahead_pos_read = 0;
  	mm_buf_reset(&io->readahead_buf);
  
  	/* maybe allocate readahead buffer and-or start io */
  	int rc;
  	rc = mm_readahead_start(io, mm_readahead_cb, io);
  	if (rc == -1)
  		return -1;
  
  	/* wait for completion */
  	mm_call(&io->call, MM_CALL_READ, time_ms);
  
  	rc = io->call.status;
  	if (rc == 0)
  		rc = io->readahead_status;
  	if (rc != 0) {
  		mm_errno_set(rc);
  		return -1;
  	}
  	ra_left = io->readahead_pos - io->readahead_pos_read;
  	if (ra_left < io->read_size) {
  		mm_errno_set(ECONNRESET);
  		return -1;
  	}
  
  	memcpy(io->read_buf + copy_pos,
  	       io->readahead_buf.start + io->readahead_pos_read,
  	       io->read_size);
  	io->readahead_pos_read += io->read_size;
  	return 0;
  }
  
  static inline int
  mm_read_chunk(mm_io_t *io, char *buf, int size, uint32_t time_ms)
  {
  	mm_errno_set(0);
  	if (mm_call_is_active(&io->call)) {
  		mm_errno_set(EINPROGRESS);
  		return -1;
  	}
  	if (! io->attached) {
  		mm_errno_set(ENOTCONN);
  		return -1;
  	}
  	io->read_buf  = buf;
  	io->read_size = size;
  	io->read_pos  = 0;
  	return mm_readahead_read(io, time_ms);
  }
  
  int mm_read(mm_io_t *io, char *buf, int size, uint32_t time_ms)
  {
  	/* split read buffer into readahead-sized chunks */
  	int total = 0;
  	while (total < size)
  	{
  		int to_read = size - total;
  		if (to_read > io->readahead_size)
  			to_read = io->readahead_size;
  		int rc;
  		rc = mm_read_chunk(io, buf + total, to_read, time_ms);
  		if (rc == -1)
  			return -1;
  		total += to_read;
  	}
  	return 0;
  }
  
  MACHINE_API int
  machine_read(machine_io_t *obj, char *buf, int size, uint32_t time_ms)
  {
  	mm_io_t *io = mm_cast(mm_io_t*, obj);
  	if (mm_tlsio_is_active(&io->tls))
  		return mm_tlsio_read(&io->tls, buf, size, time_ms);
  	return mm_read(io, buf, size, time_ms);
  }
  
  MACHINE_API int
  machine_set_readahead(machine_io_t *obj, int size)
  {
  	mm_io_t *io = mm_cast(mm_io_t*, obj);
  	mm_errno_set(0);
  	int rc;
  	rc = mm_buf_ensure(&io->readahead_buf, size);
  	if (rc == -1) {
  		mm_errno_set(ENOMEM);
  		return -1;
  	}
  	io->readahead_size = size;
  	return 0;
  }
  
  MACHINE_API int
  machine_read_pending(machine_io_t *obj)
  {
  	mm_io_t *io = mm_cast(mm_io_t*, obj);
  
  	mm_errno_set(0);
  	if (mm_call_is_active(&io->call)) {
  		mm_errno_set(EINPROGRESS);
  		return -1;
  	}
  	if (! io->attached) {
  		mm_errno_set(ENOTCONN);
  		return -1;
  	}
  
  	/* check if io has any pending read data */
  	int ra_left = io->readahead_pos - io->readahead_pos_read;
  	if (ra_left > 0)
  		return 1;
  
  	/* check if there are any data buffered inside SSL context */
  	if (mm_tlsio_is_active(&io->tls) && mm_tlsio_read_pending(&io->tls))
  		return 1;
  
  	/* check if io has error status set */
  	if (io->readahead_status != 0)
  		return 1;
  
  	/* check if io reached eof */
  	if (! io->connected)
  		return 1;
  
  	return 0;
  }
  