io: rework stream:sendfile()

5 files changed, 79 insertions, 214 deletions

diff --git a/lem/hathaway.lua b/lem/hathaway.lua
index 4e327ff..0427f86 100644
--- a/lem/hathaway.lua
+++ b/lem/hathaway.lua
@@ -268,7 +268,7 @@ do
 			local headers = res.headers
 			local file, close = res.file, false
 			if type(file) == 'string' then
-				file, err = io.sendfile(file)
+				file, err = io.open(file)
 				if file then
 					close = true
 				else
@@ -329,7 +329,7 @@ do
 
 			if method ~= 'HEAD' then
 				if file then
-					ok, err = client:sendfile(file)
+					ok, err = client:sendfile(file, headers['Content-Length'])
 					if close then file:close() end
 				else
 					ok, err = client:write(concat(res))
diff --git a/lem/io/core.c b/lem/io/core.c
index f46319e..6411094 100644
--- a/lem/io/core.c
+++ b/lem/io/core.c
@@ -62,7 +62,6 @@ io_strerror(lua_State *T, int err)
 	return 2;
 }
 
-#include "sendfile.c"
 #include "file.c"
 #include "stream.c"
 #include "server.c"
@@ -241,28 +240,6 @@ luaopen_lem_io_core(lua_State *L)
 	/* create module table */
 	lua_newtable(L);
 
-	/* create metatable for sendfile objects */
-	lua_newtable(L);
-	/* mt.__index = mt */
-	lua_pushvalue(L, -1);
-	lua_setfield(L, -2, "__index");
-	/* mt.__gc = <sendfile_gc> */
-	lua_pushcfunction(L, sendfile_gc);
-	lua_setfield(L, -2, "__gc");
-	/* mt.close = <sendfile_close> */
-	lua_pushcfunction(L, sendfile_close);
-	lua_setfield(L, -2, "close");
-	/* mt.size = <sendfile_size> */
-	lua_pushcfunction(L, sendfile_size);
-	lua_setfield(L, -2, "size");
-	/* insert table */
-	lua_setfield(L, -2, "SendFile");
-
-	/* insert sendfile function */
-	lua_getfield(L, -1, "SendFile"); /* upvalue 1 = SendFile */
-	lua_pushcclosure(L, sendfile_open, 1);
-	lua_setfield(L, -2, "sendfile");
-
 	/* create File metatable */
 	lua_newtable(L);
 	/* mt.__index = mt */
diff --git a/lem/io/file.c b/lem/io/file.c
index 2015f31..eae8b14 100644
--- a/lem/io/file.c
+++ b/lem/io/file.c
@@ -35,6 +35,11 @@ struct file {
 			off_t offset;
 			int whence;
 		} seek;
+		struct {
+			struct stream *stream;
+			off_t size;
+			off_t offset;
+		} sendfile;
 	};
 	struct lem_inputbuf buf;
 };
diff --git a/lem/io/sendfile.c b/lem/io/sendfile.c
deleted file mode 100644
index bd7d246..0000000
--- a/lem/io/sendfile.c
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * This file is part of LEM, a Lua Event Machine.
- * Copyright 2011-2012 Emil Renner Berthing
- *
- * LEM is free software: you can redistribute it and/or
- * modify it under the terms of the GNU General Public License as
- * published by the Free Software Foundation, either version 3 of
- * the License, or (at your option) any later version.
- *
- * LEM is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with LEM.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-struct lem_sendfile {
-	int fd;
-	off_t size;
-};
-
-static int
-sendfile_open(lua_State *T)
-{
-	const char *path = luaL_checkstring(T, 1);
-	int fd;
-	struct stat buf;
-	struct lem_sendfile *f;
-
-	fd = open(path, O_RDONLY | O_NONBLOCK);
-	if (fd < 0)
-		return io_strerror(T, errno);
-
-	if (fstat(fd, &buf)) {
-		int err = errno;
-		close(fd);
-		return io_strerror(T, err);
-	}
-
-	/* create userdata and set the metatable */
-	f = lua_newuserdata(T, sizeof(struct lem_sendfile));
-	lua_pushvalue(T, lua_upvalueindex(1));
-	lua_setmetatable(T, -2);
-
-	/* initialize userdata */
-	f->fd = fd;
-	f->size = buf.st_size;
-
-	return 1;
-}
-
-static int
-sendfile_gc(lua_State *T)
-{
-	struct lem_sendfile *f = lua_touserdata(T, 1);
-
-	if (f->fd >= 0)
-		close(f->fd);
-
-	return 0;
-}
-
-static int
-sendfile_close(lua_State *T)
-{
-	struct lem_sendfile *f;
-	int ret;
-
-	luaL_checktype(T, 1, LUA_TUSERDATA);
-	f = lua_touserdata(T, 1);
-	if (f->fd < 0)
-		return io_closed(T);
-
-	ret = close(f->fd);
-	f->fd = -1;
-	if (ret)
-		return io_strerror(T, errno);
-
-	lua_pushboolean(T, 1);
-	return 1;
-}
-
-static int
-sendfile_size(lua_State *T)
-{
-	struct lem_sendfile *f;
-
-	luaL_checktype(T, 1, LUA_TUSERDATA);
-	f = lua_touserdata(T, 1);
-	if (f->fd < 0)
-		return io_closed(T);
-
-	lua_pushnumber(T, (lua_Number)f->size);
-	return 1;
-}
diff --git a/lem/io/stream.c b/lem/io/stream.c
index f84c91d..4861860 100644
--- a/lem/io/stream.c
+++ b/lem/io/stream.c
@@ -302,94 +302,78 @@ stream_uncork(lua_State *T)
 	return stream_setcork(T, 0);
 }
 
-struct sendfile {
-	struct lem_sendfile *file;
-	off_t offset;
-};
-
-static int
-stream__sendfile(lua_State *T, struct stream *s, struct sendfile *sf)
+static void
+stream_sendfile_work(struct lem_async *a)
 {
-#ifdef __FreeBSD__
-	int ret;
-
-	do {
-		size_t count;
-		off_t written = 0;
-
-		count = sf->file->size - sf->offset;
-
-		if (count == 0) {
-			lua_settop(T, 0);
-			lua_pushboolean(T, 1);
-			return 1;
-		}
+	struct file *f = (struct file *)a;
+	struct stream *s = f->sendfile.stream;
+
+	/* make socket blocking */
+	if (fcntl(s->w.fd, F_SETFL, 0)) {
+		f->ret = errno;
+		close(s->w.fd);
+		s->r.fd = s->w.fd = -1;
+		return;
+	}
 
-		ret = sendfile(sf->file->fd, s->w.fd,
-				sf->offset, count,
-				NULL, &written, 0);
-		lem_debug("wrote = %ld bytes", written);
-		sf->offset += written;
-	} while (ret >= 0);
+#ifdef __FreeBSD__
+	off_t written;
+	int ret = sendfile(f->fd, s->w.fd,
+			f->sendfile.offset, f->sendfile.size,
+			NULL, &written, SF_SYNC);
+	if (ret == 0) {
+		f->ret = 0;
+		f->sendfile.size = written;
+	} else
+		f->ret = errno;
+	lem_debug("wrote = %ld bytes", written);
 #else
 #ifdef __APPLE__
-	int ret;
-
-	do {
-		off_t count = sf->file->size - sf->offset;
-
-		if (count == 0) {
-			lua_settop(T, 0);
-			lua_pushboolean(T, 1);
-			return 1;
-		}
-
-		ret = sendfile(sf->file->fd, s->w.fd,
-		               sf->offset, &count, NULL, 0);
-		lem_debug("wrote = %lld bytes", count);
-		sf->offset += count;
-	} while (ret >= 0);
+	int ret = sendfile(f->fd, s->w.fd,
+			f->sendfile.offset, &f->sendfile.size,
+			NULL, 0);
+	if (ret == 0)
+		f->ret = 0;
+	else
+		f->ret = errno;
+	lem_debug("wrote = %lld bytes", f->sendfile.size);
 #else
-	ssize_t ret;
-
-	do {
-		size_t count = sf->file->size - sf->offset;
-
-		if (count == 0) {
-			lua_settop(T, 0);
-			lua_pushboolean(T, 1);
-			return 1;
-		}
-
-		ret = sendfile(s->w.fd, sf->file->fd,
-				&sf->offset,
-				count);
-		lem_debug("wrote = %ld bytes", ret);
-	} while (ret >= 0);
+	ssize_t ret = sendfile(s->w.fd, f->fd,
+			&f->sendfile.offset, f->sendfile.size);
+	if (ret >= 0) {
+		f->ret = 0;
+		f->sendfile.size = ret;
+	} else
+		f->ret = errno;
+	lem_debug("wrote = %ld bytes", ret);
 #endif
 #endif
 
-	if (errno == EAGAIN)
-		return 0;
-
-	return io_strerror(T, errno);
+	/* make socket non-blocking again */
+	if (fcntl(s->w.fd, F_SETFL, O_NONBLOCK)) {
+		f->ret = errno;
+		close(s->w.fd);
+		s->r.fd = s->w.fd = -1;
+		return;
+	}
 }
 
 static void
-stream_sendfile_cb(EV_P_ struct ev_io *w, int revents)
+stream_sendfile_reap(struct lem_async *a)
 {
-	struct stream *s = STREAM_FROM_WATCH(w, w);
-	lua_State *T = s->w.data;
-	struct sendfile *sf = lua_touserdata(T, 3);
+	struct file *f = (struct file *)a;
+	struct stream *s = f->sendfile.stream;
+	lua_State *T = f->a.T;
 	int ret;
 
-	(void)revents;
-
-	ret = stream__sendfile(T, s, sf);
-	if (ret == 0)
-		return;
+	if (f->ret == 0) {
+		lua_pushnumber(T, f->sendfile.size);
+		ret = 1;
+	} else {
+		ret = io_strerror(T, f->ret);
+	}
 
-	ev_io_stop(EV_A_ &s->w);
+	f->a.T = NULL;
 	s->w.data = NULL;
 	lem_queue(T, ret);
 }
@@ -398,14 +382,14 @@ static int
 stream_sendfile(lua_State *T)
 {
 	struct stream *s;
-	struct lem_sendfile *f;
-	struct sendfile *sf;
+	struct file *f;
+	off_t size;
 	off_t offset;
-	int ret;
 
 	luaL_checktype(T, 1, LUA_TUSERDATA);
 	luaL_checktype(T, 2, LUA_TUSERDATA);
-	offset = (off_t)luaL_optnumber(T, 3, 0);
+	size = (off_t)luaL_checknumber(T, 3);
+	offset = (off_t)luaL_optnumber(T, 4, 0);
 
 	s = lua_touserdata(T, 1);
 	if (s->w.fd < 0)
@@ -419,24 +403,20 @@ stream_sendfile(lua_State *T)
 		lua_pushliteral(T, "file closed");
 		return 2;
 	}
+	if (f->a.T != NULL) {
+		lua_pushnil(T);
+		lua_pushliteral(T, "file busy");
+		return 2;
+	}
 
-	if (offset > f->size)
-		return luaL_error(T, "offset too big");
+	s->w.data = T;
+	f->sendfile.stream = s;
+	f->sendfile.size = size;
+	f->sendfile.offset = offset;
+	lem_async_do(&f->a, T, stream_sendfile_work, stream_sendfile_reap);
 
 	lua_settop(T, 2);
-	sf = lua_newuserdata(T, sizeof(struct sendfile));
-	sf->file = f;
-	sf->offset = offset;
-
-	ret = stream__sendfile(T, s, sf);
-	if (ret > 0)
-		return ret;
-
-	lem_debug("yielding");
-	s->w.data = T;
-	s->w.cb = stream_sendfile_cb;
-	ev_io_start(LEM_ &s->w);
-	return lua_yield(T, 3);
+	return lua_yield(T, 2);
 }
 
 static int