/*
 * Copyright (C) 2007 Tony Arcieri
 * You may redistribute this under the terms of the Ruby license.
 * See LICENSE for details
 */

#include "ruby.h"
#include "rubyio.h"

#include <assert.h>

#include <string.h>
#include <time.h>
#include <errno.h>

#ifndef GetReadFile
#define FPTR_TO_FD(fptr) (fptr->fd)
#else
#define FPTR_TO_FD(fptr) (fileno(GetReadFile(fptr)))
#endif

/* Default number of bytes in each node's buffer */
#define DEFAULT_NODE_SIZE 16384

/* Maximum age of a buffer node in a memory pool, in seconds */
#define MAX_AGE 60

/* How often to scan the pool for old nodes */
#define PURGE_INTERVAL 10

struct buffer {
	time_t last_purged_at;
  unsigned size, node_size;
  struct buffer_node *head, *tail;
  struct buffer_node *pool_head, *pool_tail;
	
};

struct buffer_node {
	time_t last_used_at;
  unsigned start, end;
  struct buffer_node *next;
  unsigned char data[0];
};

static VALUE mEm = Qnil;
static VALUE cEm_Buffer = Qnil;

static VALUE Em_Buffer_allocate(VALUE klass);
static void Em_Buffer_mark(struct buffer *);
static void Em_Buffer_free(struct buffer *);

static VALUE Em_Buffer_initialize(int argc, VALUE *argv, VALUE self);
static VALUE Em_Buffer_clear(VALUE self);
static VALUE Em_Buffer_size(VALUE self);
static VALUE Em_Buffer_empty(VALUE self);
static VALUE Em_Buffer_append(VALUE self, VALUE data);
static VALUE Em_Buffer_prepend(VALUE self, VALUE data);
static VALUE Em_Buffer_read(int argc, VALUE *argv, VALUE self);
static VALUE Em_Buffer_to_str(VALUE self);
static VALUE Em_Buffer_read_from(VALUE self, VALUE io);
static VALUE Em_Buffer_write_to(VALUE self, VALUE io);

static struct buffer *buffer_new(void);
static void buffer_clear(struct buffer *buf);
static void buffer_free(struct buffer *buf);
static void buffer_gc(struct buffer *buf);
static void buffer_prepend(struct buffer *buf, char *str, unsigned len);
static void buffer_append(struct buffer *buf, char *str, unsigned len);
static void buffer_read(struct buffer *buf, char *str, unsigned len);
static void buffer_copy(struct buffer *buf, char *str, unsigned len);
static int buffer_read_from(struct buffer *buf, int fd);
static int buffer_write_to(struct buffer *buf, int fd);

/* 
 * High speed buffering geared towards non-blocking I/O.
 *
 * Data is stored in a byte queue implemented as a linked list of equal size
 * chunks.  Since every node in the list is the same size they are easily
 * memory pooled.  Routines are provided for high speed non-blocking reads
 * and writes from Ruby IO objects.
 */
void Init_em_buffer()
{
  mEm = rb_define_module("EventMachine");
  cEm_Buffer = rb_define_class_under(mEm, "Buffer", rb_cObject);
  rb_define_alloc_func(cEm_Buffer, Em_Buffer_allocate);

  rb_define_method(cEm_Buffer, "initialize", Em_Buffer_initialize, -1);
  rb_define_method(cEm_Buffer, "clear", Em_Buffer_clear, 0);
  rb_define_method(cEm_Buffer, "size", Em_Buffer_size, 0);
  rb_define_method(cEm_Buffer, "empty?", Em_Buffer_empty, 0);
  rb_define_method(cEm_Buffer, "<<", Em_Buffer_append, 1);
  rb_define_method(cEm_Buffer, "append", Em_Buffer_append, 1);
  rb_define_method(cEm_Buffer, "prepend", Em_Buffer_prepend, 1);
  rb_define_method(cEm_Buffer, "read", Em_Buffer_read, -1);
	rb_define_method(cEm_Buffer, "to_str", Em_Buffer_to_str, 0);
	rb_define_method(cEm_Buffer, "read_from", Em_Buffer_read_from, 1);
  rb_define_method(cEm_Buffer, "write_to", Em_Buffer_write_to, 1);
}

static VALUE Em_Buffer_allocate(VALUE klass)
{
  return Data_Wrap_Struct(klass, Em_Buffer_mark, Em_Buffer_free, buffer_new());
}

static void Em_Buffer_mark(struct buffer *buf)
{
  /* Walks the pool of unused chunks and frees any that are beyond a certain age */
  buffer_gc(buf);
}

static void Em_Buffer_free(struct buffer *buf)
{
  buffer_free(buf);
}

/**
 *  call-seq:
 *    EventMachine::Buffer.new(size = DEFAULT_NODE_SIZE) -> EventMachine::Buffer
 * 
 * Create a new EventMachine::Buffer with linked segments of the given size
 */
static VALUE Em_Buffer_initialize(int argc, VALUE *argv, VALUE self)
{
  VALUE node_size_obj;
  int node_size;
  struct buffer *buf;

  if(rb_scan_args(argc, argv, "01", &node_size_obj) == 1) {
    node_size = NUM2INT(node_size_obj);

    if(node_size < 1) rb_raise(rb_eArgError, "invalid buffer size");

    Data_Get_Struct(self, struct buffer, buf);

    /* Make sure we're not changing the buffer size after data has been allocated */
    assert(!buf->head);
    assert(!buf->pool_head);

    buf->node_size = node_size;
  }

  return Qnil;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#clear -> nil
 * 
 * Clear all data from the EventMachine::Buffer
 */
static VALUE Em_Buffer_clear(VALUE self)
{
  struct buffer *buf;
  Data_Get_Struct(self, struct buffer, buf);

  buffer_clear(buf);

  return Qnil;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#size -> Integer
 * 
 * Return the size of the buffer in bytes
 */
static VALUE Em_Buffer_size(VALUE self) 
{
  struct buffer *buf;
  Data_Get_Struct(self, struct buffer, buf);

  return INT2NUM(buf->size);
}

/**
 *  call-seq:
 *    EventMachine::Buffer#empty? -> Boolean
 * 
 * Is the buffer empty?
 */
static VALUE Em_Buffer_empty(VALUE self) 
{
  struct buffer *buf;
  Data_Get_Struct(self, struct buffer, buf);

  return buf->size > 0 ? Qfalse : Qtrue;	
}

/**
 *  call-seq:
 *    EventMachine::Buffer#append(data) -> String
 * 
 * Append the given data to the end of the buffer
 */
static VALUE Em_Buffer_append(VALUE self, VALUE data)
{
  struct buffer *buf;
  Data_Get_Struct(self, struct buffer, buf);

  /* Is this needed?  Never seen anyone else do it... */
  data = rb_convert_type(data, T_STRING, "String", "to_str");
  buffer_append(buf, RSTRING_PTR(data), RSTRING_LEN(data));

  return data;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#prepend(data) -> String
 * 
 * Prepend the given data to the beginning of the buffer
 */
static VALUE Em_Buffer_prepend(VALUE self, VALUE data)
{
  struct buffer *buf;
  Data_Get_Struct(self, struct buffer, buf);

  data = rb_convert_type(data, T_STRING, "String", "to_str");
  buffer_prepend(buf, RSTRING_PTR(data), RSTRING_LEN(data));

  return data;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#read(length = nil) -> String
 * 
 * Read the specified abount of data from the buffer.  If no value
 * is given the entire contents of the buffer are returned.  Any data
 * read from the buffer is cleared.
 */
static VALUE Em_Buffer_read(int argc, VALUE *argv, VALUE self)
{
  VALUE length_obj, str;
  int length;
  struct buffer *buf;

  Data_Get_Struct(self, struct buffer, buf);

  if(rb_scan_args(argc, argv, "01", &length_obj) == 1) {
    length = NUM2INT(length_obj);
  } else {
    if(buf->size == 0)
      return rb_str_new2("");

    length = buf->size;
  }

  if(length > buf->size)
    length = buf->size;

  if(length < 1)
    rb_raise(rb_eArgError, "length must be greater than zero");

  str = rb_str_new(0, length);
  buffer_read(buf, RSTRING_PTR(str), length);

  return str;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#to_str -> String
 * 
 * Convert the Buffer to a String.  The original buffer is unmodified.
 */
static VALUE Em_Buffer_to_str(VALUE self) {
	VALUE str;
	struct buffer *buf;
	
	Data_Get_Struct(self, struct buffer, buf);
	
	str = rb_str_new(0, buf->size);
	buffer_copy(buf, RSTRING_PTR(str), buf->size);
  
  return str;
}

/**
 *  call-seq:
 *    EventMachine::Buffer#read_from(io) -> Integer
 * 
 * Perform a nonblocking read of the the given IO object and fill
 * the buffer with any data received.  The call will read as much
 * data as it can until the read would block.
 */
static VALUE Em_Buffer_read_from(VALUE self, VALUE io) {
	struct buffer *buf;
#if HAVE_RB_IO_T
  rb_io_t *fptr;
#else
  OpenFile *fptr;
#endif

  Data_Get_Struct(self, struct buffer, buf);
  GetOpenFile(rb_convert_type(io, T_FILE, "IO", "to_io"), fptr);
  rb_io_set_nonblock(fptr);

  return INT2NUM(buffer_read_from(buf, FPTR_TO_FD(fptr)));
}

/**
 *  call-seq:
 *    EventMachine::Buffer#write_to(io) -> Integer
 * 
 * Perform a nonblocking write of the buffer to the given IO object.
 * As much data as possible is written until the call would block.
 * Any data which is written is removed from the buffer.
 */
static VALUE Em_Buffer_write_to(VALUE self, VALUE io) {
  struct buffer *buf;
#if HAVE_RB_IO_T 
  rb_io_t *fptr;
#else
  OpenFile *fptr;
#endif

  Data_Get_Struct(self, struct buffer, buf);
  GetOpenFile(rb_convert_type(io, T_FILE, "IO", "to_io"), fptr);
  rb_io_set_nonblock(fptr);

  return INT2NUM(buffer_write_to(buf, FPTR_TO_FD(fptr)));
}

/*
 * Ruby bindings end here.  Below is the actual implementation of 
 * the underlying data structures.
 */

/* Create a new buffer */
static struct buffer *buffer_new(void)
{
  struct buffer *buf;

  buf = (struct buffer *)xmalloc(sizeof(struct buffer));
  buf->head = buf->tail = buf->pool_head = buf->pool_tail = 0;
  buf->size = 0;
  buf->node_size = DEFAULT_NODE_SIZE;
	time(&buf->last_purged_at);
	
  return buf;
}

/* Clear all data from a buffer */
static void buffer_clear(struct buffer *buf)
{
  struct buffer_node *tmp;

  /* Move everything into the buffer pool */
  if(!buf->pool_tail)
    buf->pool_head = buf->pool_tail = buf->head;
  else
    buf->pool_tail->next = buf->head;

  buf->head = buf->tail = 0;
  buf->size = 0;
}

/* Free a buffer */
static void buffer_free(struct buffer *buf) 
{
  struct buffer_node *tmp;

  buffer_clear(buf);

  while(buf->pool_head) {
    tmp = buf->pool_head;
    buf->pool_head = tmp->next;
    free(tmp);
  }

  free(buf);
}

/* Run through the pool and find elements that haven't been used for awhile */
static void buffer_gc(struct buffer *buf)
{
  struct buffer_node *cur, *tmp;
  time_t now;
  time(&now);

	/* Only purge if we've passed the purge interval */
	if(now - buf->last_purged_at < PURGE_INTERVAL)
		return;
		
	buf->last_purged_at = now;

  while(buf->pool_head && now - buf->pool_head->last_used_at >= MAX_AGE) {
    tmp = buf->pool_head;
    buf->pool_head = buf->pool_head->next;
    free(tmp);
  }

  if(!buf->pool_head)
		buf->pool_tail = 0;
}

/* Create a new buffer_node (or pull one from the memory pool) */
static struct buffer_node *buffer_node_new(struct buffer *buf)
{
  struct buffer_node *node;

  /* Pull from the memory pool if available */
  if(buf->pool_head) {
    node = buf->pool_head;
    buf->pool_head = node->next;

    if(node->next)
      node->next = 0;
    else
      buf->pool_tail = 0;
  } else {
    node = (struct buffer_node *)xmalloc(sizeof(struct buffer_node) + buf->node_size);
    node->next = 0;
  }

  node->start = node->end = 0;
  return node;
}

/* Free a buffer node (i.e. return it to the memory pool) */
static void buffer_node_free(struct buffer *buf, struct buffer_node *node)
{
  /* Store when the node was freed */
  time(&node->last_used_at);

  node->next = buf->pool_head;
  buf->pool_head = node;

  if(!buf->pool_tail)
    buf->pool_tail = node;
}

/* Prepend data to the front of the buffer */
static void buffer_prepend(struct buffer *buf, char *str, unsigned len)
{
  struct buffer_node *node, *tmp;
  buf->size += len;

  /* If it fits in the beginning of the head */
  if(buf->head && buf->head->start >= len) {
    buf->head->start -= len;
    memcpy(buf->head->data + buf->head->start, str, len);
  } else {
    node = buffer_node_new(buf);
    node->next = buf->head;
    buf->head = node;
    if(!buf->tail) buf->tail = node;

    while(len > buf->node_size) {
      memcpy(node->data, str, buf->node_size);
      node->end = buf->node_size;

      tmp = buffer_node_new(buf);
      tmp->next = node->next;
      node->next = tmp;

      if(buf->tail == node) buf->tail = tmp;
      node = tmp;

      str += buf->node_size;
      len -= buf->node_size;
    }

    if(len > 0) {
      memcpy(node->data, str, len);
      node->end = len;
    }
  }
}

/* Append data to the front of the buffer */
static void buffer_append(struct buffer *buf, char *str, unsigned len)
{
  unsigned nbytes;
  buf->size += len;

  /* If it fits in the remaining space in the tail */
  if(buf->tail && len <= buf->node_size - buf->tail->end) {
    memcpy(buf->tail->data + buf->tail->end, str, len);
    buf->tail->end += len;
    return;
  }

  /* Empty list needs initialized */
  if(!buf->head) {
    buf->head = buffer_node_new(buf);
    buf->tail = buf->head;
  }

  /* Build links out of the data */
  while(len > 0) {
    nbytes = buf->node_size - buf->tail->end;
    if(len < nbytes) nbytes = len;
    
    memcpy(buf->tail->data + buf->tail->end, str, nbytes);
    str += nbytes;    
    len -= nbytes;
    
    buf->tail->end += nbytes;

    if(len > 0) {
      buf->tail->next = buffer_node_new(buf);
      buf->tail = buf->tail->next;
    }
  }
}

/* Read data from the buffer (and clear what we've read) */
static void buffer_read(struct buffer *buf, char *str, unsigned len)
{
  unsigned nbytes;
  struct buffer_node *tmp;

  while(buf->size > 0 && len > 0) {
    nbytes = buf->head->end - buf->head->start;
    if(len < nbytes) nbytes = len;

    memcpy(str, buf->head->data + buf->head->start, nbytes);
    str += nbytes;
    len -= nbytes;

    buf->head->start += nbytes;
    buf->size -= nbytes;

    if(buf->head->start == buf->head->end) {
      tmp = buf->head;
      buf->head = tmp->next;
      buffer_node_free(buf, tmp);

      if(!buf->head) buf->tail = 0;
    }
  }
}

/* Copy data from the buffer without clearing it */
static void buffer_copy(struct buffer *buf, char *str, unsigned len)
{
  unsigned nbytes;
  struct buffer_node *node;

	node = buf->head;
  while(node && len > 0) {
    nbytes = node->end - node->start;
    if(len < nbytes) nbytes = len;

    memcpy(str, node->data + node->start, nbytes);
    str += nbytes;
    len -= nbytes;

    if(node->start + nbytes == node->end)
			node = node->next;
  }
}

/* Write data from the buffer to a file descriptor */
static int buffer_write_to(struct buffer *buf, int fd)
{
  int bytes_written, total_bytes_written = 0;
  struct buffer_node *tmp;

  while(buf->head) {
    bytes_written = write(fd, buf->head->data + buf->head->start, buf->head->end - buf->head->start);

    /* If the write failed... */
    if(bytes_written < 0) {
      if(errno != EAGAIN)
        rb_sys_fail("write");

      return total_bytes_written;
    }

    total_bytes_written += bytes_written;
    buf->size -= bytes_written;

    /* If the write blocked... */
    if(bytes_written < buf->head->end - buf->head->start) {
      buf->head->start += bytes_written;
      return total_bytes_written;
    }

    /* Otherwise we wrote the whole buffer */
    tmp = buf->head;
    buf->head = tmp->next;
    buffer_node_free(buf, tmp);

    if(!buf->head) buf->tail = 0;
  }

  return total_bytes_written;
}

/* Read data from a file descriptor to a buffer */
/* Append data to the front of the buffer */
static int buffer_read_from(struct buffer *buf, int fd)
{
	int bytes_read, total_bytes_read = 0;
  unsigned nbytes;

  /* Empty list needs initialized */
  if(!buf->head) {
    buf->head = buffer_node_new(buf);
    buf->tail = buf->head;
  }

	do {
	  nbytes = buf->node_size - buf->tail->end;
		bytes_read = read(fd, buf->tail->data + buf->tail->end, nbytes);
	
		if(bytes_read < 1) {
			if(errno != EAGAIN)
        rb_sys_fail("read");
			
			return total_bytes_read;
		}
		
		total_bytes_read += bytes_read; 
		buf->tail->end += nbytes;
		buf->size += nbytes;
		
		if(buf->tail->end == buf->node_size) {
      buf->tail->next = buffer_node_new(buf);
      buf->tail = buf->tail->next;
		}
	} while(bytes_read == nbytes);
	
	return total_bytes_read;
}