Index: benchmark/io/http/channel.cfa
===================================================================
--- benchmark/io/http/channel.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/channel.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,53 @@
+#include <assert.h>
+#include <stdlib.hfa>
+
+#include "channel.hfa"
+
+// forall(otype T) {
+      void  ?{}( channel & this, int _size ) with(this) {
+		buffer = anew( _size );
+		front = 0;
+		back = 0;
+		count = 0;
+		size = _size;
+
+		(lock){};
+		(prods){};
+		(cons){};
+	}
+
+      void ^?{}( channel & this ) {
+            delete( this.buffer );
+      }
+
+	void put( channel & this, int elem ) with( this ) {
+		lock( this.lock );
+		while( count ==  size) {
+			wait( prods, this.lock );
+		}
+
+		/* paranoid */ assert( count < size );
+		buffer[back] = elem;
+		count ++;
+		back = (back + 1) % size;
+
+		notify_one( cons );
+		unlock( this.lock );
+	}
+
+	int take( channel & this ) with( this ) {
+		lock( this.lock );
+		while( count == 0 ) {
+			wait( cons, this.lock );
+		}
+
+		/* paranoid */ assert( count > 0 );
+		int temp = buffer[front];
+		count --;
+		front = (front + 1) % size;
+
+		notify_one( prods );
+		unlock( this.lock );
+		return temp;
+	}
+// }
Index: benchmark/io/http/channel.hfa
===================================================================
--- benchmark/io/http/channel.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/channel.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,23 @@
+#pragma once
+
+#include <assert.h>
+#include <mutex.hfa>
+
+// forall(otype T) {
+	struct channel {
+		int * buffer;
+		int front;
+		int back;
+		int count;
+		int size;
+		mutex_lock lock;
+		condition_variable prods;
+		condition_variable cons;
+	};
+
+      void  ?{}( channel & this, int size );
+      void ^?{}( channel & this );
+
+	void put( channel & this, int elem );
+	int take( channel & this );
+// }
Index: benchmark/io/http/filecache.cfa
===================================================================
--- benchmark/io/http/filecache.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/filecache.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,174 @@
+#include "filecache.hfa"
+
+#include <assert.h>
+#include <string.h>
+
+#include <stdlib.hfa>
+
+#include <errno.h>
+#include <unistd.h>
+extern "C" {
+	#include <fcntl.h>
+	#include <ftw.h>
+}
+
+#include "options.hfa"
+
+static inline uint32_t murmur_32_scramble(uint32_t k) {
+    k *= 0xcc9e2d51;
+    k = (k << 15) | (k >> 17);
+    k *= 0x1b873593;
+    return k;
+}
+
+uint32_t murmur3_32(const uint8_t* key, size_t len, uint32_t seed)
+{
+	uint32_t h = seed;
+	uint32_t k;
+	/* Read in groups of 4. */
+	for (size_t i = len >> 2; i; i--) {
+		// Here is a source of differing results across endiannesses.
+		// A swap here has no effects on hash properties though.
+		memcpy(&k, key, sizeof(uint32_t));
+		key += sizeof(uint32_t);
+		h ^= murmur_32_scramble(k);
+		h = (h << 13) | (h >> 19);
+		h = h * 5 + 0xe6546b64;
+	}
+	/* Read the rest. */
+	k = 0;
+	for (size_t i = len & 3; i; i--) {
+		k <<= 8;
+		k |= key[i - 1];
+	}
+	// A swap is *not* necessary here because the preceding loop already
+	// places the low bytes in the low places according to whatever endianness
+	// we use. Swaps only apply when the memory is copied in a chunk.
+	h ^= murmur_32_scramble(k);
+	/* Finalize. */
+	h ^= len;
+	h ^= h >> 16;
+	h *= 0x85ebca6b;
+	h ^= h >> 13;
+	h *= 0xc2b2ae35;
+	h ^= h >> 16;
+	return h;
+}
+
+
+struct {
+	cache_line * entries;
+	size_t size;
+} file_cache;
+
+void ?{}( cache_line & this ) with( this ) {
+	file = 0p;
+	size = 0;
+	fd   = 0;
+}
+
+[int fd, size_t size] get_file( * const char file, size_t len ) {
+	uint32_t idx = murmur3_32( (const uint8_t *)file, len, options.hash_seed ) % file_cache.size;
+
+	for(int i = 0;; i++) {
+		assert( i < file_cache.size );
+		cache_line & entry = file_cache.entries[idx];
+		if( !entry.file ) return [-1, 0];
+		#if !defined(REJECT_CONFLICTS)
+			if( strncmp(entry.file, file, len) != 0 ) {
+				idx = (idx + 1) % file_cache.size;
+				continue;
+			}
+		#endif
+		return [entry.fd, entry.size];
+	}
+}
+
+int put_file( cache_line & entry ) {
+	uint32_t idx = murmur3_32( (const uint8_t *)entry.file, strlen(entry.file), options.hash_seed ) % file_cache.size;
+
+	int i = 0;
+	for(;file_cache.entries[idx].file; i++ ) {
+		assert( i < file_cache.size );
+		idx = (idx + 1) % file_cache.size;
+	}
+
+	file_cache.entries[idx] = entry;
+	return i > 0 ? 1 : 0;
+}
+
+// int ftw(const char *dirpath, int (*fn) (const char *fpath, const struct stat *sb, int typeflag), int nopenfd)
+void fill_cache( const char * path ) {
+	int ret;
+	size_t fcount = 0;
+	size_t fsize = 16;
+	cache_line * raw = 0p;
+	raw = alloc(raw, fsize, true);
+	// Step 1 get a dense array of all files
+	int walk(const char *fpath, const struct stat *sb, int typeflag) {
+		if(typeflag != FTW_F) return 0;
+
+		int idx = fcount;
+		fcount++;
+		if(fcount > fsize) {
+			fsize *= 2;
+			raw = alloc(raw, fsize, true);
+		}
+
+		raw[idx].file = strdup(fpath+2);
+		raw[idx].size = sb->st_size;
+		raw[idx].fd = open( fpath, options.open_flags );
+		if(raw[idx].fd < 0) {
+			abort( "open file error: (%d) %s\n", (int)errno, strerror(errno) );
+		}
+		return 0;
+	}
+
+	ret = ftw(path, walk, 10);
+	if(ret < 0) {
+		abort( "ftw error: (%d) %s\n", (int)errno, strerror(errno) );
+	}
+
+	if(fcount == 0) {
+		abort("No file found in path %s\n", path);
+	}
+
+	// Step 2 create the cache
+	file_cache.size = options.file_cache_size > 0 ? options.file_cache_size : fsize;
+	if( file_cache.size < fcount ) {
+		abort("File Cache too small\n");
+	}
+
+	file_cache.entries = anew(fsize);
+
+	// Step 3 fill the cache
+	int conflicts = 0;
+	for(i; fcount) {
+		printf("Added file %s\n", raw[i].file);
+		conflicts += put_file( raw[i] );
+	}
+	printf("Filled cache from path \"%s\" with %zu files\n", path, fcount);
+	if( conflicts > 0 ) {
+		printf("Found %d conflicts (seed: %u)\n", conflicts, options.hash_seed);
+		#if defined(REJECT_CONFLICTS)
+			abort("Conflicts found in the cache");
+		#endif
+	}
+
+	// Step 4 clean up
+	free( raw );
+}
+
+void close_cache() {
+	for(idx; file_cache.size) {
+		cache_line & entry = file_cache.entries[idx];
+		if( !entry.file ) continue;
+
+		int ret = close( entry.fd );
+		if(ret < 0) {
+			abort( "close file error: (%d) %s\n", (int)errno, strerror(errno) );
+		}
+	}
+
+	delete( file_cache.entries );
+}
Index: benchmark/io/http/filecache.hfa
===================================================================
--- benchmark/io/http/filecache.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/filecache.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,13 @@
+#pragma once
+
+#include <stdint.h>
+
+struct cache_line {
+	const char * file;
+	size_t size;
+	int fd;
+};
+
+[int fd, size_t size] get_file( * const char file, size_t len );
+void fill_cache( const char * path );
+void close_cache();
Index: benchmark/io/http/main.cfa
===================================================================
--- benchmark/io/http/main.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/main.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,141 @@
+#define __USE_GNU
+
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+extern "C" {
+	#include <sys/socket.h>
+	#include <sys/types.h>
+	#include <netinet/in.h>
+}
+
+#include <kernel.hfa>
+#include <stats.hfa>
+#include <thread.hfa>
+
+#include "channel.hfa"
+#include "filecache.hfa"
+#include "options.hfa"
+#include "worker.hfa"
+
+//=============================================================================================
+// Globals
+//=============================================================================================
+Options options @= {
+	0,
+	42u,
+	0,
+	false,
+	false,
+	0
+};
+
+channel & wait_connect;
+
+struct ServerProc {
+	processor self;
+};
+
+void ?{}( ServerProc & this ) {
+	/* paranoid */ assert( options.the_cluster != 0p );
+	(this.self){ "Benchmark Processor", *options.the_cluster };
+
+	#if !defined(__CFA_NO_STATISTICS__)
+		if( options.procstats ) {
+			print_stats_at_exit( this.self, options.the_cluster->print_stats );
+		}
+		if( options.viewhalts ) {
+			print_halts( this.self );
+		}
+	#endif
+}
+
+//=============================================================================================
+// Main
+//============================================================================================='
+int main( int argc, char * argv[] ) {
+	int port      = 8080;
+	int ret       = 0;
+	int backlog   = 10;
+	int nprocs    = 1;
+	int nworkers  = 1;
+	int cl_flags  = 0;
+	int chan_size = 10;
+	const char * path = ".";
+	//===================
+	// Parse args
+
+	//===================
+	// Open Files
+	printf("Filling cache from %s\n", path);
+	fill_cache( path );
+
+	//===================
+	// Open Socket
+	printf("Listening on port %d\n", port);
+	int server_fd = socket(AF_INET, SOCK_STREAM, 0);
+	if(server_fd < 0) {
+		abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
+	}
+
+	struct sockaddr_in address;
+	int addrlen = sizeof(address);
+	memset( (char *)&address, '\0' );
+	address.sin_family = AF_INET;
+	address.sin_addr.s_addr = htonl(INADDR_ANY);
+	address.sin_port = htons( port );
+
+	ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
+	if(ret < 0) {
+		abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
+	}
+
+	ret = listen( server_fd, backlog );
+	if(ret < 0) {
+		abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
+	}
+
+	//===================
+	// Run Server Cluster
+	{
+		cluster cl = { "Server Cluster", cl_flags };
+		#if !defined(__CFA_NO_STATISTICS__)
+			print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
+		#endif
+		options.the_cluster = &cl;
+
+		channel chan = { chan_size };
+		&wait_connect = &chan;
+
+		{
+			ServerProc procs[nprocs];
+			{
+				Worker workers[nworkers];
+				printf("%d workers started on %d processors\n", nworkers, nprocs);
+				{
+					Acceptor acceptor = { server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen, 0 };
+				}
+				printf("Shutting Down\n");
+
+				// Clean-up the workers
+				for(nworkers) {
+					put( wait_connect, -1 );
+				}
+			}
+		}
+	}
+
+	//===================
+	// Close Socket
+	printf("Closing Socket\n");
+	ret = close( server_fd );
+	if(ret < 0) {
+		abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
+	}
+
+	//===================
+	// Close Files
+	printf("Closing Files\n");
+	close_cache();
+}
Index: benchmark/io/http/options.hfa
===================================================================
--- benchmark/io/http/options.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/options.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,16 @@
+#pragma once
+
+#include <stdint.h>
+
+struct cluster;
+
+struct Options {
+	int open_flags;
+	uint32_t hash_seed;
+      size_t file_cache_size;
+	bool procstats;
+	bool viewhalts;
+	cluster * the_cluster;
+};
+
+extern Options options;
Index: benchmark/io/http/protocol.cfa
===================================================================
--- benchmark/io/http/protocol.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/protocol.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,84 @@
+#include "protocol.hfa"
+
+#include <iofwd.hfa>
+
+#include <assert.h>
+// #include <stdio.h> // Don't use stdio.h, too slow to compile
+extern "C" {
+      int snprintf ( char * s, size_t n, const char * format, ... );
+}
+#include <string.h>
+
+#include <errno.h>
+
+
+const char * http_msgs[] = {
+	"HTTP/1.1 200 OK\nContent-Type: text/plain\nContent-Length: %zu\n\n",
+	"HTTP/1.1 400 Bad Request\nContent-Type: text/plain\nContent-Length: 0\n\n",
+	"HTTP/1.1 404 Not Found\nContent-Type: text/plain\nContent-Length: 0\n\n",
+	"HTTP/1.1 413 Payload Too Large\nContent-Type: text/plain\nContent-Length: 0\n\n",
+	"HTTP/1.1 414 URI Too Long\nContent-Type: text/plain\nContent-Length: 0\n\n",
+};
+
+_Static_assert( KNOWN_CODES == (sizeof(http_msgs) / sizeof(http_msgs[0])));
+
+static inline int answer( int fd, const char * it, int len) {
+	while(len > 0) {
+		// Call write
+		int ret = write(fd, it, len);
+		if( ret < 0 ) { if( errno != EAGAIN && errno != EWOULDBLOCK) abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) ); }
+
+		// update it/len
+		it  += ret;
+		len -= ret;
+	}
+	return 0;
+}
+
+int answer_error( int fd, HttpCode code ) {
+	/* paranoid */ assert( code < KNOWN_CODES && code != OK200 );
+	int idx = (int)code;
+	return answer( fd, http_msgs[idx], strlen( http_msgs[idx] ) );
+}
+
+int answer_header( int fd, size_t size ) {
+	const char * fmt = http_msgs[OK200];
+	int len = 100;
+	char buffer[len];
+	len = snprintf(buffer, len, fmt, size);
+	return answer( fd, buffer, len );
+}
+
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
+	char * it = buffer;
+	size_t count = len - 1;
+	int rlen = 0;
+	READ:
+	for() {
+		int ret = cfa_read(fd, it, count);
+		if(ret < 0 ) {
+			if( errno ) return [OK200, true, 0p, 0];
+			if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
+			abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
+		}
+		it[ret + 1] = '\0';
+		rlen += ret;
+
+		if( strstr( it, "\r\n\r\n" ) ) break;
+
+		it += ret;
+		count -= ret;
+
+		if( count < 1 ) return [E414, false, 0p, 0];
+	}
+
+	printf("%.*s\n", rlen, buffer);
+
+	it = buffer;
+	int ret = memcmp(it, "GET /", 5);
+	if( ret != 0 ) return [E400, false, 0p, 0];
+	it += 5;
+
+	char * end = strstr( it, " " );
+	return [OK200, false, it, end - it];
+}
Index: benchmark/io/http/protocol.hfa
===================================================================
--- benchmark/io/http/protocol.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/protocol.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,15 @@
+#pragma once
+
+enum HttpCode {
+	OK200 = 0,
+	E400,
+	E404,
+	E413,
+	E414,
+	KNOWN_CODES
+};
+
+int answer_error( int fd, HttpCode code );
+int answer_header( int fd, size_t size );
+
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len);
Index: benchmark/io/http/worker.cfa
===================================================================
--- benchmark/io/http/worker.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/worker.cfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,143 @@
+#include "worker.hfa"
+
+#define __USE_GNU
+
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+extern "C" {
+	#include <sys/socket.h>
+	#include <sys/types.h>
+	#include <linux/fcntl.h>
+	#include <linux/stat.h>
+	#include <netinet/in.h>
+}
+
+#include <iofwd.hfa>
+
+#include "options.hfa"
+#include "protocol.hfa"
+#include "filecache.hfa"
+
+extern "C" {
+// extern ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);
+extern ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out, size_t len, unsigned int flags);
+}
+
+ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count) {
+	return splice(in_fd, offset, out_fd, 0p, count, 0);
+}
+
+
+//=============================================================================================
+// Worker Thread
+//=============================================================================================
+void sendfile( Worker & this, int fd, int ans_fd, size_t count ) {
+	off_t offset = 0;
+	ssize_t ret;
+	SPLICE1: while(count > 0) {
+		ret = cfa_splice(ans_fd, &offset, this.pipe[1], 0p, count, 5);
+		if( ret < 0 ) {
+			if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1;
+			abort( "splice [0] error: (%d) %s\n", (int)errno, strerror(errno) );
+		}
+
+		count -= ret;
+		offset += ret;
+		size_t in_pipe = ret;
+		SPLICE2: while(in_pipe > 0) {
+			ret = cfa_splice(this.pipe[0], 0p, fd, 0p, in_pipe, 5);
+			if( ret < 0 ) {
+				if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2;
+				abort( "splice [1] error: (%d) %s\n", (int)errno, strerror(errno) );
+			}
+			in_pipe -= ret;
+		}
+
+	}
+}
+
+void ?{}( Worker & this ) {
+	((thread&)this){ "Server Worker Thread", *options.the_cluster };
+	int ret = pipe(this.pipe);
+	if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
+}
+
+void main( Worker & this ) {
+	CONNECTION:
+	while( int fd = take(wait_connect); fd >= 0) {
+            printf("New connection, waiting for requests\n");
+		REQUEST:
+		for() {
+			bool closed;
+			HttpCode code;
+			const char * file;
+			size_t name_size;
+
+			// Read the http request
+			size_t len = 1024;
+			char buffer[len];
+			printf("Reading request\n");
+			[code, closed, file, name_size] = http_read(fd, buffer, len);
+
+			// if we are done, break out of the loop
+			if( closed ) {
+				printf("Connection closed\n");
+				continue CONNECTION;
+			}
+
+			// If this wasn't a request retrun 400
+			if( code != OK200 ) {
+				printf("Invalid Request\n");
+				answer_error(fd, code);
+				continue REQUEST;
+			}
+
+			printf("Request for file %.*s\n", name_size, file);
+
+			// Get the fd from the file cache
+			int ans_fd;
+			size_t count;
+			[ans_fd, count] = get_file( file, name_size );
+
+			// If we can't find the file, return 404
+			if( ans_fd < 0 ) {
+				printf("File Not Found\n");
+				answer_error(fd, E404);
+				continue REQUEST;
+			}
+
+			// Send the header
+			answer_header(fd, count);
+
+			// Send the desired file
+			sendfile( this, fd, ans_fd, count);
+
+			printf("File sent\n");
+		}
+	}
+}
+
+//=============================================================================================
+// Acceptor Thread
+//=============================================================================================
+void ?{}( Acceptor & this, int sockfd, struct sockaddr * addr, socklen_t * addrlen, int flags ) {
+	((thread&)this){ "Acceptor Thread", *options.the_cluster };
+	this.sockfd  = sockfd;
+	this.addr    = addr;
+	this.addrlen = addrlen;
+	this.flags   = flags;
+}
+
+void main( Acceptor & this ) {
+	for() {
+		int ret = cfa_accept4( this.[sockfd, addr, addrlen, flags] );
+		if(ret < 0) {
+			if( errno == ECONNABORTED ) break;
+			abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
+		}
+
+            printf("New connection accepted\n");
+		put( wait_connect, ret );
+      }
+}
Index: benchmark/io/http/worker.hfa
===================================================================
--- benchmark/io/http/worker.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
+++ benchmark/io/http/worker.hfa	(revision 0aec496b4b9d7e9afdb62632be73923517922a41)
@@ -0,0 +1,34 @@
+#pragma once
+
+#include <thread.hfa>
+
+extern "C" {
+	#include <sys/socket.h>
+}
+
+#include "channel.hfa"
+
+extern channel & wait_connect;
+
+//=============================================================================================
+// Worker Thread
+//=============================================================================================
+
+thread Worker {
+	int pipe[2];
+};
+void ?{}( Worker & this );
+void main( Worker & );
+
+//=============================================================================================
+// Acceptor Thread
+//=============================================================================================
+thread Acceptor {
+	int sockfd;
+	struct sockaddr * addr;
+	socklen_t * addrlen;
+	int flags;
+};
+
+void ?{}( Acceptor & this, int sockfd, struct sockaddr * addr, socklen_t * addrlen, int flags );
+void main( Acceptor & this );