moonbridge

view moonbridge_io.c @ 103:4f9e4c6109f4

Different buffering model for I/O writer
author jbe
date Wed Apr 08 17:43:31 2015 +0200 (2015-04-08)
parents 51ff6ad11677
children 5e8eeb5b6c84
line source
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <stdint.h>
5 #include <errno.h>
6 #include <string.h>
7 #include <sys/socket.h>
8 #include <sys/select.h>
9 #include <fcntl.h>
10 #include <netinet/in.h>
11 #include <netinet/tcp.h>
12 #include <sys/types.h>
13 #include <netdb.h>
15 #include <lua.h>
16 #include <lauxlib.h>
17 #include <lualib.h>
19 #define MOONBR_IO_MAXSTRERRORLEN 80
20 #define MOONBR_IO_READBUFLEN 4096
21 #define MOONBR_IO_WRITEBUFLEN 4096
23 #define moonbr_io_errmsg() \
24 char errmsg[MOONBR_IO_MAXSTRERRORLEN]; \
25 strerror_r(errno, errmsg, MOONBR_IO_MAXSTRERRORLEN)
27 #define MOONBR_IO_HANDLE_MT_REGKEY "moonbridge_io_handle"
28 #define MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY "moonbridge_io_handle_public"
30 typedef struct {
31 int fd;
32 int isnetwork;
33 int finished;
34 int closed;
35 int nonblocking;
36 int nopush;
37 int readerr;
38 int readbufcnt;
39 int writeerr;
40 size_t writeleft;
41 #if LUA_VERSION_NUM >= 503
42 lua_Integer writeqin;
43 lua_Integer writeqout;
44 #else
45 int writeqin;
46 int writeqout;
47 #endif
48 size_t writeqoff;
49 int writebufin;
50 int writebufout;
51 char readbuf[MOONBR_IO_READBUFLEN];
52 char writebuf[MOONBR_IO_WRITEBUFLEN];
53 } moonbr_io_handle_t;
55 static void moonbr_io_handle_set_nonblocking(lua_State *L, moonbr_io_handle_t *handle, int nonblocking) {
56 int flags;
57 if (handle->nonblocking == nonblocking) return;
58 flags = fcntl(handle->fd, F_GETFL, 0);
59 if (flags == -1) {
60 moonbr_io_errmsg();
61 close(handle->fd);
62 handle->fd = -1;
63 handle->closed = 1;
64 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);
65 }
66 if (nonblocking) flags |= O_NONBLOCK;
67 else flags &= ~O_NONBLOCK;
68 if (fcntl(handle->fd, F_SETFL, flags) == -1) {
69 moonbr_io_errmsg();
70 close(handle->fd);
71 handle->fd = -1;
72 handle->closed = 1;
73 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);
74 }
75 handle->nonblocking = nonblocking;
76 }
78 static void moonbr_io_handle_set_linger(lua_State *L, moonbr_io_handle_t *handle, int timeout) {
79 struct linger lingerval = { 0, };
80 if (!handle->isnetwork) return;
81 if (timeout >= 0) {
82 lingerval.l_onoff = 1;
83 lingerval.l_linger = timeout;
84 }
85 if (setsockopt(handle->fd, SOL_SOCKET, SO_LINGER, &lingerval, sizeof(lingerval))) {
86 moonbr_io_errmsg();
87 close(handle->fd);
88 handle->fd = -1;
89 handle->closed = 1;
90 luaL_error(L, "Unexpected error while setting SO_LINGER with setsockopt: %s", errmsg);
91 }
92 }
94 static void moonbr_io_handle_set_nopush(lua_State *L, moonbr_io_handle_t *handle, int nopush) {
95 #if defined(TCP_NOPUSH) || defined(TCP_CORK)
96 if (!handle->isnetwork || handle->nopush == nopush) return;
97 #if defined(TCP_NOPUSH)
98 if (setsockopt(handle->fd, IPPROTO_TCP, TCP_NOPUSH, &nopush, sizeof(nopush))) {
99 moonbr_io_errmsg();
100 close(handle->fd);
101 handle->fd = -1;
102 handle->closed = 1;
103 luaL_error(L, "Unexpected error while setting TCP_NOPUSH with setsockopt: %s", errmsg);
104 }
105 #elif defined(TCP_CORK)
106 if (setsockopt(handle->fd, IPPROTO_TCP, TCP_CORK, &nopush, sizeof(nopush))) {
107 moonbr_io_errmsg();
108 close(handle->fd);
109 handle->fd = -1;
110 handle->closed = 1;
111 luaL_error(L, "Unexpected error while setting TCP_CORK with setsockopt: %s", errmsg);
112 }
113 #endif
114 handle->nopush = nopush;
115 #else
116 #warning Neither TCP_NOPUSH nor TCP_CORK is available
117 #endif
118 }
120 static int moonbr_io_read_impl(lua_State *L, int nonblocking, int drain) {
121 moonbr_io_handle_t *handle;
122 lua_Integer maxread;
123 const char *terminatorstr;
124 size_t terminatorlen;
125 char terminator;
126 luaL_Buffer luabuf;
127 size_t luabufcnt = 0;
128 int endcnt;
129 char *terminatorpos;
130 ssize_t bytesread;
131 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
132 maxread = luaL_optinteger(L, 2, 0);
133 terminatorstr = luaL_optlstring(L, 3, "", &terminatorlen);
134 if (terminatorlen) {
135 luaL_argcheck(L, terminatorlen == 1, 3, "single byte expected");
136 terminator = terminatorstr[0];
137 }
138 lua_settop(L, 1); /* return handle on drain, terminator string may be garbage collected */
139 if (handle->closed) luaL_error(L, "Attempt to read from a closed I/O handle");
140 if (handle->fd < 0) goto moonbr_io_read_impl_eof; /* fake EOF to simulate shutdown */
141 if (handle->readerr) {
142 lua_pushnil(L);
143 lua_pushliteral(L, "Previous read error");
144 return 2;
145 }
146 moonbr_io_handle_set_nonblocking(L, handle, nonblocking);
147 if (!drain) luaL_buffinit(L, &luabuf);
148 while (1) {
149 endcnt = -1;
150 if (maxread > 0 && handle->readbufcnt >= maxread - luabufcnt) {
151 endcnt = maxread - luabufcnt;
152 } else if (terminatorlen) {
153 terminatorpos = memchr(handle->readbuf, terminator, handle->readbufcnt);
154 if (terminatorpos) endcnt = 1 + (terminatorpos - handle->readbuf);
155 }
156 if (endcnt >= 0) {
157 if (!drain) {
158 luaL_addlstring(&luabuf, handle->readbuf, endcnt);
159 luaL_pushresult(&luabuf);
160 } else {
161 luabufcnt += handle->readbufcnt;
162 lua_pushinteger(L, luabufcnt);
163 }
164 handle->readbufcnt -= endcnt;
165 memmove(handle->readbuf, handle->readbuf + endcnt, handle->readbufcnt);
166 return 1;
167 }
168 if (!drain) luaL_addlstring(&luabuf, handle->readbuf, handle->readbufcnt);
169 luabufcnt += handle->readbufcnt;
170 handle->readbufcnt = 0;
171 do {
172 bytesread = read(handle->fd, handle->readbuf, MOONBR_IO_READBUFLEN);
173 } while (bytesread < 0 && (errno == EINTR));
174 if (bytesread == 0 || (nonblocking && bytesread < 0 && (errno == EAGAIN || errno == EWOULDBLOCK))) break;
175 if (bytesread < 0) {
176 moonbr_io_errmsg();
177 handle->readerr = 1;
178 lua_pushnil(L);
179 lua_pushstring(L, errmsg);
180 return 2;
181 }
182 handle->readbufcnt += bytesread;
183 }
184 if (!drain) {
185 luaL_addlstring(&luabuf, handle->readbuf, handle->readbufcnt);
186 luaL_pushresult(&luabuf);
187 }
188 luabufcnt += handle->readbufcnt;
189 handle->readbufcnt = 0;
190 if (!drain) {
191 if (!luabufcnt && bytesread == 0) {
192 moonbr_io_read_impl_eof:
193 lua_pushboolean(L, 0);
194 lua_pushliteral(L, "End of file");
195 return 2;
196 }
197 } else {
198 if (!luabufcnt && bytesread == 0) lua_pushboolean(L, 1);
199 else lua_pushboolean(L, luabufcnt);
200 }
201 return 1;
202 }
204 static int moonbr_io_read(lua_State *L) {
205 return moonbr_io_read_impl(L, 0, 0);
206 }
208 static int moonbr_io_read_nb(lua_State *L) {
209 return moonbr_io_read_impl(L, 1, 0);
210 }
212 static int moonbr_io_drain(lua_State *L) {
213 return moonbr_io_read_impl(L, 0, 1);
214 }
216 static int moonbr_io_drain_nb(lua_State *L) {
217 return moonbr_io_read_impl(L, 1, 1);
218 }
220 static int moonbr_io_write_impl(lua_State *L, int nonblocking, int flush) {
221 moonbr_io_handle_t *handle;
222 int i, top;
223 const char *str;
224 size_t strlen;
225 ssize_t written;
226 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
227 if (handle->closed) luaL_error(L, "Attempt to write to a closed I/O handle");
228 if (handle->finished) luaL_error(L, "Attempt to write to a finished I/O handle");
229 if (handle->writeerr) {
230 lua_pushnil(L);
231 lua_pushliteral(L, "Previous write error");
232 return 2;
233 }
234 handle->writeerr = 1;
235 moonbr_io_handle_set_nonblocking(L, handle, nonblocking);
236 top = lua_gettop(L);
237 lua_getuservalue(L, 1);
238 lua_getfield(L, -1, "writequeue");
239 for (i=2; i<=top; i++) {
240 luaL_checklstring(L, i, &strlen);
241 lua_pushvalue(L, i);
242 lua_rawseti(L, -2, handle->writeqin++);
243 handle->writeleft += strlen;
244 }
245 while (handle->writeqout != handle->writeqin) {
246 lua_rawgeti(L, -1, handle->writeqout);
247 str = lua_tolstring(L, -1, &strlen);
248 while (handle->writeqoff < strlen) {
249 if (
250 strlen - handle->writeqoff <=
251 MOONBR_IO_WRITEBUFLEN - handle->writebufin
252 ) {
253 memcpy(
254 handle->writebuf + handle->writebufin,
255 str + handle->writeqoff,
256 strlen - handle->writeqoff
257 );
258 handle->writebufin += strlen - handle->writeqoff;
259 break;
260 } else {
261 moonbr_io_handle_set_nopush(L, handle, 1);
262 memcpy(
263 handle->writebuf + handle->writebufin,
264 str + handle->writeqoff,
265 MOONBR_IO_WRITEBUFLEN - handle->writebufin
266 );
267 handle->writeqoff += MOONBR_IO_WRITEBUFLEN - handle->writebufin;
268 while (handle->writebufout < MOONBR_IO_WRITEBUFLEN) {
269 written = write(
270 handle->fd,
271 handle->writebuf + handle->writebufout,
272 MOONBR_IO_WRITEBUFLEN - handle->writebufout
273 );
274 if (written < 0) {
275 if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {
276 goto moonbr_io_write_impl_block;
277 } else if (errno != EINTR) {
278 moonbr_io_errmsg();
279 lua_pushnil(L);
280 lua_pushstring(L, errmsg);
281 return 2;
282 }
283 } else {
284 handle->writebufout += written;
285 handle->writeleft -= written;
286 }
287 }
288 handle->writebufin = 0;
289 handle->writebufout = 0;
290 }
291 }
292 handle->writeqoff = 0;
293 lua_pop(L, 1);
294 lua_pushnil(L);
295 lua_rawseti(L, -2, handle->writeqout++);
296 }
297 if (flush) {
298 moonbr_io_handle_set_nopush(L, handle, 0);
299 while (handle->writebufout < handle->writebufin) {
300 written = write(
301 handle->fd,
302 handle->writebuf + handle->writebufout,
303 handle->writebufin - handle->writebufout
304 );
305 if (written < 0) {
306 if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {
307 goto moonbr_io_write_impl_block;
308 } else if (errno != EINTR) {
309 moonbr_io_errmsg();
310 lua_pushnil(L);
311 lua_pushstring(L, errmsg);
312 return 2;
313 }
314 } else {
315 handle->writebufout += written;
316 handle->writeleft -= written;
317 }
318 }
319 handle->writebufin = 0;
320 handle->writebufout = 0;
321 }
322 if (nonblocking) lua_pushinteger(L, 0);
323 else lua_pushvalue(L, 1);
324 handle->writeerr = 0;
325 return 1;
326 moonbr_io_write_impl_block:
327 lua_pushinteger(L, handle->writeleft);
328 handle->writeerr = 0;
329 return 1;
330 }
332 static int moonbr_io_write(lua_State *L) {
333 return moonbr_io_write_impl(L, 0, 0);
334 }
336 static int moonbr_io_write_nb(lua_State *L) {
337 return moonbr_io_write_impl(L, 1, 0);
338 }
340 static int moonbr_io_flush(lua_State *L) {
341 return moonbr_io_write_impl(L, 0, 1);
342 }
344 static int moonbr_io_flush_nb(lua_State *L) {
345 return moonbr_io_write_impl(L, 1, 1);
346 }
348 static int moonbr_io_finish(lua_State *L) {
349 moonbr_io_handle_t *handle;
350 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
351 if (handle->closed) luaL_error(L, "Attempt to finish a closed I/O handle");
352 if (handle->finished) luaL_error(L, "Attempt to finish a finished I/O handle");
353 if (handle->writeleft) {
354 lua_pushcfunction(L, moonbr_io_flush);
355 lua_pushvalue(L, 1);
356 lua_call(L, 1, 2);
357 if (!lua_toboolean(L, -2)) {
358 handle->finished = 1;
359 return 2;
360 }
361 }
362 handle->finished = 1;
363 if (handle->isnetwork) {
364 if (shutdown(handle->fd, SHUT_WR)) {
365 moonbr_io_errmsg();
366 lua_pushnil(L);
367 lua_pushstring(L, errmsg);
368 return 2;
369 }
370 } else {
371 if (close(handle->fd)) {
372 moonbr_io_errmsg();
373 handle->fd = -1;
374 lua_pushnil(L);
375 lua_pushstring(L, errmsg);
376 return 2;
377 }
378 handle->fd = -1; /* fake EOF on read */
379 }
380 lua_pushboolean(L, 1);
381 return 1;
382 }
384 static int moonbr_io_close_impl(lua_State *L, int reset) {
385 moonbr_io_handle_t *handle;
386 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
387 if (handle->closed) luaL_error(L, "Attempt to close a closed I/O handle");
388 if (!reset) {
389 if (handle->writeleft) {
390 lua_pushcfunction(L, moonbr_io_flush);
391 lua_pushvalue(L, 1);
392 lua_call(L, 1, 2);
393 if (!lua_toboolean(L, -2)) {
394 close(handle->fd);
395 handle->fd = -1;
396 return 2;
397 }
398 }
399 moonbr_io_handle_set_linger(L, handle, -1);
400 }
401 if (handle->fd >= 0) {
402 if (close(handle->fd)) {
403 moonbr_io_errmsg();
404 handle->fd = -1;
405 lua_pushnil(L);
406 lua_pushstring(L, errmsg);
407 return 2;
408 }
409 }
410 handle->fd = -1;
411 lua_pushboolean(L, 1);
412 return 1;
414 }
416 static int moonbr_io_close(lua_State *L) {
417 return moonbr_io_close_impl(L, 0);
418 }
420 static int moonbr_io_reset(lua_State *L) {
421 return moonbr_io_close_impl(L, 1);
422 }
424 static int moonbr_io_gc(lua_State *L) {
425 moonbr_io_handle_t *handle;
426 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
427 if (handle->fd >= 0) {
428 lua_pushcfunction(L, moonbr_io_close);
429 lua_pushvalue(L, 1);
430 lua_pushinteger(L, 0);
431 lua_call(L, 2, 0);
432 }
433 return 0;
434 }
436 void moonbr_io_closehandle(lua_State *L, int idx, int reset) {
437 moonbr_io_handle_t *handle;
438 handle = luaL_checkudata(L, idx, MOONBR_IO_HANDLE_MT_REGKEY);
439 if (!handle->closed) {
440 lua_pushcfunction(L, reset ? moonbr_io_reset : moonbr_io_close);
441 lua_pushvalue(L, idx < 0 ? idx-1 : idx);
442 lua_call(L, 1, 0);
443 }
444 }
446 void moonbr_io_pushhandle(lua_State *L, int fd, int isnetwork) {
447 moonbr_io_handle_t *handle;
448 handle = lua_newuserdata(L, sizeof(moonbr_io_handle_t));
449 handle->fd = fd;
450 handle->isnetwork = isnetwork;
451 handle->finished = 0;
452 handle->closed = 0;
453 handle->nonblocking = -1;
454 handle->nopush = -1;
455 handle->readerr = 0;
456 handle->readbufcnt = 0;
457 handle->writeerr = 0;
458 handle->writeleft = 0;
459 handle->writeqin = 0;
460 handle->writeqout = 0;
461 handle->writeqoff = 0;
462 handle->writebufin = 0;
463 handle->writebufout = 0;
464 moonbr_io_handle_set_linger(L, handle, 0);
465 luaL_getmetatable(L, MOONBR_IO_HANDLE_MT_REGKEY);
466 lua_setmetatable(L, -2);
467 lua_newtable(L); // uservalue
468 lua_newtable(L);
469 lua_setfield(L, -2, "writequeue");
470 lua_newtable(L); // public
471 luaL_getmetatable(L, MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY);
472 lua_setmetatable(L, -2);
473 lua_setfield(L, -2, "public");
474 lua_setuservalue(L, -2);
475 }
477 static int moonbr_io_handleindex(lua_State *L) {
478 luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
479 lua_getuservalue(L, 1);
480 lua_getfield(L, -1, "public");
481 lua_pushvalue(L, 2);
482 lua_gettable(L, -2);
483 return 1;
484 }
486 static int moonbr_io_handlenewindex(lua_State *L) {
487 luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);
488 lua_getuservalue(L, 1);
489 lua_getfield(L, -1, "public");
490 lua_pushvalue(L, 2);
491 lua_pushvalue(L, 3);
492 lua_settable(L, -3);
493 return 0;
494 }
496 static int moonbr_io_tcpconnect_impl(lua_State *L, int nonblocking) {
497 const char *host, *port;
498 struct addrinfo hints = { 0, };
499 struct addrinfo *res, *addrinfo;
500 int errcode;
501 int sock;
502 host = luaL_checkstring(L, 1);
503 port = luaL_checkstring(L, 2);
504 hints.ai_family = AF_UNSPEC;
505 hints.ai_socktype = SOCK_STREAM;
506 hints.ai_protocol = IPPROTO_TCP;
507 hints.ai_flags = AI_ADDRCONFIG;
508 errcode = getaddrinfo(host, port, &hints, &res);
509 if (errcode) {
510 freeaddrinfo(res);
511 if (errcode == EAI_SYSTEM) {
512 moonbr_io_errmsg();
513 lua_pushnil(L);
514 lua_pushfstring(L, "%s: %s", gai_strerror(errcode), errmsg);
515 } else {
516 lua_pushnil(L);
517 lua_pushstring(L, gai_strerror(errcode));
518 }
519 return 2;
520 }
521 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {
522 if (addrinfo->ai_family == PF_INET6) goto moonbr_io_tcpconnect_found;
523 }
524 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {
525 if (addrinfo->ai_family == PF_INET) goto moonbr_io_tcpconnect_found;
526 }
527 addrinfo = res;
528 moonbr_io_tcpconnect_found:
529 sock = socket(
530 addrinfo->ai_family,
531 addrinfo->ai_socktype | SOCK_CLOEXEC | (nonblocking ? SOCK_NONBLOCK : 0),
532 addrinfo->ai_protocol
533 );
534 freeaddrinfo(res);
535 if (sock < 0) {
536 moonbr_io_errmsg();
537 lua_pushnil(L);
538 lua_pushstring(L, errmsg);
539 }
540 if (connect(sock, addrinfo->ai_addr, addrinfo->ai_addrlen)) {
541 if (!nonblocking && errno == EINTR) {
542 moonbr_io_errmsg();
543 close(sock);
544 lua_pushnil(L);
545 lua_pushstring(L, errmsg);
546 return 2;
547 } else if (!(nonblocking && (errno == EINPROGRESS || errno == EINTR))) {
548 moonbr_io_errmsg();
549 lua_pushnil(L);
550 lua_pushstring(L, errmsg);
551 return 2;
552 }
553 }
554 moonbr_io_pushhandle(L, sock, 1);
555 return 1;
556 }
558 static int moonbr_io_tcpconnect(lua_State *L) {
559 return moonbr_io_tcpconnect_impl(L, 0);
560 }
562 static int moonbr_io_tcpconnect_nb(lua_State *L) {
563 return moonbr_io_tcpconnect_impl(L, 1);
564 }
566 static const struct luaL_Reg moonbr_io_handle_methods[] = {
567 {"read", moonbr_io_read},
568 {"read_nb", moonbr_io_read_nb},
569 {"drain", moonbr_io_drain},
570 {"drain_nb", moonbr_io_drain_nb},
571 {"write", moonbr_io_write},
572 {"write_nb", moonbr_io_write_nb},
573 {"flush", moonbr_io_flush},
574 {"flush_nb", moonbr_io_flush_nb},
575 {"finish", moonbr_io_finish},
576 {"close", moonbr_io_close},
577 {"reset", moonbr_io_reset},
578 {NULL, NULL}
579 };
581 static const struct luaL_Reg moonbr_io_handle_metamethods[] = {
582 {"__index", moonbr_io_handleindex},
583 {"__newindex", moonbr_io_handlenewindex},
584 {"__gc", moonbr_io_gc},
585 {NULL, NULL}
586 };
588 static const struct luaL_Reg moonbr_io_module_funcs[] = {
589 {"tcpconnect", moonbr_io_tcpconnect},
590 {"tcpconnect_nb", moonbr_io_tcpconnect_nb},
591 {NULL, NULL}
592 };
594 int luaopen_moonbridge_io(lua_State *L) {
596 lua_newtable(L); // module
598 lua_newtable(L); // public metatable
599 lua_newtable(L); // handle methods
600 luaL_setfuncs(L, moonbr_io_handle_methods, 0);
601 lua_pushvalue(L, -1);
602 lua_setfield(L, -4, "handle");
603 lua_setfield(L, -2, "__index");
604 lua_setfield(L, LUA_REGISTRYINDEX, MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY);
606 lua_newtable(L); // handle metatable
607 luaL_setfuncs(L, moonbr_io_handle_metamethods, 0);
608 lua_setfield(L, LUA_REGISTRYINDEX, MOONBR_IO_HANDLE_MT_REGKEY);
610 luaL_setfuncs(L, moonbr_io_module_funcs, 0);
611 return 1;
613 }

Impressum / About Us