/* Copyright (c) 2015 Google Inc.
 * Barret Rhoden <brho@cs.berkeley.edu>
 * See LICENSE for details.
 *
 * Coalescing Event Queue: encapuslates the essence of epoll/kqueue in shared
 * memory: a dense array of sticky status bits.
 *
 * Kernel side (producer)
 *
 * All of the printks are just us helping the user debug their CEQs. */

#include <ceq.h>
#include <process.h>
#include <stdio.h>
#include <umem.h>

static void error_addr(struct ceq *ceq, struct proc *p, void *addr)
{
	printk("[kernel] Invalid ceq (%p) bad addr %p for proc %d\n", ceq,
	       addr, p->pid);
}

void send_ceq_msg(struct ceq *ceq, struct proc *p, struct event_msg *msg)
{
	struct ceq_event *ceq_ev;
	int32_t *ring_slot;
	unsigned long my_slot;
	int loops = 0;
	#define NR_RING_TRIES 10

	/* should have been checked by the kernel func that called us */
	assert(is_user_rwaddr(ceq, sizeof(struct ceq)));
	if (msg->ev_type >= ceq->nr_events) {
		printk("[kernel] CEQ %p too small.  Wanted %d, had %d\n", ceq,
		       msg->ev_type, ceq->nr_events);
		return;
	}
	/* ACCESS_ONCE, prevent the compiler from rereading ceq->events later, and
	 * possibly getting a new, illegal version after our check */
	ceq_ev = &(ACCESS_ONCE(ceq->events))[msg->ev_type];
	if (!is_user_rwaddr(ceq_ev, sizeof(struct ceq_event))) {
		error_addr(ceq, p, ceq);
		return;
	}
	/* ideally, we'd like the blob to be posted after the coal, so that the
	 * 'reason' for the blob is present when the blob is.  but we can't
	 * guarantee that.  after we write the coal, the cons could consume that.
	 * then the next time it looks at us, it could just see the blob - so
	 * there's no good way to keep them together.  the user will just have to
	 * deal with it.  in that case, we might as well do it first, to utilize the
	 * atomic ops's memory barrier. */
	ceq_ev->blob_data = (uint64_t)msg->ev_arg3;
	switch (ceq->operation) {
		case (CEQ_OR):
			atomic_or(&ceq_ev->coalesce, msg->ev_arg2);
			break;
		case (CEQ_ADD):
			atomic_add(&ceq_ev->coalesce, msg->ev_arg2);
			break;
		default:
			printk("[kernel] CEQ %p invalid op%d\n", ceq, ceq->operation);
			return;
	}
	/* write before checking if we need to post (covered by the atomic) */
	if (ceq_ev->idx_posted) {
		/* our entry was updated and posted was still set: we know the consumer
		 * will still check it, so we can safely leave.  If we ever have exit
		 * codes or something from send_*_msg, then we can tell the kernel to
		 * not bother with INDIRS/IPIs/etc.  This is unnecessary now since
		 * INDIRs are throttled */
		return;
	}
	/* at this point, we need to make sure the cons looks at our entry.  it may
	 * have already done so while we were mucking around, but 'poking' them to
	 * look again can't hurt */
	ceq_ev->idx_posted = TRUE;
	/* idx_posted write happens before the writes posting it.  the following
	 * atomic provides the cpu mb() */
	cmb();
	/* I considered checking the buffer for full-ness or the ceq overflow here.
	 * Those would be reads, which would require a wrmb() right above for every
	 * ring post, all for something we check for later anyways and for something
	 * that should be rare.  In return, when we are overflowed, which should be
	 * rare if the user sizes their ring buffer appropriately, we go through a
	 * little more hassle below. */
	/* I tried doing this with fetch_and_add to avoid the while loop and picking
	 * a number of times to try.  The trick is that you need to back out, and
	 * could have multiple producers working on the same slot.  Although the
	 * overflow makes it okay for the producers idxes to be clobbered, it's not
	 * okay to have two producers on the same slot, since there'd only be one
	 * consumer.  Theoretically, you could have a producer delayed a long time
	 * that just clobbers an index at some point in the future, or leaves an
	 * index in the non-init state (-1).  It's a mess. */
	do {
		cmb();	/* reread the indices */
		my_slot = atomic_read(&ceq->prod_idx);
		if (__ring_full(ceq->ring_sz, my_slot,
		                atomic_read(&ceq->cons_pub_idx))) {
			ceq->ring_overflowed = TRUE;
			return;
		}
		if (loops++ == NR_RING_TRIES) {
			ceq->ring_overflowed = TRUE;
			return;
		}
	} while (!atomic_cas(&ceq->prod_idx, my_slot, my_slot + 1));
	/* ring_slot is a user pointer, calculated by ring, my_slot, and sz */
	ring_slot = &(ACCESS_ONCE(ceq->ring))[my_slot & (ceq->ring_sz - 1)];
	if (!is_user_rwaddr(ring_slot, sizeof(int32_t))) {
		/* This is a serious user error.  We're just bailing out, and any
		 * consumers might be spinning waiting on us to produce.  Probably not
		 * though, since the ring slot is bad memory. */
		error_addr(ceq, p, ring_slot);
		return;
	}
	/* At this point, we have a valid slot */
	*ring_slot = msg->ev_type;
}