Skip to content

Commit 8d78fb1

Browse files
committed
changed net/connect to be non-blocking / asynchronous
1 parent 148917d commit 8d78fb1

File tree

3 files changed

+63
-11
lines changed

3 files changed

+63
-11
lines changed

src/core/ev.c

+41-1
Original file line numberDiff line numberDiff line change
@@ -2456,7 +2456,8 @@ void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, in
24562456
typedef enum {
24572457
JANET_ASYNC_WRITEMODE_WRITE,
24582458
JANET_ASYNC_WRITEMODE_SEND,
2459-
JANET_ASYNC_WRITEMODE_SENDTO
2459+
JANET_ASYNC_WRITEMODE_SENDTO,
2460+
JANET_ASYNC_WRITEMODE_CONNECT
24602461
} JanetWriteMode;
24612462

24622463
typedef struct {
@@ -2480,6 +2481,31 @@ typedef struct {
24802481
#endif
24812482
} StateWrite;
24822483

2484+
static JanetAsyncStatus handle_connect(JanetListenerState *s) {
2485+
#ifdef JANET_WINDOWS
2486+
int res = 0;
2487+
int size = sizeof(res);
2488+
int r = getsockopt((SOCKET)s->stream->handle, SOL_SOCKET, SO_ERROR, (char *)&res, &size);
2489+
#else
2490+
int res = 0;
2491+
socklen_t size = sizeof res;
2492+
int r = getsockopt(s->stream->handle, SOL_SOCKET, SO_ERROR, &res, &size);
2493+
#endif
2494+
if (r == 0) {
2495+
if (res == 0) {
2496+
janet_schedule(s->fiber, janet_wrap_abstract(s->stream));
2497+
} else {
2498+
// TODO help needed. janet_stream_close(s->stream);
2499+
janet_cancel(s->fiber, janet_cstringv(strerror(res)));
2500+
}
2501+
} else {
2502+
// TODO help needed. janet_stream_close(s->stream);
2503+
janet_cancel(s->fiber, janet_ev_lasterr());
2504+
}
2505+
return JANET_ASYNC_STATUS_DONE;
2506+
}
2507+
2508+
24832509
JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event) {
24842510
StateWrite *state = (StateWrite *) s;
24852511
switch (event) {
@@ -2509,6 +2535,11 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
25092535
}
25102536
break;
25112537
case JANET_ASYNC_EVENT_USER: {
2538+
#ifdef JANET_NET
2539+
if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
2540+
return handle_connect(s);
2541+
}
2542+
#endif
25122543
/* Begin write */
25132544
int32_t len;
25142545
const uint8_t *bytes;
@@ -2572,6 +2603,11 @@ JanetAsyncStatus ev_machine_write(JanetListenerState *s, JanetAsyncEvent event)
25722603
janet_cancel(s->fiber, janet_cstringv("stream hup"));
25732604
return JANET_ASYNC_STATUS_DONE;
25742605
case JANET_ASYNC_EVENT_WRITE: {
2606+
#ifdef JANET_NET
2607+
if (state->mode == JANET_ASYNC_WRITEMODE_CONNECT) {
2608+
return handle_connect(s);
2609+
}
2610+
#endif
25752611
int32_t start, len;
25762612
const uint8_t *bytes;
25772613
start = state->start;
@@ -2674,6 +2710,10 @@ void janet_ev_sendto_buffer(JanetStream *stream, JanetBuffer *buf, void *dest, i
26742710
void janet_ev_sendto_string(JanetStream *stream, JanetString str, void *dest, int flags) {
26752711
janet_ev_write_generic(stream, (void *) str, dest, JANET_ASYNC_WRITEMODE_SENDTO, 0, flags);
26762712
}
2713+
2714+
void janet_ev_connect(JanetStream *stream, int flags) {
2715+
janet_ev_write_generic(stream, NULL, NULL, JANET_ASYNC_WRITEMODE_CONNECT, 0, flags);
2716+
}
26772717
#endif
26782718

26792719
/* For a pipe ID */

src/core/net.c

+21-10
Original file line numberDiff line numberDiff line change
@@ -477,32 +477,43 @@ JANET_CORE_FN(cfun_net_connect,
477477
}
478478
}
479479

480+
/* Wrap socket in abstract type JanetStream */
481+
JanetStream *stream = make_stream(sock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
482+
483+
/* Set the socket to non-blocking mode */
484+
janet_net_socknoblock(sock);
485+
480486
/* Connect to socket */
481487
#ifdef JANET_WINDOWS
482488
int status = WSAConnect(sock, addr, addrlen, NULL, NULL, NULL, NULL);
483-
Janet lasterr = janet_ev_lasterr();
489+
int err = WSAGetLastError();
484490
freeaddrinfo(ai);
485491
#else
486492
int status = connect(sock, addr, addrlen);
487-
Janet lasterr = janet_ev_lasterr();
493+
int err = errno;
488494
if (is_unix) {
489495
janet_free(ai);
490496
} else {
491497
freeaddrinfo(ai);
492498
}
493499
#endif
494500

495-
if (status == -1) {
496-
JSOCKCLOSE(sock);
497-
janet_panicf("could not connect socket: %V", lasterr);
501+
if (status != 0) {
502+
#ifdef JANET_WINDOWS
503+
if (err != WSAEWOULDBLOCK) {
504+
#else
505+
if (err != EINPROGRESS) {
506+
#endif
507+
JSOCKCLOSE(sock);
508+
Janet lasterr = janet_ev_lasterr();
509+
janet_panicf("could not connect socket: %V", lasterr);
510+
}
498511
}
499512

500-
/* Set up the socket for non-blocking IO after connect - TODO - non-blocking connect? */
501-
janet_net_socknoblock(sock);
513+
/* Handle the connect() result in the event loop*/
514+
janet_ev_connect(stream, MSG_NOSIGNAL);
502515

503-
/* Wrap socket in abstract type JanetStream */
504-
JanetStream *stream = make_stream(sock, JANET_STREAM_READABLE | JANET_STREAM_WRITABLE);
505-
return janet_wrap_abstract(stream);
516+
janet_await();
506517
}
507518

508519
static const char *serverify_socket(JSock sfd) {

src/include/janet.h

+1
Original file line numberDiff line numberDiff line change
@@ -1479,6 +1479,7 @@ JANET_API void janet_ev_readchunk(JanetStream *stream, JanetBuffer *buf, int32_t
14791479
JANET_API void janet_ev_recv(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
14801480
JANET_API void janet_ev_recvchunk(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
14811481
JANET_API void janet_ev_recvfrom(JanetStream *stream, JanetBuffer *buf, int32_t nbytes, int flags);
1482+
JANET_API void janet_ev_connect(JanetStream *stream, int flags);
14821483
#endif
14831484

14841485
/* Write async to a stream */

0 commit comments

Comments
 (0)