diff options
| author | Garrett D'Amore <garrett@damore.org> | 2018-07-09 09:59:46 -0700 |
|---|---|---|
| committer | Garrett D'Amore <garrett@damore.org> | 2018-07-16 10:06:50 -0700 |
| commit | b44e20c80c936a29bfeaf964ec94bc62ac0386f5 (patch) | |
| tree | 87b2b5b999046b7f10789d4bae863eeea9354e44 /src/supplemental/http | |
| parent | 05f404b917ddaf9fee70208a796cdf66ee747050 (diff) | |
| download | nng-b44e20c80c936a29bfeaf964ec94bc62ac0386f5.tar.gz nng-b44e20c80c936a29bfeaf964ec94bc62ac0386f5.tar.bz2 nng-b44e20c80c936a29bfeaf964ec94bc62ac0386f5.zip | |
fixes #523 dialers could support multiple outstanding dial requests
fixes #179 DNS resolution should be done at connect time
fixes #586 Windows IO completion port work could be better
fixes #339 Windows iocp could use synchronous completions
fixes #280 TCP abstraction improvements
This is a rather monstrous set of changes, which refactors TCP, and
the underlying Windows I/O completion path logic, in order to obtain
a cleaner, simpler API, with support for asynchronous DNS lookups performed
on connect rather than initialization time, the ability to have multiple
connects or accepts pending, as well as fewer extraneous function calls.
The Windows code also benefits from greatly reduced context switching,
fewer lock operations performed, and a reduced number of system calls
on the hot code path. (We use automatic event resetting instead of manual.)
Some dead code was removed as well, and a few potential edge case leaks
on failure paths (in the websocket code) were plugged.
Note that all TCP based transports benefit from this work. The IPC code
on Windows still uses the legacy IOCP for now, as does the UDP code (used
for ZeroTier.) We will be converting those soon too.
Diffstat (limited to 'src/supplemental/http')
| -rw-r--r-- | src/supplemental/http/http_api.h | 4 | ||||
| -rw-r--r-- | src/supplemental/http/http_client.c | 129 | ||||
| -rw-r--r-- | src/supplemental/http/http_conn.c | 57 | ||||
| -rw-r--r-- | src/supplemental/http/http_server.c | 75 |
4 files changed, 138 insertions, 127 deletions
diff --git a/src/supplemental/http/http_api.h b/src/supplemental/http/http_api.h index 4b515ca5..cf2c78bf 100644 --- a/src/supplemental/http/http_api.h +++ b/src/supplemental/http/http_api.h @@ -67,9 +67,9 @@ extern void *nni_http_conn_get_ctx(nni_http_conn *); // These initialization functions create stream for HTTP transactions. // They should only be used by the server or client HTTP implementations, // and are not for use by other code. -extern int nni_http_conn_init_tcp(nni_http_conn **, void *); +extern int nni_http_conn_init_tcp(nni_http_conn **, nni_tcp_conn *); extern int nni_http_conn_init_tls( - nni_http_conn **, struct nng_tls_config *, void *); + nni_http_conn **, struct nng_tls_config *, nni_tcp_conn *); extern void nni_http_conn_close(nni_http_conn *); extern void nni_http_conn_fini(nni_http_conn *); diff --git a/src/supplemental/http/http_client.c b/src/supplemental/http/http_client.c index 4034bbff..6842b271 100644 --- a/src/supplemental/http/http_client.c +++ b/src/supplemental/http/http_client.c @@ -20,72 +20,97 @@ #include "http_api.h" struct nng_http_client { - nni_list aios; - nni_mtx mtx; - bool closed; - struct nng_tls_config *tls; - nni_aio * connaio; - nni_plat_tcp_ep * tep; + nni_list aios; + nni_mtx mtx; + bool closed; + bool resolving; + nng_tls_config *tls; + nni_aio * aio; + nng_sockaddr sa; + nni_tcp_dialer *dialer; + char * host; + char * port; + nni_url * url; }; static void -http_conn_start(nni_http_client *c) +http_dial_start(nni_http_client *c) { - nni_plat_tcp_ep_connect(c->tep, c->connaio); + nni_aio *aio; + + if ((aio = nni_list_first(&c->aios)) == NULL) { + return; + } + c->resolving = true; + nni_aio_set_input(c->aio, 0, &c->sa); + nni_tcp_resolv(c->host, c->port, NNG_AF_UNSPEC, 0, c->aio); } static void -http_conn_done(void *arg) +http_dial_cb(void *arg) { - nni_http_client * c = arg; - nni_aio * aio; - int rv; - nni_plat_tcp_pipe *p; - nni_http_conn * conn; + nni_http_client *c = arg; + nni_aio * aio; + int rv; + nni_tcp_conn * tcp; + nni_http_conn * conn; nni_mtx_lock(&c->mtx); - rv = nni_aio_result(c->connaio); - p = rv == 0 ? nni_aio_get_output(c->connaio, 0) : NULL; + rv = nni_aio_result(c->aio); + if ((aio = nni_list_first(&c->aios)) == NULL) { - if (p != NULL) { - nni_plat_tcp_pipe_fini(p); - } + // User abandoned request, and no residuals left. nni_mtx_unlock(&c->mtx); + if ((rv == 0) && !c->resolving) { + tcp = nni_aio_get_output(c->aio, 0); + nni_tcp_conn_fini(tcp); + } return; } - nni_aio_list_remove(aio); if (rv != 0) { + nni_aio_list_remove(aio); + http_dial_start(c); + nni_mtx_unlock(&c->mtx); nni_aio_finish_error(aio, rv); + return; + } + + if (c->resolving) { + // This was a DNS lookup -- advance to normal TCP connect. + c->resolving = false; + nni_tcp_dialer_dial(c->dialer, &c->sa, c->aio); nni_mtx_unlock(&c->mtx); return; } + nni_aio_list_remove(aio); + tcp = nni_aio_get_output(c->aio, 0); + NNI_ASSERT(tcp != NULL); + if (c->tls != NULL) { - rv = nni_http_conn_init_tls(&conn, c->tls, p); + rv = nni_http_conn_init_tls(&conn, c->tls, tcp); } else { - rv = nni_http_conn_init_tcp(&conn, p); + rv = nni_http_conn_init_tcp(&conn, tcp); } + http_dial_start(c); + nni_mtx_unlock(&c->mtx); + if (rv != 0) { + // the conn_init function will have already discard tcp. nni_aio_finish_error(aio, rv); - nni_mtx_unlock(&c->mtx); return; } nni_aio_set_output(aio, 0, conn); nni_aio_finish(aio, 0, 0); - - if (!nni_list_empty(&c->aios)) { - http_conn_start(c); - } - nni_mtx_unlock(&c->mtx); } void nni_http_client_fini(nni_http_client *c) { - nni_aio_fini(c->connaio); - nni_plat_tcp_ep_fini(c->tep); + nni_aio_fini(c->aio); + nni_tcp_dialer_fini(c->dialer); nni_mtx_fini(&c->mtx); #ifdef NNG_SUPP_TLS if (c->tls != NULL) { @@ -100,10 +125,6 @@ nni_http_client_init(nni_http_client **cp, const nni_url *url) { int rv; nni_http_client *c; - nni_aio * aio; - nni_sockaddr sa; - char * host; - char * port; if (strlen(url->u_hostname) == 0) { // We require a valid hostname. @@ -118,29 +139,17 @@ nni_http_client_init(nni_http_client **cp, const nni_url *url) return (NNG_EADDRINVAL); } - // For now we are looking up the address. We would really like - // to do this later, but we need TcP support for this. One - // imagines the ability to create a tcp dialer that does the - // necessary DNS lookups, etc. all asynchronously. - if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) { - return (rv); - } - nni_aio_set_input(aio, 0, &sa); - host = (strlen(url->u_hostname) != 0) ? url->u_hostname : NULL; - port = (strlen(url->u_port) != 0) ? url->u_port : NULL; - nni_plat_tcp_resolv(host, port, NNG_AF_UNSPEC, false, aio); - nni_aio_wait(aio); - rv = nni_aio_result(aio); - nni_aio_fini(aio); - if (rv != 0) { - return (rv); - } - if ((c = NNI_ALLOC_STRUCT(c)) == NULL) { return (NNG_ENOMEM); } nni_mtx_init(&c->mtx); nni_aio_list_init(&c->aios); + if (((c->host = nni_strdup(url->u_hostname)) == NULL) || + ((strlen(url->u_port) != 0) && + ((c->port = nni_strdup(url->u_port)) == NULL))) { + nni_http_client_fini(c); + return (NNG_ENOMEM); + } #ifdef NNG_SUPP_TLS if ((strcmp(url->u_scheme, "https") == 0) || @@ -167,16 +176,12 @@ nni_http_client_init(nni_http_client **cp, const nni_url *url) } #endif - rv = nni_plat_tcp_ep_init(&c->tep, NULL, &sa, NNI_EP_MODE_DIAL); - if (rv != 0) { + if (((rv = nni_tcp_dialer_init(&c->dialer)) != 0) || + ((rv = nni_aio_init(&c->aio, http_dial_cb, c)) != 0)) { nni_http_client_fini(c); return (rv); } - if ((rv = nni_aio_init(&c->connaio, http_conn_done, c)) != 0) { - nni_http_client_fini(c); - return (rv); - } *cp = c; return (0); } @@ -224,7 +229,7 @@ nni_http_client_get_tls(nni_http_client *c, struct nng_tls_config **tlsp) } static void -http_connect_cancel(nni_aio *aio, int rv) +http_dial_cancel(nni_aio *aio, int rv) { nni_http_client *c = nni_aio_get_prov_data(aio); nni_mtx_lock(&c->mtx); @@ -233,7 +238,7 @@ http_connect_cancel(nni_aio *aio, int rv) nni_aio_finish_error(aio, rv); } if (nni_list_empty(&c->aios)) { - nni_aio_abort(c->connaio, rv); + nni_aio_abort(c->aio, rv); } nni_mtx_unlock(&c->mtx); } @@ -246,14 +251,14 @@ nni_http_client_connect(nni_http_client *c, nni_aio *aio) return; } nni_mtx_lock(&c->mtx); - if ((rv = nni_aio_schedule(aio, http_connect_cancel, c)) != 0) { + if ((rv = nni_aio_schedule(aio, http_dial_cancel, c)) != 0) { nni_mtx_unlock(&c->mtx); nni_aio_finish_error(aio, rv); return; } nni_list_append(&c->aios, aio); if (nni_list_first(&c->aios) == aio) { - http_conn_start(c); + http_dial_start(c); } nni_mtx_unlock(&c->mtx); } diff --git a/src/supplemental/http/http_conn.c b/src/supplemental/http/http_conn.c index 15d1f776..169918e9 100644 --- a/src/supplemental/http/http_conn.c +++ b/src/supplemental/http/http_conn.c @@ -699,17 +699,23 @@ http_init(nni_http_conn **connp, nni_http_tran *tran, void *data) if ((conn = NNI_ALLOC_STRUCT(conn)) == NULL) { return (NNG_ENOMEM); } - conn->rd_bufsz = HTTP_BUFSIZE; - if ((conn->rd_buf = nni_alloc(conn->rd_bufsz)) == NULL) { - NNI_FREE_STRUCT(conn); - return (NNG_ENOMEM); - } nni_mtx_init(&conn->mtx); nni_aio_list_init(&conn->rdq); nni_aio_list_init(&conn->wrq); + if ((conn->rd_buf = nni_alloc(HTTP_BUFSIZE)) == NULL) { + nni_http_conn_fini(conn); + return (NNG_ENOMEM); + } + conn->rd_bufsz = HTTP_BUFSIZE; + + if (((rv = nni_aio_init(&conn->wr_aio, http_wr_cb, conn)) != 0) || + ((rv = nni_aio_init(&conn->rd_aio, http_rd_cb, conn)) != 0)) { + nni_http_conn_fini(conn); + return (rv); + } + conn->sock = data; - conn->rd_bufsz = HTTP_BUFSIZE; conn->rd = tran->h_read; conn->wr = tran->h_write; conn->close = tran->h_close; @@ -718,12 +724,6 @@ http_init(nni_http_conn **connp, nni_http_tran *tran, void *data) conn->peer_addr = tran->h_peer_addr; conn->verified = tran->h_verified; - if (((rv = nni_aio_init(&conn->wr_aio, http_wr_cb, conn)) != 0) || - ((rv = nni_aio_init(&conn->rd_aio, http_rd_cb, conn)) != 0)) { - nni_http_conn_fini(conn); - return (rv); - } - *connp = conn; return (0); @@ -737,19 +737,23 @@ nni_http_verified_tcp(void *arg) } static nni_http_tran http_tcp_ops = { - .h_read = (http_read_fn) nni_plat_tcp_pipe_recv, - .h_write = (http_write_fn) nni_plat_tcp_pipe_send, - .h_close = (http_close_fn) nni_plat_tcp_pipe_close, - .h_fini = (http_fini_fn) nni_plat_tcp_pipe_fini, - .h_sock_addr = (http_addr_fn) nni_plat_tcp_pipe_sockname, - .h_peer_addr = (http_addr_fn) nni_plat_tcp_pipe_peername, + .h_read = (http_read_fn) nni_tcp_conn_recv, + .h_write = (http_write_fn) nni_tcp_conn_send, + .h_close = (http_close_fn) nni_tcp_conn_close, + .h_fini = (http_fini_fn) nni_tcp_conn_fini, + .h_sock_addr = (http_addr_fn) nni_tcp_conn_sockname, + .h_peer_addr = (http_addr_fn) nni_tcp_conn_peername, .h_verified = (http_verified_fn) nni_http_verified_tcp, }; int -nni_http_conn_init_tcp(nni_http_conn **connp, void *tcp) +nni_http_conn_init_tcp(nni_http_conn **connp, nni_tcp_conn *tcp) { - return (http_init(connp, &http_tcp_ops, tcp)); + int rv; + if ((rv = http_init(connp, &http_tcp_ops, tcp)) != 0) { + nni_tcp_conn_fini(tcp); + } + return (rv); } #ifdef NNG_SUPP_TLS @@ -765,26 +769,29 @@ static nni_http_tran http_tls_ops = { int nni_http_conn_init_tls( - nni_http_conn **connp, struct nng_tls_config *cfg, void *tcp) + nni_http_conn **connp, struct nng_tls_config *cfg, nni_tcp_conn *tcp) { nni_tls *tls; int rv; if ((rv = nni_tls_init(&tls, cfg, tcp)) != 0) { - nni_plat_tcp_pipe_fini(tcp); + nni_tcp_conn_fini(tcp); return (rv); } - return (http_init(connp, &http_tls_ops, tls)); + if ((rv = http_init(connp, &http_tls_ops, tls)) != 0) { + nni_tls_fini(tls); + } + return (rv); } #else int nni_http_conn_init_tls( - nni_http_conn **connp, struct nng_tls_config *cfg, void *tcp) + nni_http_conn **connp, struct nng_tls_config *cfg, nni_tcp_conn *tcp) { NNI_ARG_UNUSED(connp); NNI_ARG_UNUSED(cfg); - nni_plat_tcp_pipe_fini(tcp); + nni_tcp_conn_fini(tcp); return (NNG_ENOTSUP); } #endif // NNG_SUPP_TLS diff --git a/src/supplemental/http/http_server.c b/src/supplemental/http/http_server.c index 4a07d544..9d1313b1 100644 --- a/src/supplemental/http/http_server.c +++ b/src/supplemental/http/http_server.c @@ -64,22 +64,22 @@ typedef struct http_error { } http_error; struct nng_http_server { - nng_sockaddr addr; - nni_list_node node; - int refcnt; - int starts; - nni_list handlers; - nni_list conns; - nni_mtx mtx; - nni_cv cv; - bool closed; - nng_tls_config * tls; - nni_aio * accaio; - nni_plat_tcp_ep *tep; - char * port; - char * hostname; - nni_list errors; - nni_mtx errors_mtx; + nng_sockaddr addr; + nni_list_node node; + int refcnt; + int starts; + nni_list handlers; + nni_list conns; + nni_mtx mtx; + nni_cv cv; + bool closed; + nng_tls_config * tls; + nni_aio * accaio; + nni_tcp_listener *listener; + char * port; + char * hostname; + nni_list errors; + nni_mtx errors_mtx; }; int @@ -682,13 +682,13 @@ http_sconn_cbdone(void *arg) } static int -http_sconn_init(http_sconn **scp, nni_http_server *s, nni_plat_tcp_pipe *tcp) +http_sconn_init(http_sconn **scp, nni_http_server *s, nni_tcp_conn *tcp) { http_sconn *sc; int rv; if ((sc = NNI_ALLOC_STRUCT(sc)) == NULL) { - nni_plat_tcp_pipe_fini(tcp); + nni_tcp_conn_fini(tcp); return (NNG_ENOMEM); } @@ -720,17 +720,17 @@ http_sconn_init(http_sconn **scp, nni_http_server *s, nni_plat_tcp_pipe *tcp) static void http_server_acccb(void *arg) { - nni_http_server * s = arg; - nni_aio * aio = s->accaio; - nni_plat_tcp_pipe *tcp; - http_sconn * sc; - int rv; + nni_http_server *s = arg; + nni_aio * aio = s->accaio; + nni_tcp_conn * tcp; + http_sconn * sc; + int rv; nni_mtx_lock(&s->mtx); if ((rv = nni_aio_result(aio)) != 0) { if (!s->closed) { // try again? - nni_plat_tcp_ep_accept(s->tep, s->accaio); + nni_tcp_listener_accept(s->listener, s->accaio); } nni_mtx_unlock(&s->mtx); return; @@ -738,14 +738,14 @@ http_server_acccb(void *arg) tcp = nni_aio_get_output(aio, 0); if (s->closed) { // If we're closing, then reject this one. - nni_plat_tcp_pipe_fini(tcp); + nni_tcp_conn_fini(tcp); nni_mtx_unlock(&s->mtx); return; } if (http_sconn_init(&sc, s, tcp) != 0) { // The TCP structure is already cleaned up. // Start another accept attempt. - nni_plat_tcp_ep_accept(s->tep, s->accaio); + nni_tcp_listener_accept(s->listener, s->accaio); nni_mtx_unlock(&s->mtx); return; } @@ -753,7 +753,7 @@ http_server_acccb(void *arg) nni_list_append(&s->conns, sc); nni_http_read_req(sc->conn, sc->req, sc->rxaio); - nni_plat_tcp_ep_accept(s->tep, s->accaio); + nni_tcp_listener_accept(s->listener, s->accaio); nni_mtx_unlock(&s->mtx); } @@ -769,8 +769,8 @@ http_server_fini(nni_http_server *s) while (!nni_list_empty(&s->conns)) { nni_cv_wait(&s->cv); } - if (s->tep != NULL) { - nni_plat_tcp_ep_fini(s->tep); + if (s->listener != NULL) { + nni_tcp_listener_fini(s->listener); } while ((h = nni_list_first(&s->handlers)) != NULL) { nni_list_remove(&s->handlers, h); @@ -875,7 +875,7 @@ http_server_init(nni_http_server **serverp, const nni_url *url) return (rv); } nni_aio_set_input(aio, 0, &s->addr); - nni_plat_tcp_resolv(s->hostname, s->port, NNG_AF_UNSPEC, true, aio); + nni_tcp_resolv(s->hostname, s->port, NNG_AF_UNSPEC, true, aio); nni_aio_wait(aio); rv = nni_aio_result(aio); nni_aio_fini(aio); @@ -921,16 +921,15 @@ static int http_server_start(nni_http_server *s) { int rv; - rv = nni_plat_tcp_ep_init(&s->tep, &s->addr, NULL, NNI_EP_MODE_LISTEN); - if (rv != 0) { + if ((rv = nni_tcp_listener_init(&s->listener)) != 0) { return (rv); } - if ((rv = nni_plat_tcp_ep_listen(s->tep, NULL)) != 0) { - nni_plat_tcp_ep_fini(s->tep); - s->tep = NULL; + if ((rv = nni_tcp_listener_listen(s->listener, &s->addr)) != 0) { + nni_tcp_listener_fini(s->listener); + s->listener = NULL; return (rv); } - nni_plat_tcp_ep_accept(s->tep, s->accaio); + nni_tcp_listener_accept(s->listener, s->accaio); return (0); } @@ -961,8 +960,8 @@ http_server_stop(nni_http_server *s) s->closed = true; // Close the TCP endpoint that is listening. - if (s->tep) { - nni_plat_tcp_ep_close(s->tep); + if (s->listener) { + nni_tcp_listener_close(s->listener); } // Stopping the server is a hard stop -- it aborts any work |
