Re: [PATCH v3 00/13] epoll: support pollable epoll from userspace

[Peter Zijlstra <peterz@infradead.org>]

On Thu, May 16, 2019 at 10:57:57AM +0200, Roman Penyaev wrote:
> When new event comes for some epoll item kernel does the following:
> 
>  struct epoll_uitem *uitem;
> 
>  /* Each item has a bit (index in user items array), discussed later */
>  uitem = user_header->items[epi->bit];
> 
>  if (!atomic_fetch_or(uitem->ready_events, pollflags)) {
>      i = atomic_add(&ep->user_header->tail, 1);

So this is where you increment tail

> 
>      item_idx = &user_index[i & index_mask];
> 
>      /* Signal with a bit, user spins on index expecting value > 0 */
>      *item_idx = idx + 1;

IUC, this is where you write the idx into shared memory, which is
_after_ tail has already been incremented.

>  }
> 
> Important thing here is that ring can't infinitely grow and corrupt other
> elements, because kernel always checks that item was marked as ready, so
> userspace has to clear ready_events field.
> 
> On userside events the following code should be used in order to consume
> events:
> 
>  tail = READ_ONCE(header->tail);
>  for (i = 0; header->head != tail; header->head++) {
>      item_idx_ptr = &index[idx & indeces_mask];
> 
>      /*
>       * Spin here till we see valid index
>       */
>      while (!(idx = __atomic_load_n(item_idx_ptr, __ATOMIC_ACQUIRE)))
>          ;

Which you then try and fix up by busy waiting for @idx to become !0 ?!

Why not write the idx first, then increment the ->tail, such that when
we see ->tail, we already know idx must be correct?

> 
>      item = &header->items[idx - 1];
> 
>      /*
>       * Mark index as invalid, that is for userspace only, kernel does not care
>       * and will refill this pointer only when observes that event is cleared,
>       * which happens below.
>       */
>      *item_idx_ptr = 0;

That avoids this store too.

> 
>      /*
>       * Fetch data first, if event is cleared by the kernel we drop the data
>       * returning false.
>       */
>      event->data = item->event.data;
>      event->events = __atomic_exchange_n(&item->ready_events, 0,
>                          __ATOMIC_RELEASE);
> 
>  }

Aside from that, you have to READ/WRITE_ONCE() on ->head, to avoid
load/store tearing.


That would give something like:

kernel:

	slot = atomic_fetch_inc(&ep->slot);
	item_idx = &user_index[slot & idx_mask];
	WRITE_ONCE(*item_idx, idx);
	smp_store_release(&ep->user_header->tail, slot);

userspace:

	tail = smp_load_acquire(&header->tail);
	for (head = READ_ONCE(header->head); head != tail; head++) {
		idx = READ_ONCE(index[head & idx_mask]);
		itemp = &header->items[idx];

		...
	}
	smp_store_release(&header->head, head);