From 1178aa02f933b61b47cc94b1d7eceed721c57e10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Zago=C5=BEen?= Date: Thu, 16 Nov 2023 09:11:53 +0100 Subject: [PATCH 1/4] Improve http.Client connection handling HTTP 1.0 servers close the connection automatically after sending the response. We updated the http.Client actor such that the users can use the high-level API like making several requests in a row without having to worry about reconnects. http.Client will buffer the requests and flush the buffer once it (re)connects the socket. Writing to a closed socket is bad and guaranteed to fail, so the user (in this case the http.Client actor) should be notified ASAP before making further writes. Rather than using the on_error callback from the underlying TCPConnection and TLSConnection actors we raise an exception from their write() methods. This implies the calls to write() are synchronous, but since we're not waiting on network I/O it should be fine?! --- base/src/http.act | 54 +++++++++++++++++++++++++++++++--------------- base/src/net.ext.c | 24 +++++++++++++++++---- 2 files changed, 57 insertions(+), 21 deletions(-) diff --git a/base/src/http.act b/base/src/http.act index 92da6f0c4..a21def66e 100644 --- a/base/src/http.act +++ b/base/src/http.act @@ -447,6 +447,7 @@ actor Listener(cap: net.TCPListenCap, address: str, port: int, on_listen_error: # TODO: default schema="https" # TODO: default port=None # TODO: default tls_verify=True +# TODO: add arguments for configuring request buffering (size) and reconnection attempts and timeouts actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_verify: bool, on_connect: action(Client) -> None, on_error: action(Client, str) -> None, log_handler: ?logging.Handler): """HTTP(S) Client @@ -457,10 +458,11 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ var _on_response: list[(bytes, action(Client, Response) -> None)] = [] var version: ?bytes = None var buf = b"" - var close_connection: bool = True var tcp_conn: ?net.TCPConnection = None var tls_conn: ?net.TLSConnection = None + var connecting: bool = True + def _connect(): if scheme == "http": _log.verbose("Using http scheme and port 80", None) @@ -475,9 +477,18 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ def _on_conn_connect(): # If there are outstanding requests, it probably means we were + # disconnected or have not connected yet + # TODO: do not flush entire buffer if we know the server will close the + # connection. If the latency is so big that we're able to send the + # entire buffer before the server closes the connection we're just + # wasting resources. for r in _on_response: + _log.trace("Sending outstanding request", {"request": r.0}) _conn_write(r.0) - await async on_connect(self) + if connecting: + # Dispatch the on_connect callback on first connect but not for reconnects + await async on_connect(self) + connecting = False def _on_tcp_connect(conn: net.TCPConnection) -> None: _on_conn_connect() @@ -500,10 +511,10 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ r, buf = parse_response(buf, _log) if r is not None: if "connection" in r.headers and r.headers["connection"] == "close": - close_connection = True - _conn_close() + # Is this really the right thing to do here? If the client + # does not make a new request we just reconnected for nothing! _log.debug("Closing TCP connection due to header: Connection: close", None) - _connect() + _conn_reconnect() if len(_on_response) == 0: _log.notice("Data received with no on_response callback set", None) break @@ -524,22 +535,31 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ def _on_con_error(error: str) -> None: on_error(self, error) - def _conn_close() -> None: + def _conn_reconnect() -> None: if tcp_conn is not None: - def _noop(c): - pass - tcp_conn.close(_noop) + tcp_conn.reconnect() elif tls_conn is not None: - def _noop(c): - pass - tls_conn.close(_noop) + tls_conn.reconnect() def _conn_write(data: bytes) -> None: - _log.trace("Sending data", {"data": data}) - if tcp_conn is not None: - tcp_conn.write(data) - elif tls_conn is not None: - tls_conn.write(data) + try: + _log.trace("Sending data", {"data": data}) + # We call the write method on the TCP or TLS connection actor + # synchronously because we want to be able to catch exceptions + # signaling the socket was closed. Note that this is *not* waiting + # for network I/O, just waiting on system I/O for writing to the + # local socket buffer. + if tcp_conn is not None: + await async tcp_conn.write(data) + elif tls_conn is not None: + await async tls_conn.write(data) + except RuntimeError as exc: + # HTTP/1.0 servers close the connection after each request by default + if "bad file descriptor" in str(exc) or "bad stream" in str(exc): + _log.debug("TCP connection closed, reconnecting", {"error": str(exc)}) + _conn_reconnect() + else: + _on_con_error(str(exc)) # HTTP methods def get(path: str, headers: dict[str, str], on_response: action(Client, Response) -> None): diff --git a/base/src/net.ext.c b/base/src/net.ext.c index 091463588..da3645e41 100644 --- a/base/src/net.ext.c +++ b/base/src/net.ext.c @@ -276,6 +276,12 @@ void on_connect6(uv_connect_t *connect_req, int status) { char errmsg[1024] = "Failed to write to TCP socket: "; uv_strerror_r(r, errmsg + strlen(errmsg), sizeof(errmsg)-strlen(errmsg)); log_warn(errmsg); + if (strstr(errmsg, "bad file descriptor")) { + // This can happen if the socket is closed, se we raise an exception + // and let the caller retry + $RAISE(((B_BaseException)B_RuntimeErrorG_new(to$str(errmsg)))); + return $R_CONT(c$cont, B_None); + } $action2 f = ($action2)self->on_error; f->$class->__asyn__(f, self, to$str(errmsg)); } @@ -311,6 +317,9 @@ static void after_shutdown(uv_shutdown_t* req, int status) { uv_stream_t *stream = (uv_stream_t *)from$int(self->_sock); // fd == -1 means invalid FD and can happen after __resume__ if (stream == -1) + // TODO: should we dispatch the on_close callback here too even though + // we did not close anything? This is what we do for TLSConnection too + // and it allows for chaining the callbacks return $R_CONT(c$cont, B_None); log_debug("Closing TCP connection"); @@ -629,9 +638,11 @@ void tls_write_cb(uv_write_t *wreq, int status) { $R netQ_TLSConnectionD_closeG_local (netQ_TLSConnection self, $Cont c$cont, $action on_close) { uv_stream_t *stream = (uv_stream_t *)from$int(self->_stream); - // fd == -1 means invalid FD and can happen after __resume__ - if (stream == -1) + // fd == -1 means invalid FD and can happen after __resume__ or if the socket is closed + if (stream == -1) { + on_close->$class->__asyn__(on_close, self); return $R_CONT(c$cont, B_None); + } self->_on_close = on_close; @@ -645,9 +656,14 @@ void tls_write_cb(uv_write_t *wreq, int status) { $R netQ_TLSConnectionD_writeG_local (netQ_TLSConnection self, $Cont c$cont, B_bytes data) { uv_stream_t *stream = (uv_stream_t *)from$int(self->_stream); - // fd == -1 means invalid FD and can happen after __resume__ - if (stream == -1) + // fd == -1 means invalid FD and can happen after __resume__ or if the socket is closed + if (stream == -1) { + // Raise an exception and let the caller retry + char errmsg[] = "Failed to write to TLS TCP socket: bad stream"; + log_debug(errmsg); + $RAISE(((B_BaseException)B_RuntimeErrorG_new(to$str(errmsg)))); return $R_CONT(c$cont, B_None); + } uv_write_t *wreq = (uv_write_t *)malloc(sizeof(uv_write_t)); wreq->data = self; From b632e7ef1a35c5479062606eb0818b607db03cc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Zago=C5=BEen?= Date: Mon, 20 Nov 2023 05:45:44 +0100 Subject: [PATCH 2/4] Add more TODOs for limiting auto reconnects Add some basic heuristics, like when we create a http.Client the first time we obviously connect directly then we expect to run at least one query. Maybe we should not reconnect after that (in case of HTTP / 1.0 or not having persistent in later HTTP versions) until there is a second query. Now if we see a second query we can assume "there are multiple requests" and thus reconnect directly after the 2nd and subsequent requests. --- base/src/http.act | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/base/src/http.act b/base/src/http.act index a21def66e..e5b708d6c 100644 --- a/base/src/http.act +++ b/base/src/http.act @@ -513,6 +513,10 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ if "connection" in r.headers and r.headers["connection"] == "close": # Is this really the right thing to do here? If the client # does not make a new request we just reconnected for nothing! + # TODO: inspect the _on_response queue to see if there are more requests + # TODO: for HTTP/1.0 do not reconnect right after the first + # request, wait until the 2nd query and then assume there + # will be more. _log.debug("Closing TCP connection due to header: Connection: close", None) _conn_reconnect() if len(_on_response) == 0: From 6f95d979d4ba1bfc65fa3590ba03b5c9a9e90a78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Zago=C5=BEen?= Date: Mon, 20 Nov 2023 06:41:02 +0100 Subject: [PATCH 3/4] Add ConnectionError to builtins --- base/builtin/registration.c | 1 + base/builtin/registration.h | 25 +++++++++++++------------ base/src/__builtin__.act | 3 +++ 3 files changed, 17 insertions(+), 12 deletions(-) diff --git a/base/builtin/registration.c b/base/builtin/registration.c index 657161e67..803acd0a1 100644 --- a/base/builtin/registration.c +++ b/base/builtin/registration.c @@ -79,6 +79,7 @@ void $register_builtin() { $register_force(MEMORYERROR_ID,&B_MemoryErrorG_methods); $register_force(OSERROR_ID,&B_OSErrorG_methods); $register_force(RUNTIMEERROR_ID,&B_RuntimeErrorG_methods); + $register_force(CONNECTIONERROR_ID,&B_ConnectionErrorG_methods); $register_force(NOTIMPLEMENTEDERROR_ID,&B_NotImplementedErrorG_methods); $register_force(VALUEERROR_ID,&B_ValueErrorG_methods); // $register_builtin_protocols(); diff --git a/base/builtin/registration.h b/base/builtin/registration.h index 5af9ad16d..b168036e0 100644 --- a/base/builtin/registration.h +++ b/base/builtin/registration.h @@ -54,22 +54,23 @@ #define KEYERROR_ID 39 #define MEMORYERROR_ID 40 #define OSERROR_ID 41 -#define RUNTIMEERROR_ID 42 -#define NOTIMPLEMENTEDERROR_ID 43 -#define VALUEERROR_ID 44 +#define CONNECTIONERROR_ID 42 +#define RUNTIMEERROR_ID 43 +#define NOTIMPLEMENTEDERROR_ID 44 +#define VALUEERROR_ID 45 -#define PROC_ID 45 -#define ACTION_ID 46 -#define MUT_ID 47 -#define PURE_ID 48 +#define PROC_ID 46 +#define ACTION_ID 47 +#define MUT_ID 48 +#define PURE_ID 49 -#define SEQ_ID 49 -#define BRK_ID 50 -#define CNT_ID 51 -#define RET_ID 52 +#define SEQ_ID 50 +#define BRK_ID 51 +#define CNT_ID 52 +#define RET_ID 53 -#define PREASSIGNED 53 +#define PREASSIGNED 54 /* diff --git a/base/src/__builtin__.act b/base/src/__builtin__.act index 5cd2688b3..16398529d 100644 --- a/base/src/__builtin__.act +++ b/base/src/__builtin__.act @@ -344,6 +344,9 @@ class MemoryError (Exception): class OSError (Exception): pass +class ConnectionError (OSError): + pass + class RuntimeError (Exception): pass From 268f073796c426d7899015537dd42e46523dda84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Zago=C5=BEen?= Date: Mon, 20 Nov 2023 06:41:52 +0100 Subject: [PATCH 4/4] Raise ConnectionError when writing to closed socket By grouping the socket errors under ConnectionError we can be more precise in handling the error in consumers of TCP connection actors. --- base/src/http.act | 9 +++------ base/src/net.ext.c | 12 ++++++------ 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/base/src/http.act b/base/src/http.act index e5b708d6c..174fc3c01 100644 --- a/base/src/http.act +++ b/base/src/http.act @@ -557,13 +557,10 @@ actor Client(cap: net.TCPConnectCap, scheme: str, address: str, port: ?int, tls_ await async tcp_conn.write(data) elif tls_conn is not None: await async tls_conn.write(data) - except RuntimeError as exc: + except ConnectionError as exc: # HTTP/1.0 servers close the connection after each request by default - if "bad file descriptor" in str(exc) or "bad stream" in str(exc): - _log.debug("TCP connection closed, reconnecting", {"error": str(exc)}) - _conn_reconnect() - else: - _on_con_error(str(exc)) + _log.debug("TCP connection closed, reconnecting", {"error": str(exc)}) + _conn_reconnect() # HTTP methods def get(path: str, headers: dict[str, str], on_response: action(Client, Response) -> None): diff --git a/base/src/net.ext.c b/base/src/net.ext.c index da3645e41..37a3a0bd6 100644 --- a/base/src/net.ext.c +++ b/base/src/net.ext.c @@ -275,13 +275,13 @@ void on_connect6(uv_connect_t *connect_req, int status) { if (r < 0) { char errmsg[1024] = "Failed to write to TCP socket: "; uv_strerror_r(r, errmsg + strlen(errmsg), sizeof(errmsg)-strlen(errmsg)); - log_warn(errmsg); - if (strstr(errmsg, "bad file descriptor")) { - // This can happen if the socket is closed, se we raise an exception - // and let the caller retry - $RAISE(((B_BaseException)B_RuntimeErrorG_new(to$str(errmsg)))); + if (r == UV_EBADF) { + // "bad file descriptor" error occurs when the socket is closed, se + // we raise an exception and let the caller retry + $RAISE(((B_BaseException)B_ConnectionErrorG_new(to$str(errmsg)))); return $R_CONT(c$cont, B_None); } + log_warn(errmsg); $action2 f = ($action2)self->on_error; f->$class->__asyn__(f, self, to$str(errmsg)); } @@ -661,7 +661,7 @@ void tls_write_cb(uv_write_t *wreq, int status) { // Raise an exception and let the caller retry char errmsg[] = "Failed to write to TLS TCP socket: bad stream"; log_debug(errmsg); - $RAISE(((B_BaseException)B_RuntimeErrorG_new(to$str(errmsg)))); + $RAISE(((B_BaseException)B_ConnectionErrorG_new(to$str(errmsg)))); return $R_CONT(c$cont, B_None); }