evl/monitor: fix polling support

Requires a couple of ABI changes.

Signed-off-by: Philippe Gerum <rpm@xenomai.org>

include/evl/poll.h

@@ -17,6 +17,8 @@
 #include <evl/factory.h>
 #include <uapi/evl/poll.h>
 
+struct file;
+
 #define EVL_POLLHEAD_INITIALIZER(__name) {				\
 		.watchpoints = LIST_HEAD_INIT((__name).watchpoints),	\
 		lock = __EVL_SPIN_LOCK_INITIALIZER((__name).lock),	\
@@ -31,6 +33,22 @@ struct evl_poll_node {
 	struct list_head next;	/* in evl_fd->poll_nodes */
 };
 
+/*
+ * The watchpoint struct linked to poll heads by drivers. This watches
+ * files not elements, so that we can monitor any type of EVL files.
+ */
+struct evl_poll_watchpoint {
+	unsigned int fd;
+	int events_polled;
+	int events_received;
+	struct oob_poll_wait wait;
+	struct evl_flag *flag;
+	struct file *filp;
+	struct evl_poll_head *head;
+	void (*unwatch)(struct file *filp);
+	struct evl_poll_node node;
+};
+
 static inline
 void evl_init_poll_head(struct evl_poll_head *head)
 {
@@ -39,7 +57,8 @@ void evl_init_poll_head(struct evl_poll_head *head)
 }
 
 void evl_poll_watch(struct evl_poll_head *head,
-		    struct oob_poll_wait *wait);
+		struct oob_poll_wait *wait,
+		void (*unwait)(struct file *filp));
 
 void __evl_signal_poll_events(struct evl_poll_head *head,
 			      int events);

include/uapi/evl/control.h

@@ -10,7 +10,7 @@
 #include <linux/types.h>
 #include <uapi/evl/sched.h>
 
-#define EVL_ABI_LEVEL  2
+#define EVL_ABI_LEVEL  3
 
 struct evl_core_info {
 	__u32 abi_level;

include/uapi/evl/monitor.h

@@ -43,8 +43,9 @@ struct evl_monitor_state {
 			__u32 ceiling;
 		} gate;
 		struct {
-			__u32 gate_offset;
 			atomic_t value;
+			atomic_t pollrefs;
+			__u32 gate_offset;
 		} event;
 	} u;
 };

kernel/evl/clock.c

@@ -688,7 +688,7 @@ static __poll_t timerfd_oob_poll(struct file *filp,
 {
 	struct evl_timerfd *timerfd = filp->private_data;
 
-	evl_poll_watch(&timerfd->poll_head, wait);
+	evl_poll_watch(&timerfd->poll_head, wait, NULL);
 
 	return timerfd->ticked ? POLLIN|POLLRDNORM : 0;
 }

kernel/evl/monitor.c

@@ -61,7 +61,7 @@ int evl_signal_monitor_targeted(struct evl_thread *target, int monfd)
 	struct evl_monitor *event;
 	struct evl_file *efilp;
 	unsigned long flags;
-	int ret = -ESRCH;
+	int ret = 0;
 
 	event = get_monitor_by_fd(monfd, &efilp);
 	if (event == NULL)
@@ -83,7 +83,6 @@ int evl_signal_monitor_targeted(struct evl_thread *target, int monfd)
 		target->info |= T_SIGNAL;
 		event->state->flags |= (EVL_MONITOR_TARGETED|
 					EVL_MONITOR_SIGNALED);
-		ret = 0;
 	}
 
 	xnlock_put_irqrestore(&nklock, flags);
@@ -309,21 +308,6 @@ static inline bool test_event_mask(struct evl_monitor_state *state,
 	}
 }
 
-static inline s32 set_event_mask(struct evl_monitor_state *state,
-				s32 addval)
-{
-	int oldval, newval;
-
-	for (;;) {
-		oldval = atomic_read(&state->u.event.value);
-		newval = oldval | (int)addval;
-		if (atomic_cmpxchg(&state->u.event.value, oldval, newval) == oldval)
-			break;
-	}
-
-	return oldval;
-}
-
 /*
  * Special forms of the wait operation which are not protected by a
  * lock but behave either as a semaphore P operation based on the
@@ -358,6 +342,9 @@ static int wait_monitor_ungated(struct evl_monitor *event,
 		ret = evl_wait_event_timeout(&event->wait_queue,
 					timeout, tmode,
 					test_event_mask(state, r_value));
+		if (!ret) /* POLLOUT if flags have been received. */
+			evl_signal_poll_events(&event->poll_head,
+					POLLOUT|POLLWRNORM);
 		break;
 	default:
 		ret = -EINVAL;	/* uh? brace for rollercoaster. */
@@ -366,24 +353,45 @@ static int wait_monitor_ungated(struct evl_monitor *event,
 	return ret;
 }
 
+static inline s32 set_event_mask(struct evl_monitor_state *state,
+				s32 addval)
+{
+	int prev, val, next;
+
+	val = atomic_read(&state->u.event.value);
+	do {
+		prev = val;
+		next = prev | (int)addval;
+		val = atomic_cmpxchg(&state->u.event.value, prev, next);
+	} while (val != prev);
+
+	return next;
+}
+
 static int signal_monitor_ungated(struct evl_monitor *event, s32 sigval)
 {
 	struct evl_monitor_state *state = event->state;
 	bool pollable = true;
 	unsigned long flags;
-	int ret = 0, oldval;
+	int ret = 0, val;
 
 	if (event->type != EVL_MONITOR_EVENT)
 		return -EINVAL;
 
 	/*
-	 * Still serializing on the nklock until we can get rid of it,
-	 * using the per-waitqueue lock instead. In any case, we have
-	 * to serialize against the read side not to lose wake up
-	 * events.
+	 * We might receive a null sigval for the purpose of
+	 * triggering a wakeup check and/or poll notification without
+	 * changing the event value.
+	 *
+	 * Also, we still serialize on the nklock until we can get rid
+	 * of it, using the per-waitqueue lock instead. In any case,
+	 * we have to serialize against the read side not to lose wake
+	 * up events.
 	 */
 	switch (event->protocol) {
 	case EVL_EVENT_COUNT:
+		if (!sigval)
+			break;
 		xnlock_get_irqsave(&nklock, flags);
 		if (atomic_inc_return(&state->u.event.value) <= 0) {
 			evl_wake_up_head(&event->wait_queue);
@@ -392,11 +400,12 @@ static int signal_monitor_ungated(struct evl_monitor *event, s32 sigval)
 		xnlock_put_irqrestore(&nklock, flags);
 		break;
 	case EVL_EVENT_MASK:
-		if (!sigval)
-			return -EINVAL;
 		xnlock_get_irqsave(&nklock, flags);
-		oldval = set_event_mask(state, (int)sigval);
-		evl_wake_up_head(&event->wait_queue);
+		val = set_event_mask(state, (int)sigval);
+		if (val)
+			evl_flush_wait_locked(&event->wait_queue, 0);
+		else
+			pollable = false;
 		xnlock_put_irqrestore(&nklock, flags);
 		break;
 	default:
@@ -404,7 +413,7 @@ static int signal_monitor_ungated(struct evl_monitor *event, s32 sigval)
 	}
 
 	if (pollable)
-		evl_signal_poll_events(&event->poll_head, POLLIN);
+		evl_signal_poll_events(&event->poll_head, POLLIN|POLLRDNORM);
 
 	evl_schedule();
 
@@ -535,12 +544,12 @@ static long monitor_common_ioctl(struct file *filp, unsigned int cmd,
 				unsigned long arg)
 {
 	struct evl_monitor *mon = element_of(filp, struct evl_monitor);
-	__s32 sigval = 0;
+	__s32 sigval;
 	int ret;
 
 	switch (cmd) {
 	case EVL_MONIOC_SIGNAL:
-		if (arg && raw_get_user(sigval, (__s32 __user *)arg))
+		if (raw_get_user(sigval, (__s32 __user *)arg))
 			return -EFAULT;
 		ret = signal_monitor_ungated(mon, sigval);
 		break;
@@ -621,6 +630,13 @@ static long monitor_oob_ioctl(struct file *filp, unsigned int cmd,
 	return ret;
 }
 
+static void monitor_unwatch(struct file *filp)
+{
+	struct evl_monitor *mon = element_of(filp, struct evl_monitor);
+
+	atomic_dec(&mon->state->u.event.pollrefs);
+}
+
 static __poll_t monitor_oob_poll(struct file *filp,
 				struct oob_poll_wait *wait)
 {
@@ -628,36 +644,49 @@ static __poll_t monitor_oob_poll(struct file *filp,
 	struct evl_monitor_state *state = mon->state;
 	__poll_t ret = 0;
 
-	evl_poll_watch(&mon->poll_head, wait);
-
+	/*
+	 * NOTE: for ungated events, we close a race window by queuing
+	 * the caller into the poll queue _before_ incrementing the
+	 * pollrefs count which userland checks.
+	 */
 	switch (mon->type) {
 	case EVL_MONITOR_EVENT:
 		switch (mon->protocol) {
 		case EVL_EVENT_COUNT:
+			evl_poll_watch(&mon->poll_head, wait, monitor_unwatch);
+			atomic_inc(&state->u.event.pollrefs);
 			if (atomic_read(&state->u.event.value) > 0)
 				ret = POLLIN|POLLRDNORM;
 			break;
 		case EVL_EVENT_MASK:
+			evl_poll_watch(&mon->poll_head, wait, monitor_unwatch);
+			atomic_inc(&state->u.event.pollrefs);
 			if (atomic_read(&state->u.event.value))
 				ret = POLLIN|POLLRDNORM;
+			else
+				ret = POLLOUT|POLLWRNORM;
 			break;
 		case EVL_EVENT_GATED:
 			/*
 			 * The poll interface does not cope with the
-			 * event one, we cannot figure out which gate
-			 * protects the event, so polling an event
-			 * will block indefinitely.
+			 * gated event one, we cannot figure out which
+			 * gate protects the event when signaling it
+			 * from userland in order to mark that gate,
+			 * so we cannot force a kernel entry upon gate
+			 * release. Therefore, polling such event will
+			 * block indefinitely.
 			 */
 			break;
 		}
 		break;
 	case EVL_MONITOR_GATE:
 		/*
-		 * We don't poll for lock ownership, because this
-		 * would slow down the fast release path in
-		 * user-space. A mutex should be held for a short
-		 * period of time anyway, so assume it is always
-		 * readable.
+		 * A mutex should be held only for a short period of
+		 * time, with the locked state appearing as a discrete
+		 * event to users. Assume a gate lock is always
+		 * readable then. If this is about probing for a mutex
+		 * state from userland then trylock() should be used
+		 * instead of poll().
 		 */
 		ret = POLLIN|POLLRDNORM;
 		break;

kernel/evl/poll.c

@@ -43,27 +43,14 @@ struct poll_waiter {
 	struct list_head next;
 };
 
-/*
- * The watchpoint struct linked to poll heads by drivers. This watches
- * files not elements, so that we can monitor any type of EVL files.
- */
-struct evl_poll_watchpoint {
-	unsigned int fd;
-	int events_polled;
-	int events_received;
-	struct oob_poll_wait wait;
-	struct evl_flag *flag;
-	struct evl_poll_head *head;
-	struct evl_poll_node node;
-};
-
 /* Maximum nesting depth (poll group watching other group(s)) */
 #define POLLER_NEST_MAX  4
 
 static const struct file_operations poll_fops;
 
 void evl_poll_watch(struct evl_poll_head *head,
-		struct oob_poll_wait *wait)
+		struct oob_poll_wait *wait,
+		void (*unwatch)(struct file *filp))
 {
 	struct evl_poll_watchpoint *wpt;
 	unsigned long flags;
@@ -73,6 +60,7 @@ void evl_poll_watch(struct evl_poll_head *head,
 	evl_spin_lock_irqsave(&head->lock, flags);
 	wpt->head = head;
 	wpt->events_received = 0;
+	wpt->unwatch = unwatch;
 	list_add(&wait->next, &head->watchpoints);
 	evl_spin_unlock_irqrestore(&head->lock, flags);
 }
@@ -291,18 +279,25 @@ void evl_drop_watchpoints(struct list_head *drop_list)
 	struct evl_poll_node *node;
 
 	/*
-	 * Drop the watchpoints attached to a closed file
-	 * descriptor. A watchpoint found in @drop_list was registered
-	 * via a call to evl_watch_fd() from wait_events() but not
-	 * unregistered by calling evl_ignore_fd() from clear_wait()
-	 * yet, so we know it is still valid.
+	 * Drop the watchpoints attached to a closed file descriptor
+	 * upon release from inband. A watchpoint found in @drop_list
+	 * was registered via a call to evl_watch_fd() from
+	 * wait_events() but not unregistered by calling
+	 * evl_ignore_fd() from clear_wait() yet, so we know it is
+	 * still valid. Since a polled EVL fd has to be passed to this
+	 * routine before the file it references can be dismantled, we
+	 * may keep and use a direct pointer to this file in the
+	 * watchpoint struct until we return.
 	 */
 	list_for_each_entry(node, drop_list, next) {
 		wpt = container_of(node, struct evl_poll_watchpoint, node);
 		evl_spin_lock(&wpt->head->lock);
 		wpt->events_received |= POLLFREE;
+		if (wpt->unwatch)
+			wpt->unwatch(wpt->filp);
 		evl_raise_flag_nosched(wpt->flag);
 		evl_spin_unlock(&wpt->head->lock);
+		wpt->filp = NULL;
 	}
 }
 
@@ -430,6 +425,7 @@ static int collect_events(struct poll_group *group,
 			if (efilp == NULL)
 				goto stale;
 			filp = efilp->filp;
+			wpt->filp = filp;
 			if (filp->f_op->oob_poll)
 				ready = filp->f_op->oob_poll(filp, &wpt->wait);
 			evl_put_file(efilp);
@@ -476,6 +472,8 @@ static inline void clear_wait(void)
 	 * Current stopped waiting for events, remove the watchpoints
 	 * we have been monitoring so far from their poll heads.
 	 * wpt->head->lock serializes with __evl_signal_poll_events().
+	 * Any watchpoint which does not bear the POLLFREE bit is
+	 * monitoring a still valid file by construction.
 	 */
 	for (n = 0, wpt = curr->poll_context.table;
 	     n < curr->poll_context.nr; n++, wpt++) {
@@ -484,6 +482,8 @@ static inline void clear_wait(void)
 		if (!list_empty(&wpt->wait.next)) {
 			evl_spin_lock_irqsave(&wpt->head->lock, flags);
 			list_del(&wpt->wait.next);
+			if (!(wpt->events_received & POLLFREE) && wpt->unwatch)
+				wpt->unwatch(wpt->filp);
 			evl_spin_unlock_irqrestore(&wpt->head->lock, flags);
 		}
 	}
@@ -494,7 +494,7 @@ int wait_events(struct file *filp,
 		struct poll_group *group,
 		struct evl_poll_waitreq *wreq)
 {
-	struct poll_waiter wait;
+	struct poll_waiter waiter;
 	enum evl_tmode tmode;
 	unsigned long flags;
 	ktime_t timeout;
@@ -509,9 +509,9 @@ int wait_events(struct file *filp,
 	if (wreq->nrset == 0)
 		return 0;
 
-	evl_init_flag(&wait.flag);
+	evl_init_flag(&waiter.flag);
 
-	count = collect_events(group, wreq->pollset, wreq->nrset, &wait.flag);
+	count = collect_events(group, wreq->pollset, wreq->nrset, &waiter.flag);
 	if (count > 0 || (count == -EFAULT || count == -EBADF))
 		goto unwait;
 	if (count < 0)
@@ -526,11 +526,11 @@ int wait_events(struct file *filp,
 	tmode = timeout ? EVL_ABS : EVL_REL;
 
 	evl_spin_lock_irqsave(&group->wait_lock, flags);
-	list_add(&wait.next, &group->waiter_list);
+	list_add(&waiter.next, &group->waiter_list);
 	evl_spin_unlock_irqrestore(&group->wait_lock, flags);
-	ret = evl_wait_flag_timeout(&wait.flag, timeout, tmode);
+	ret = evl_wait_flag_timeout(&waiter.flag, timeout, tmode);
 	evl_spin_lock_irqsave(&group->wait_lock, flags);
-	list_del(&wait.next);
+	list_del(&waiter.next);
 	evl_spin_unlock_irqrestore(&group->wait_lock, flags);
 
 	count = ret;
@@ -540,7 +540,7 @@ int wait_events(struct file *filp,
 unwait:
 	clear_wait();
 out:
-	evl_destroy_flag(&wait.flag);
+	evl_destroy_flag(&waiter.flag);
 
 	return count;
 }

kernel/evl/xbuf.c

@@ -531,7 +531,7 @@ static __poll_t xbuf_oob_poll(struct file *filp,
 	unsigned long flags;
 	__poll_t ready = 0;
 
-	evl_poll_watch(&xbuf->poll_head, wait);
+	evl_poll_watch(&xbuf->poll_head, wait, NULL);
 
 	xnlock_get_irqsave(&nklock, flags);