From 7c78155484932b57b2fe12dfdda288d7cf23eff6 Mon Sep 17 00:00:00 2001 From: Andrea Date: Wed, 1 Jan 2020 22:26:49 +0100 Subject: [PATCH 1/4] Added additional check for EAGAIN in recv. To be added for send to --- src/network.c | 17 +++++++- src/server.c | 110 +++++++++++++++++++++++++++----------------------- src/server.h | 1 + 3 files changed, 75 insertions(+), 53 deletions(-) diff --git a/src/network.c b/src/network.c index a7fff11..dbb39c0 100644 --- a/src/network.c +++ b/src/network.c @@ -43,6 +43,7 @@ #include "util.h" #include "config.h" #include "network.h" +#include "server.h" /* Set non-blocking socket */ int set_nonblocking(int fd) { @@ -265,7 +266,7 @@ ssize_t recv_bytes(int fd, unsigned char *buf, size_t bufsize) { if ((n = recv(fd, buf, bufsize - total, 0)) < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { - break; + goto eagain; } else goto err; } @@ -279,6 +280,12 @@ ssize_t recv_bytes(int fd, unsigned char *buf, size_t bufsize) { return total; +eagain: + + if (total == 0) + return -ERREAGAIN; + return total; + err: fprintf(stderr, "recv(2) - error reading data: %s\n", strerror(errno)); @@ -460,7 +467,7 @@ ssize_t ssl_recv_bytes(SSL *ssl, unsigned char *buf, size_t bufsize) { || (err == SSL_ERROR_SYSCALL && !errno)) return 0; // Connection closed if (errno == EAGAIN || errno == EWOULDBLOCK) - break; + goto eagain; else goto err; } @@ -474,6 +481,12 @@ ssize_t ssl_recv_bytes(SSL *ssl, unsigned char *buf, size_t bufsize) { return total; +eagain: + + if (total == 0) + return -ERREAGAIN; + return total; + err: fprintf(stderr, "SSL_read(2) - error reading data: %s\n", strerror(errno)); diff --git a/src/server.c b/src/server.c index 34e042f..5759e07 100644 --- a/src/server.c +++ b/src/server.c @@ -299,6 +299,9 @@ static ssize_t recv_packet(struct connection *c, unsigned char **buf, */ ret = recv_data(c, *buf, 2); + if (ret == -ERREAGAIN) + return -ERREAGAIN; + if (ret <= 0) return -ERRCLIENTDC; @@ -493,58 +496,63 @@ static void *io_worker(void *arg) { */ struct connection *c = event->client->conn; int rc = read_data(c, buffer, &event->data); - if (rc == 0) { - /* - * All is ok, raise an event to the worker poll EPOLL and - * link it with the IO event containing the decode payload - * ready to be processed - */ - eventfd_t ev = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); - event->eventfd = ev; - epoll_add(epoll->w_epollfd, ev, EPOLLIN, event); - - /* Record last action as of now */ - event->client->last_action_time = time(NULL); - - /* Fire an event toward the worker thread pool */ - eventfd_write(ev, 1); - - } else if (rc == -ERRCLIENTDC || rc == -ERRPACKETERR) { - - /* - * We got an unexpected error or a disconnection from the - * client side, remove client from the global map and - * free resources allocated such as io_event structure and - * paired payload - */ - pthread_spin_lock(&io_spinlock); - log_error("Closing connection with %s: %s", - event->client->conn->ip, solerr(rc)); - // Publish, if present, LWT message - if (event->client->lwt_msg) { - char *tname = (char *) event->client->lwt_msg->topic; - struct topic *t = sol_topic_get(&sol, tname); - publish_message(event->client->lwt_msg, t, event->epollfd); - } - event->client->online = false; - // Clean resources - close_conn(event->client->conn); - // Remove from subscriptions for now - if (event->client->session) { - struct list *subs = event->client->session->subscriptions; - struct iterator *it = iter_new(subs, list_iter_next); - FOREACH (it) { - log_debug("Deleting %s from topic %s", - event->client->client_id, - ((struct topic *) it->ptr)->name); - topic_del_subscriber(it->ptr, event->client, false); + switch (rc) { + case 0: + /* + * All is ok, raise an event to the worker poll EPOLL and + * link it with the IO event containing the decode payload + * ready to be processed + */ + event->eventfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + epoll_add(epoll->w_epollfd, + event->eventfd, EPOLLIN, event); + + /* Record last action as of now */ + event->client->last_action_time = time(NULL); + + /* Fire an event toward the worker thread pool */ + eventfd_write(event->eventfd, 1); + break; + case -ERRCLIENTDC: + case -ERRPACKETERR: + /* + * We got an unexpected error or a disconnection from the + * client side, remove client from the global map and + * free resources allocated such as io_event structure and + * paired payload + */ + pthread_spin_lock(&io_spinlock); + log_error("Closing connection with %s: %s", + event->client->conn->ip, solerr(rc)); + // Publish, if present, LWT message + if (event->client->lwt_msg) { + char *tname = (char *) event->client->lwt_msg->topic; + struct topic *t = sol_topic_get(&sol, tname); + publish_message(event->client->lwt_msg, t, event->epollfd); } - iter_destroy(it); - } - info.nclients--; - info.nconnections--; - xfree(event); - pthread_spin_unlock(&io_spinlock); + event->client->online = false; + // Clean resources + close_conn(event->client->conn); + // Remove from subscriptions for now + if (event->client->session) { + struct list *subs = event->client->session->subscriptions; + struct iterator *it = iter_new(subs, list_iter_next); + FOREACH (it) { + log_debug("Deleting %s from topic %s", + event->client->client_id, + ((struct topic *) it->ptr)->name); + topic_del_subscriber(it->ptr, event->client, false); + } + iter_destroy(it); + } + info.nclients--; + info.nconnections--; + xfree(event); + pthread_spin_unlock(&io_spinlock); + break; + case -ERREAGAIN: + epoll_mod(epoll->io_epollfd, c->fd, EPOLLIN, event->client); + break; } } else if (e_events[i].events & EPOLLOUT) { diff --git a/src/server.h b/src/server.h index ad7827c..19586d9 100644 --- a/src/server.h +++ b/src/server.h @@ -50,6 +50,7 @@ #define ERRCLIENTDC 1 #define ERRPACKETERR 2 #define ERRMAXREQSIZE 3 +#define ERREAGAIN 4 /* Return code of handler functions, signaling if there's data payload to be * sent out or if the server just need to re-arm closure for reading incoming From a7ec977c1114b3c0f0b9f9937df7ca8ab0075192 Mon Sep 17 00:00:00 2001 From: Andrea Date: Fri, 3 Jan 2020 00:09:40 +0100 Subject: [PATCH 2/4] Fixed bug in publish over 128 bytes len --- src/mqtt.c | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/src/mqtt.c b/src/mqtt.c index d826b8a..230af07 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -119,20 +119,21 @@ static mqtt_pack_handler *pack_handlers[13] = { int mqtt_encode_length(unsigned char *buf, size_t len) { int bytes = 0; + short encoded = 0; do { if (bytes + 1 > MAX_LEN_BYTES) return bytes; - char d = len % 128; + encoded = len % 128; len /= 128; /* if there are more digits to encode, set the top bit of this digit */ if (len > 0) - d |= 0x80; + encoded |= 128; - buf[bytes++] = d; + buf[bytes++] = encoded; } while (len > 0); @@ -456,13 +457,23 @@ static unsigned char *pack_mqtt_publish(const union mqtt_packet *pkt) { if (pkt->publish.header.bits.qos > AT_MOST_ONCE) pktlen += sizeof(uint16_t); + int remaninglen_offset = 0; + if ((pktlen - 1) > 0x200000) + remaninglen_offset = 3; + else if ((pktlen - 1) > 0x4000) + remaninglen_offset = 2; + else if ((pktlen - 1) > 0x80) + remaninglen_offset = 1; + + pktlen += remaninglen_offset; + unsigned char *packed = xmalloc(pktlen); unsigned char *ptr = packed; pack(ptr++, "B", pkt->publish.header.byte); // Total len of the packet excluding fixed header len - len += (pktlen - MQTT_HEADER_LEN); + len += (pktlen - MQTT_HEADER_LEN - remaninglen_offset); /* * TODO handle case where step is > 1, e.g. when a message longer than 128 From 0698a7978545723992b3736a565e1500765f4894 Mon Sep 17 00:00:00 2001 From: Andrea Date: Fri, 3 Jan 2020 00:21:44 +0100 Subject: [PATCH 3/4] Temporary fix for payload over 128 bytes len on publish handler --- src/handlers.c | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/src/handlers.c b/src/handlers.c index 6f0c009..2a92c2f 100644 --- a/src/handlers.c +++ b/src/handlers.c @@ -94,6 +94,17 @@ void publish_message(struct mqtt_publish *p, p->topiclen + p->payloadlen; // Add 2 bytes to make space for packet identifier publen += sizeof(uint16_t); + + int remaninglen_offset = 0; + if ((publen - 1) > 0x200000) + remaninglen_offset = 3; + else if ((publen - 1) > 0x4000) + remaninglen_offset = 2; + else if ((publen - 1) > 0x80) + remaninglen_offset = 1; + + publen += remaninglen_offset; + pkt.publish.pkt_id = next_free_mid(sc->i_msgs); pkt.publish.header.bits.qos = p->header.bits.qos; pub = pack_mqtt_packet(&pkt, PUBLISH); @@ -122,6 +133,16 @@ void publish_message(struct mqtt_publish *p, */ publen = MQTT_HEADER_LEN + sizeof(uint16_t) + p->topiclen + p->payloadlen; + + int remaninglen_offset = 0; + if ((publen - 1) > 0x200000) + remaninglen_offset = 3; + else if ((publen - 1) > 0x4000) + remaninglen_offset = 2; + else if ((publen - 1) > 0x80) + remaninglen_offset = 1; + + publen += remaninglen_offset; pkt.publish.header.bits.qos = 0; pkt.publish.pkt_id = 0; pub = pack_mqtt_packet(&pkt, PUBLISH); From f250f7fc4536fcdf4a12fe235ffd49d6c2182a7d Mon Sep 17 00:00:00 2001 From: Andrea Date: Fri, 3 Jan 2020 00:36:37 +0100 Subject: [PATCH 4/4] Fixed autocomplete bug --- src/pack.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pack.c b/src/pack.c index cfbe04c..139ff38 100644 --- a/src/pack.c +++ b/src/pack.c @@ -463,7 +463,7 @@ long long unpack_integer(unsigned char **buf, char size) { *buf += 8; break; case 'Q': - val = unpacku16(*buf); + val = unpacku64(*buf); *buf += 8; break; }