temp

sys/kern/kern_fork.c

@@ -690,7 +690,7 @@ do_fork(struct thread *td, struct fork_req *fr, struct proc *p2, struct thread *
 	/*
 	 * Tell any interested parties about the new process.
 	 */
-	knote_fork(p1->p_klist, p2->p_pid);
+	knote_fork(p1->p_klist, td, p2->p_pid);
 
 	/*
 	 * Now can be swapped.

sys/kern/kern_thread.c

@@ -56,6 +56,7 @@ __FBSDID("$FreeBSD$");
 #include <sys/umtx.h>
 #include <sys/vmmeter.h>
 #include <sys/cpuset.h>
+#include <sys/event.h>
 #ifdef	HWPMC_HOOKS
 #include <sys/pmckern.h>
 #endif
@@ -82,9 +83,9 @@ _Static_assert(offsetof(struct thread, td_flags) == 0xfc,
     "struct thread KBI td_flags");
 _Static_assert(offsetof(struct thread, td_pflags) == 0x104,
     "struct thread KBI td_pflags");
-_Static_assert(offsetof(struct thread, td_frame) == 0x478,
+_Static_assert(offsetof(struct thread, td_frame) == 0x478 + 0x8,
     "struct thread KBI td_frame");
-_Static_assert(offsetof(struct thread, td_emuldata) == 0x530,
+_Static_assert(offsetof(struct thread, td_emuldata) == 0x530 + 0x8,
     "struct thread KBI td_emuldata");
 _Static_assert(offsetof(struct proc, p_flag) == 0xb0,
     "struct proc KBI p_flag");
@@ -444,6 +445,9 @@ thread_free(struct thread *td)
 	cpu_thread_free(td);
 	if (td->td_kstack != 0)
 		vm_thread_dispose(td);
+	if (td->td_kevq_thred != NULL) {
+		kevq_thred_drain(td->td_kevq_thred);
+	}
 	callout_drain(&td->td_slpcallout);
 	uma_zfree(thread_zone, td);
 }

sys/sys/event.h

@@ -33,6 +33,8 @@
 
 #include <sys/_types.h>
 #include <sys/queue.h>
+#include <sys/lock.h>
+#include <sys/mutex.h>
 
 #define EVFILT_READ		(-1)
 #define EVFILT_WRITE		(-2)
@@ -143,6 +145,7 @@ struct kevent32_freebsd11 {
 #define EV_CLEAR	0x0020		/* clear event state after reporting */
 #define EV_RECEIPT	0x0040		/* force EV_ERROR on success, data=0 */
 #define EV_DISPATCH	0x0080		/* disable event after reporting */
+#define EV_AFFINITY	0x0200		/* in multithreaded mode, this event has hard affinity for the registering thread */
 
 #define EV_SYSFLAGS	0xF000		/* reserved by system */
 #define	EV_DROP		0x1000		/* note should be dropped */
@@ -220,6 +223,9 @@ struct knote;
 SLIST_HEAD(klist, knote);
 struct kqueue;
 TAILQ_HEAD(kqlist, kqueue);
+struct kevq;
+SLIST_HEAD(kevqlist, kevq);
+
 struct knlist {
 	struct	klist	kl_list;
 	void    (*kl_lock)(void *);	/* lock function */
@@ -269,6 +275,10 @@ struct filterops {
 	void	(*f_touch)(struct knote *kn, struct kevent *kev, u_long type);
 };
 
+/* The ioctl to set multithreaded mode
+ */
+#define	FKQMULTI	_IO('f', 89)
+
 /*
  * An in-flux knote cannot be dropped from its kq while the kq is
  * unlocked.  If the KN_SCAN flag is not set, a thread can only set
@@ -283,7 +293,9 @@ struct knote {
 	SLIST_ENTRY(knote)	kn_selnext;	/* for struct selinfo */
 	struct			knlist *kn_knlist;	/* f_attach populated */
 	TAILQ_ENTRY(knote)	kn_tqe;
-	struct			kqueue *kn_kq;	/* which queue we are on */
+	struct			kqueue *kn_kq;	/* which kqueue we are on */
+	struct			kevq *kn_org_kevq; /* the kevq that registered the knote */
+	struct			kevq *kn_kevq; /* the kevq the knote is on */
 	struct 			kevent kn_kevent;
 	void			*kn_hook;
 	int			kn_hookid;
@@ -295,7 +307,9 @@ struct knote {
 #define KN_MARKER	0x20			/* ignore this knote */
 #define KN_KQUEUE	0x40			/* this knote belongs to a kq */
 #define	KN_SCAN		0x100			/* flux set in kqueue_scan() */
+	int			kn_fluxwait;
 	int			kn_influx;
+	struct 		mtx kn_fluxlock;
 	int			kn_sfflags;	/* saved filter flags */
 	int64_t			kn_sdata;	/* saved data field */
 	union {
@@ -321,6 +335,14 @@ struct kevent_copyops {
 	size_t	kevent_size;
 };
 
+struct kevq_thred {
+	u_long		kevq_hashmask;		/* hash mask for kevqs */
+	struct		kevqlist *kevq_hash;	    /* hash table for kevqs */
+	TAILQ_HEAD(, kevq) kevq_tq;
+	struct		mtx lock;		/* the lock for the kevq*/
+};
+
+
 struct thread;
 struct proc;
 struct knlist;
@@ -328,7 +350,7 @@ struct mtx;
 struct rwlock;
 
 void	knote(struct knlist *list, long hint, int lockflags);
-void	knote_fork(struct knlist *list, int pid);
+void knote_fork(struct knlist *list, struct thread *td, int pid);
 struct knlist *knlist_alloc(struct mtx *lock);
 void	knlist_detach(struct knlist *knl);
 void	knlist_add(struct knlist *knl, struct knote *kn, int islocked);
@@ -352,6 +374,7 @@ int 	kqfd_register(int fd, struct kevent *kev, struct thread *p,
 int	kqueue_add_filteropts(int filt, struct filterops *filtops);
 int	kqueue_del_filteropts(int filt);
 
+void kevq_thred_drain(struct kevq_thred *kevq_th);
 #else 	/* !_KERNEL */
 
 #include <sys/cdefs.h>

sys/sys/eventvar.h

@@ -40,27 +40,47 @@
 #define KQ_NEVENTS	8		/* minimize copy{in,out} calls */
 #define KQEXTENT	256		/* linear growth by this amount */
 
+struct kevq {
+	SLIST_ENTRY(kevq)	kevq_th_e; /* entry into kevq_thred's hashtable */
+	TAILQ_ENTRY(kevq)	kq_e;   /* entry into kqueue's list */
+	TAILQ_ENTRY(kevq)	kevq_th_tqe; /* entry into kevq_thred's TAILQ */
+	struct		kqueue *kq;     /* the kq that the kevq belongs to */
+	struct		kevq_thred *kevq_th; /* the thread that the kevq belongs to */
+	struct		mtx lock;		/* the lock for the kevq */
+	TAILQ_HEAD(, knote) kn_head;	/* list of pending knotes */
+	int		kn_count;				/* number of pending knotes */
+#define KEVQ_SLEEP	0x01
+#define KEVQ_CLOSING  0x02
+#define KEVQ_RDY	0x04
+	int		kevq_state;
+	int		kevq_refcnt;
+};
+
 struct kqueue {
 	struct		mtx kq_lock;
 	int		kq_refcnt;
-	TAILQ_ENTRY(kqueue)	kq_list;
-	TAILQ_HEAD(, knote)	kq_head;	/* list of pending event */
-	int		kq_count;		/* number of pending events */
 	struct		selinfo kq_sel;
-	struct		sigio *kq_sigio;
-	struct		filedesc *kq_fdp;
 	int		kq_state;
 #define KQ_SEL		0x01
-#define KQ_SLEEP	0x02
+#define KQ_ASYNC	0x02
-#define KQ_FLUXWAIT	0x04			/* waiting for a in flux kn */
+#define	KQ_TASKSCHED	0x04			/* task scheduled */
-#define KQ_ASYNC	0x08
+#define	KQ_TASKDRAIN	0x08			/* waiting for task to drain */
 #define KQ_CLOSING	0x10
-#define	KQ_TASKSCHED	0x20			/* task scheduled */
+#define KQ_FLAG_INIT 0x20		/* kqueue has been initialized. this flag is set after the first kevent structure is processed */
-#define	KQ_TASKDRAIN	0x40			/* waiting for task to drain */
+#define KQ_FLAG_MULTI 0x40		/* Multi-threaded mode */
+	TAILQ_ENTRY(kqueue)	kq_list;
+	struct		sigio *kq_sigio;
+	struct		filedesc *kq_fdp;
 	int		kq_knlistsize;		/* size of knlist */
 	struct		klist *kq_knlist;	/* list of knotes */
 	u_long		kq_knhashmask;		/* size of knhash */
 	struct		klist *kq_knhash;	/* hash table for knotes */
+	/* only-set: in multithreaded mode */
+	TAILQ_HEAD(, kevq)	kq_kevqlist; /* list of kevqs interested in the kqueue */
+	struct		kevq	*kq_ckevq; /* current kevq for multithreaded kqueue */
+	/* only-set: in single threaded mode */
+	struct		kevq *kq_kevq;
+	/* End only-set */
 	struct		task kq_task;
 	struct		ucred *kq_cred;
 };

sys/sys/proc.h

@@ -302,6 +302,7 @@ struct thread {
 	int		td_rtcgen;	/* (s) rtc_generation of abs. sleep */
 	size_t		td_vslock_sz;	/* (k) amount of vslock-ed space */
 	struct kcov_info *td_kcov_info;	/* (*) Kernel code coverage data */
+	struct kevq_thred *td_kevq_thred;
 #define	td_endzero td_sigmask
 
 /* Copied during fork1() or create_thread(). */

tests/sys/kqueue/libkqueue/Makefile

@@ -15,7 +15,8 @@ SRCS.kqtest=		\
 	vnode.c		\
 	proc.c		\
 	signal.c	\
+	read_m.c 	\
 	user.c
 WARNS?=	2
-
+LDADD+= -lthr
 .include <bsd.test.mk>

tests/sys/kqueue/libkqueue/main.c

@@ -29,6 +29,7 @@ extern void test_evfilt_read();
 extern void test_evfilt_signal();
 extern void test_evfilt_vnode();
 extern void test_evfilt_timer();
+extern void test_evfilt_read_m();
 extern void test_evfilt_proc();
 #if HAVE_EVFILT_USER
 extern void test_evfilt_user();
@@ -322,6 +323,7 @@ main(int argc, char **argv)
     int test_signal = 1;
     int test_vnode = 1;
     int test_timer = 1;
+    int test_socket_m = 1;
 #ifdef __FreeBSD__
     int test_user = 1;
 #else
@@ -342,6 +344,8 @@ main(int argc, char **argv)
             test_vnode = 0;
         if (strcmp(argv[0], "--no-user") == 0)
             test_user = 0;
+        if (strcmp(argv[0], "--no-socket_m") == 0)
+            test_socket_m = 0;
         argv++;
         argc--;
     }
@@ -360,6 +364,8 @@ main(int argc, char **argv)
 
     if (test_socket) 
         test_evfilt_read();
+    if (test_socket_m)
+        test_evfilt_read_m();
     if (test_signal) 
         test_evfilt_signal();
     if (test_vnode) 

tests/sys/kqueue/libkqueue/read_m.c

@@ -0,0 +1,222 @@
+/*
+ * Copyright (c) 2009 Mark Heily <mark@heily.com>
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ *
+ * $FreeBSD$
+ */
+
+#include "common.h"
+
+#include <sys/ioctl.h>
+#include <semaphore.h>
+#include <pthread.h>
+
+#define THREAD_CNT (100)
+#define PACKET_CNT (5000)
+#define	FKQMULTI	_IO('f', 89)
+#define TEST_DEBUG
+
+struct thread_info {
+    pthread_t thrd;
+    int evcnt;
+    int tid;
+};
+
+int g_kqfd;
+int g_sockfd[2];
+int g_end;
+struct thread_info g_thrd_info[THREAD_CNT];
+/* Test threads signals this upon receiving events */
+sem_t g_sem_driver;
+
+static int
+check_sched(void)
+{
+    int max = 0, min = 999999;
+
+    for(int i = 0; i < THREAD_CNT; i++) {
+        int cur = g_thrd_info[i].evcnt;
+        if (cur > max) {
+            max = cur;
+        }
+        if (cur < min) {
+            min = cur;
+        }
+    }
+
+#ifdef TEST_DEBUG
+        printf("READ_M: check_sched: max difference is %d\n", max - min);
+#endif
+
+    return (max - min) <= 1;
+}
+
+static void
+socket_pop()
+{
+    char buf[1];
+
+    /* Drain the read buffer, then make sure there are no more events. */
+#ifdef TEST_DEBUG
+    printf("READ_M: popping the read buffer\n");
+#endif
+    if (read(g_sockfd[0], &buf[0], 1) < 1)
+        err(1, "read(2)");
+}
+
+static void
+socket_push()
+{
+#ifdef TEST_DEBUG
+    printf("READ_M: pushing to socket\n");
+#endif
+    if (write(g_sockfd[1], ".", 1) < 1) {
+#ifdef TEST_DEBUG
+    printf("READ_M: write failed with %d\n", errno);
+#endif     
+        err(1, "write(2)");
+    }
+}
+
+/* for multi threaded read */
+static void*
+test_socket_read_thrd(void* args)
+{
+    struct thread_info *info = (struct thread_info *) args;
+
+    struct kevent kev;
+    EV_SET(&kev, g_sockfd[0], EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, &g_sockfd[0]);
+    kev.data = 1;
+
+    while (1) {
+#ifdef TEST_DEBUG
+        printf("READ_M: thread %d waiting for events\n", info->tid); 
+#endif
+        kevent_cmp(&kev, kevent_get(g_kqfd));
+#ifdef TEST_DEBUG
+        printf("READ_M: thread %d woke up\n", info->tid); 
+#endif
+        socket_pop();
+        info->evcnt++;
+
+        /* signal the driver */
+        sem_post(&g_sem_driver);
+
+        if (g_end) {
+#ifdef TEST_DEBUG
+    printf("READ_M: thread %d exiting...\n", info->tid);
+#endif   
+            break;
+        }
+    }
+    pthread_exit(0);
+}
+
+static void
+test_socket_read(void)
+{
+    int error = 0;
+    struct kevent kev;
+    EV_SET(&kev, g_sockfd[0], EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, &g_sockfd[0]);
+
+    const char *test_id = "[Multi]kevent(EVFILT_READ)";
+    sem_init(&g_sem_driver, 0, 0);
+    g_end = 0;
+
+    test_begin(test_id);
+
+    error = kevent(g_kqfd, &kev, 1, NULL, 0, NULL);
+
+    if (error == -1) {
+#ifdef TEST_DEBUG
+        printf("READ_M: kevent add failed with %d\n", errno);
+#endif   
+        err(1, "kevent_add");     
+    }
+
+#ifdef TEST_DEBUG
+    printf("READ_M: creating %d threads...\n", THREAD_CNT);
+#endif
+    for (int i = 0; i < THREAD_CNT; i++) {
+        g_thrd_info[i].tid = i;
+        g_thrd_info[i].evcnt = 0;
+        pthread_create(&g_thrd_info[i].thrd, NULL, test_socket_read_thrd, &g_thrd_info[i]);
+    }
+
+#ifdef TEST_DEBUG
+    printf("READ_M: waiting for threads to wait on KQ...\n");
+#endif
+    
+    sleep(3);
+
+    for(int i = 0; i < PACKET_CNT; i++) {
+#ifdef TEST_DEBUG
+    printf("READ_M: processing packet %d\n", i);
+#endif        
+        socket_push();
+        /* wait for thread events */
+        sem_wait(&g_sem_driver);
+
+        if (!check_sched()) {
+#ifdef TEST_DEBUG
+            printf("READ_M: check_sched failed...\n");
+#endif
+            g_end = 1;
+            error = 2;
+            break;
+        }
+
+        if ((i + THREAD_CNT) == PACKET_CNT - 1) {
+            g_end = 1;
+        }
+    }
+
+    /* shutdown the systems */
+
+#ifdef TEST_DEBUG
+    printf("READ_M: finished testing, system shutting down...\n");
+#endif   
+    for (int i = 0; i < THREAD_CNT; i++) {
+        pthread_join(g_thrd_info[i].thrd, NULL);
+    }
+
+    if (!error)
+        success();
+    else
+        err(error, "kevent");
+}
+
+
+void
+test_evfilt_read_m()
+{
+    /* Create a connected pair of full-duplex sockets for testing socket events */
+    if (socketpair(AF_UNIX, SOCK_STREAM, 0, &g_sockfd[0]) < 0) 
+        abort();
+
+    g_kqfd = kqueue();
+    int error = ioctl(g_kqfd, FKQMULTI);
+    if (error == -1) {
+#ifdef TEST_DEBUG
+        printf("READ_M: ioctl failed with %d\n", errno);
+#endif
+        err(1, "ioctl");
+    }
+
+    test_socket_read();
+
+    close(g_kqfd);
+    close(g_sockfd[0]);
+    close(g_sockfd[1]);
+}