Mercurial > luaevent-prosody / changeset

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/doc/PLAN	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,19 @@
+-- Listener Scenario
+create socket, perform bind, set listening
++add read-event listener
++start loop
+
+-- Comm Scenario
+provided socket
+-inside loop
+begin coro
+	read - data not ready
+	yield sock, needRead
+end coro
+coro parent == callback
+(coro parent
+	call coro(sock, event)
+	if not ok, go back.. end
+	if newEvent ~= event then
+		unset event, reset newEvent
+	end)
\ No newline at end of file

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/include/luaevent.h	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,17 @@
+#ifndef LUAEVENT_H
+#define LUAEVENT_H
+
+#include <lua.h>
+#include <sys/types.h>
+#include <sys/time.h>
+#include <event.h>
+
+typedef struct {
+	struct event ev;
+	lua_State* L;
+	int callbackRef;
+} le_callback;
+
+int luaopen_luaevent(lua_State* L);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/luaevent.lua	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,130 @@
+module("luaevent", package.seeall)
+require("luaevent.core")
+
+local EV_READ = luaevent.core.EV_READ
+local EV_WRITE = luaevent.core.EV_WRITE
+local fair = false
+
+-- Weak keys.. the keys are the client sockets
+local clientTable = {} or setmetatable({}, {'__mode', 'k'})
+
+local function getWrapper()
+	local running = coroutine.running()
+	return function(...)
+		print(coroutine.running(), running)
+		print(debug.traceback())
+		if coroutine.running() == running then return end
+		return select(2, coroutine.resume(running, ...))
+	end
+end
+
+function send(sock, data, start, stop)
+	local s, err
+	local from = start or 1
+	local sent = 0
+	repeat
+		from = from + sent
+		s, err, sent = sock:send(data, from, stop)
+		-- Add extra coro swap for fairness
+		-- CURRENTLY DISABLED FOR TESTING......
+		if fair and math.random(100) > 90 then
+			coroutine.yield(EV_WRITE)
+		end
+		if s or err ~= "timeout" then return s, err, sent end
+		if not clientTable[sock] then clientTable[sock] = luaevent.core.addevent(sock, EV_WRITE, getWrapper()) end
+		coroutine.yield(EV_WRITE)
+	until false
+end
+function receive(sock, pattern, part)
+	local s, err
+	pattern = pattern or '*l'
+	repeat
+		s, err, part = sock:receive(pattern, part)
+		if s or err ~= "timeout" then return s, err, part end
+		if not clientTable[sock] then clientTable[sock] = luaevent.core.addevent(sock, EV_READ, getWrapper()) end
+		coroutine.yield(EV_READ)
+	until false
+end
+-- same as above but with special treatment when reading chunks,
+-- unblocks on any data received.
+function receivePartial(client, pattern)
+	local s, err, part
+	pattern = pattern or "*l"
+	repeat
+	s, err, part = client:receive(pattern)
+	if s or ( (type(pattern)=="number") and part~="" and part ~=nil ) or 
+		err ~= "timeout" then return s, err, part end
+		if not clientTable[sock] then clientTable[sock] = luaevent.core.addevent(sock, EV_READ, getWrapper()) end
+		coroutine.yield(EV_READ)
+	until false
+end
+function connect(sock, ...)
+	sock:settimeout(0)
+	local ret, err = sock:connect(...)
+	if ret or err ~= "timeout" then return ret, err end
+	if not clientTable[sock] then clientTable[sock] = luaevent.core.addevent(sock, EV_WRITE, getWrapper()) end
+	coroutine.yield(EV_WRITE)
+	ret, err = sock:connect(...)
+	if err == "already connected" then
+		return 1
+	end
+	return ret, err
+end
+-- Deprecated..
+function flush(sock)
+end
+local function clientCoroutine(sock, handler)
+	-- Figure out what to do ......
+	return handler(sock)
+end
+local function handleClient(co, client, handler)
+	local ok, res, event = coroutine.resume(co, client, handler)
+end
+local function serverCoroutine(sock, callback)
+	local listenItem = luaevent.core.addevent(sock, EV_READ, getWrapper())
+	repeat
+		local event = coroutine.yield(EV_READ)
+		-- Get new socket
+		local client = sock:accept()
+		if client then
+			client:settimeout(0)
+			local co = coroutine.create(clientCoroutine)
+			handleClient(co, client, callback)
+		end
+	until false
+end
+function addserver(sock, callback)
+	local coro = coroutine.create(serverCoroutine)
+	assert(coroutine.resume(coro, sock, callback))
+end
+function addthread(func, ...)
+	return coroutine.resume(coroutine.create(func), ...)
+end
+local _skt_mt = {__index = {
+	connect = function(self, ...)
+		return connect(self.socket, ...)
+	end,
+	send = function (self, data)
+		return send (self.socket, data)
+	end,
+	
+	receive = function (self, pattern)
+		if (self.timeout==0) then
+  			return receivePartial(self.socket, pattern)
+  		end
+		return receive (self.socket, pattern)
+	end,
+	
+	flush = function (self)
+		return flush (self.socket)
+	end,
+	
+	settimeout = function (self,time)
+		self.timeout=time
+		return
+	end,
+}}
+function wrap(sock)
+	return setmetatable({socket = sock}, _skt_mt)
+end
+loop = luaevent.core.loop
\ No newline at end of file

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/premake.lua	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,32 @@
+project.name = "luaevent.core"
+project.libdir = "lib"
+project.bindir = "bin"
+
+package = newpackage()
+package.kind = "dll"
+package.language = "c++"
+package.targetprefix = ""
+package.target = "core"
+
+package.links = {
+	"event"
+}
+
+package.includepaths = {
+	"include",
+}
+if linux then
+	package.buildoptions = { "-Wall" }
+	package.config["Debug"].buildoptions = { "-O0" }
+	package.linkoptions =  { "-Wall -L/usr/local/lib" }
+	package.postbuildcommands = { "mkdir -p test/luaevent", "cp bin/* test/luaevent", "cp luaevent.lua test" }
+else
+	print([[Other environements currently untested, may need tweaking]])
+end
+
+package.files = {
+	matchrecursive(
+		"src/*.c",
+		"include/*.h"
+	)
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/src/luaevent.c	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,164 @@
+#include "luaevent.h"
+
+#include <lua.h>
+#include <lauxlib.h>
+
+#define EVENT_BASE_MT "EVENT_BASE_MT"
+#define EVENT_CALLBACK_ARG_MT "EVENT_CALLBACK_ARG_MT"
+#define EVENT_BASE_LOCATION 1
+
+void setEventBase(lua_State* L, struct event_base* base) {
+	struct event_base** pbase = lua_newuserdata(L, sizeof(base));
+	*pbase = base;
+	luaL_getmetatable(L, EVENT_BASE_MT);
+	lua_setmetatable(L, -2);
+	lua_rawseti(L, LUA_ENVIRONINDEX, EVENT_BASE_LOCATION);
+}
+struct event_base* getEventBase(lua_State* L) {
+	struct event_base* base;
+	lua_rawgeti(L, LUA_ENVIRONINDEX, EVENT_BASE_LOCATION);
+	base = *(struct event_base**)lua_topointer(L, -1);
+	lua_pop(L, 1);
+	return base;
+}
+
+void freeCallbackArgs(le_callback* arg) {
+	if(arg->L) {
+		lua_State* L = arg->L;
+		arg->L = NULL;
+		event_del(&arg->ev);
+		luaL_unref(L, LUA_REGISTRYINDEX, arg->callbackRef);
+	}
+}
+/* le_callback is allocated at the beginning of the coroutine in which it
+is used, no need to manually de-allocate */
+
+/* Index for coroutine is fd as integer for *nix, as lightuserdata for Win */
+static void luaevent_callback(int fd, short event, void* p) {
+	le_callback* arg = p;
+	lua_State* L = arg->L;
+	int ret;
+	lua_rawgeti(L, LUA_REGISTRYINDEX, arg->callbackRef);
+	lua_pushinteger(L, event);
+	lua_call(L, 1, 1);
+	ret = lua_tointeger(L, -1);
+	lua_pop(L, 1);
+	if(ret == -1) {
+		freeCallbackArgs(arg);
+	} else {
+		struct event *ev = &arg->ev;
+		int newEvent = ret;
+		if(newEvent != event) { // Need to hook up new event...
+			event_del(ev);
+			event_set(ev, fd, EV_PERSIST | newEvent, luaevent_callback, arg);
+			event_add(ev, NULL);
+		}
+	}
+}
+
+static int luaevent_base_gc(lua_State* L) {
+	struct event_base** pbase = luaL_checkudata(L, 1, EVENT_BASE_MT);
+	if(*pbase) {
+		event_base_free(*pbase);
+		*pbase = NULL;
+	}
+	return 0;
+}
+
+static int luaevent_cb_gc(lua_State* L) {
+	le_callback* arg = luaL_checkudata(L, 1, EVENT_CALLBACK_ARG_MT);
+	freeCallbackArgs(arg);
+	return 0;
+}
+
+int getSocketFd(lua_State* L, int idx) {
+	int fd;
+	luaL_checktype(L, idx, LUA_TUSERDATA);
+	lua_getfield(L, idx, "getfd");
+	if(lua_isnil(L, -1))
+		return luaL_error(L, "Socket type missing 'getfd' method");
+	lua_pushvalue(L, idx);
+	lua_call(L, 1, 1);
+	fd = lua_tointeger(L, -1);
+	lua_pop(L, 1);
+	return fd;
+}
+
+/* Expected to be called at the beginning of the coro that uses it.. 
+Value must be kept until coro is complete....
+*/
+/* sock, event, callback */
+static int luaevent_addevent(lua_State* L) {
+	int fd, event, callbackRef;
+	le_callback* arg;
+	fd = getSocketFd(L, 1);
+	event = luaL_checkinteger(L, 2);
+	luaL_checktype(L, 3, LUA_TFUNCTION);
+	lua_pushvalue(L, 3);
+	callbackRef = luaL_ref(L, LUA_REGISTRYINDEX);
+	arg = lua_newuserdata(L, sizeof(*arg));
+	luaL_getmetatable(L, EVENT_CALLBACK_ARG_MT);
+	lua_setmetatable(L, -2);
+	
+	arg->L = L;
+	arg->callbackRef = callbackRef;
+	/* Setup event... */
+	event_set(&arg->ev, fd, event | EV_PERSIST, luaevent_callback, arg);
+	event_base_set(getEventBase(L), &arg->ev);
+	event_add(&arg->ev, NULL);
+	return 1;
+}
+
+static int luaevent_loop(lua_State* L) {
+	int ret = event_base_loop(getEventBase(L), 0);
+	lua_pushinteger(L, ret);
+	return 1;
+}
+
+static luaL_Reg funcs[] = {
+	{ "addevent", luaevent_addevent },
+	{ "loop", luaevent_loop },
+	{ NULL, NULL }
+};
+
+typedef struct {
+	const char* name;
+	int value;
+} namedInteger;
+
+static namedInteger consts[] = {
+	{"LEAVE", -1},
+	{"EV_READ", EV_READ},
+	{"EV_WRITE", EV_WRITE},
+	{NULL, 0}
+};
+
+void setNamedIntegers(lua_State* L, namedInteger* p) {
+	while(p->name) {
+		lua_pushinteger(L, p->value);
+		lua_setfield(L, -2, p->name);
+		p++;
+	}
+}
+
+/* Verified ok */
+int luaopen_luaevent_core(lua_State* L) {
+	/* Setup environ table */
+	lua_createtable(L, 1, 0);
+	lua_replace(L, LUA_ENVIRONINDEX);
+	/* Setup metatable */
+	luaL_newmetatable(L, EVENT_BASE_MT);
+	lua_pushcfunction(L, luaevent_base_gc);
+	lua_setfield(L, -2, "__gc");
+	lua_pop(L, 1);
+	luaL_newmetatable(L, EVENT_CALLBACK_ARG_MT);
+	lua_pushcfunction(L, luaevent_cb_gc);
+	lua_setfield(L, -2, "__gc");
+	lua_pop(L, 1);
+
+	setEventBase(L, event_init());
+	
+	luaL_register(L, "luaevent.core", funcs);
+	setNamedIntegers(L, consts);
+	return 1;
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/test/test.lua	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,34 @@
+-- Tests Copas with a simple Echo server
+--
+-- Run the test file and the connect to the server by telnet on the used port
+-- to stop the test just send the command "quit"
+
+require"luaevent"
+require"socket"
+local function echoHandler(skt)
+  while true do
+  print(skt)
+    local data,ret = luaevent.receive(skt, 10)
+    print("GOT: ", data, ret)
+    if data == "quit" or ret == 'closed' then
+      break
+    end
+    luaevent.send(skt, data)
+  end
+  print("DONE")
+end
+local function setupHook(thread)
+	if not thread then debug.sethook(function(event) print("TRACE >: ", debug.getinfo(2, 'n').name) end, 'c')
+	else debug.sethook(thread, function(event) print("TRACE ", thread,">: ", debug.getinfo(2, 'n').name) end, 'c') end
+end
+local server = assert(socket.bind("localhost", 20000))
+server:settimeout(0)
+setupHook()
+local coro = coroutine.create
+coroutine.create = function(...)
+	local ret = coro(...)
+	setupHook(ret)
+	return ret
+end
+luaevent.addserver(server, echoHandler)
+luaevent.loop()

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/luaevent/test/testClient.lua	Sun Jun 10 16:01:26 2007 +0000
@@ -0,0 +1,15 @@
+require"luaevent"
+require"socket"
+
+local function func()
+	print("ACTIVATED")
+	local sock = socket.tcp()
+	--sock:
+	sock = luaevent.wrap(sock)
+	print(assert(sock:connect("localhost", 20000)))
+	for i = 1, 100000 do assert(sock:send("Greet me  ")) assert(sock:receive(10)) collectgarbage() end
+end
+
+luaevent.addthread(func)
+
+luaevent.loop()
\ No newline at end of file