polled streams proof of concept

README

wrt Smoothie Firmware:

Polled Streams proof of concept

Instead of each stream firing an event when a line is available, the kernel instead polls each stream, examines the line then chooses whether or not to accept it.

If the line is accepted, it is popped from the stream buffer, and that stream is moved to the end of the kernel's list ensuring that a fast stream cannot starve other streams.

Allowing the kernel to choose whether or not to accept a line means that we can still issue non-queued commands when the queue is full.

In this proof of concept, I deliberately create a condition where the serial buffer will overflow, by filling it more quickly than the queue is emptying.

Output:
Abc
Abc
[...]
Abc
Abc
Abc
Abc
Abc
Abc
Abc
Serial: rx buffer overflow!
Abc
Abc
Abc
Abc
Abc
Abc
Abc
Abc
Serial: rx buffer overflow!
[...]

action.cpp

#include "action.h"

#include "actiondata.h"

#include <cstddef>
#include <cstdio>

#include <cstdlib>

Action::Action()
{
}

void Action::add(ActionData* data)
{
	ActionData* d = first_data;

	while (d)
		d = d->next;

	d->next = data;

	data->next = NULL;
	data->action = this;
}

void Action::remove(ActionData* data)
{
	// remove the supplied data from LL
	if (first_data == data)
		first_data = first_data->next;
	else
	{
		ActionData* d = first_data;
		while (d)
		{
			if (d->next == data)
			{
				d->next = data->next;
				data->next = NULL;
				d = NULL;
			}
			else
				d = d->next;
		}
		if (!d)
			// strange, didn't find that data
			return;
	}

	ActionData* g = gc_data;
	while (g->next)
		g = g->next;
	g->next = data;
}

void Action::invoke()
{
	ActionData* d = first_data;
	ActionData* temp;

	while (d)
	{
		// advance pointer first because on_action_invoke can remove this data
		// and we don't want to inadvertently skip to gc chain
		temp = d;
		d = d->next;

		temp->owner->on_action_invoke(temp);
	}
}

void Action::clean()
{
	ActionData* d = first_data;
	ActionData* temp;

	if (first_data)
	{
		fprintf(stderr, "ERROR! Action %p cleaned while it still holds valid data!\n", this);
		exit(1);
	}

	while (d)
	{
		temp = d;
		d = d->next;

		delete temp;
	}
}

action.h

#ifndef _ACTION_H
#define _ACTION_H

class ActionData;

class ActionReceiver
{
public:
	virtual void on_action_invoke(ActionData*) = 0;
};

class Action
{
public:
	Action();

	void add(ActionData*);
	void remove(ActionData*);

	void invoke(void);

	void clean(void);
protected:
	ActionData* first_data;
	ActionData* gc_data;
};

#endif /* _ACTION_H */

actiondata.cpp

#include "actiondata.h"

#include "action.h"

ActionData::ActionData(ActionReceiver* o)
{
	owner = o;
}

void ActionData::invoke()
{
	if (owner)
		owner->on_action_invoke(this);
}

void ActionData::remove()
{
	if (action)
		action->remove(this);
}

actiondata.h

#ifndef _ACTIONDATA_H
#define _ACTIONDATA_H

class Action;
class ActionData;
class ActionReceiver;

class ActionData
{
	friend class Action;
public:
	ActionData(ActionReceiver*);
	virtual void invoke(void);
	virtual void remove(void);
protected:
	Action* action;
	ActionReceiver* owner;

	ActionData* next;
};

#endif /* _ACTIONDATA_H */

arbiter.cpp

#include "arbiter.h"

#include <cstddef>

#include "kernel.h"
#include "streamio.h"

Arbiter::Arbiter()
{
	first_stream = NULL;
}

void Arbiter::on_module_added(Kernel* k)
{
	this->Module::on_module_added(k);
	k->arbiter = this;
}

void Arbiter::receive_event(Event event, void* data)
{
	switch(event)
	{
		case EVENT_IDLE:
			{
			// TODO: if queue is empty, poll streams
				if (((qhead + 1) % QUEUE_SIZE) != qtail)
				{
					poll_streams();
				}
			}
			break;
		case EVENT_QUEUE_PUSH:
			break;
		case EVENT_QUEUE_POP:
			{
				// unset queue block flags
				StreamIO* s = first_stream;
				while (s)
				{
					s->flag_queue = 0;
					s = s->next_stream;
				}
			}
			break;
		default:
			break;
	}
}

int Arbiter::add_stream(StreamIO* streamio)
{
	StreamIO* s = first_stream;

	if (s == NULL)
	{
		first_stream = s = streamio;
	}
	else
	{
		while (s->next_stream)
			s = s->next_stream;
		s->next_stream = streamio;
	}
	streamio->next_stream = NULL;

	return 0;
}

int Arbiter::poll_streams()
{
	StreamIO* s = first_stream;
	StreamIO* prev = s;
	StreamIO* last;

	while (s) {
		if ((s->block_flags == 0) && (s->has_line()))
		{
			char linebuf[512];

			s->peek_line(linebuf, 512);
			fprintf(stderr, "%s", linebuf);
			s->accept_line();

			// remove stream from list
			if (prev == first_stream)
			{
				first_stream = prev->next_stream;
				last = first_stream;
			}
			else
			{
				prev->next_stream = s->next_stream;
				last = prev;
			}

			// place stream at end of list
			if (last == NULL)
				first_stream = s;
			else
			{
				while(last->next_stream)
					last = last->next_stream;
				last->next_stream = s;
			}

			return 1;
		}
		prev = s;
		s = s->next_stream;
	}

	return 0;
}

arbiter.h

#ifndef _ARBITER_H
#define _ARBITER_H

#include "module.h"
#include "action.h"

#define QUEUE_SIZE 32

class StreamIO;

class Arbiter : public Module
{
public:
	Arbiter();

	void on_module_added(Kernel* k);
	void receive_event(Event event, void* data);

	int   add_stream(StreamIO*);

	int   poll_streams();
protected:
	StreamIO* first_stream;

	Action Queue[QUEUE_SIZE];
	int qhead;
	int qtail;
private:

};

#endif /* _ARBITER_H */

event.h

#ifndef _EVENT_H
#define _EVENT_H

typedef enum {
	EVENT_TEST,
	EVENT_IDLE,
	EVENT_QUEUE_PUSH,
	EVENT_QUEUE_POP,

	N_EVENTS
} Event;

#endif /* _EVENT_H */

fat.cpp

#include "fat.h"

#include <cstdlib>
#include <cstdio>
#include <cstddef>
#include <cstring>

#include <unistd.h>

#include "sd.h"

void dump_buffer(uint8_t* buffer)
{
	for (int i = 0; i < 32; i++)
	{
		fprintf(stderr, "0x%04x| ", i * 16);

		for (int j = 0; j < 16; j++)
			fprintf(stderr, "%02X ", buffer[(i * 16) + j]);
		fprintf(stderr, " | ");
		for (int j = 0; j < 16; j++)
			if (buffer[(i * 16) + j] >= 32 && buffer[(i * 16) + j] < 127)
				fprintf(stderr, "%c", buffer[(i * 16) + j]);
			else
				fprintf(stderr, ".");

			fprintf(stderr, "\n");
	}
}

const char* action_name(_fat_ioaction i)
{
	switch(i)
	{
		case IOACTION_NULL:
			return str(IOACTION_NULL);
		case IOACTION_MOUNT:
			return str(IOACTION_MOUNT);
		case IOACTION_OPEN:
			return str(IOACTION_OPEN);
		case IOACTION_READ_ONE:
			return str(IOACTION_READ_ONE);
		default:
			return "?";
	}
}

Fat::Fat()
{
	sd = NULL;
	fat_buf = dentry_buf = NULL;
	work_queue = NULL;

	fat_lba = dentry_lba = 0xFFFFFFFF;

	fat_begin_lba       = 0;
	cluster_begin_lba   = 0;
	sectors_per_cluster = 0;
	root_dir_cluster    = 0;
}

void Fat::f_mount(_fat_mount_ioresult* w, Sd& sd)
{
	if (fat_buf)
		free(fat_buf);
	if (dentry_buf && dentry_buf != fat_buf)
		free(dentry_buf);

	this->sd = &sd;

	fat_buf = (uint8_t*) malloc(512);
	fat_lba = -1;

// 	dentry_buf = (uint8_t*) malloc(512);
	dentry_buf = fat_buf;
	dentry_lba = -1;

	w->action   = IOACTION_MOUNT;
	w->lba      = 0;
	w->buffer   = dentry_buf;
	w->buflen   = 512;
	w->owner    = NULL;
	w->ready    = 1;
	w->fini     = 0;
	w->next     = NULL;

	if (work_queue)
	{
		fprintf(stderr, "Error! Already mounted and I/O in progress! umount first!\n");
// 		f_umount();
		exit(1);
	}

	work_queue = w;

	this->sd->begin_read(w->lba, w->buffer, w->buflen, this);
}

int Fat::f_mounted()
{
	if (fat_begin_lba == 0)
		return 0;
	if (cluster_begin_lba == 0)
		return 0;
	if (sectors_per_cluster == 0)
		return 0;
	if (root_dir_cluster == 0)
		return 0;
	return 1;
}

uint32_t Fat::cluster_to_lba(uint32_t cluster)
{
	return cluster_begin_lba + (cluster - 2) * sectors_per_cluster;
}

uint32_t Fat::lba_to_cluster(uint32_t lba)
{
	return (lba - cluster_begin_lba) / sectors_per_cluster + 2;
}

void Fat::enqueue(_fat_ioresult* ior)
{
	queue_walk();

	// TODO: atomic
	_fat_ioresult* w = work_queue;
	ior->next = NULL;
	ior->fini = 0;
	if (w)
	{
		while (w->next)
		{
			if (w == ior)
				return;
			w = w->next;
		}
		if (w == ior)
			return;
		w->next = ior;
	}
	else
	{
		work_queue = w = ior;

		if ((w->lba == dentry_lba) && (w->buffer == dentry_buf))
			process_buffer(w->buffer, w->lba);
		else if ((w->lba == fat_lba) && (w->buffer == fat_buf))
			process_buffer(w->buffer, w->lba);
		else
			sd->begin_read(w->lba, w->buffer, w->buflen, this);

		// else do nothing
	}

	queue_walk();
}

void Fat::dequeue(_fat_ioresult* w)
{
	if (work_queue == w)
		work_queue = w->next;
	else
	{
		_fat_ioresult* j = work_queue;
		while (j->next && j->next != w)
			j = j->next;
		if (j)
			j->next = w->next;
	}

	w->fini = 1;
}

int  Fat::f_open( _fat_file_ioresult* ior, const char* path)
{
	// skip leading slash
	while (path[0] == '/')
		path++;

	int l = strlen(path);
	ior->file.path = (char*) malloc(l + 1);
	strncpy(ior->file.path, path, l);

	ior->action    = IOACTION_OPEN;
	ior->ready     = 1;
	if (ior->buffer == NULL)
	{
		ior->buffer    = dentry_buf;
		ior->buflen    = 512;
	}
	ior->lba       = cluster_to_lba(root_dir_cluster);

	ior->file.root_cluster     = 0;
	ior->file.direntry_cluster = 0;
	ior->file.direntry_index   = 0;
	ior->file.current_cluster  = root_dir_cluster;
	ior->file.byte_in_cluster  = 0;
	ior->file.pathname_traversed_bytes = 0;

	ior->next = NULL;

	fprintf(stderr, "FAT: Open %s\n", path);
	enqueue(ior);

	return 0;
}

int  Fat::f_read( _fat_file_ioresult* ior, void* buffer, uint32_t buflen)
{
	fprintf(stderr, "FAT: READ %s (%p)!\n", ior->file.path, ior);

	ior->action = IOACTION_READ_ONE;

	ior->buffer = (uint8_t*) buffer;
	ior->buflen = buflen;

	if (ior->file.byte_in_cluster >= bytes_per_sector * sectors_per_cluster)
	{
		if (fat_cache(fat_begin_lba + (ior->file.current_cluster >> 7)))
		{
			ior->file.current_cluster = ((uint32_t*) fat_buf)[ior->file.current_cluster & 0x7F];
			ior->file.cluster_index++;
			ior->file.byte_in_cluster -= (bytes_per_sector * sectors_per_cluster);
		}
	}

	ior->bytes_remaining = ior->buflen;

	ior->lba    = cluster_to_lba(ior->file.current_cluster) + (ior->file.byte_in_cluster >> 9);

	enqueue(ior);

	return 0;
}

int  Fat::f_write(_fat_file_ioresult* ior, void* buffer, uint32_t buflen)
{
	fprintf(stderr, "f_write: unimplementeed\n");
	return 0;
}

int  Fat::f_close(_fat_file_ioresult* ior)
{
	return 0;
}

void Fat::_sd_callback(_sd_work_stack* w)
{
	fprintf(stderr, "FAT: block %d read ok\n", w->lba);

	process_buffer((uint8_t*) w->buffer, w->lba);

	fprintf(stderr, "FAT: end process lba %u\n", w->lba);
}

void Fat::process_buffer(uint8_t* buffer, uint32_t lba)
{
	fprintf(stderr, "FAT: --PROCBUF-- (%p lba %u)\n", buffer, lba);

	if (work_queue == NULL)
		return;

	_fat_ioresult* w = work_queue;

	if (buffer == fat_buf)
		fat_lba = lba;
	if (buffer == dentry_buf)
		dentry_lba = lba;

	fprintf(stderr, "FAT: action is %u (%s)\n", w->action, action_name((_fat_ioaction) w->action));

	switch(w->action)
	{
		case IOACTION_MOUNT:
			ioaction_mount((_fat_mount_ioresult*) w, buffer, lba);
			break;
		case IOACTION_OPEN:
			ioaction_open((_fat_file_ioresult*) w, buffer, lba);
			break;
		case IOACTION_READ_ONE:
			ioaction_read_one((_fat_file_ioresult*) w, buffer, lba);
			break;
	}
}

void Fat::ioaction_mount(_fat_mount_ioresult* w, uint8_t* buffer, uint32_t lba)
{
	if (lba == cluster_to_lba(root_dir_cluster))
	{
		_fat_direntry* d = (_fat_direntry*) buffer;
		// TODO: don't assume that volume label is in root cluster
		for (int i = 0; i < 16; i++)
		{
			if (d[i].attr == 0x08)
			{
				memcpy(w->label, d[i].name, 11);
				w->label[11] = 0;
				w->fini = 1;

				fprintf(stderr, "FAT: mount succeeded! label is %s\n", w->label);

				dequeue(w);

				return;
			}
		}
	}

	_fat_bootblock* bootblock = (_fat_bootblock*) buffer;

	if (bootblock->magic != 0xAA55)
	{
		fprintf(stderr, "bad magic at LBA %u, corrupt disk?\n", lba);
		return;
	}

	fprintf(stderr, "FAT: magic ok, trying partition table\n");
	for (int i = 0; i < 4; i++)
	{
		fprintf(stderr, "FAT: Partition table %u:\n\ttype: %02X\n\tlba_begin: %u\n\tn_sectors: %u\n\tend: %u\n\tdisk blocks: %u\n",
			i,
			bootblock->partition[i].type,
			bootblock->partition[i].lba_begin,
			bootblock->partition[i].n_sectors,
			bootblock->partition[i].lba_begin + bootblock->partition[i].n_sectors,
			sd->disk_blocks()
		);
		if (
			(
				bootblock->partition[i].type == 0x01 ||
				bootblock->partition[i].type == 0x04 ||
				bootblock->partition[i].type == 0x0B ||
				bootblock->partition[i].type == 0x0C ||
				bootblock->partition[i].type == 0x0E ||
				bootblock->partition[i].type == 0x0F
			) &&
			bootblock->partition[i].lba_begin < sd->disk_blocks() &&
			bootblock->partition[i].n_sectors + bootblock->partition[i].lba_begin <= sd->disk_blocks()
		)
		{
			fprintf(stderr, "FAT: Found a partition!\n");

			if (dentry_cache(bootblock->partition[i].lba_begin) == 0) return;
		}
	}

	fprintf(stderr, "FAT: didn't look like a partition table, trying FAT superblock\n");

	_fat32_volid*   volid     = (_fat32_volid  *) buffer;
	fprintf(stderr, "FAT: superblock:\n\tid: %c%c%c%c%c%c%c%c\n\tbytes_per_sector: %u\n\tn_fats: %u\n\tsectors_per_cluster: %u\n\tn_reserved_sectors: %u\n\tsectors_per_fat: %u\n\troot_dir_cluster: %u\n\ttotal_sectors: %u (%uMB)\n",
			volid->oem_id[0],volid->oem_id[1],volid->oem_id[2],volid->oem_id[3],volid->oem_id[4],volid->oem_id[5],volid->oem_id[6],volid->oem_id[7],
		volid->bytes_per_sector,
		volid->n_fats,
		volid->sectors_per_cluster,
		volid->n_reserved_sectors,
		volid->sectors_per_fat,
		volid->root_dir_cluster,
		volid->total_sectors,
		volid->total_sectors / 2048
	);

	if (volid->bytes_per_sector == 512 &&
		volid->n_fats == 2             &&
		(	volid->sectors_per_cluster == 1   ||
		volid->sectors_per_cluster == 2   ||
		volid->sectors_per_cluster == 4   ||
		volid->sectors_per_cluster == 8   ||
		volid->sectors_per_cluster == 16  ||
		volid->sectors_per_cluster == 32  ||
		volid->sectors_per_cluster == 64  ||
		volid->sectors_per_cluster == 128
		)
	)
	{
		fprintf(stderr, "FAT: Found a VolID!\n");
		// looks like a volid

		fat_begin_lba       = lba + volid->n_reserved_sectors;
		cluster_begin_lba   = lba + volid->n_reserved_sectors + (volid->n_fats * volid->sectors_per_fat);
		sectors_per_cluster = volid->sectors_per_cluster;
		root_dir_cluster    = volid->root_dir_cluster;
		bytes_per_sector    = volid->bytes_per_sector;

		w->lba = cluster_to_lba(root_dir_cluster);

		if (dentry_cache(cluster_to_lba(root_dir_cluster)) == 0) return;
	}

	fprintf(stderr, "did not recognise disk image: looks like neither FAT volid or partition table.\n");
	dequeue(w);
	return;
}

void Fat::ioaction_open(_fat_file_ioresult* w, uint8_t* buffer, uint32_t lba)
{
	// in f_open, we pre-request the root dir
	// so now we're free to scan each direntry and traverse as necessary

	// scan for file of interest

	char* fn = w->file.path + w->file.pathname_traversed_bytes;
	char matchname[14];
	int i, j;
	uint8_t dir = 0;
	for (i = 0, j = 0; i < 11; i++, j++)
	{
		if (fn[i] == '/')
		{
			dir = i;
			break;
		}
		if (fn[i] == '.')
		{
			for (;i < 8; i++)
				matchname[i] = 32;
			i--;
		}
		else
			matchname[i] = fn[j];
	}
	matchname[i] = 0;

	fprintf(stderr, "FAT: Matchname is '%s'\n", matchname);

	dump_buffer(buffer);

	_fat_direntry* d = (_fat_direntry*) buffer;
	_fat_lfnentry* l = (_fat_lfnentry*) buffer;
	int lfn_match = 0;
	for (i = 0; i < 16 && d[i].name[0] != 0; i++)
	{
		if (dir == 0 && (d[i].attr & 0x1F) == 0 && d[i].name[0] != 0xE5)
		{
			// FILE entry
			fprintf(stderr, "checking '%s' vs '%s'(%d): %d\n", matchname, d[i].name, d[i].attr, strncasecmp(matchname, (char*) d[i].name, 11));
			if (strncasecmp(matchname, (char*) d[i].name, 11) == 0 || lfn_match)
			{
				// found it!
				fprintf(stderr, "Found! First cluster: %u, size: %ub\n", (d[i].ch << 16) | d[i].cl, d[i].size);

				w->action                = IOACTION_READ_ONE;

				w->file.direntry_cluster = lba_to_cluster(lba);
				w->file.direntry_index   = i;

				w->file.root_cluster     = (d[i].ch << 16) | d[i].cl;

				w->file.current_cluster  = w->file.root_cluster;
				w->file.byte_in_cluster  = 0;
				w->file.cluster_index    = 0;

				w->file.size             = d[i].size;

				w->lba                   = cluster_to_lba(w->file.root_cluster);

				dequeue(w);

				break;
			}
		}
		// 				else if (d[i].attr & 0x8)
		// 				{
		// VOLUME LABEL
		// 				}
		else if (dir == 0 && d[i].attr == 0xF)
		{
			// LFN entry
			if (l[i].final || lfn_match)
			{
// 				char lfn[14];
// 				lfn[0]  = l[i].name0[0];
// 				lfn[1]  = l[i].name0[1];
// 				lfn[2]  = l[i].name0[2];
// 				lfn[3]  = l[i].name0[3];
// 				lfn[4]  = l[i].name0[4];
// 				lfn[5]  = l[i].name1[0];
// 				lfn[6]  = l[i].name1[1];
// 				lfn[7]  = l[i].name1[2];
// 				lfn[8]  = l[i].name1[3];
// 				lfn[9]  = l[i].name1[4];
// 				lfn[10] = l[i].name1[5];
// 				lfn[11] = l[i].name2[0];
// 				lfn[12] = l[i].name2[1];
// 				lfn[13] = 0;
//
// 				int off = (l[i].sequence - 1) * 13;
			}
		}
		else if (dir == 1 && d[i].attr & 0x10)
		{
			// FOLDER entry
			if (strncasecmp(matchname, (char*) d[i].name, 11) == 0 || lfn_match)
			{
				w->file.pathname_traversed_bytes += dir + 1;

				w->file.direntry_cluster = (d[i].ch << 16) | d[i].cl;

				w->lba = cluster_to_lba(w->file.direntry_cluster);

				if (dentry_cache(cluster_to_lba(w->file.direntry_cluster)) == 0) return;
			}
		}
	}
}

void Fat::ioaction_read_one(_fat_file_ioresult* w, uint8_t* buffer, uint32_t lba)
{
	if (w->file.byte_in_cluster >= bytes_per_sector * sectors_per_cluster)
	{
		if (fat_cache(fat_begin_lba + (w->file.current_cluster >> 7)))
		{
			w->file.current_cluster = ((uint32_t*) fat_buf)[w->file.current_cluster & 0x7F];
			w->file.cluster_index++;
			w->file.byte_in_cluster -= (bytes_per_sector * sectors_per_cluster);
		}
		else
			return;
	}

	uint32_t l = cluster_to_lba(w->file.current_cluster) + (w->file.byte_in_cluster >> 9);

	if (l == lba)
	{
		w->file.byte_in_cluster += 512;
		w->bytes_remaining -= 512;
		if (w->bytes_remaining == 0)
			dequeue(w);
			if (w->owner)
				w->owner->_fat_io(w);
	}
}

void Fat::queue_walk()
{
	fprintf(stderr, "FAT: Queue walk\n");
	_fat_ioresult* j = work_queue;
	while (j)
	{
		fprintf(stderr, "FAT: Queue item %p:\n\taction : %d (%s)\n\tbuffer : %p\n\tbuflen : %u\n\tcluster: %u\n\tlba    : %u\n\towner  : %p\n\tnext   : %p\n", j, j->action, action_name((_fat_ioaction) j->action), j->buffer, j->buflen, lba_to_cluster(j->lba), j->lba, j->owner, j->next);
		j = j->next;
	}
	fprintf(stderr, "FAT: end queue\n");
}

int Fat::fat_cache(uint32_t lba)
{
	fprintf(stderr, "Fat cache: %s on %u\n", (fat_lba == lba)?"hit":"miss", lba);

	if (fat_lba == lba)
		return 1;

	sd->begin_read(lba, fat_buf, 512, this);

	return 0;
}

int Fat::dentry_cache(uint32_t lba)
{
	fprintf(stderr, "Dentry cache: %s on %u\n", (dentry_lba == lba)?"hit":"miss", lba);

	if (dentry_lba == lba)
		return 1;

	sd->begin_read(lba, dentry_buf, 512, this);

	return 0;
}

fat.h

#ifndef _FAT_H
#define _FAT_H

#include "sd.h"

#include "fat_struct.h"

/*
 * Asynchronous FAT Filesystem
 * for DMA driven microcontroller applications
 * Copyright (c) Michael Moon 2013
 *
 * Released under Gnu GPL
 */

/*
 * Operation of Asynchronous FAT:
 *
 * All calls return immediately. We NEVER busyloop, waiting for data
 *
 * Whenever new data is required, we request a DMA transfer and return
 *
 * We maintain a queue of tasks. External tasks (eg open, read, readdir) are
 *     placed at the bottom of the queue.
 *     Internal tasks (eg traverse FAT, read directory, etc) are placed at the
 *     head of the queue.
 *
 * When a new block of data arrives, we examine the first item in the queue.
 * Usually, it can do something useful with the newly arrived data.
 *
 * When a queue item has finished, it is removed from the queue, and the next
 *     item is activated.
 *
 * This, in combination with placing internal tasks on the top of the queue,
 *     means that the following sequence might occur:
 *
 * external -> f_open(&FIL, "/path/to/file.txt")
 *
 * [ add file_open action to TAIL of stack ]
 * cache_test(root_dir_cluster) -> fail
 * [ PUSH directory_traverse action to HEAD of stack ]
 * request dir cluster from disk, return
 *
 * disk->read_complete()
 * [ POP request from head of stack -> directory_traverse action ]
 * cache_test(root_dir_cluster) -> success
 * find "path/" in root_dir_cluster.
 * cache_test(cluster for "path/") -> fail
 * [ PUSH directory_traverse action to HEAD of stack ]
 * request cluster from disk, return
 *
 * disk->read_complete()
 * [ POP request from head of stack -> directory_traverse action ]
 * cache_test(cluster for "path/") -> success
 * find "to/" in cluster for "path/"
 * cache_test(cluster for "to/") -> fail
 * [ PUSH directory_traverse action to HEAD of stack ]
 * request from disk, return
 *
 * disk->read_complete()
 * [ POP request from head of stack -> directory_traverse action ]
 * cache_test(cluster for "to/") -> success
 * find "file.txt" in cluster for "to/" -> fail
 * not in first 16 entries, cache_test(2nd cluster for "to/") ->fail
 * [ PUSH directory_traverse action to HEAD of stack ]
 * request from disk, return
 *
 * disk->read_complete()
 * [ POP request from head of stack -> directory_traverse action ]
 * cache_test(2nd cluster for "to/") -> success
 * find "file.txt" in cluster -> success
 * [ POP request from head of stack -> file_open action ]
 * fill the FIL structure stored in the file_open action with data from the
 *     dir cluster
 *
 * fire callback stored in file_open action
 *
 * stack empty!
 *
 * external -> f_read(&FIL, buffer, size)
 * cache_test(first cluster, from &FIL) -> fail
 * [ PUSH file_read to TAIL of stack ]
 * request from disk, return
 *
 * disk->read_complete()
 * [ POP request from stack -> file_read ]
 * cache_test(first cluster from &FIL) -> success
 *
 * fire read_complete callback
 *
 * stack empty!
 *
 * external -> f_read(&FIL, buffer, size)
 * [ PUSH file_read to TAIL of stack ]
 * find next cluster
 * cache_test(FAT cluster map) -> fail
 * [ PUSH cluster_traverse to HEAD of stack ]
 * request cluster map from disk, return
 *
 * disk->read_complete()
 * [ POP stack -> cluster_traverse ]
 * find relevant cluster
 * cache_test(FAT cluster map) -> success
 * update file_read action
 * [ POP stack -> file_read ]
 * cache_test(2nd cluster) -> fail
 * [ PUSH file_read action back to HEAD of stack (not finished) ]
 * request cluster from disk, return
 *
 * disk->read_complete()
 * [ POP stack -> file_read ]
 * cache_test(file data) -> success!
 *
 * trigger file_read callback
 *
 * stack empty!
 *
 */

class _fat_ioreceiver
{
public:
	virtual void _fat_io(_fat_ioresult*) = 0;
};

class Fat : public _sd_event_receiver
{
public:
	Fat();

	/*
	 * should be a fairly familiar set of functions
	 *
	 * however, note that they all return *immediately*, having queued the
	 *     requested action to happen at a later time
	 *
	 * applications should either poll the 'fini' flag (discouraged)
	 * or inherit _sd_event_receiver and register as owner (preferred)
	 *
	 */
	void f_mount(_fat_mount_ioresult*, Sd&);
	int  f_open( _fat_file_ioresult*, const char*);
	int  f_seek( _fat_file_ioresult*, uint32_t);
	int  f_read( _fat_file_ioresult*, void*, uint32_t);
	int  f_write(_fat_file_ioresult*, void*, uint32_t);
	int  f_close(_fat_file_ioresult*);

	int  f_mounted(void);

	/*
	 * this method receives completion messages from the disk
	 */
	void _sd_callback(_sd_work_stack*);

	/*
	 * this is where we crunch received (or cached) data
	 */
	void process_buffer(uint8_t* buffer, uint32_t lba);

	void ioaction_mount(    _fat_mount_ioresult* w, uint8_t* buffer, uint32_t lba);
	void ioaction_open(     _fat_file_ioresult*  w, uint8_t* buffer, uint32_t lba);
	void ioaction_read_one( _fat_file_ioresult*  w, uint8_t* buffer, uint32_t lba);
	void ioaction_write_one(_fat_file_ioresult*  w, uint8_t* buffer, uint32_t lba);

	/*
	 * debug function, prints queue contents
	 */
	void queue_walk(void);
protected:
	/*
	 * these four variables uniquely define a FAT filesystem at some position on a disk
	 */
	uint32_t fat_begin_lba;
	uint32_t cluster_begin_lba;
	uint32_t sectors_per_cluster;
	uint32_t root_dir_cluster;

	uint32_t bytes_per_sector; // I don't feel comfortable assuming that this is always 512

	/*
	 * conversion between cluster and lba
	 */
	uint32_t cluster_to_lba(uint32_t cluster);
	uint32_t lba_to_cluster(uint32_t lba);

	/*
	 * queueing functions
	 */
	void     enqueue(_fat_ioresult*);
	void     dequeue(_fat_ioresult*);
	void     byte2cluster(_fat_ioresult*, uint32_t);

	int      fat_cache(   uint32_t lba);
	int      dentry_cache(uint32_t lba);

private:
	Sd* sd;
	/*
	 * these buffers are used internally for traversing cluster chains and directory entries
	 */
	uint8_t* fat_buf;
	uint32_t fat_lba;
	uint8_t* dentry_buf;
	uint32_t dentry_lba;

	/*
	 * this is the head of the queue, which is a linked list
	 */
	_fat_ioresult* work_queue;
};

#endif /* _FAT_H */

fat_struct.h

#ifndef _FAT_STRUCT_H
#define _FAT_STRUCT_H

#define _str(x) #x
#define str(x) _str(x)

/*
 * data structures on disk
 */

// partition headers in MBR
typedef struct __attribute__ ((packed))
{
	uint8_t  boot_flag;
	uint8_t  chs_begin[3];
	uint8_t  type;
	uint8_t  chs_end[3];
	uint32_t lba_begin;
	uint32_t n_sectors;
} _fat_partition;

// MBR itself
typedef struct __attribute__ ((packed))
{
	uint8_t mbr[446];
	_fat_partition partition[4];
	uint16_t magic;
} _fat_bootblock;

/*
 * FAT12 superblock
 */
typedef struct __attribute__ ((packed))
{
	uint8_t  jump[3];
	char     oem_id[8];
	uint16_t bytes_per_sector;		// usually 512
	uint8_t  sectors_per_cluster;
	uint16_t num_boot_sectors;		// usually 1
	uint8_t  num_fats;				// usually 2
	uint16_t num_root_dir_ents;
	uint16_t total_sectors;			// 0 if num_sectors > 65535
	uint8_t  media_id;				// usually 0xF0
	uint16_t sectors_per_fat;
	uint16_t sectors_per_track;
	uint16_t heads;
	uint32_t hidden_sectors;		// root dir entry? usually 2
	uint32_t total_sectors_32;		// 0 if num_sectors < 65536
	uint8_t  mbr[474];
	uint16_t magic;					// always 0xAA55
} _fat12_volid;

/*
 * FAT32 superblock
 * first sector of filesystem
 */
typedef struct __attribute__ ((packed))
{
	uint8_t  irrelevant0[3];
	uint8_t  oem_id[8];				// eg "MSWIN4.0"
	uint16_t bytes_per_sector;		// usually 512
	uint8_t  sectors_per_cluster;	// 1,2,4,8,...,128
	uint16_t n_reserved_sectors;	// usually 32
	uint8_t  n_fats;				// always 2
	uint8_t  irrelevant1[19];
	uint32_t sectors_per_fat;		// depends on disk size
	uint8_t  irrelevant2[4];
	uint32_t root_dir_cluster;		// usually 2
	uint32_t total_sectors;			// disk size / bytes_per_sector
	uint8_t  irrelevant3[458];
	uint16_t magic;
} _fat32_volid;

/*
 * Directory entries
 *
 * 32 bytes long
 */
typedef struct __attribute__ ((packed))
{
	uint8_t  name[11];
	uint8_t  attr;
	uint8_t  irrelevant0[8];
	uint16_t ch;
	uint8_t  irrelevant1[4];
	uint16_t cl;
	uint32_t size;
} _fat_direntry;

typedef struct __attribute__ ((packed))
{
	union {
		uint8_t flags;
		struct {
			uint8_t  sequence    :4;
			uint8_t  irrelevant0 :1;
			uint8_t  final       :1;
			uint8_t  irrelevant1 :2;
		};
	};

	uint16_t name0[5];

	uint8_t  attr; // always 0xF
	uint8_t  type; // always 0
	uint8_t  checksum;
	uint16_t name1[6];
	uint16_t cs;   // always 0
	uint16_t name2[2];
} _fat_lfnentry;

typedef struct __attribute__ ((packed))
{
	char* path;

	uint32_t root_cluster;

	uint32_t direntry_cluster;
	uint8_t  direntry_index;

	/*
	 * where are we now?
	 */
	uint32_t current_cluster;
	uint32_t byte_in_cluster;
	uint32_t cluster_index;

	// overall position = cluster_index * sectors_per_cluster * bytes_per_sector + byte_in_cluster
	// when (byte_in_cluster >= sectors_per_cluster * bytes_per_sector + byte_in_cluster)
	// it's time to fetch next cluster

	union {
		uint32_t size;
		uint32_t pathname_traversed_bytes;
	};
} FIL;

typedef enum
{
	IOACTION_NULL,
	IOACTION_MOUNT,
	IOACTION_OPEN,
	IOACTION_READ_ONE,
	IOACTION_WRITE_ONE,
	IOACTION_SEEK,
} _fat_ioaction;

class Fat;
class _fat_ioreceiver;
struct __fat_ioresult;

struct __attribute__ ((packed))
_fat_ioresult
{
	uint32_t lba;
	uint8_t  action:6;
	uint8_t  ready :1;
	uint8_t  fini  :1;

	uint8_t* buffer;
	uint32_t buflen;

	_fat_ioreceiver* owner;

	_fat_ioresult* next;
};

struct _fat_file_ioresult;
struct __attribute__ ((packed))
_fat_traverse_ioresult : _fat_ioresult
{
	uint32_t size;
	uint32_t cluster;

	_fat_file_ioresult* child;
};

struct __attribute__ ((packed))
_fat_file_ioresult : _fat_ioresult
{
	char*    path;
	uint8_t  path_traverse;

	uint32_t bytes_remaining;

	FIL      file;

	_fat_traverse_ioresult traverse;
};

struct __attribute__ ((packed))
_fat_mount_ioresult : _fat_ioresult
{
	char label[12];
	uint32_t lba_start;
	_fat_ioreceiver* owner;
};


#endif /* _FAT_STRUCT_H */

kernel.cpp

#include "kernel.h"

#include <cstddef>
#include <cstdlib>

#include <strings.h>

#include "module.h"
#include "event.h"
#include "streamio.h"

Kernel::Kernel()
{
	first_module = NULL;
	for (int i = 0; i < N_EVENTS; i++)
	{
		event_registry_size[i] = 0;
		event_registry[i] = NULL;
	}
}

int   Kernel::add_module(Module* module)
{
	Module* m = first_module;

	if (m == NULL)
	{
		first_module = m = module;
	}
	else
	{
		while (m->next_module)
			m = m->next_module;
		m->next_module = module;
	}
	module->next_module = NULL;

	module->on_module_added(this);

	return 0;
}

int   Kernel::register_for_event(Module* registree, Event event)
{
	if (event >= N_EVENTS)
		return -1;

	if (event_registry_size[event] == 0 || event_registry[event][event_registry_size[event] - 1] != NULL)
	{
		event_registry_size[event] += 4;
		event_registry[event] = (Module**) realloc(event_registry[event], event_registry_size[event] * sizeof(Module*));
		if (event_registry[event] == NULL)
		{
			// TODO: flag error
			return -1;
		}
	}

	for (int j = 0; j < event_registry_size[event]; j++)
	{
		if (event_registry[event][j] == NULL)
		{
			event_registry[event][j] = registree;
			return j;
		}
	}
	return -1;
}

void Kernel::fire_event(Event event, void* data)
{
	if (event_registry[event] == NULL)
		return;

	for (int i = 0; event_registry[event][i] != NULL; i++)
	{
		if (event_registry[event][i] != NULL)
			event_registry[event][i]->receive_event(event, data);
	}
}

kernel.h

#ifndef _KERNEL_H
#define _KERNEL_H

#include "event.h"

class Arbiter;
class Module;
class StreamIO;

#define N_REGISTRABLES 32;

class Kernel
{
public:
	Kernel();
	~Kernel();

	int   add_module(Module*);
// 	int   add_stream(StreamIO*);

	int   register_for_event(Module* registree, Event event);
	void  fire_event(Event event, void* data);

	int   poll_streams(void);

	Arbiter* arbiter;

private:
	Module* first_module;

	int event_registry_size[N_EVENTS];
	Module** event_registry[N_EVENTS];
};

#endif /* _KERNEL_H */

main.cpp

#include <cstdio>
#include <cstdlib>

#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>

#include <unistd.h>

#include "kernel.h"
#include "serial.h"
#include "arbiter.h"
#include "sd.h"
#include "fat.h"

Sd sd;
Fat fat;

class readump : public _fat_ioreceiver
{
public:
	uint8_t buffer[512];
	void _fat_io(_fat_ioresult* w) {
		fprintf(stderr, "main: buffer: %p/%p\n", w->buffer, buffer);
		for (int i = 0; i < 32; i++)
		{
			fprintf(stderr, "0x%04x| ", i * 16);

			for (int j = 0; j < 16; j++)
				fprintf(stderr, "%02X ", buffer[(i * 16) + j]);
			fprintf(stderr, " | ");
			for (int j = 0; j < 16; j++)
				if (buffer[(i * 16) + j] >= 32 && buffer[(i * 16) + j] < 127)
					fprintf(stderr, "%c", buffer[(i * 16) + j]);
				else
					fprintf(stderr, ".");

			fprintf(stderr, "\n");
		}

		fprintf(stderr, "w byte_in_cluster: %u\n", ((_fat_file_ioresult*)w)->file.byte_in_cluster);

		fat.f_read((_fat_file_ioresult*) w, buffer, 512);
	};
};

void sleep_u(uint32_t micros)
{
	struct timeval t;
	t.tv_sec = micros / 1000000;
	t.tv_usec = micros - (t.tv_sec * 1000000);
	select(0, NULL, NULL, NULL, &t);
}

int main()
{
	Kernel* k = new Kernel();
	Arbiter* a = new Arbiter();

	k->add_module(a);

	Serial* s = new Serial();

	k->add_module(s);

	_fat_mount_ioresult m;
	fat.f_mount(&m, sd);

	while(!m.fini);

	readump r;

	do {
		_fat_file_ioresult ior;

		ior.ready  = 1;
		ior.owner  = &r;

		fat.f_open(&ior, "fan.gco");

		while (ior.fini == 0)
			sleep_u(25000);

		ior.buffer = r.buffer;
		ior.buflen = 512;

		fat.f_read(&ior, r.buffer, 512);
	} while (0);

// 	int i = 0, j = 0;

	while (1) {
		k->fire_event(EVENT_IDLE, NULL);

		// Serial emulates receiving data with EVENT_TEST
// 		if (j >= 15)
// 		{
// 			uint8_t* c = (uint8_t*) "Abc\n";
// 			k->fire_event(EVENT_TEST, c);
// 			j = 0;
// 		}
// 		j++;
	}
}

Makefile

PROJECT = silken

FLAGS=-Wall -ffunction-sections -fdata-sections -g

CFLAGS=$(FLAGS) -std=gnu99 -g
CXXFLAGS=$(FLAGS) -std=gnu++0x
LDFLAGS=$(FLAGS) -Wl,-gc-sections

CC=gcc
CPP=g++
LD=g++

O=build

CSRC=$(shell find . -iname '*.c')
CPPSRC=$(shell find . -iname '*.cpp')
OBJ=$(patsubst %.c,%.o,$(CSRC)) $(patsubst %.cpp,%.o,$(CPPSRC))

.PHONY: all clean
VPATH=$(O)

all: $(PROJECT)

clean:
	@rm -rf $(O)

$(PROJECT): $(OBJ)
	@echo "  LD    " $(O)/$@ "<" $^
	@$(LD) $(LDFLAGS) $(patsubst %.o,$(O)/%.o,$(OBJ)) -o $(O)/$(PROJECT)

$(O):
	mkdir $@

%.o: %.c Makefile $(O)
	@echo "  CC    " $<
	@$(CC) $(CFLAGS) -c -o $(O)/$@ $<

%.o: %.cpp Makefile $(O)
	@echo "  CPP   " $<
	@$(CPP) $(CXXFLAGS) -c -o $(O)/$@ $<

module.h

#ifndef _MODULE_H
#define _MODULE_H

#include "event.h"

class Module;
class Kernel;

class Module
{
	friend class Kernel;
public:
	virtual void on_module_added(Kernel* k) { kernel = k; };
	virtual void receive_event(Event event, void* data) = 0;
protected:
	Kernel* kernel;
private:
	Module* next_module;
};

#endif /* _MODULE_H */

ringbuffer.cpp

#include "ringbuffer.h"

#include <cstdlib>

RingBuffer::RingBuffer(int size)
{
	head = tail = 0;
	buf = (uint8_t*) malloc(size);
	this->size = size;
}

RingBuffer::~RingBuffer()
{
	free(buf);
}

uint8_t RingBuffer::popc()
{
	if (can_read())
	{
		uint8_t c = buf[tail];
		tail = incwrap(tail + 1);
		return c;
	}
	return 0;
}

uint8_t RingBuffer::peekc(int index)
{
	if (index > can_read())
		return 0;

	return buf[incwrap(tail + index)];
}

void RingBuffer::pushc_drop(uint8_t c)
{
	if (can_write())
	{
		buf[head] = c;
		head = incwrap(head + 1);
	}
}

void RingBuffer::pushc_wait(uint8_t c)
{
	while (can_write() == 0);

	buf[head] = c;
	head = incwrap(head + 1);
}

int RingBuffer::can_read()
{
	if (head >= tail)
		return head - tail;
	return head + size - tail;
}

int RingBuffer::can_write()
{
	if (tail > head)
		return tail - head - 1;
	return tail + size - head - 1;
}

int RingBuffer::incwrap(int r)
{
	while (r < 0)
		r += size;
	while (r >= size)
		r -= size;

	return r;
}

ringbuffer.h

#ifndef _RINGBUFFER_H
#define _RINGBUFFER_H

#include <cstdint>

class RingBuffer
{
public:
	RingBuffer(int);
	~RingBuffer();

	uint8_t popc(void);
	uint8_t peekc(int);

	void pushc_drop(uint8_t c);
	void pushc_wait(uint8_t c);

	int can_read(void);
	int can_write(void);
// protected:
	int incwrap(int);

	int head;
	int tail;
	int size;

	uint8_t* buf;
};

#endif /* _RINGBUFFER_H */

sd.cpp

#include "sd.h"

#include <cstddef>
#include <cstdio>
#include <cstdlib>

#include <sys/time.h>
#include <sys/stat.h>

#include <unistd.h>
#include <fcntl.h>
#include <signal.h>

Sd* Sd::instance = NULL;

void sig_alarm(int)
{
	if (Sd::instance)
	{
		fprintf(stderr, "!!-!! ");
		Sd::instance->_DMA();
	}
}

void dump_stack(_sd_work_stack* w)
{
	while (w)
	{
		fprintf(stderr, "[SD] work item %p:\n\taction: %d\n\tbuffer: %p\n\tcount: %d\n\tlba: %u\n\tnext: %p\n", w, w->action, w->buffer, w->count, w->lba, w->next);
		w = w->next;
	}
}

Sd::Sd()
{
	instance = this;

	fd = open("diskimg", O_RDWR);
	if (fd < 0)
	{
		perror("open(diskimg, O_RDWR)");
		exit(1);
	}

	signal(SIGALRM, sig_alarm);

	work_stack = NULL;
	high_prio = NULL;
}

Sd::~Sd()
{
	close(fd);
}

uint32_t Sd::disk_blocks(void)
{
	if (fd)
	{
		struct stat s;
		if (fstat(fd, &s) == 0)
			return s.st_size >> 9;
	}
	perror("stat");
	fprintf(stderr, "failed to stat disk to get size!\n");
	exit(1);
}

int Sd::begin_read(uint32_t lba, void* buffer, int bufsize, _sd_event_receiver* owner)
{
	if (bufsize < 512)
		return -1;

	_sd_work_stack* w = (_sd_work_stack*) malloc(sizeof(__sd_work_stack));
	if (bufsize == 512)
		w->action = SD_ITEM_READ_BLOCK;
	else
		w->action = SD_ITEM_READ_MULTI;

	w->buffer = (uint8_t*) buffer;
	w->count  = bufsize >> 9;
	w->lba    = lba;
	w->owner  = owner;
	w->next   = NULL;

	_sd_work_stack* x = work_stack;

	if (x)
	{
		while (x->next)
			x = x->next;
		x->next = w;
	}
	else
	{
		work_stack = w;
		begin_work();
	}

// 	dump_stack(work_stack);

	return 0;
}

void Sd::begin_work()
{
	if (work_stack)
		ualarm(1000, 0);
	else
		fprintf(stderr, "Sd::begin_work called with empty work stack!");
}

void Sd::_DMA()
{
	if (work_stack)
	{
		_sd_work_stack* completed = work_stack;

		switch(completed->action)
		{
			case SD_ITEM_INIT:
				break;
			case SD_ITEM_READ_BLOCK:
			case SD_ITEM_READ_MULTI:
			{
				completed->lba = lseek(fd, completed->lba * 512, SEEK_SET) / 512;
				int r = read(fd, completed->buffer, 512);
				if (r == 512)
					completed->count--;
				else
					perror("read");
				break;
			}
			case SD_ITEM_WRITE_BLOCK:
			case SD_ITEM_WRITE_MULTI:
			{
				completed->lba = lseek(fd, completed->lba * 512, SEEK_SET) / 512;
				int r = write(fd, completed->buffer, 512);
				if (r == 512)
					completed->count--;
				else
					perror("write");
				break;
			}
			default:
				break;
		}

		if (completed->count == 0)
		{
			work_stack = work_stack->next;

			completed->next = NULL;

			if (completed->owner)
				completed->owner->_sd_callback(completed);

			free(completed);

			if (work_stack)
				begin_work();
		}
		else
		{
			completed->lba += 512;
			completed->buffer += 512;
			begin_work();
		}
	}
}

sd.h

#ifndef _SD_H
#define _SD_H

#include <cstdint>

class Sd;
class _sd_event_receiver;

typedef enum
{
	SD_ITEM_NULL,
	SD_ITEM_INIT,
	SD_ITEM_READ_BLOCK,
	SD_ITEM_READ_MULTI,
	SD_ITEM_WRITE_BLOCK,
	SD_ITEM_WRITE_MULTI,

	SD_NUM_ITEMS
} __sd_work_item;

struct __sd_work_stack;
struct __sd_work_stack
{
	__sd_work_item action;
	uint8_t*       buffer;
	uint32_t       count;
	uint32_t       lba;

	_sd_event_receiver* owner;

	struct __sd_work_stack* next;
};

typedef struct __sd_work_stack _sd_work_stack;

class _sd_event_receiver
{
	friend class Sd;
	virtual void _sd_callback(_sd_work_stack*) = 0;
};

class Sd
{
public:
	Sd(void);
	~Sd(void);

	int begin_read(uint32_t lba, void* buffer, int bufsize, _sd_event_receiver* owner);
	void begin_work(void);

	void _DMA(void);

	static Sd* instance;

	uint32_t disk_blocks(void);

protected:
	struct __sd_work_stack* work_stack;
	struct __sd_work_stack* high_prio;

private:
	int fd;
};

#endif /* _SD_H */

serial.cpp

#include "serial.h"

#include <cstdio>
#include <cstdlib>
#include <cstring>

#include "kernel.h"
#include "arbiter.h"

Serial::Serial() : rxbuf(256), txbuf(256)
{
	block_flags = 0;
}

void Serial::on_module_added(Kernel* k)
{
	this->Module::on_module_added(k);

	kernel->register_for_event(this, EVENT_IDLE);
	kernel->register_for_event(this, EVENT_TEST);

	kernel->arbiter->add_stream(this);

	block_flags = 0;
}

void Serial::receive_event(Event event, void* data)
{
	switch(event)
	{
		case EVENT_IDLE:
			{
				// TODO: check for tx data
			};
			break;
		case EVENT_TEST:
			{
				if (data)
				{
					char* c = (char*) data;
					int l = strlen(c);
					if (l <= rxbuf.can_write())
					{
						while (*c)
						{
							rxbuf.pushc_drop(*c);
							if ((*c == 13) || (*c == 10))
								nl++;
							c++;
						}
					}
					else
					{
						fprintf(stderr, "Serial: rx buffer overflow!\n");
					}
// 					fprintf(stderr, "Serial rxbuf: %d\n", rxbuf.can_read());
				}
			};
			break;
		default:
			break;
	}
}

int   Serial::puts(const char* str)
{
	const char* c = str;
	int l = 0;
	while (*c)
	{
		txbuf.pushc_wait(*c);
		c++;
		l++;
	}
	return l;
}

int   Serial::has_line(void)
{
	return nl;
}

int Serial::peek_line(char* buf, int bufsize)
{
// 	fprintf(stderr, "peek\n");
	for (int l = 0, t = rxbuf.tail; t != rxbuf.head; t = rxbuf.incwrap(t + 1))
	{
		l++;
		if (rxbuf.buf[t] == 13 || rxbuf.buf[t] == 10)
		{
			int c = l;
			if (c >= bufsize)
				c = bufsize - 1;
			for (int i = 0; i < c; i++)
				buf[i] = rxbuf.buf[rxbuf.incwrap(rxbuf.tail + i)];
			buf[c] = 0;
			return l;
		}
	}
	return -1;
}

void  Serial::accept_line()
{
	uint8_t c;
	do {
		c = rxbuf.popc();
		if (c == 13 || c == 10)
		{
			nl--;
			return;
		}
	} while (c != 0);
}

void Serial::on_action_invoke(ActionData* data)
{
	SerialData* s = (SerialData*) data;
	printf("%s\n", s->str);
	free(s->str);
	s->str = NULL;
	s->remove();
}

SerialData::SerialData(ActionReceiver* owner, char* str) : ActionData(owner)
{
	this->str = str;
}

serial.h

#ifndef _SERIAL_H
#define _SERIAL_H

#include "module.h"
#include "ringbuffer.h"
#include "streamio.h"
#include "action.h"
#include "actiondata.h"

class SerialData : public ActionData
{
	friend class Serial;
public:
	SerialData(ActionReceiver*, char*);
protected:
	char* str;
};

class Serial : public Module, public StreamIO, public ActionReceiver
{
public:
	Serial();
	~Serial();

	void on_module_added(Kernel*);
	void receive_event(Event event, void* data);

	// implementation of StreamIO, see comments in streamio.h
	int   puts(const char*);
	int   has_line(void);
	int   peek_line(char*, int);
	void  accept_line();

	// implementation of ActionReceiver
	void on_action_invoke(ActionData*);

private:
	RingBuffer rxbuf;
	RingBuffer txbuf;
	int nl;
};

#endif /* _SERIAL_H */

streamio.cpp

streamio.h

#ifndef _STREAMIO_H
#define _STREAMIO_H

#include <cstdint>
#include <cstdarg>
#include <cstdio>

class StreamIO;

class StreamIO
{
	friend class Arbiter;
public:
	/*
	 * inheritors must implement these methods
	 */
	// puts: copy data into txbuf until a zero is encountered
	virtual int   puts(const char*) = 0;

	// return nonzero if a line is available in the rxbuf
	virtual int   has_line(void) = 0;

	// provide the next line. DO NOT remove it from rxbuf yet
	virtual int   peek_line(char*, int) = 0;

	// kernel has accepted the line, remove it from rxbuf
	virtual void  accept_line(void) = 0;

	// here we store any reason that this stream might be blocked.
	// These bits are manipulated externally, no need for inheritors to touch them
	union {
		uint8_t block_flags;
		struct {
			uint8_t flag_queue :1;
			uint8_t flag_delay :1;
			uint8_t flag_temperature :1;
		};
	};

	// utility printf, calls puts() so no need to overload it
	virtual int   printf(const char* format, ...)
		__attribute__ ((format(printf, 2, 3)))
	{
		char *buffer;
		// Make the message
		va_list args;
		va_start(args, format);

		int size = vsnprintf(NULL, 0, format, args)
		+ 1; // we add one to take into account space for the terminating \0

		buffer = new char[size];
		vsnprintf(buffer, size, format, args);
		va_end(args);

		puts(buffer);

		delete[] buffer;
		return size - 1;
	}
protected:
	StreamIO* next_stream;
};

#endif /* _STREAMIO_H */