Move PDU parsing to RTR server threads

PDU parsing has been done by the poller up to this point.
I probably intended this to encapsulate all the stream operations in
the poller.

During 4e0f81dd5405f8c6c48d2e74c0468d16b44f9a50, the poller's block of
code that contained PDU parsing had to be moved to the mutex time.
This created a substantial incentive to migrate PDU parsing out of the
poller.

So now it's done by the RTR server threads instead.

diff --git a/src/rtr/pdu_stream.c b/src/rtr/pdu_stream.c
index d0e7cf36afcd2faa8c65b35314cd74e473ca75c2..64c2a9f87f95b76705ba0fa57ae8284453912ec4 100644 (file)
--- a/src/rtr/pdu_stream.c
+++ b/src/rtr/pdu_stream.c
@@ -47,30 +47,13 @@ struct pdu_stream *pdustream_create(int fd, char const *addr)
        result->start = result->buffer;
        result->end = result->buffer;
 
-       TAILQ_INIT(&result->requests);
-       result->reqcount = 0;
-
        return result;
 }
 
-void
-pdustream_clear_requests(struct pdu_stream *stream)
-{
-       struct rtr_request *req;
-
-       while ((req = TAILQ_FIRST(&stream->requests)) != NULL) {
-               TAILQ_REMOVE(&stream->requests, req, lh);
-               rtreq_destroy(req);
-       }
-
-       stream->reqcount = 0;
-}
-
 void
 pdustream_destroy(struct pdu_stream *stream)
 {
        close(stream->fd);
-       pdustream_clear_requests(stream);
        free(stream);
 }
 
@@ -454,22 +437,22 @@ create_request(struct pdu_stream *stream, struct pdu_header *hdr,
 }
 
 static int
-queue_request(struct pdu_stream *stream, struct rtr_request *req)
+queue_request(struct rtr_request_list *list, struct rtr_request *req)
 {
-       if (TAILQ_EMPTY(&stream->requests)) {
-               TAILQ_INSERT_HEAD(&stream->requests, req, lh);
-               stream->reqcount++;
+       if (TAILQ_EMPTY(&list->nodes)) {
+               TAILQ_INSERT_HEAD(&list->nodes, req, lh);
+               list->count++;
                return 0;
        }
 
-       if (stream->reqcount >= 4) {
+       if (list->count >= 4) {
                pr_op_err("%s: Too many simultaneous requests; Dropping RTR connection.",
-                   stream->addr);
+                   req->client_addr);
                return ENOSPC;
        }
 
-       TAILQ_INSERT_TAIL(&stream->requests, req, lh);
-       stream->reqcount++;
+       TAILQ_INSERT_TAIL(&list->nodes, req, lh);
+       list->count++;
        return 0;
 }
 
@@ -485,7 +468,7 @@ queue_request(struct pdu_stream *stream, struct rtr_request *req)
  * false: Input stream ended or broken; handle PDUs and end connection.
  */
 bool
-pdustream_parse(struct pdu_stream *stream, bool *wakeup)
+pdustream_parse(struct pdu_stream *stream, struct rtr_request_list *reqs)
 {
        enum buffer_state state;
        struct pdu_header hdr;
@@ -565,10 +548,9 @@ again:
                        goto fail;
                }
 
-               error = queue_request(stream, request);
+               error = queue_request(reqs, request);
                if (error)
                        goto fail;
-               *wakeup = true;
        }
 
        switch (state) {
@@ -589,9 +571,46 @@ fail:      if (request != NULL)
        return false;
 }
 
+void
+pdustream_disable_read(struct pdu_stream *stream)
+{
+       pr_op_debug("Shutting down input stream of client %s.", stream->addr);
+       PS_ENABLE(stream, EOS);
+       if (shutdown(stream->fd, SHUT_RD) < 0)
+               pr_op_warn("Can't shut down read end of client socket: %s",
+                   strerror(errno));
+}
+
 void
 rtreq_destroy(struct rtr_request *request)
 {
        free(request->pdu.raw.bytes);
        free(request);
 }
+
+struct rtr_request *
+rtreqlist_pop(struct rtr_request_list *reqs)
+{
+       struct rtr_request *req;
+
+       req = TAILQ_FIRST(&reqs->nodes);
+       if (req) {
+               TAILQ_REMOVE(&reqs->nodes, req, lh);
+               reqs->count--;
+       }
+
+       return req;
+}
+
+void
+rtreqlist_clear(struct rtr_request_list *list)
+{
+       struct rtr_request *req;
+
+       while ((req = TAILQ_FIRST(&list->nodes)) != NULL) {
+               TAILQ_REMOVE(&list->nodes, req, lh);
+               rtreq_destroy(req);
+       }
+
+       list->count = 0;
+}

diff --git a/src/rtr/pdu_stream.h b/src/rtr/pdu_stream.h
index 93acd396993ef73d8c123f17726a20044502167e..56d37eb060773509cb99deab1c78054c32745f66 100644 (file)
--- a/src/rtr/pdu_stream.h
+++ b/src/rtr/pdu_stream.h
@@ -15,6 +15,7 @@ struct rtr_request;
 #define PS_POLLIN(s) ((s)->flags & PSF_POLLIN)
 #define PS_CLAIMED(s) ((s)->flags & PSF_CLAIMED)
 #define PS_EOS(s) ((s)->flags & PSF_EOS)
+#define PS_NEED_POLL(s) (((s)->flags & (PSF_POLLIN | PSF_CLAIMED | PSF_EOS)) == 0)
 
 #define PS_ENABLE(s, f) (s)->flags |= PSF_##f
 #define PS_DISABLE(s, f) (s)->flags &= ~PSF_##f
@@ -22,7 +23,7 @@ struct rtr_request;
 struct pdu_stream { /* It's an *input* stream. */
        int fd;
        char addr[INET6_ADDRSTRLEN];    /* Printable address of the client. */
-       int flags;
+       int flags;                      /* Requires lock */
        int rtr_version;                /* -1: unset; > 0: version number */
        int session;                    /* -1: unset; > 0: session */
 
@@ -30,9 +31,6 @@ struct pdu_stream { /* It's an *input* stream. */
        /* buffer's active bytes */
        unsigned char *start;
        unsigned char *end;
-
-       TAILQ_HEAD(, rtr_request) requests; /* No more than 4 nodes */
-       unsigned int reqcount;
 };
 
 struct rtr_request {
@@ -62,12 +60,20 @@ struct rtr_request {
        } pdu;
 };
 
+struct rtr_request_list {
+       TAILQ_HEAD(rtr_requests, rtr_request) nodes;
+       unsigned int count;
+};
+
 struct pdu_stream *pdustream_create(int, char const *);
-void pdustream_clear_requests(struct pdu_stream *);
 void pdustream_destroy(struct pdu_stream *);
 
-bool pdustream_parse(struct pdu_stream *, bool *);
+bool pdustream_parse(struct pdu_stream *, struct rtr_request_list *);
+void pdustream_disable_read(struct pdu_stream *);
 
 void rtreq_destroy(struct rtr_request *);
 
+struct rtr_request *rtreqlist_pop(struct rtr_request_list *);
+void rtreqlist_clear(struct rtr_request_list *);
+
 #endif /* SRC_RTR_PDU_STREAM_H_ */

diff --git a/src/rtr/rtr.c b/src/rtr/rtr.c
index 1c7f4b727c23290235e89c8071759c275ece07a2..2d6b790fb9d2d250e5b01295a29d799f588bd0a0 100644 (file)
--- a/src/rtr/rtr.c
+++ b/src/rtr/rtr.c
@@ -40,7 +40,8 @@ static struct server_arraylist servers;
 static struct client_arraylist clients;
 
 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
-static pthread_cond_t parent2worker;
+static pthread_cond_t poller2worker;
+static int worker2poller[2];
 
 enum poll_verdict {
        PV_CONTINUE,
@@ -345,7 +346,8 @@ claim_client(void)
 
        ARRAYLIST_FOREACH(&clients, _client) {
                client = *_client;
-               if (!TAILQ_EMPTY(&client->requests) && !PS_CLAIMED(client)) {
+               if (PS_POLLIN(client) && !PS_CLAIMED(client)) {
+                       PS_DISABLE(client, POLLIN);
                        PS_ENABLE(client, CLAIMED);
                        return client;
                }
@@ -354,20 +356,6 @@ claim_client(void)
        return NULL;
 }
 
-static struct rtr_request *
-next_request(struct pdu_stream *client)
-{
-       struct rtr_request *req;
-
-       req = TAILQ_FIRST(&client->requests);
-       if (req) {
-               TAILQ_REMOVE(&client->requests, req, lh);
-               client->reqcount--;
-       }
-
-       return req;
-}
-
 static void
 handle_request(struct rtr_request *req)
 {
@@ -390,26 +378,49 @@ handle_request(struct rtr_request *req)
 static void *
 handle_clients(void *arg)
 {
+       static const unsigned char one = 1;
+
        struct pdu_stream *client;
+       struct rtr_request_list reqs;
        struct rtr_request *req;
 
+       TAILQ_INIT(&reqs.nodes);
+       reqs.count = 0;
+
        mutex_lock(&lock);
 
        while (!fort_end) {
                client = claim_client();
                if (!client) {
-                       panic_on_fail(pthread_cond_wait(&parent2worker, &lock),
+                       panic_on_fail(pthread_cond_wait(&poller2worker, &lock),
                            "pthread_cond_wait");
                        continue;
                }
 
-               while ((req = next_request(client)) != NULL) {
-                       mutex_unlock(&lock);
-                       handle_request(req);
-                       mutex_lock(&lock);
-               }
+               mutex_unlock(&lock);
 
-               PS_DISABLE(client, CLAIMED);
+               do {
+                       if (!pdustream_parse(client, &reqs)) {
+                               rtreqlist_clear(&reqs);
+                               mutex_lock(&lock);
+                               PS_DISABLE(client, CLAIMED);
+                               pdustream_disable_read(client);
+                               break;
+                       }
+
+                       if (TAILQ_EMPTY(&reqs.nodes)) {
+                               mutex_lock(&lock);
+                               PS_DISABLE(client, CLAIMED);
+                               break;
+                       }
+
+                       while ((req = rtreqlist_pop(&reqs)) != NULL)
+                               handle_request(req);
+               } while (true);
+
+               if (write(worker2poller[1], &one, 1) < 0)
+                       pr_op_warn("Cannot wake up RTR server thread: %s",
+                           strerror(errno));
        }
 
        mutex_unlock(&lock);
@@ -599,17 +610,6 @@ apply_pollfds(struct pollfd *pollfds, size_t nclients)
        delete_dead_clients();
 }
 
-static void
-disable_read(struct pdu_stream *stream)
-{
-       pr_op_debug("Shutting down input stream of client %s.", stream->addr);
-       if (shutdown(stream->fd, SHUT_RD) < 0)
-               pr_op_warn("Can't shut down read end of client socket: %s",
-                   strerror(errno));
-       pdustream_clear_requests(stream);
-       PS_ENABLE(stream, EOS);
-}
-
 static enum poll_verdict
 fddb_poll(void)
 {
@@ -618,20 +618,26 @@ fddb_poll(void)
        struct pdu_stream *client;
        size_t nclients;
        size_t i;
-       bool wakeup;
+       bool wakeup_threads;
+       unsigned char trash[16];
        int error;
 
-       pollfds = pcalloc(servers.len + clients.len, sizeof(struct pollfd));
+       pollfds = pcalloc(servers.len + clients.len + 1, sizeof(struct pollfd));
 
        ARRAYLIST_FOREACH_IDX(&servers, i)
                init_pollfd(&pollfds[i], servers.array[i].fd);
-       ARRAYLIST_FOREACH_IDX(&clients, i)
+       mutex_lock(&lock);
+       ARRAYLIST_FOREACH_IDX(&clients, i) {
+               client = clients.array[i];
                init_pollfd(
                    &pollfds[servers.len + i],
-                   PS_EOS(clients.array[i]) ? -1 : clients.array[i]->fd
+                   PS_NEED_POLL(client) ? client->fd : -1
                );
+       }
+       mutex_unlock(&lock);
+       init_pollfd(&pollfds[servers.len + clients.len], worker2poller[0]);
 
-       error = poll(pollfds, servers.len + clients.len, 1000);
+       error = poll(pollfds, servers.len + clients.len + 1, 1000);
 
        if (fort_end)
                goto stop;
@@ -658,6 +664,19 @@ fddb_poll(void)
        /* accept_new_client() might change this number, so store a backup. */
        nclients = clients.len;
 
+       /* Client thread to poller wakeupper */
+       fd = &pollfds[servers.len + clients.len];
+       if (fd->revents & POLLIN) {
+               if (read(fd->fd, trash, sizeof(trash)) < 0)
+                       pr_op_warn("Cannot read RTR server wakeupper: %s",
+                           strerror(errno));
+       }
+       if (fd->revents & (POLLHUP | POLLERR | POLLNVAL)) {
+               print_poll_failure(fd, "Wakeupper", "local");
+               close(worker2poller[0]);
+               worker2poller[0] = -1;
+       }
+
        /* New connections */
        for (i = 0; i < servers.len; i++) {
                /* This fd is a listening socket. */
@@ -685,7 +704,7 @@ fddb_poll(void)
        }
 
        /* Client requests */
-       wakeup = false;
+       wakeup_threads = false;
        mutex_lock(&lock);
 
        for (i = 0; i < nclients; i++) {
@@ -701,19 +720,19 @@ fddb_poll(void)
 
                if (fd->revents & (POLLHUP | POLLERR | POLLNVAL)) {
                        print_poll_failure(fd, "Client", client->addr);
-                       disable_read(client);
+                       pdustream_disable_read(client);
 
                } else if (fd->revents & POLLIN) {
-                       if (!pdustream_parse(client, &wakeup))
-                               disable_read(client);
+                       PS_ENABLE(client, POLLIN);
+                       wakeup_threads = true;
                }
        }
 
        apply_pollfds(pollfds, nclients);
        nclients = clients.len;
 
-       if (wakeup)
-               panic_on_fail(pthread_cond_broadcast(&parent2worker),
+       if (wakeup_threads)
+               panic_on_fail(pthread_cond_broadcast(&poller2worker),
                    "pthread_cond_broadcast");
 
        mutex_unlock(&lock);
@@ -755,13 +774,16 @@ end_server_threads(size_t count)
        int error;
 
        fort_end = true;
-       pthread_cond_broadcast(&parent2worker);
+       pthread_cond_broadcast(&poller2worker);
        for (i = 0; i < count; i++) {
                error = pthread_join(server_threads[i], NULL);
                if (error)
                        pr_op_warn("pthread_join: %s", strerror(error));
                pr_op_debug("Ended RTR server thread #%zu.", i + 1);
        }
+
+       free(server_threads);
+       server_threads = NULL;
 }
 
 int
@@ -777,13 +799,20 @@ rtr_start(void)
 
        error = init_server_fds();
        if (error)
-               goto revert_fds;
+               goto fds;
 
-       error = pthread_cond_init(&parent2worker, NULL);
+       error = pthread_cond_init(&poller2worker, NULL);
        if (error) {
                pr_op_err_st("pthread_cond_init(p2w) returned error %d: %s",
                    error, strerror(error));
-               goto revert_fds;
+               goto fds;
+       }
+
+       if (pipe(worker2poller) < 0) {
+               error = errno;
+               pr_op_err_st("Cannot create RTR worker-poller pipe: %s",
+                   strerror(error));
+               goto cond;
        }
 
        server_threads = pcalloc(config_get_thread_pool_server_max(),
@@ -794,7 +823,7 @@ rtr_start(void)
                if (error) {
                        pr_op_err_st("pthread_create() returned error %d: %s",
                            error, strerror(error));
-                       goto revert_threads;
+                       goto threads;
                }
 
                pr_op_debug("Spawned RTR server thread #%zu.", i + 1);
@@ -802,13 +831,17 @@ rtr_start(void)
 
        error = pthread_create(&control_thread, NULL, control_cb, NULL);
        if (error)
-               goto revert_threads;
+               goto threads;
 
        return 0;
 
-revert_threads:
+threads:
        end_server_threads(i);
-revert_fds:
+       close(worker2poller[0]);
+       close(worker2poller[1]);
+cond:
+       pthread_cond_destroy(&poller2worker);
+fds:
        destroy_db();
        return error;
 }
@@ -825,8 +858,11 @@ void rtr_stop(void)
                    strerror(error));
        }
 
-       free(server_threads);
-       pthread_cond_destroy(&parent2worker);
+       if (worker2poller[0] != -1)
+               close(worker2poller[0]);
+       close(worker2poller[1]);
+
+       pthread_cond_destroy(&poller2worker);
        destroy_db();
 }