From abc288a43527bc83fa239602692481c83674cc11 Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Tue, 9 Mar 2021 10:03:27 +0000 Subject: [PATCH] Experimental kqueue support. --- CMakeLists.txt | 1 + ChangeLog.txt | 1 + lib/mosquitto_internal.h | 8 +- src/CMakeLists.txt | 6 +- src/Makefile | 4 + src/bridge.c | 1 + src/context.c | 2 +- src/mosquitto.c | 4 +- src/mosquitto_broker_internal.h | 7 +- src/mux.c | 14 ++ src/mux.h | 8 + src/mux_kqueue.c | 266 ++++++++++++++++++++++++++++++++ src/mux_poll.c | 2 +- 13 files changed, 316 insertions(+), 8 deletions(-) create mode 100644 src/mux_kqueue.c diff --git a/CMakeLists.txt b/CMakeLists.txt index 327280ba..f4a945ad 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -24,6 +24,7 @@ if(APPLE) endif(APPLE) include(GNUInstallDirs) +include(CheckIncludeFiles) option(WITH_BUNDLED_DEPS "Build with bundled dependencies?" ON) option(WITH_TLS diff --git a/ChangeLog.txt b/ChangeLog.txt index 594d6d58..98488236 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -9,6 +9,7 @@ Broker: available, which means they are self contained. - Increase maximum connection count on Windows from 2048 to 8192 where supported. Closes #2122. +- Add kqueue support. Client library: - Add MOSQ_OPT_DISABLE_SOCKETPAIR to allow the disabling of the socketpair diff --git a/lib/mosquitto_internal.h b/lib/mosquitto_internal.h index 699d0f3e..fc6eab16 100644 --- a/lib/mosquitto_internal.h +++ b/lib/mosquitto_internal.h @@ -207,7 +207,7 @@ struct mosquitto_msg_data{ struct mosquitto { -#if defined(WITH_BROKER) && defined(WITH_EPOLL) +#if defined(WITH_BROKER) && (defined(WITH_EPOLL) || defined(WITH_KQUEUE)) /* This *must* be the first element in the struct. */ int ident; #endif @@ -352,7 +352,13 @@ struct mosquitto { struct session_expiry_list *expiry_list_item; uint16_t remote_port; #endif +#ifdef WITH_EPOLL uint32_t events; +#elif defined(WITH_KQUEUE) + short events; +#else + uint32_t events; +#endif }; #define STREMPTY(str) (str[0] == '\0') diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index f283c8d0..ecda70a1 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -32,7 +32,7 @@ set (MOSQ_SRCS mosquitto.c ../include/mosquitto_broker.h mosquitto_broker_internal.h ../lib/misc_mosq.c ../lib/misc_mosq.h - mux.c mux.h mux_epoll.c mux_poll.c + mux.c mux.h mux_epoll.c mux_kqueue.c mux_poll.c net.c ../lib/net_mosq_ocsp.c ../lib/net_mosq.c ../lib/net_mosq.h ../lib/packet_datatypes.c @@ -72,6 +72,10 @@ set (MOSQ_SRCS will_delay.c ../lib/will_mosq.c ../lib/will_mosq.h) +CHECK_INCLUDE_FILES(sys/event.h HAVE_SYS_EVENT_H) +if (HAVE_SYS_EVENT_H) + add_definitions("-DWITH_KQUEUE") +endif (HAVE_SYS_EVENT_H) if (WITH_BUNDLED_DEPS) include_directories(${mosquitto_SOURCE_DIR} ${mosquitto_SOURCE_DIR}/deps) diff --git a/src/Makefile b/src/Makefile index 92bad827..f3163550 100644 --- a/src/Makefile +++ b/src/Makefile @@ -34,6 +34,7 @@ OBJS= mosquitto.o \ misc_mosq.o \ mux.o \ mux_epoll.o \ + mux_kqueue.o \ mux_poll.o \ net.o \ net_mosq.o \ @@ -174,6 +175,9 @@ mux.o : mux.c mosquitto_broker_internal.h mux_epoll.o : mux_epoll.c mosquitto_broker_internal.h ${CROSS_COMPILE}${CC} $(BROKER_CPPFLAGS) $(BROKER_CFLAGS) -c $< -o $@ +mux_kqueue.o : mux_kqueue.c mosquitto_broker_internal.h + ${CROSS_COMPILE}${CC} $(BROKER_CPPFLAGS) $(BROKER_CFLAGS) -c $< -o $@ + mux_poll.o : mux_poll.c mosquitto_broker_internal.h ${CROSS_COMPILE}${CC} $(BROKER_CPPFLAGS) $(BROKER_CFLAGS) -c $< -o $@ diff --git a/src/bridge.c b/src/bridge.c index 823d9f5f..4702bfea 100644 --- a/src/bridge.c +++ b/src/bridge.c @@ -26,6 +26,7 @@ Contributors: #ifndef WIN32 #include #include +#include #include #else #include diff --git a/src/context.c b/src/context.c index 083a07d0..5a48f5c0 100644 --- a/src/context.c +++ b/src/context.c @@ -40,7 +40,7 @@ struct mosquitto *context__init(mosq_sock_t sock) context = mosquitto__calloc(1, sizeof(struct mosquitto)); if(!context) return NULL; -#ifdef WITH_EPOLL +#if defined(WITH_EPOLL) || defined(WITH_KQUEUE) context->ident = id_client; #else context->pollfd_index = -1; diff --git a/src/mosquitto.c b/src/mosquitto.c index 0fe8ba02..0df3afdb 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -229,7 +229,7 @@ int listeners__start_single_mqtt(struct mosquitto__listener *listener) } listensock[listensock_index].sock = listener->socks[i]; listensock[listensock_index].listener = listener; -#ifdef WITH_EPOLL +#if defined(WITH_EPOLL) || defined(WITH_KQUEUE) listensock[listensock_index].ident = id_listener; #endif listensock_index++; @@ -268,7 +268,7 @@ void listeners__add_websockets(struct lws_context *ws_context, mosq_sock_t fd) listensock[listensock_index].sock = fd; listensock[listensock_index].listener = listener; -#ifdef WITH_EPOLL +#if defined(WITH_EPOLL) || defined(WITH_KQUEUE) listensock[listensock_index].ident = id_listener_ws; #endif listensock_index++; diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index 492dc584..6936aba2 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -182,7 +182,7 @@ struct mosquitto__security_options { mosquitto_plugin_id_t *pid; /* For registering as a "plugin" */ }; -#ifdef WITH_EPOLL +#if defined(WITH_EPOLL) || defined(WITH_KQUEUE) enum struct_ident{ id_invalid = 0, id_listener = 1, @@ -238,7 +238,7 @@ struct mosquitto__listener { struct mosquitto__listener_sock{ -#ifdef WITH_EPOLL +#if defined(WITH_EPOLL) || defined(WITH_KQUEUE) /* This *must* be the first element in the struct. */ int ident; #endif @@ -469,6 +469,9 @@ struct mosquitto_db{ struct mosquitto *ll_for_free; #ifdef WITH_EPOLL int epollfd; +#endif +#ifdef WITH_KQUEUE + int kqueuefd; #endif struct mosquitto_message_v5 *plugin_msgs; }; diff --git a/src/mux.c b/src/mux.c index 74483201..672a56ec 100644 --- a/src/mux.c +++ b/src/mux.c @@ -23,6 +23,8 @@ int mux__init(struct mosquitto__listener_sock *listensock, int listensock_count) { #ifdef WITH_EPOLL return mux_epoll__init(listensock, listensock_count); +#elif defined(WITH_KQUEUE) + return mux_kqueue__init(listensock, listensock_count); #else return mux_poll__init(listensock, listensock_count); #endif @@ -32,6 +34,8 @@ int mux__add_out(struct mosquitto *context) { #ifdef WITH_EPOLL return mux_epoll__add_out(context); +#elif defined(WITH_KQUEUE) + return mux_kqueue__add_out(context); #else return mux_poll__add_out(context); #endif @@ -42,6 +46,8 @@ int mux__remove_out(struct mosquitto *context) { #ifdef WITH_EPOLL return mux_epoll__remove_out(context); +#elif defined(WITH_KQUEUE) + return mux_kqueue__remove_out(context); #else return mux_poll__remove_out(context); #endif @@ -52,6 +58,8 @@ int mux__add_in(struct mosquitto *context) { #ifdef WITH_EPOLL return mux_epoll__add_in(context); +#elif defined(WITH_KQUEUE) + return mux_kqueue__add_in(context); #else return mux_poll__add_in(context); #endif @@ -62,6 +70,8 @@ int mux__delete(struct mosquitto *context) { #ifdef WITH_EPOLL return mux_epoll__delete(context); +#elif defined(WITH_KQUEUE) + return mux_kqueue__delete(context); #else return mux_poll__delete(context); #endif @@ -72,6 +82,8 @@ int mux__handle(struct mosquitto__listener_sock *listensock, int listensock_coun { #ifdef WITH_EPOLL return mux_epoll__handle(listensock, listensock_count); +#elif defined(WITH_KQUEUE) + return mux_kqueue__handle(listensock, listensock_count); #else return mux_poll__handle(listensock, listensock_count); #endif @@ -82,6 +94,8 @@ int mux__cleanup(void) { #ifdef WITH_EPOLL return mux_epoll__cleanup(); +#elif defined(WITH_KQUEUE) + return mux_kqueue__cleanup(); #else return mux_poll__cleanup(); #endif diff --git a/src/mux.h b/src/mux.h index 6b416c99..c3ad04f4 100644 --- a/src/mux.h +++ b/src/mux.h @@ -29,6 +29,14 @@ int mux_epoll__delete(struct mosquitto *context); int mux_epoll__handle(struct mosquitto__listener_sock *listensock, int listensock_count); int mux_epoll__cleanup(void); +int mux_kqueue__init(struct mosquitto__listener_sock *listensock, int listensock_count); +int mux_kqueue__add_out(struct mosquitto *context); +int mux_kqueue__remove_out(struct mosquitto *context); +int mux_kqueue__add_in(struct mosquitto *context); +int mux_kqueue__delete(struct mosquitto *context); +int mux_kqueue__handle(struct mosquitto__listener_sock *listensock, int listensock_count); +int mux_kqueue__cleanup(void); + int mux_poll__init(struct mosquitto__listener_sock *listensock, int listensock_count); int mux_poll__add_out(struct mosquitto *context); int mux_poll__remove_out(struct mosquitto *context); diff --git a/src/mux_kqueue.c b/src/mux_kqueue.c new file mode 100644 index 00000000..b1c8e7cf --- /dev/null +++ b/src/mux_kqueue.c @@ -0,0 +1,266 @@ +/* +Copyright (c) 2021 Roger Light + +All rights reserved. This program and the accompanying materials +are made available under the terms of the Eclipse Public License 2.0 +and Eclipse Distribution License v1.0 which accompany this distribution. + +The Eclipse Public License is available at + https://www.eclipse.org/legal/epl-2.0/ +and the Eclipse Distribution License is available at + http://www.eclipse.org/org/documents/edl-v10.php. + +SPDX-License-Identifier: EPL-2.0 OR EDL-1.0 + +Contributors: + Roger Light - initial implementation and documentation. +*/ + +#include "config.h" + +#ifdef WITH_KQUEUE + +#define MAX_EVENTS 1000 + +#include +#include +#include + +#include "mosquitto_broker_internal.h" +#include "packet_mosq.h" +#include "util_mosq.h" + +static void loop_handle_reads_writes(struct mosquitto *context, short events); + +static sigset_t my_sigblock; + +static struct kevent event_list[MAX_EVENTS]; + +int mux_kqueue__init(struct mosquitto__listener_sock *listensock, int listensock_count) +{ + struct kevent ev; + int i; + + sigemptyset(&my_sigblock); + sigaddset(&my_sigblock, SIGINT); + sigaddset(&my_sigblock, SIGTERM); + sigaddset(&my_sigblock, SIGUSR1); + sigaddset(&my_sigblock, SIGUSR2); + sigaddset(&my_sigblock, SIGHUP); + + memset(&event_list, 0, sizeof(struct kevent)*MAX_EVENTS); + + db.kqueuefd = 0; + if ((db.kqueuefd = kqueue()) == -1) { + log__printf(NULL, MOSQ_LOG_ERR, "Error in kqueue creating: %s", strerror(errno)); + return MOSQ_ERR_UNKNOWN; + } + for(i=0; ievents != EVFILT_WRITE){ + EV_SET(&ev, context->sock, EVFILT_WRITE, EV_ADD, 0, 0, &context); + if(kevent(db.kqueuefd, &ev, 1, NULL, 0, NULL) == -1){ + log__printf(NULL, MOSQ_LOG_DEBUG, "Error in kqueue re-registering to EVFILT_WRITE: %s", strerror(errno)); + } + + context->events = EVFILT_WRITE; + } + return MOSQ_ERR_SUCCESS; +} + + +int mux_kqueue__remove_out(struct mosquitto *context) +{ + struct kevent ev; + + if(context->events == EVFILT_WRITE){ + EV_SET(&ev, context->sock, EVFILT_WRITE, EV_DELETE, 0, 0, &context); + if(kevent(db.kqueuefd, &ev, 1, NULL, 0, NULL) == -1){ + log__printf(NULL, MOSQ_LOG_DEBUG, "Error in kqueue removing EVFILT_WRITE: %s", strerror(errno)); + } + + context->events = 0; + } + return MOSQ_ERR_SUCCESS; +} + + +int mux_kqueue__add_in(struct mosquitto *context) +{ + struct kevent ev; + + EV_SET(&ev, context->sock, EVFILT_READ, EV_ADD, 0, 0, context); + if(kevent(db.kqueuefd, &ev, 1, NULL, 0, NULL) == -1){ + log__printf(NULL, MOSQ_LOG_ERR, "Error in kqueue accepting: %s", strerror(errno)); + } + + context->events = 0; + return MOSQ_ERR_SUCCESS; +} + + +int mux_kqueue__delete(struct mosquitto *context) +{ + struct kevent ev[2]; + + if(context->sock != INVALID_SOCKET){ + EV_SET(&ev[0], context->sock, EVFILT_READ, EV_DELETE, 0, 0, &context); + EV_SET(&ev[1], context->sock, EVFILT_WRITE, EV_DELETE, 0, 0, &context); + if(kevent(db.kqueuefd, ev, 2, NULL, 0, NULL) == -1){ + return 1; + } + } + return 0; +} + + +int mux_kqueue__handle(void) +{ + int i; + sigset_t origsig; + struct mosquitto *context; + struct mosquitto__listener_sock *listensock; + int event_count; + struct timespec timeout = {0, 100000000}; /* 100 ms */ + + memset(&event_list, 0, sizeof(event_list)); + sigprocmask(SIG_SETMASK, &my_sigblock, &origsig); + event_count = kevent(db.kqueuefd, + NULL, 0, + event_list, MAX_EVENTS, + &timeout); + sigprocmask(SIG_SETMASK, &origsig, NULL); + + db.now_s = mosquitto_time(); + db.now_real_s = time(NULL); + + switch(event_count){ + case -1: + if(errno != EINTR){ + log__printf(NULL, MOSQ_LOG_ERR, "Error in kqueue waiting: %s.", strerror(errno)); + } + break; + case 0: + break; + default: + for(i=0; iident == id_client){ + loop_handle_reads_writes(context, event_list[i].filter); + }else if(context->ident == id_listener){ + listensock = event_list[i].udata; + + if(event_list[i].filter == EVFILT_READ){ + while((context = net__socket_accept(listensock)) != NULL){ + context->events = EVFILT_READ; + mux__add_in(context); + } + } +#ifdef WITH_WEBSOCKETS + }else if(context->ident == id_listener_ws){ + /* Nothing needs to happen here, because we always call lws_service in the loop. + * The important point is we've been woken up for this listener. */ +#endif + } + } + } + return MOSQ_ERR_SUCCESS; +} + + +int mux_kqueue__cleanup(void) +{ + (void)close(db.kqueuefd); + db.kqueuefd = 0; + return MOSQ_ERR_SUCCESS; +} + + +static void loop_handle_reads_writes(struct mosquitto *context, short event) +{ + int err; + socklen_t len; + int rc; + + if(context->sock == INVALID_SOCKET){ + return; + } + +#ifdef WITH_WEBSOCKETS + if(context->wsi){ + struct lws_pollfd wspoll; + wspoll.fd = context->sock; + wspoll.events = (int16_t)context->events; + wspoll.revents = (int16_t)events; + lws_service_fd(lws_get_context(context->wsi), &wspoll); + return; + } +#endif + + if(event == EVFILT_WRITE +#ifdef WITH_TLS + || context->want_write + || (context->ssl && context->state == mosq_cs_new) +#endif + ){ + + if(context->state == mosq_cs_connect_pending){ + len = sizeof(int); + if(!getsockopt(context->sock, SOL_SOCKET, SO_ERROR, (char *)&err, &len)){ + if(err == 0){ + mosquitto__set_state(context, mosq_cs_new); +#if defined(WITH_ADNS) && defined(WITH_BRIDGE) + if(context->bridge){ + bridge__connect_step3(context); + } +#endif + } + }else{ + do_disconnect(context, MOSQ_ERR_CONN_LOST); + return; + } + } + rc = packet__write(context); + if(rc){ + do_disconnect(context, rc); + return; + } + } + + if(event == EVFILT_READ +#ifdef WITH_TLS + || (context->ssl && context->state == mosq_cs_new) +#endif + ){ + + do{ + rc = packet__read(context); + if(rc){ + do_disconnect(context, rc); + return; + } + }while(SSL_DATA_PENDING(context)); + } +} +#endif diff --git a/src/mux_poll.c b/src/mux_poll.c index 3e0977dc..875c4e3e 100644 --- a/src/mux_poll.c +++ b/src/mux_poll.c @@ -18,7 +18,7 @@ Contributors: #include "config.h" -#ifndef WITH_EPOLL +#if !defined(WITH_EPOLL) && !defined(WITH_KQUEUE) #ifndef WIN32 # define _GNU_SOURCE