9ns: Add tree_chan_ctl()
[akaros.git] / kern / src / ceq.c
1 /* Copyright (c) 2015 Google Inc.
2  * Barret Rhoden <brho@cs.berkeley.edu>
3  * See LICENSE for details.
4  *
5  * Coalescing Event Queue: encapuslates the essence of epoll/kqueue in shared
6  * memory: a dense array of sticky status bits.
7  *
8  * Kernel side (producer)
9  *
10  * All of the printks are just us helping the user debug their CEQs. */
11
12 #include <ceq.h>
13 #include <process.h>
14 #include <stdio.h>
15 #include <umem.h>
16
17 static void error_addr(struct ceq *ceq, struct proc *p, void *addr)
18 {
19         printk("[kernel] Invalid ceq (%p) bad addr %p for proc %d\n", ceq,
20                addr, p->pid);
21 }
22
23 static void ceq_update_max_event(struct ceq *ceq, unsigned int new_max)
24 {
25         unsigned int old_max;
26
27         do {
28                 old_max = atomic_read(&ceq->max_event_ever);
29                 if (new_max <= old_max)
30                         return;
31         } while (!atomic_cas(&ceq->max_event_ever, old_max, new_max));
32 }
33
34 void send_ceq_msg(struct ceq *ceq, struct proc *p, struct event_msg *msg)
35 {
36         struct ceq_event *ceq_ev;
37         int32_t *ring_slot;
38         unsigned long my_slot;
39         int loops = 0;
40         #define NR_RING_TRIES 10
41
42         /* should have been checked by the kernel func that called us */
43         assert(is_user_rwaddr(ceq, sizeof(struct ceq)));
44         if (msg->ev_type >= ceq->nr_events) {
45                 printk("[kernel] CEQ %p too small.  Wanted %d, had %d\n", ceq,
46                        msg->ev_type, ceq->nr_events);
47                 return;
48         }
49         ceq_update_max_event(ceq, msg->ev_type);
50         /* ACCESS_ONCE, prevent the compiler from rereading ceq->events later, and
51          * possibly getting a new, illegal version after our check */
52         ceq_ev = &(ACCESS_ONCE(ceq->events))[msg->ev_type];
53         if (!is_user_rwaddr(ceq_ev, sizeof(struct ceq_event))) {
54                 error_addr(ceq, p, ceq);
55                 return;
56         }
57         /* ideally, we'd like the blob to be posted after the coal, so that the
58          * 'reason' for the blob is present when the blob is.  but we can't
59          * guarantee that.  after we write the coal, the cons could consume that.
60          * then the next time it looks at us, it could just see the blob - so
61          * there's no good way to keep them together.  the user will just have to
62          * deal with it.  in that case, we might as well do it first, to utilize the
63          * atomic ops's memory barrier. */
64         ceq_ev->blob_data = (uint64_t)msg->ev_arg3;
65         switch (ceq->operation) {
66                 case (CEQ_OR):
67                         atomic_or(&ceq_ev->coalesce, msg->ev_arg2);
68                         break;
69                 case (CEQ_ADD):
70                         atomic_add(&ceq_ev->coalesce, msg->ev_arg2);
71                         break;
72                 default:
73                         printk("[kernel] CEQ %p invalid op %d\n", ceq, ceq->operation);
74                         return;
75         }
76         /* write before checking if we need to post (covered by the atomic) */
77         if (ceq_ev->idx_posted) {
78                 /* our entry was updated and posted was still set: we know the consumer
79                  * will still check it, so we can safely leave.  If we ever have exit
80                  * codes or something from send_*_msg, then we can tell the kernel to
81                  * not bother with INDIRS/IPIs/etc.  This is unnecessary now since
82                  * INDIRs are throttled */
83                 return;
84         }
85         /* at this point, we need to make sure the cons looks at our entry.  it may
86          * have already done so while we were mucking around, but 'poking' them to
87          * look again can't hurt */
88         ceq_ev->idx_posted = TRUE;
89         /* idx_posted write happens before the writes posting it.  the following
90          * atomic provides the cpu mb() */
91         cmb();
92         /* I considered checking the buffer for full-ness or the ceq overflow here.
93          * Those would be reads, which would require a wrmb() right above for every
94          * ring post, all for something we check for later anyways and for something
95          * that should be rare.  In return, when we are overflowed, which should be
96          * rare if the user sizes their ring buffer appropriately, we go through a
97          * little more hassle below. */
98         /* I tried doing this with fetch_and_add to avoid the while loop and picking
99          * a number of times to try.  The trick is that you need to back out, and
100          * could have multiple producers working on the same slot.  Although the
101          * overflow makes it okay for the producers idxes to be clobbered, it's not
102          * okay to have two producers on the same slot, since there'd only be one
103          * consumer.  Theoretically, you could have a producer delayed a long time
104          * that just clobbers an index at some point in the future, or leaves an
105          * index in the non-init state (-1).  It's a mess. */
106         do {
107                 cmb();  /* reread the indices */
108                 my_slot = atomic_read(&ceq->prod_idx);
109                 if (__ring_full(ceq->ring_sz, my_slot,
110                                 atomic_read(&ceq->cons_pub_idx))) {
111                         ceq->ring_overflowed = TRUE;
112                         return;
113                 }
114                 if (loops++ == NR_RING_TRIES) {
115                         ceq->ring_overflowed = TRUE;
116                         return;
117                 }
118         } while (!atomic_cas(&ceq->prod_idx, my_slot, my_slot + 1));
119         /* ring_slot is a user pointer, calculated by ring, my_slot, and sz */
120         ring_slot = &(ACCESS_ONCE(ceq->ring))[my_slot & (ceq->ring_sz - 1)];
121         if (!is_user_rwaddr(ring_slot, sizeof(int32_t))) {
122                 /* This is a serious user error.  We're just bailing out, and any
123                  * consumers might be spinning waiting on us to produce.  Probably not
124                  * though, since the ring slot is bad memory. */
125                 error_addr(ceq, p, ring_slot);
126                 return;
127         }
128         /* At this point, we have a valid slot */
129         *ring_slot = msg->ev_type;
130 }
131
132 void ceq_dumper(int pid, struct event_queue *ev_q)
133 {
134         struct proc *p;
135         uintptr_t switch_state;
136         struct ceq *ceq;
137
138         p = pid2proc(pid);
139         if (!p) {
140                 printk("No such proc %d\n", pid);
141                 return;
142         }
143         switch_state = switch_to(p);
144         if (ev_q->ev_mbox->type != EV_MBOX_CEQ) {
145                 printk("Not a CEQ evq (type %d)\n", ev_q->ev_mbox->type);
146                 goto out;
147         }
148         ceq = &ev_q->ev_mbox->ceq;
149         printk("CEQ %p\n---------------\n"
150                "\tevents ptr %p\n"
151                "\tnr_events %d\n"
152                "\tlast_recovered %d\n"
153                "\tmax_event_ever %ld\n"
154                "\tring %p\n"
155                "\tring_sz %d\n"
156                "\toperation %d\n"
157                "\tring_overflowed %d\n"
158                "\toverflow_recovery %d\n"
159                "\tprod_idx %lu\n"
160                "\tcons_pub_idx %lu\n"
161                "\tcons_pvt_idx %lu\n"
162                "\n",
163                    ceq,
164                ceq->events,
165                ceq->nr_events,
166                ceq->last_recovered,
167                atomic_read(&ceq->max_event_ever),
168                ceq->ring,
169                ceq->ring_sz,
170                ceq->operation,
171                ceq->ring_overflowed,
172                ceq->overflow_recovery,
173                atomic_read(&ceq->prod_idx),
174                atomic_read(&ceq->cons_pub_idx),
175                atomic_read(&ceq->cons_pvt_idx));
176         for (int i = 0; i < atomic_read(&ceq->max_event_ever) + 1; i++)
177                 printk("\tEvent %3d, coal %p, blob %p, idx_posted %d, user %p\n", i,
178                        atomic_read(&ceq->events[i].coalesce),
179                        ceq->events[i].blob_data,
180                        ceq->events[i].idx_posted,
181                        ceq->events[i].user_data);
182 out:
183         switch_back(p, switch_state);
184         proc_decref(p);
185 }