moonbridge: 293e666a41e0 moonbridge

moonbridge

view moonbridge_io.c @ 132:293e666a41e0

Revised flushing behavior of write_nb(...)

author	jbe
date	Thu Apr 16 04:24:21 2015 +0200 (2015-04-16)
parents	df08e63dc44b
children	891cdc4b43eb

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/un.h>

9 #include <sys/select.h>

10 #include <fcntl.h>

11 #include <netinet/in.h>

12 #include <netinet/tcp.h>

13 #include <arpa/inet.h>

14 #include <sys/types.h>

15 #include <sys/stat.h>

16 #include <netdb.h>

17 #include <signal.h>

18 #include <time.h>

20 #include <lua.h>

21 #include <lauxlib.h>

22 #include <lualib.h>

24 #define MOONBR_IO_MAXSTRERRORLEN 80

25 #define MOONBR_IO_READBUFLEN 4096

26 #define MOONBR_IO_WRITEBUFLEN 4096

28 #define MOONBR_IO_LISTEN_BACKLOG 1024

30 #define moonbr_io_errmsg() \

31 char errmsg[MOONBR_IO_MAXSTRERRORLEN]; \

32 strerror_r(errno, errmsg, MOONBR_IO_MAXSTRERRORLEN)

34 #define MOONBR_IO_HANDLE_MT_REGKEY "moonbridge_io_handle"

35 #define MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY "moonbridge_io_handle_public"

36 #define MOONBR_IO_LISTENER_MT_REGKEY "moonbridge_io_listener"

38 typedef struct {

39 int fd;

40 int issock;

41 sa_family_t addrfam;

42 int finished;

43 int closed;

44 int nonblocking;

45 int nopush;

46 int readerr;

47 int readbufin;

48 int readbufout;

49 int writeerr;

50 size_t writeleft;

51 size_t flushedleft;

52 #if LUA_VERSION_NUM >= 503

53 lua_Integer writeqin;

54 lua_Integer writeqout;

55 #else

56 int writeqin;

57 int writeqout;

58 #endif

59 size_t writeqoff;

60 int writebufin;

61 int writebufout;

62 char readbuf[MOONBR_IO_READBUFLEN];

63 char writebuf[MOONBR_IO_WRITEBUFLEN];

64 } moonbr_io_handle_t;

66 typedef struct {

67 int fd;

68 sa_family_t addrfam;

69 int nonblocking;

70 } moonbr_io_listener_t;

72 static void moonbr_io_handle_set_nonblocking(lua_State *L, moonbr_io_handle_t *handle, int nonblocking) {

73 int flags;

74 if (handle->nonblocking == nonblocking) return;

75 flags = fcntl(handle->fd, F_GETFL, 0);

76 if (flags == -1) {

77 moonbr_io_errmsg();

78 close(handle->fd);

79 handle->fd = -1;

80 handle->closed = 1;

81 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);

82 }

83 if (nonblocking) flags |= O_NONBLOCK;

84 else flags &= ~O_NONBLOCK;

85 if (fcntl(handle->fd, F_SETFL, flags) == -1) {

86 moonbr_io_errmsg();

87 close(handle->fd);

88 handle->fd = -1;

89 handle->closed = 1;

90 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);

91 }

92 handle->nonblocking = nonblocking;

93 }

95 static void moonbr_io_handle_set_linger(lua_State *L, moonbr_io_handle_t *handle, int timeout) {

96 struct linger lingerval = { 0, };

97 if (!handle->issock) return;

98 if (timeout >= 0) {

99 lingerval.l_onoff = 1;

100 lingerval.l_linger = timeout;

101 }

102 if (setsockopt(handle->fd, SOL_SOCKET, SO_LINGER, &lingerval, sizeof(lingerval))) {

103 moonbr_io_errmsg();

104 close(handle->fd);

105 handle->fd = -1;

106 handle->closed = 1;

107 luaL_error(L, "Unexpected error while setting SO_LINGER with setsockopt: %s", errmsg);

108 }

109 }

110

111 static void moonbr_io_handle_set_nopush(lua_State *L, moonbr_io_handle_t *handle, int nopush) {

112 #if defined(TCP_NOPUSH) || defined(TCP_CORK)

113 if (

114 !(handle->addrfam == AF_INET6 || handle->addrfam == AF_INET) ||

115 handle->nopush == nopush

116 ) return;

117 #if defined(TCP_NOPUSH)

118 if (setsockopt(handle->fd, IPPROTO_TCP, TCP_NOPUSH, &nopush, sizeof(nopush))) {

119 #elif defined(TCP_CORK)

120 if (setsockopt(handle->fd, IPPROTO_TCP, TCP_CORK, &nopush, sizeof(nopush))) {

121 #endif

122 moonbr_io_errmsg();

123 close(handle->fd);

124 handle->fd = -1;

125 handle->closed = 1;

126 #if defined(TCP_NOPUSH)

127 luaL_error(L, "Unexpected error while setting TCP_NOPUSH with setsockopt: %s", errmsg);

128 #elif defined(TCP_CORK)

129 luaL_error(L, "Unexpected error while setting TCP_CORK with setsockopt: %s", errmsg);

130 #endif

131 }

132 handle->nopush = nopush;

133 #else

134 #warning Neither TCP_NOPUSH nor TCP_CORK is available

135 #endif

136 }

137

138 static int moonbr_io_read_impl(lua_State *L, int nonblocking, int drain) {

139 moonbr_io_handle_t *handle;

140 lua_Integer maxread;

141 const char *terminatorstr;

142 size_t terminatorlen;

143 char terminator;

144 luaL_Buffer luabuf;

145 size_t luabufcnt = 0;

146 int remaining;

147 char *terminatorpos;

148 ssize_t bytesread;

149 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

150 maxread = luaL_optinteger(L, 2, 0);

151 terminatorstr = luaL_optlstring(L, 3, "", &terminatorlen);

152 if (terminatorlen) {

153 luaL_argcheck(L, terminatorlen == 1, 3, "single byte expected");

154 terminator = terminatorstr[0];

155 }

156 lua_settop(L, 1); /* return handle on drain, terminator string may be garbage collected */

157 if (handle->closed) luaL_error(L, "Attempt to read from a closed I/O handle");

158 if (handle->readerr) {

159 lua_pushnil(L);

160 lua_pushliteral(L, "Previous read error");

161 return 2;

162 }

163 handle->readerr = 1;

164 if (handle->fd < 0) goto moonbr_io_read_impl_eof; /* fake EOF to simulate shutdown */

165 moonbr_io_handle_set_nonblocking(L, handle, nonblocking);

166 if (!drain) luaL_buffinit(L, &luabuf);

167 while (1) {

168 remaining = -1;

169 if (

170 maxread > 0 &&

171 handle->readbufin - handle->readbufout >= (size_t)maxread - luabufcnt

172 ) {

173 remaining = maxread - luabufcnt;

174 } else if (terminatorlen) {

175 terminatorpos = memchr(

176 handle->readbuf + handle->readbufout,

177 terminator,

178 handle->readbufin - handle->readbufout

179 );

180 if (terminatorpos) remaining = 1 + (

181 terminatorpos - (handle->readbuf + handle->readbufout)

182 );

183 }

184 if (remaining >= 0) {

185 if (!drain) {

186 luaL_addlstring(

187 &luabuf,

188 handle->readbuf + handle->readbufout,

189 remaining

190 );

191 luaL_pushresult(&luabuf);

192 } else {

193 luaL_pushresult(&luabuf);

194 lua_pop(L, 1);

195 lua_pushinteger(L, luabufcnt + remaining);

196 }

197 handle->readbufout += remaining;

198 if (handle->readbufout == handle->readbufin) {

199 handle->readbufin = 0;

200 handle->readbufout =0;

201 }

202 handle->readerr = 0;

203 return 1;

204 }

205 if (!drain) luaL_addlstring(

206 &luabuf,

207 handle->readbuf + handle->readbufout,

208 handle->readbufin - handle->readbufout

209 );

210 luabufcnt += handle->readbufin - handle->readbufout;

211 do {

212 bytesread = read(handle->fd, handle->readbuf, MOONBR_IO_READBUFLEN);

213 } while (bytesread < 0 && (errno == EINTR));

214 if (

215 bytesread == 0 || (

216 nonblocking &&

217 bytesread < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)

218 )

219 ) {

220 handle->readbufin = 0;

221 handle->readbufout = 0;

222 if (!drain) {

223 luaL_pushresult(&luabuf);

224 if (!luabufcnt && bytesread == 0) {

225 lua_pop(L, 1);

226 moonbr_io_read_impl_eof:

227 lua_pushboolean(L, 0);

228 lua_pushliteral(L, "End of file");

229 handle->readerr = 0;

230 return 2;

231 }

232 } else {

233 if (!luabufcnt && bytesread == 0) lua_pushboolean(L, 1);

234 else lua_pushboolean(L, luabufcnt);

235 }

236 handle->readerr = 0;

237 return 1;

238 }

239 if (bytesread < 0) {

240 moonbr_io_errmsg();

241 lua_pushnil(L);

242 lua_pushstring(L, errmsg);

243 return 2;

244 }

245 handle->readbufin = bytesread;

246 handle->readbufout = 0;

247 }

248 }

249

250 static int moonbr_io_read(lua_State *L) {

251 return moonbr_io_read_impl(L, 0, 0);

252 }

253

254 static int moonbr_io_read_nb(lua_State *L) {

255 return moonbr_io_read_impl(L, 1, 0);

256 }

257

258 static int moonbr_io_drain(lua_State *L) {

259 return moonbr_io_read_impl(L, 0, 1);

260 }

261

262 static int moonbr_io_drain_nb(lua_State *L) {

263 return moonbr_io_read_impl(L, 1, 1);

264 }

265

266 static int moonbr_io_write_impl(lua_State *L, int nonblocking, int flush) {

267 moonbr_io_handle_t *handle;

268 int i, top;

269 const char *str;

270 size_t strlen;

271 ssize_t written;

272 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

273 if (handle->closed) luaL_error(L, "Attempt to write to a closed I/O handle");

274 if (handle->finished) luaL_error(L, "Attempt to write to a finished I/O handle");

275 if (handle->writeerr) {

276 lua_pushnil(L);

277 lua_pushliteral(L, "Previous write error");

278 return 2;

279 }

280 handle->writeerr = 1;

281 moonbr_io_handle_set_nonblocking(L, handle, nonblocking);

282 top = lua_gettop(L);

283 lua_getuservalue(L, 1);

284 lua_getfield(L, -1, "writequeue");

285 for (i=2; i<=top; i++) {

286 luaL_checklstring(L, i, &strlen);

287 lua_pushvalue(L, i);

288 lua_rawseti(L, -2, handle->writeqin++);

289 handle->writeleft += strlen;

290 }

291 if (flush) handle->flushedleft = handle->writeleft;

292 if (handle->flushedleft) moonbr_io_handle_set_nopush(L, handle, 0);

293 while (handle->writeqout != handle->writeqin) {

294 lua_rawgeti(L, -1, handle->writeqout);

295 str = lua_tolstring(L, -1, &strlen);

296 while (handle->writeqoff < strlen) {

297 if (

298 strlen - handle->writeqoff <

299 MOONBR_IO_WRITEBUFLEN - handle->writebufin

300 ) {

301 memcpy(

302 handle->writebuf + handle->writebufin,

303 str + handle->writeqoff,

304 strlen - handle->writeqoff

305 );

306 handle->writebufin += strlen - handle->writeqoff;

307 break;

308 } else {

309 memcpy(

310 handle->writebuf + handle->writebufin,

311 str + handle->writeqoff,

312 MOONBR_IO_WRITEBUFLEN - handle->writebufin

313 );

314 handle->writeqoff += MOONBR_IO_WRITEBUFLEN - handle->writebufin;

315 while (handle->writebufout < MOONBR_IO_WRITEBUFLEN) {

316 if (!handle->flushedleft) moonbr_io_handle_set_nopush(L, handle, 1);

317 written = write(

318 handle->fd,

319 handle->writebuf + handle->writebufout,

320 MOONBR_IO_WRITEBUFLEN - handle->writebufout

321 );

322 if (written < 0) {

323 if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {

324 goto moonbr_io_write_impl_block;

325 } else if (errno != EINTR) {

326 moonbr_io_errmsg();

327 lua_pushnil(L);

328 lua_pushstring(L, errmsg);

329 return 2;

330 }

331 } else {

332 handle->writebufout += written;

333 handle->writeleft -= written;

334 if (written >= handle->flushedleft) handle->flushedleft = 0;

335 else handle->flushedleft -= written;

336 }

337 }

338 handle->writebufin = 0;

339 handle->writebufout = 0;

340 }

341 }

342 handle->writeqoff = 0;

343 lua_pop(L, 1);

344 lua_pushnil(L);

345 lua_rawseti(L, -2, handle->writeqout++);

346 }

347 while (handle->flushedleft) {

348 written = write(

349 handle->fd,

350 handle->writebuf + handle->writebufout,

351 handle->writebufin - handle->writebufout

352 );

353 if (written < 0) {

354 if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {

355 goto moonbr_io_write_impl_block;

356 } else if (errno != EINTR) {

357 moonbr_io_errmsg();

358 lua_pushnil(L);

359 lua_pushstring(L, errmsg);

360 return 2;

361 }

362 } else {

363 handle->writebufout += written;

364 handle->writeleft -= written;

365 if (written >= handle->flushedleft) handle->flushedleft = 0;

366 else handle->flushedleft -= written;

367 }

368 }

369 if (handle->writebufout == handle->writebufin) {

370 handle->writebufin = 0;

371 handle->writebufout = 0;

372 }

373 if (nonblocking) lua_pushinteger(L, 0);

374 else lua_pushvalue(L, 1);

375 handle->writeerr = 0;

376 return 1;

377 moonbr_io_write_impl_block:

378 lua_pushinteger(L, handle->writeleft);

379 handle->writeerr = 0;

380 return 1;

381 }

382

383 static int moonbr_io_write(lua_State *L) {

384 return moonbr_io_write_impl(L, 0, 0);

385 }

386

387 static int moonbr_io_write_nb(lua_State *L) {

388 return moonbr_io_write_impl(L, 1, 0);

389 }

390

391 static int moonbr_io_flush(lua_State *L) {

392 return moonbr_io_write_impl(L, 0, 1);

393 }

394

395 static int moonbr_io_flush_nb(lua_State *L) {

396 return moonbr_io_write_impl(L, 1, 1);

397 }

398

399 static int moonbr_io_finish(lua_State *L) {

400 moonbr_io_handle_t *handle;

401 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

402 if (handle->closed) luaL_error(L, "Attempt to finish a closed I/O handle");

403 if (handle->finished) luaL_error(L, "Attempt to finish a finished I/O handle");

404 if (handle->writeleft) {

405 lua_pushcfunction(L, moonbr_io_flush);

406 lua_pushvalue(L, 1);

407 if (lua_pcall(L, 1, 2, 0)) {

408 handle->finished = 1;

409 lua_error(L);

410 }

411 if (!lua_toboolean(L, -2)) {

412 handle->finished = 1;

413 return 2;

414 }

415 }

416 handle->finished = 1;

417 if (handle->addrfam == AF_INET6 || handle->addrfam == AF_INET) {

418 if (shutdown(handle->fd, SHUT_WR)) {

419 moonbr_io_errmsg();

420 lua_pushnil(L);

421 lua_pushstring(L, errmsg);

422 return 2;

423 }

424 } else {

425 if (close(handle->fd)) {

426 moonbr_io_errmsg();

427 handle->fd = -1;

428 lua_pushnil(L);

429 lua_pushstring(L, errmsg);

430 return 2;

431 }

432 handle->fd = -1; /* fake EOF on read */

433 }

434 lua_pushboolean(L, 1);

435 return 1;

436 }

437

438 static int moonbr_io_close_impl(lua_State *L, int reset) {

439 moonbr_io_handle_t *handle;

440 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

441 if (handle->closed) luaL_error(L, "Attempt to close a closed I/O handle");

442 if (!reset) {

443 if (handle->writeleft) {

444 lua_pushcfunction(L, moonbr_io_flush);

445 lua_pushvalue(L, 1);

446 if (lua_pcall(L, 1, 2, 0)) {

447 handle->closed = 1;

448 close(handle->fd);

449 handle->fd = -1;

450 lua_error(L);

451 }

452 handle->closed = 1;

453 if (!lua_toboolean(L, -2)) {

454 close(handle->fd);

455 handle->fd = -1;

456 return 2;

457 }

458 } else {

459 handle->closed = 1;

460 moonbr_io_handle_set_linger(L, handle, -1);

461 }

462 } else {

463 handle->closed = 1;

464 }

465 if (handle->fd >= 0) {

466 if (close(handle->fd)) {

467 moonbr_io_errmsg();

468 handle->fd = -1;

469 lua_pushnil(L);

470 lua_pushstring(L, errmsg);

471 return 2;

472 }

473 handle->fd = -1;

474 }

475 lua_pushboolean(L, 1);

476 return 1;

477

478 }

479

480 static int moonbr_io_close(lua_State *L) {

481 return moonbr_io_close_impl(L, 0);

482 }

483

484 static int moonbr_io_reset(lua_State *L) {

485 return moonbr_io_close_impl(L, 1);

486 }

487

488 static int moonbr_io_handlegc(lua_State *L) {

489 moonbr_io_handle_t *handle;

490 handle = luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

491 if (handle->fd >= 0) {

492 lua_pushcfunction(L, moonbr_io_close);

493 lua_pushvalue(L, 1);

494 lua_pushinteger(L, 0);

495 lua_call(L, 2, 0);

496 }

497 return 0;

498 }

499

500 void moonbr_io_closehandle(lua_State *L, int idx, int reset) {

501 moonbr_io_handle_t *handle;

502 handle = luaL_checkudata(L, idx, MOONBR_IO_HANDLE_MT_REGKEY);

503 if (!handle->closed) {

504 lua_pushcfunction(L, reset ? moonbr_io_reset : moonbr_io_close);

505 lua_pushvalue(L, idx < 0 ? idx-1 : idx);

506 lua_call(L, 1, 0);

507 }

508 }

509

510 void moonbr_io_pushhandle(lua_State *L, int fd) {

511 moonbr_io_handle_t *handle;

512 struct sockaddr addr;

513 socklen_t addrlen;

514 handle = lua_newuserdata(L, sizeof(moonbr_io_handle_t));

515 handle->fd = fd;

516 addrlen = sizeof(addr);

517 if (getsockname(fd, &addr, &addrlen)) {

518 if (errno != ENOTSOCK) {

519 moonbr_io_errmsg();

520 luaL_error(L, "Unexpected error when examining socket: %s", errmsg);

521 }

522 handle->issock = 0;

523 } else {

524 handle->issock = 1;

525 handle->addrfam = addr.sa_family;

526 }

527 handle->finished = 0;

528 handle->closed = 0;

529 handle->nonblocking = -1;

530 handle->nopush = -1;

531 handle->readerr = 0;

532 handle->readbufin = 0;

533 handle->readbufout = 0;

534 handle->writeerr = 0;

535 handle->writeleft = 0;

536 handle->flushedleft = 0;

537 handle->writeqin = 0;

538 handle->writeqout = 0;

539 handle->writeqoff = 0;

540 handle->writebufin = 0;

541 handle->writebufout = 0;

542 moonbr_io_handle_set_linger(L, handle, 0);

543 luaL_getmetatable(L, MOONBR_IO_HANDLE_MT_REGKEY);

544 lua_setmetatable(L, -2);

545 lua_newtable(L); // uservalue

546 lua_newtable(L);

547 lua_setfield(L, -2, "writequeue");

548 lua_newtable(L); // public

549 if (handle->addrfam == AF_INET6) {

550 struct sockaddr_in6 addr_in6;

551 char addrstrbuf[INET6_ADDRSTRLEN];

552 const char *addrstr;

553 addrlen = sizeof(addr_in6);

554 if (getsockname(fd, (struct sockaddr *)&addr_in6, &addrlen)) {

555 moonbr_io_errmsg();

556 luaL_error(L, "Could not determine local IP address/port: %s", errmsg);

557 }

558 if (addrlen > sizeof(addr_in6)) {

559 luaL_error(L, "Could not determine local IP address/port: buffer size exceeded");

560 }

561 addrstr = inet_ntop(AF_INET6, addr_in6.sin6_addr.s6_addr, addrstrbuf, sizeof(addrstrbuf));

562 if (!addrstr) {

563 moonbr_io_errmsg();

564 luaL_error(L, "Could not format local IP address: %s", errmsg);

565 } else {

566 lua_pushstring(L, addrstr);

567 lua_setfield(L, -2, "local_ip6");

568 }

569 lua_pushinteger(L, ntohs(addr_in6.sin6_port));

570 lua_setfield(L, -2, "local_tcpport");

571 if (getpeername(fd, (struct sockaddr *)&addr_in6, &addrlen)) {

572 moonbr_io_errmsg();

573 luaL_error(L, "Could not determine remote IP address/port: %s", errmsg);

574 }

575 if (addrlen > sizeof(addr_in6)) {

576 luaL_error(L, "Could not determine remote IP address/port: buffer size exceeded");

577 }

578 addrstr = inet_ntop(AF_INET6, addr_in6.sin6_addr.s6_addr, addrstrbuf, sizeof(addrstrbuf));

579 if (!addrstr) {

580 moonbr_io_errmsg();

581 luaL_error(L, "Could not format remote IP address: %s", errmsg);

582 } else {

583 lua_pushstring(L, addrstr);

584 lua_setfield(L, -2, "remote_ip6");

585 }

586 lua_pushinteger(L, ntohs(addr_in6.sin6_port));

587 lua_setfield(L, -2, "remote_tcpport");

588 } else if (handle->addrfam == AF_INET) {

589 struct sockaddr_in addr_in;

590 char addrstrbuf[INET_ADDRSTRLEN];

591 const char *addrstr;

592 addrlen = sizeof(addr_in);

593 if (getsockname(fd, (struct sockaddr *)&addr_in, &addrlen)) {

594 moonbr_io_errmsg();

595 luaL_error(L, "Could not determine local IP address/port: %s", errmsg);

596 }

597 if (addrlen > sizeof(addr_in)) {

598 luaL_error(L, "Could not determine local IP address/port: buffer size exceeded");

599 }

600 addrstr = inet_ntop(AF_INET, &addr_in.sin_addr.s_addr, addrstrbuf, sizeof(addrstrbuf));

601 if (!addrstr) {

602 moonbr_io_errmsg();

603 luaL_error(L, "Could not format local IP address: %s", errmsg);

604 } else {

605 lua_pushstring(L, addrstr);

606 lua_setfield(L, -2, "local_ip4");

607 }

608 lua_pushinteger(L, ntohs(addr_in.sin_port));

609 lua_setfield(L, -2, "local_tcpport");

610 if (getpeername(fd, (struct sockaddr *)&addr_in, &addrlen)) {

611 moonbr_io_errmsg();

612 luaL_error(L, "Could not determine remote IP address/port: %s", errmsg);

613 }

614 if (addrlen > sizeof(addr_in)) {

615 luaL_error(L, "Could not determine remote IP address/port: buffer size exceeded");

616 }

617 addrstr = inet_ntop(AF_INET, &addr_in.sin_addr.s_addr, addrstrbuf, sizeof(addrstrbuf));

618 if (!addrstr) {

619 moonbr_io_errmsg();

620 luaL_error(L, "Could not format remote IP address: %s", errmsg);

621 } else {

622 lua_pushstring(L, addrstr);

623 lua_setfield(L, -2, "remote_ip4");

624 }

625 lua_pushinteger(L, ntohs(addr_in.sin_port));

626 lua_setfield(L, -2, "remote_tcpport");

627 }

628 luaL_getmetatable(L, MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY);

629 lua_setmetatable(L, -2);

630 lua_setfield(L, -2, "public");

631 lua_setuservalue(L, -2);

632 }

633

634 static int moonbr_io_handleindex(lua_State *L) {

635 luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

636 lua_getuservalue(L, 1);

637 lua_getfield(L, -1, "public");

638 lua_pushvalue(L, 2);

639 lua_gettable(L, -2);

640 return 1;

641 }

642

643 static int moonbr_io_handlenewindex(lua_State *L) {

644 luaL_checkudata(L, 1, MOONBR_IO_HANDLE_MT_REGKEY);

645 lua_getuservalue(L, 1);

646 lua_getfield(L, -1, "public");

647 lua_pushvalue(L, 2);

648 lua_pushvalue(L, 3);

649 lua_settable(L, -3);

650 return 0;

651 }

652

653 static int moonbr_io_localconnect_impl(lua_State *L, int nonblocking) {

654 const char *path;

655 struct sockaddr_un sockaddr = { .sun_family = AF_LOCAL };

656 const int path_maxlen = sizeof(struct sockaddr_un) - (

657 (void *)sockaddr.sun_path - (void *)&sockaddr

658 ) - 1; /* one byte for termination */

659 int sock;

660 path = luaL_checkstring(L, 1);

661 if (strlen(path) > path_maxlen) luaL_error(L, "Path too long; only %i characters allowed", path_maxlen);

662 strcpy(sockaddr.sun_path, path);

663 sock = socket(

664 PF_LOCAL,

665 SOCK_STREAM | SOCK_CLOEXEC | (nonblocking ? SOCK_NONBLOCK : 0),

666 0

667 );

668 if (sock < 0) {

669 moonbr_io_errmsg();

670 lua_pushnil(L);

671 lua_pushstring(L, errmsg);

672 return 2;

673 }

674 if (connect(sock, (struct sockaddr *)&sockaddr, sizeof(sockaddr))) {

675 if (!nonblocking && errno == EINTR) {

676 moonbr_io_errmsg();

677 close(sock);

678 lua_pushnil(L);

679 lua_pushstring(L, errmsg);

680 return 2;

681 } else if (!(nonblocking && (errno == EINPROGRESS || errno == EINTR))) {

682 moonbr_io_errmsg();

683 lua_pushnil(L);

684 lua_pushstring(L, errmsg);

685 return 2;

686 }

687 }

688 moonbr_io_pushhandle(L, sock);

689 return 1;

690 }

691

692 static int moonbr_io_localconnect(lua_State *L) {

693 return moonbr_io_localconnect_impl(L, 0);

694 }

695

696 static int moonbr_io_localconnect_nb(lua_State *L) {

697 return moonbr_io_localconnect_impl(L, 1);

698 }

699

700 static int moonbr_io_tcpconnect_impl(lua_State *L, int nonblocking) {

701 const char *host, *port;

702 struct addrinfo hints = { 0, };

703 struct addrinfo *res, *addrinfo;

704 int errcode;

705 int sock;

706 host = luaL_checkstring(L, 1);

707 port = luaL_checkstring(L, 2);

708 hints.ai_family = AF_UNSPEC;

709 hints.ai_socktype = SOCK_STREAM;

710 hints.ai_protocol = IPPROTO_TCP;

711 hints.ai_flags = AI_ADDRCONFIG;

712 errcode = getaddrinfo(host, port, &hints, &res);

713 if (errcode) {

714 freeaddrinfo(res);

715 if (errcode == EAI_SYSTEM) {

716 moonbr_io_errmsg();

717 lua_pushnil(L);

718 lua_pushfstring(L, "%s: %s", gai_strerror(errcode), errmsg);

719 } else {

720 lua_pushnil(L);

721 lua_pushstring(L, gai_strerror(errcode));

722 }

723 return 2;

724 }

725 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {

726 if (addrinfo->ai_family == PF_INET6) goto moonbr_io_tcpconnect_found;

727 }

728 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {

729 if (addrinfo->ai_family == PF_INET) goto moonbr_io_tcpconnect_found;

730 }

731 addrinfo = res;

732 moonbr_io_tcpconnect_found:

733 sock = socket(

734 addrinfo->ai_family,

735 addrinfo->ai_socktype | SOCK_CLOEXEC | (nonblocking ? SOCK_NONBLOCK : 0),

736 addrinfo->ai_protocol

737 );

738 if (sock < 0) {

739 moonbr_io_errmsg();

740 freeaddrinfo(res);

741 lua_pushnil(L);

742 lua_pushstring(L, errmsg);

743 return 2;

744 }

745 if (connect(sock, addrinfo->ai_addr, addrinfo->ai_addrlen)) {

746 freeaddrinfo(res);

747 if (!nonblocking && errno == EINTR) {

748 moonbr_io_errmsg();

749 close(sock);

750 lua_pushnil(L);

751 lua_pushstring(L, errmsg);

752 return 2;

753 } else if (!(nonblocking && (errno == EINPROGRESS || errno == EINTR))) {

754 moonbr_io_errmsg();

755 lua_pushnil(L);

756 lua_pushstring(L, errmsg);

757 return 2;

758 }

759 } else {

760 freeaddrinfo(res);

761 }

762 moonbr_io_pushhandle(L, sock);

763 return 1;

764 }

765

766 static int moonbr_io_tcpconnect(lua_State *L) {

767 return moonbr_io_tcpconnect_impl(L, 0);

768 }

769

770 static int moonbr_io_tcpconnect_nb(lua_State *L) {

771 return moonbr_io_tcpconnect_impl(L, 1);

772 }

773

774 static int moonbr_io_locallisten(lua_State *L) {

775 moonbr_io_listener_t *listener;

776 const char *path;

777 struct stat sb;

778 struct sockaddr_un sockaddr = { .sun_family = AF_LOCAL };

779 const int path_maxlen = sizeof(struct sockaddr_un) - (

780 (void *)sockaddr.sun_path - (void *)&sockaddr

781 ) - 1; /* one byte for termination */

782 int sock;

783 path = luaL_checkstring(L, 1);

784 if (strlen(path) > path_maxlen) luaL_error(L, "Path too long; only %i characters allowed", path_maxlen);

785 strcpy(sockaddr.sun_path, path);

786 if (stat(path, &sb) == 0) {

787 if (S_ISSOCK(sb.st_mode)) unlink(path);

788 }

789 listener = lua_newuserdata(L, sizeof(moonbr_io_listener_t));

790 listener->fd = -1;

791 luaL_setmetatable(L, MOONBR_IO_LISTENER_MT_REGKEY);

792 sock = socket(

793 PF_LOCAL,

794 SOCK_STREAM | SOCK_CLOEXEC,

795 0

796 );

797 if (sock < 0) {

798 moonbr_io_errmsg();

799 lua_pushnil(L);

800 lua_pushstring(L, errmsg);

801 return 2;

802 }

803 if (bind(sock, (struct sockaddr *)&sockaddr, sizeof(sockaddr))) {

804 moonbr_io_errmsg();

805 close(sock);

806 lua_pushnil(L);

807 lua_pushstring(L, errmsg);

808 return 2;

809 }

810 if (listen(sock, MOONBR_IO_LISTEN_BACKLOG)) {

811 moonbr_io_errmsg();

812 close(sock);

813 lua_pushnil(L);

814 lua_pushstring(L, errmsg);

815 return 2;

816 }

817 listener->fd = sock;

818 listener->addrfam = AF_LOCAL;

819 listener->nonblocking = -1;

820 return 1;

821 }

822

823 static int moonbr_io_tcplisten(lua_State *L) {

824 moonbr_io_listener_t *listener;

825 const char *host, *port;

826 struct addrinfo hints = { 0, };

827 struct addrinfo *res, *addrinfo;

828 int errcode;

829 int sock;

830 host = luaL_optstring(L, 1, NULL);

831 port = luaL_checkstring(L, 2);

832 listener = lua_newuserdata(L, sizeof(moonbr_io_listener_t));

833 listener->fd = -1;

834 luaL_setmetatable(L, MOONBR_IO_LISTENER_MT_REGKEY);

835 hints.ai_family = AF_UNSPEC;

836 hints.ai_socktype = SOCK_STREAM;

837 hints.ai_protocol = IPPROTO_TCP;

838 hints.ai_flags = AI_ADDRCONFIG | AI_PASSIVE;

839 errcode = getaddrinfo(host, port, &hints, &res);

840 if (errcode) {

841 freeaddrinfo(res);

842 if (errcode == EAI_SYSTEM) {

843 moonbr_io_errmsg();

844 lua_pushnil(L);

845 lua_pushfstring(L, "%s: %s", gai_strerror(errcode), errmsg);

846 } else {

847 lua_pushnil(L);

848 lua_pushstring(L, gai_strerror(errcode));

849 }

850 return 2;

851 }

852 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {

853 if (addrinfo->ai_family == PF_INET6) goto moonbr_io_tcpconnect_found;

854 }

855 for (addrinfo=res; addrinfo; addrinfo=addrinfo->ai_next) {

856 if (addrinfo->ai_family == PF_INET) goto moonbr_io_tcpconnect_found;

857 }

858 addrinfo = res;

859 moonbr_io_tcpconnect_found:

860 listener->addrfam = addrinfo->ai_family;

861 sock = socket(

862 addrinfo->ai_family,

863 addrinfo->ai_socktype | SOCK_CLOEXEC,

864 addrinfo->ai_protocol

865 );

866 if (sock < 0) {

867 moonbr_io_errmsg();

868 freeaddrinfo(res);

869 lua_pushnil(L);

870 lua_pushstring(L, errmsg);

871 return 2;

872 }

873 {

874 static const int reuseval = 1;

875 if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuseval, sizeof(reuseval))) {

876 moonbr_io_errmsg();

877 freeaddrinfo(res);

878 close(sock);

879 lua_pushnil(L);

880 lua_pushfstring(L, "Error while setting SO_REUSEADDR with setsockopt: %s", errmsg);

881 return 2;

882 }

883 }

884 if (bind(sock, addrinfo->ai_addr, addrinfo->ai_addrlen)) {

885 moonbr_io_errmsg();

886 freeaddrinfo(res);

887 close(sock);

888 lua_pushnil(L);

889 lua_pushstring(L, errmsg);

890 return 2;

891 }

892 freeaddrinfo(res);

893 if (listen(sock, MOONBR_IO_LISTEN_BACKLOG)) {

894 moonbr_io_errmsg();

895 close(sock);

896 lua_pushnil(L);

897 lua_pushstring(L, errmsg);

898 return 2;

899 }

900 listener->fd = sock;

901 listener->nonblocking = -1;

902 return 1;

903 }

904

905 static int moonbr_io_accept_impl(lua_State *L, int nonblocking) {

906 moonbr_io_listener_t *listener;

907 int fd;

908 listener = luaL_checkudata(L, 1, MOONBR_IO_LISTENER_MT_REGKEY);

909 if (listener->fd < 0) luaL_error(L, "Attempt to use a closed listener");

910 if (listener->nonblocking != nonblocking) {

911 int flags;

912 flags = fcntl(listener->fd, F_GETFL, 0);

913 if (flags == -1) {

914 moonbr_io_errmsg();

915 close(listener->fd);

916 listener->fd = -1;

917 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);

918 }

919 if (nonblocking) flags |= O_NONBLOCK;

920 else flags &= ~O_NONBLOCK;

921 if (fcntl(listener->fd, F_SETFL, flags) == -1) {

922 moonbr_io_errmsg();

923 close(listener->fd);

924 listener->fd = -1;

925 luaL_error(L, "Unexpected error in fcntl call: %s", errmsg);

926 }

927 listener->nonblocking = nonblocking;

928 }

929 while (1) {

930 fd = accept4(listener->fd, NULL, NULL, SOCK_CLOEXEC);

931 if (fd < 0) {

932 if (nonblocking && (errno == EAGAIN || errno == EWOULDBLOCK)) {

933 lua_pushboolean(L, 0);

934 lua_pushliteral(L, "No incoming connection pending");

935 return 2;

936 } else if (errno != EINTR) {

937 moonbr_io_errmsg();

938 lua_pushnil(L);

939 lua_pushstring(L, errmsg);

940 return 2;

941 }

942 } else {

943 moonbr_io_pushhandle(L, fd);

944 return 1;

945 }

946 }

947 }

948

949 static int moonbr_io_accept(lua_State *L) {

950 return moonbr_io_accept_impl(L, 0);

951 }

952

953 static int moonbr_io_accept_nb(lua_State *L) {

954 return moonbr_io_accept_impl(L, 1);

955 }

956

957 static int moonbr_io_unlisten(lua_State *L) {

958 moonbr_io_listener_t *listener;

959 struct sockaddr_un addr;

960 socklen_t addrlen;

961 struct stat sb;

962 listener = luaL_checkudata(L, 1, MOONBR_IO_LISTENER_MT_REGKEY);

963 if (listener->fd < 0) luaL_error(L, "Attempt to close a closed listener");

964 addrlen = sizeof(addr);

965 if (getsockname(listener->fd, (struct sockaddr *)&addr, &addrlen)) addrlen = 0;

966 if (close(listener->fd)) {

967 moonbr_io_errmsg();

968 listener->fd = -1;

969 if (addrlen && addrlen <= sizeof(addr)) {

970 if (stat(addr.sun_path, &sb) == 0) {

971 if (S_ISSOCK(sb.st_mode)) unlink(addr.sun_path);

972 }

973 }

974 lua_pushnil(L);

975 lua_pushstring(L, errmsg);

976 return 2;

977 }

978 listener->fd = -1;

979 if (addrlen && addrlen <= sizeof(addr)) {

980 if (stat(addr.sun_path, &sb) == 0) {

981 if (S_ISSOCK(sb.st_mode)) unlink(addr.sun_path);

982 }

983 }

984 lua_pushboolean(L, 1);

985 return 1;

986 }

987

988 static int moonbr_io_listenergc(lua_State *L) {

989 moonbr_io_listener_t *listener;

990 listener = luaL_checkudata(L, 1, MOONBR_IO_LISTENER_MT_REGKEY);

991 if (listener->fd >= 0) close(listener->fd);

992 listener->fd = -1;

993 return 0;

994 }

995

996 static int moonbr_io_poll(lua_State *L) {

997 moonbr_io_handle_t *handle;

998 moonbr_io_listener_t *listener;

999 int fd, isnum;

1000 int nfds = 0;

1001 fd_set readfds, writefds, exceptfds;

1002 struct timeval timeout = {0, };

1003 int status;

1004 FD_ZERO(&readfds);

1005 FD_ZERO(&writefds);

1006 FD_ZERO(&exceptfds);

1007 if (!lua_isnoneornil(L, 1)) {

1008 luaL_checktype(L, 1, LUA_TTABLE);

1009 for (lua_pushnil(L); lua_next(L, 1); lua_pop(L, 1)) {

1010 if (lua_toboolean(L, -1)) {

1011 handle = luaL_testudata(L, -2, MOONBR_IO_HANDLE_MT_REGKEY);

1012 if (handle) {

1013 if (handle->closed) luaL_error(L, "Attempt to poll a closed connection");

1014 fd = handle->fd;

1015 if (

1016 fd < 0 || /* fake EOF to simulate shutdown if fd < 0 */

1017 handle->readbufin != handle->readbufout /* data pending in buffer */

1018 ) {

1019 lua_pushboolean(L, 1);

1020 return 1;

1021 }

1022 } else {

1023 listener = luaL_testudata(L, -2, MOONBR_IO_LISTENER_MT_REGKEY);

1024 if (listener) {

1025 fd = listener->fd;

1026 if (fd < 0) luaL_error(L, "Attempt to poll a closed listener");

1027 } else {

1028 fd = lua_tointegerx(L, -2, &isnum);

1029 if (!isnum) luaL_error(L, "Expected integer (file descriptor), I/O handle, or listener in table key");

1030 }

1031 }

1032 FD_SET(fd, &readfds);

1033 if (fd+1 > nfds) nfds = fd+1;

1034 }

1035 }

1036 }

1037 if (!lua_isnoneornil(L, 2)) {

1038 luaL_checktype(L, 2, LUA_TTABLE);

1039 for (lua_pushnil(L); lua_next(L, 2); lua_pop(L, 1)) {

1040 if (lua_toboolean(L, -1)) {

1041 handle = luaL_testudata(L, -2, MOONBR_IO_HANDLE_MT_REGKEY);

1042 if (handle) {

1043 if (handle->closed) luaL_error(L, "Attempt to poll a closed connection");

1044 if (handle->finished) luaL_error(L, "Attempt to write-poll a finished connection");

1045 fd = handle->fd;

1046 } else {

1047 listener = luaL_testudata(L, -2, MOONBR_IO_LISTENER_MT_REGKEY);

1048 if (listener) luaL_error(L, "Attempt to write-poll a listener");

1049 fd = lua_tointegerx(L, -2, &isnum);

1050 if (!isnum) luaL_error(L, "Expected integer (file descriptor) or I/O handle in table key");

1051 }

1052 FD_SET(fd, &writefds);

1053 if (fd+1 > nfds) nfds = fd+1;

1054 }

1055 }

1056 }

1057 if (!lua_isnoneornil(L, 3)) {

1058 lua_Number n;

1059 n = lua_tonumberx(L, 3, &isnum);

1060 if (isnum && n>=0 && n<100000000) {

1061 timeout.tv_sec = n;

1062 timeout.tv_usec = 1e6 * (n - timeout.tv_sec);

1063 } else {

1064 luaL_argcheck(L, 0, 3, "not a valid timeout");

1065 }

1066 status = select(nfds, &readfds, &writefds, &exceptfds, &timeout);

1067 } else {

1068 status = select(nfds, &readfds, &writefds, &exceptfds, NULL);

1069 }

1070 if (status == -1) {

1071 if (errno == EINTR) {

1072 lua_pushboolean(L, 0);

1073 lua_pushliteral(L, "Signal received while polling file descriptors");

1074 return 2;

1075 } else {

1076 moonbr_io_errmsg();

1077 return luaL_error(L, "Unexpected error during \"select\" system call: %s", errmsg);

1078 }

1079 } else if (status == 0) {

1080 lua_pushboolean(L, 0);

1081 lua_pushliteral(L, "Timeout while polling file descriptors");

1082 return 2;

1083 } else {

1084 lua_pushboolean(L, 1);

1085 return 1;

1086 }

1087 }

1088

1089 static int moonbr_io_timeref(lua_State *L) {

1090 lua_Number sub;

1091 struct timespec tp;

1092 sub = luaL_optnumber(L, 1, 0);

1093 if (clock_gettime(CLOCK_MONOTONIC, &tp)) {

1094 return luaL_error(L, "Could not access CLOCK_MONOTONIC");

1095 }

1096 lua_pushnumber(L, tp.tv_sec + tp.tv_nsec / 1.0e9 - sub);

1097 return 1;

1098 }

1099

1100 static const struct luaL_Reg moonbr_io_handle_methods[] = {

1101 {"read", moonbr_io_read},

1102 {"read_nb", moonbr_io_read_nb},

1103 {"drain", moonbr_io_drain},

1104 {"drain_nb", moonbr_io_drain_nb},

1105 {"write", moonbr_io_write},

1106 {"write_nb", moonbr_io_write_nb},

1107 {"flush", moonbr_io_flush},

1108 {"flush_nb", moonbr_io_flush_nb},

1109 {"finish", moonbr_io_finish},

1110 {"close", moonbr_io_close},

1111 {"reset", moonbr_io_reset},

1112 {NULL, NULL}

1113 };

1114

1115 static const struct luaL_Reg moonbr_io_handle_metamethods[] = {

1116 {"__index", moonbr_io_handleindex},

1117 {"__newindex", moonbr_io_handlenewindex},

1118 {"__gc", moonbr_io_handlegc},

1119 {NULL, NULL}

1120 };

1121

1122 static const struct luaL_Reg moonbr_io_listener_methods[] = {

1123 {"accept", moonbr_io_accept},

1124 {"accept_nb", moonbr_io_accept_nb},

1125 {"close", moonbr_io_unlisten},

1126 {NULL, NULL}

1127 };

1128

1129 static const struct luaL_Reg moonbr_io_listener_metamethods[] = {

1130 {"__gc", moonbr_io_listenergc},

1131 {NULL, NULL}

1132 };

1133

1134 static const struct luaL_Reg moonbr_io_module_funcs[] = {

1135 {"localconnect", moonbr_io_localconnect},

1136 {"localconnect_nb", moonbr_io_localconnect_nb},

1137 {"tcpconnect", moonbr_io_tcpconnect},

1138 {"tcpconnect_nb", moonbr_io_tcpconnect_nb},

1139 {"locallisten", moonbr_io_locallisten},

1140 {"tcplisten", moonbr_io_tcplisten},

1141 {"poll", moonbr_io_poll},

1142 {"timeref", moonbr_io_timeref},

1143 {NULL, NULL}

1144 };

1145

1146 int luaopen_moonbridge_io(lua_State *L) {

1147

1148 signal(SIGPIPE, SIG_IGN); /* generate I/O errors instead of signal 13 */

1149

1150 lua_newtable(L); // module

1151

1152 lua_newtable(L); // public metatable

1153 lua_newtable(L); // handle methods

1154 luaL_setfuncs(L, moonbr_io_handle_methods, 0);

1155 lua_pushvalue(L, -1);

1156 lua_setfield(L, -4, "prototype_handle");

1157 lua_setfield(L, -2, "__index");

1158 lua_setfield(L, LUA_REGISTRYINDEX, MOONBR_IO_HANDLE_PUBLIC_MT_REGKEY);

1159

1160 lua_newtable(L); // handle metatable

1161 luaL_setfuncs(L, moonbr_io_handle_metamethods, 0);

1162 lua_setfield(L, LUA_REGISTRYINDEX, MOONBR_IO_HANDLE_MT_REGKEY);

1163

1164 lua_newtable(L); // listener metatable

1165 luaL_setfuncs(L, moonbr_io_listener_metamethods, 0);

1166 lua_newtable(L); // listener methods

1167 luaL_setfuncs(L, moonbr_io_listener_methods, 0);

1168 lua_pushvalue(L, -1);

1169 lua_setfield(L, -4, "prototype_listener");

1170 lua_setfield(L, -2, "__index");

1171 lua_setfield(L, LUA_REGISTRYINDEX, MOONBR_IO_LISTENER_MT_REGKEY);

1172

1173 luaL_setfuncs(L, moonbr_io_module_funcs, 0);

1174 return 1;

1175

1176 }

1177