daemon: server-side tls: use asynchronous network io model

diff --git a/daemon/tls.c b/daemon/tls.c
index 03b601fd735babe7583dd2f2e9629c72797d2189..b5378f4d4c6771cdac1ea69bbbac9ad75d63ba3b 100644 (file)
--- a/daemon/tls.c
+++ b/daemon/tls.c
@@ -38,21 +38,6 @@
 #define EPHEMERAL_CERT_EXPIRATION_SECONDS_RENEW_BEFORE 60*60*24*7
 #define GNUTLS_PIN_MIN_VERSION  0x030400
 
-/* gnutls_record_recv and gnutls_record_send */
-struct tls_ctx_t {
-       gnutls_session_t session;
-       bool handshake_done;
-
-       uv_stream_t *handle;
-
-       /* for reading from the network */
-       const uint8_t *buf;
-       ssize_t nread;
-       ssize_t consumed;
-       uint8_t recv_buf[4096];
-       struct tls_credentials *credentials;
-};
-
 struct tls_client_ctx_t {
        gnutls_session_t tls_session;
        tls_client_hs_state_t handshake_state;
@@ -201,8 +186,10 @@ struct tls_ctx_t *tls_new(struct worker_ctx *worker)
                return NULL;
        }
 
+       tls->worker = worker;
+
        gnutls_transport_set_pull_function(tls->session, kres_gnutls_pull);
-       gnutls_transport_set_push_function(tls->session, kres_gnutls_push);
+       gnutls_transport_set_push_function(tls->session, worker_gnutls_push);
        gnutls_transport_set_ptr(tls->session, tls);
        return tls;
 }
@@ -237,6 +224,10 @@ int tls_push(struct qr_task *task, uv_handle_t *handle, knot_pkt_t *pkt)
                return kr_error(ENOENT);
        }
 
+       assert(gnutls_record_check_corked(tls_p->session) == 0);
+
+       tls_p->task = task;
+
        gnutls_record_cork(tls_p->session);
        ssize_t count = 0;
        if ((count = gnutls_record_send(tls_p->session, &pkt_size, sizeof(pkt_size)) < 0) ||
@@ -260,14 +251,19 @@ int tls_push(struct qr_task *task, uv_handle_t *handle, knot_pkt_t *pkt)
                                             retries, gnutls_strerror_name(count), count);
                                return kr_error(EIO);
                        }
-               } else {
-                       retries = 0;
+               } else if (count != 0) {
                        submitted += count;
-                       if (count == 0 && submitted != sizeof(pkt_size) + pkt->size) {
-                               kr_log_error("[tls] gnutls_record_uncork didn't send all data: %s (%zd)\n",
-                                            gnutls_strerror_name(count), count);
+                       retries = 0;
+               } else if (gnutls_record_check_corked(tls_p->session) != 0) {
+                       if (++retries > TLS_MAX_UNCORK_RETRIES) {
+                               kr_log_error("[tls] gnutls_record_uncork: too many retries (%zd)\n",
+                                            retries);
                                return kr_error(EIO);
                        }
+               } else if (submitted != sizeof(pkt_size) + pkt->size) {
+                       kr_log_error("[tls] gnutls_record_uncork didn't send all data(%zd of %zd)\n",
+                                    submitted, sizeof(pkt_size) + pkt->size);
+                       return kr_error(EIO);
                }
        } while (submitted != sizeof(pkt_size) + pkt->size);
 

diff --git a/daemon/tls.h b/daemon/tls.h
index 79f9665045f0f10846d8e0a08b28e8c98199423e..45ffcb142cc286ae85790bc1e6a4f8d9e64e7fcc 100644 (file)
--- a/daemon/tls.h
+++ b/daemon/tls.h
@@ -44,6 +44,26 @@ struct tls_client_paramlist_entry {
        gnutls_certificate_credentials_t credentials;
 };
 
+struct worker_ctx;
+struct qr_task;
+
+/* gnutls_record_recv and gnutls_record_send */
+struct tls_ctx_t {
+       gnutls_session_t session;
+       bool handshake_done;
+
+       uv_stream_t *handle;
+
+       /* for reading from the network */
+       const uint8_t *buf;
+       ssize_t nread;
+       ssize_t consumed;
+       uint8_t recv_buf[4096];
+       struct tls_credentials *credentials;
+       struct worker_ctx *worker;
+       struct qr_task *task;
+};
+
 typedef enum tls_client_hs_state {
        TLS_HS_NOT_STARTED = 0,
        TLS_HS_IN_PROGRESS,

diff --git a/daemon/worker.c b/daemon/worker.c
index 7f099741fc8f8809a11271b03c3d254c954f2dc7..e35524b1471e6ef75bb7ef81dbf62cb373195e9f 100644 (file)
--- a/daemon/worker.c
+++ b/daemon/worker.c
@@ -870,7 +870,7 @@ static void on_send(uv_udp_send_t *req, int status)
        iorequest_release(worker, req);
 }
 
-static void on_write(uv_write_t *req, int status)
+void on_write(uv_write_t *req, int status)
 {
        uv_handle_t *handle = (uv_handle_t *)(req->handle);
        uv_loop_t *loop = handle->loop;
@@ -882,6 +882,71 @@ static void on_write(uv_write_t *req, int status)
        iorequest_release(worker, req);
 }
 
+ssize_t worker_gnutls_push(gnutls_transport_ptr_t h, const void *buf, size_t len)
+{
+       struct tls_ctx_t *t = (struct tls_ctx_t *)h;
+       const uv_buf_t ub = {(void *)buf, len};
+
+       VERBOSE_MSG(NULL,"[tls] push %zu <%p>\n", len, h);
+       if (t == NULL) {
+               errno = EFAULT;
+               return -1;
+       }
+
+       assert(t->handle);
+       assert(t->handle->type == UV_TCP);
+
+       if (!t->handshake_done) {
+               int ret = uv_try_write(t->handle, &ub, 1);
+               if (ret > 0) {
+                       return (ssize_t) ret;
+               }
+               if (ret == UV_EAGAIN) {
+                       errno = EAGAIN;
+               } else {
+                       kr_log_error("[tls] uv_try_write: %s\n", uv_strerror(ret));
+                       errno = EIO;
+               }
+               return -1;
+       }
+
+       struct worker_ctx *worker = t->worker;
+       struct qr_task *task = t->task;
+
+       assert(worker && task);
+
+       void *ioreq = worker_iohandle_borrow(worker);
+       if (!ioreq) {
+               errno = EFAULT;
+               return -1;
+       }
+
+       uv_write_t *write_req = (uv_write_t *)ioreq;
+       uv_buf_t uv_buf[1] = {
+               { (char *)buf, len }
+       };
+
+       write_req->data = task;
+
+       ssize_t ret = -1;
+       int res = uv_write(write_req, t->handle, uv_buf, 1, &on_write);
+       if (res == 0) {
+               qr_task_ref(task); /* Pending ioreq on current task */
+               if (worker->too_many_open &&
+                   worker->stats.rconcurrent <
+                       worker->rconcurrent_highwatermark - 10) {
+                       worker->too_many_open = false;
+               }
+               ret = len;
+       } else {
+               VERBOSE_MSG(NULL,"[tls] uv_write: %s\n", uv_strerror(res));
+               iorequest_release(worker, ioreq);
+               errno = EIO;
+               /* TODO ret == UV_EMFILE */
+       }
+       return ret;
+}
+
 static int qr_task_send(struct qr_task *task, uv_handle_t *handle, struct sockaddr *addr, knot_pkt_t *pkt)
 {
        if (!handle) {
@@ -904,7 +969,7 @@ static int qr_task_send(struct qr_task *task, uv_handle_t *handle, struct sockad
                        }
                        ret = tls_client_push(task, handle, pkt);
                }
-               return qr_task_on_send(task, handle, ret);
+               return ret;
        }
 
        int ret = 0;

diff --git a/daemon/worker.h b/daemon/worker.h
index 56e04b4dd59acf15e1ad9907ade25375740e86bc..cac2a624423814a7ba4caba27584704c7907b83b 100644 (file)
--- a/daemon/worker.h
+++ b/daemon/worker.h
@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include <gnutls/gnutls.h>
+
 #include "daemon/engine.h"
 #include "lib/generic/array.h"
 #include "lib/generic/map.h"
@@ -88,7 +90,7 @@ void *worker_iohandle_borrow(struct worker_ctx *worker);
 
 void worker_iohandle_release(struct worker_ctx *worker, void *h);
 
-
+ssize_t worker_gnutls_push(gnutls_transport_ptr_t h, const void *buf, size_t len);
 
 /** @cond internal */