Experimental kqueue support.

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

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

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')

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)

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 $@
 

src/bridge.c

@@ -26,6 +26,7 @@ Contributors:
 #ifndef WIN32
 #include <netdb.h>
 #include <sys/socket.h>
+#include <netinet/in.h>
 #include <netinet/tcp.h>
 #else
 #include <winsock2.h>

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;

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++;

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;
 };

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

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);

src/mux_kqueue.c

@@ -0,0 +1,266 @@
+/*
+Copyright (c) 2021 Roger Light <roger@atchoo.org>
+
+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 <signal.h>
+#include <sys/event.h>
+#include <sys/socket.h>
+
+#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; i<listensock_count; i++){
+		EV_SET(&ev, listensock[i].sock, EVFILT_READ, EV_ADD, 0, 0, &listensock[i]);
+		if(kevent(db.kqueuefd, &ev, 1, NULL, 0, NULL) == -1){
+			log__printf(NULL, MOSQ_LOG_ERR, "Error in kqueue initial registering: %s", strerror(errno));
+			(void)close(db.kqueuefd);
+			db.kqueuefd = 0;
+			return MOSQ_ERR_UNKNOWN;
+		}
+	}
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+int mux_kqueue__loop_setup(void)
+{
+	return MOSQ_ERR_SUCCESS;
+}
+
+
+int mux_kqueue__add_out(struct mosquitto *context)
+{
+	struct kevent ev;
+
+	if(context->events != 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; i<event_count; i++){
+			context = event_list[i].udata;
+			if(context->ident == 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

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