Adjustments of IPC to match new convention (static buffers).
[barrelfish] / lib / barrelfish / waitset.c
1 /**
2  * \file
3  * \brief Waitset and low-level event handling mechanism
4  *
5  * A "wait set" is a collection of channels to wait on, much like an
6  * FDSET in POSIX. There should be a default, static wait set for each
7  * dispatcher. Threads which wait for events specify the wait set they
8  * are waiting on.
9  */
10
11 /*
12  * Copyright (c) 2009-2012, ETH Zurich.
13  * Copyright (c) 2015, Hewlett Packard Enterprise Development LP.
14  * All rights reserved.
15  *
16  * This file is distributed under the terms in the attached LICENSE file.
17  * If you do not find this file, copies can be found by writing to:
18  * ETH Zurich D-INFK, Universitaetstr. 6, CH-8092 Zurich. Attn: Systems Group.
19  */
20
21 #include <barrelfish/barrelfish.h>
22 #include <barrelfish/waitset.h>
23 #include <barrelfish/waitset_chan.h>
24 #include <barrelfish/threads.h>
25 #include <barrelfish/dispatch.h>
26 #include "threads_priv.h"
27 #include "waitset_chan_priv.h"
28 #include <stdio.h>
29 #include <string.h>
30
31 #include <flounder/flounder.h>
32
33 #ifdef CONFIG_INTERCONNECT_DRIVER_UMP
34 #  include <barrelfish/ump_endpoint.h>
35 #endif
36
37 /// Dequeue a chanstate from a queue
38 static void dequeue(struct waitset_chanstate **queue, struct waitset_chanstate *chan)
39 {
40     if (chan->next == chan) {
41         assert(chan->prev == chan);
42         assert(*queue == chan);
43         *queue = NULL;
44     } else {
45         chan->prev->next = chan->next;
46         chan->next->prev = chan->prev;
47         if (*queue == chan) {
48             *queue = chan->next;
49         }
50     }
51     chan->prev = chan->next = NULL;
52 }
53
54 /// Enqueue a chanstate on a queue
55 static void enqueue(struct waitset_chanstate **queue, struct waitset_chanstate *chan)
56 {
57     if (*queue == NULL) {
58         *queue = chan;
59         chan->next = chan->prev = chan;
60     } else {
61         chan->next = *queue;
62         chan->prev = (*queue)->prev;
63         chan->next->prev = chan;
64         chan->prev->next = chan;
65     }
66 }
67
68 /// Dequeue a chanstate from polled queue
69 static void dequeue_polled(struct waitset_chanstate **queue,
70                             struct waitset_chanstate *chan)
71 {
72     if (chan->polled_next == chan) {
73         assert(chan->polled_prev == chan);
74         assert(*queue == chan);
75         *queue = NULL;
76     } else {
77         chan->polled_prev->polled_next = chan->polled_next;
78         chan->polled_next->polled_prev = chan->polled_prev;
79         if (*queue == chan) {
80             *queue = chan->polled_next;
81         }
82     }
83     chan->polled_prev = chan->polled_next = NULL;
84 }
85
86 /// Enqueue a chanstate on polled queue
87 static void enqueue_polled(struct waitset_chanstate **queue,
88                             struct waitset_chanstate *chan)
89 {
90     if (*queue == NULL) {
91         *queue = chan;
92         chan->polled_next = chan->polled_prev = chan;
93     } else {
94         chan->polled_next = *queue;
95         chan->polled_prev = (*queue)->polled_prev;
96         chan->polled_next->polled_prev = chan;
97         chan->polled_prev->polled_next = chan;
98     }
99 }
100
101 /**
102  * \brief Initialise a new waitset
103  */
104 void waitset_init(struct waitset *ws)
105 {
106     assert(ws != NULL);
107     ws->pending = ws->polled = ws->idle = ws->waiting = NULL;
108     ws->waiting_threads = NULL;
109 }
110
111 /**
112  * \brief Destroy a previously initialised waitset
113  */
114 errval_t waitset_destroy(struct waitset *ws)
115 {
116     assert(ws != NULL);
117
118     // FIXME: do we want to support cancelling all the pending events/channels?
119     if (ws->pending || ws->waiting_threads) {
120         return LIB_ERR_WAITSET_IN_USE;
121     }
122
123     // remove idle and polled channels from waitset
124     struct waitset_chanstate *chan, *next;
125     for (chan = ws->idle; chan != NULL; chan = next) {
126         next = chan->next;
127         assert(chan->state == CHAN_IDLE);
128         assert(chan->waitset == ws);
129         chan->waitset = NULL;
130         chan->next = chan->prev = NULL;
131
132         if (next == ws->idle) {
133             break;
134         }
135     }
136     ws->idle = NULL;
137
138     for (chan = ws->polled; chan != NULL; chan = next) {
139         next = chan->next;
140         assert(chan->state == CHAN_POLLED);
141         assert(chan->waitset == ws);
142         chan->waitset = NULL;
143         chan->next = chan->prev = NULL;
144
145         if (next == ws->polled) {
146             break;
147         }
148     }
149     ws->polled = NULL;
150
151     return SYS_ERR_OK;
152 }
153
154 /// Check if a thread can receive an event
155 static bool waitset_check_token(struct waitset_chanstate *chan,
156                                 struct thread *thread)
157 {
158     bool res = false;
159
160     if (chan->wait_for) // if a thread is waiting for this specific event
161         res = chan->wait_for == thread;
162     else
163         res = (chan->token & 1 && !thread->token) // incoming token is a request
164             // and a thread is not waiting for a token
165             || (!chan->token && chan != thread->channel) // there's no token
166             // and a thread is not waiting specifically for that event
167             || (chan->token == thread->token && chan == thread->channel);
168             // there is a token and it matches thread's token and event
169     return res;
170 }
171
172 /// Returns a channel with a pending event on the given waitset matching
173 /// our thread
174 static struct waitset_chanstate *get_pending_event_disabled(struct waitset *ws,
175                                     struct waitset_chanstate *chan)
176 {
177     struct thread *me = thread_self();
178
179     if (chan) { // channel that we wait for
180         if (chan->state == CHAN_PENDING && waitset_check_token(chan, me)) {
181             return chan;
182         }
183         if (chan->state == CHAN_WAITING && waitset_check_token(chan, me)) {
184             return chan;
185         }
186     }
187     // check a waiting queue for matching event
188     for (chan = ws->waiting; chan; ) {
189         if (waitset_check_token(chan, me)) {
190             assert_disabled(chan->state == CHAN_WAITING);
191             return chan;
192         }
193         chan = chan->next;
194         if (chan == ws->waiting)
195             break;
196     }
197     // check a pending queue for matching event
198     for (chan = ws->pending; chan;) {
199         if (waitset_check_token(chan, me)) {
200             assert_disabled(chan->state == CHAN_PENDING);
201             return chan;
202         }
203         chan = chan->next;
204         if (chan == ws->pending)
205             break;
206     }
207     return NULL;
208 }
209
210 void arranet_polling_loop_proxy(void) __attribute__((weak));
211 void arranet_polling_loop_proxy(void)
212 {
213     USER_PANIC("Network polling not available without Arranet!\n");
214 }
215
216 /// Check polled channels
217 void poll_channels_disabled(dispatcher_handle_t handle) {
218     struct dispatcher_generic *dp = get_dispatcher_generic(handle);
219     struct waitset_chanstate *chan;
220
221     if (!dp->polled_channels)
222         return;
223     chan = dp->polled_channels;
224     do {
225         switch (chan->chantype) {
226 #ifdef CONFIG_INTERCONNECT_DRIVER_UMP
227         case CHANTYPE_UMP_IN: {
228             if (ump_endpoint_poll(chan)) {
229                 errval_t err = waitset_chan_trigger_disabled(chan, handle);
230                 assert(err_is_ok(err)); // should not fail
231                 if (!dp->polled_channels) // restart scan
232                     return;
233                 chan = dp->polled_channels;
234                 continue;
235             } else
236                 chan = chan->polled_next;
237         } break;
238 #endif // CONFIG_INTERCONNECT_DRIVER_UMP
239         case CHANTYPE_LWIP_SOCKET:
240             arranet_polling_loop_proxy();
241             break;
242
243         default:
244             assert(!"invalid channel type to poll!");
245         }
246     } while (chan != dp->polled_channels);
247 }
248
249 /// Re-register a channel (if persistent)
250 static void reregister_channel(struct waitset *ws, struct waitset_chanstate *chan,
251                                 dispatcher_handle_t handle)
252 {
253     assert(chan->waitset == ws);
254     if (chan->state == CHAN_PENDING) {
255         dequeue(&ws->pending, chan);
256     } else {
257         assert(chan->state == CHAN_WAITING);
258         dequeue(&ws->waiting, chan);
259     }
260
261     chan->token = 0;
262     if (chan->chantype == CHANTYPE_UMP_IN) {
263         enqueue(&ws->polled, chan);
264         enqueue_polled(&get_dispatcher_generic(handle)->polled_channels, chan);
265         chan->state = CHAN_POLLED;
266     } else {
267         enqueue(&ws->idle, chan);
268         chan->state = CHAN_IDLE;
269     }
270 }
271
272 /// Find a thread that is able to receive an event
273 static struct thread * find_recipient(struct waitset *ws,
274                         struct waitset_chanstate *channel, struct thread *me)
275 {
276     struct thread *t = ws->waiting_threads;
277
278     if (!t)
279         return NULL;
280     do {
281         if (waitset_check_token(channel, t))
282             return t;
283         t = t->next;
284     } while (t != ws->waiting_threads);
285     return ws->waiting_threads;
286 }
287
288 /// Wake up other thread if there's more pending events
289 static void wake_up_other_thread(dispatcher_handle_t handle, struct waitset *ws)
290 {
291     if (ws->pending && ws->waiting_threads) {
292         struct thread *t;
293
294         t = thread_unblock_one_disabled(handle, &ws->waiting_threads, NULL);
295         assert_disabled(t == NULL); // shouldn't see a remote thread
296     }
297 }
298
299 /**
300  * \brief Get next pending event
301  *
302  * Check if there is a pending event that matches current thread and return it.
303  * Pending events are in a pending queue and in a waiting queue.
304  * A pending event then will be removed from a pending/waiting queue and become
305  * unregistered or, if it's persistent, will be re-registered to an idle queue
306  * or a polled queue (UMP channels) of a waitset.
307  * If there's no pending event, block this thread.
308  * If there's a pending event but it doesn't match our thread, don't remove it
309  * from a pending queue and wake up a matching thread.
310  * If there's no matching thread, add it to a waiting queue.
311  *
312  * \param ws Waitset with sources of events
313  * \param retchannel Holder of returned event
314  * \param retclosure Holder of returned closure
315  * \param waitfor Specific event that we're waiting for (can be NULL)
316  * \param handle Dispatcher's handle
317  * \param debug Debug mode (not used)
318  */
319
320 errval_t get_next_event_disabled(struct waitset *ws,
321     struct waitset_chanstate **retchannel, struct event_closure *retclosure,
322     struct waitset_chanstate *waitfor, dispatcher_handle_t handle, bool debug)
323 {
324     struct waitset_chanstate * chan;
325
326     for (;;) {
327         chan = get_pending_event_disabled(ws, waitfor); // get our event
328         if (chan) {
329             *retchannel = chan;
330             *retclosure = chan->closure;
331             chan->wait_for = NULL;
332             chan->token = 0;
333             if (chan->persistent)
334                 reregister_channel(ws, chan, handle);
335             else
336                 waitset_chan_deregister_disabled(chan, handle);
337             wake_up_other_thread(handle, ws);
338             return SYS_ERR_OK;
339         }
340         chan = ws->pending; // check a pending queue
341         if (!chan) { // if nothing then wait
342             thread_block_disabled(handle, &ws->waiting_threads);
343             disp_disable();
344         } else { // something but it's not our event
345             if (!ws->waiting_threads) { // no other thread interested in
346                 dequeue(&ws->pending, chan);
347                 enqueue(&ws->waiting, chan);
348                 chan->state = CHAN_WAITING;
349                 chan->waitset = ws;
350             } else {
351                 // find a matching thread
352                 struct thread *t;
353                 for (t = ws->waiting_threads; t; ) {
354                     if (waitset_check_token(chan, t)) { // match found, wake it
355                         ws->waiting_threads = t;
356                         t = thread_unblock_one_disabled(handle,
357                                                     &ws->waiting_threads, chan);
358                         assert_disabled(t == NULL); // shouldn't see a remote thread
359                         break;
360                     }
361                     t = t->next;
362                     if (t == ws->waiting_threads) { // no recipient found
363                         dequeue(&ws->pending, chan);
364                         enqueue(&ws->waiting, chan);
365                         chan->state = CHAN_WAITING;
366                         chan->waitset = ws;
367                         break;
368                     }
369                 }
370             }
371         }
372     }
373 }
374
375 /**
376  * \brief Wait for (block) and return next event on given waitset
377  *
378  * Wait until something happens, either activity on some channel, or a deferred
379  * call, and then return the corresponding closure. This is the core of the
380  * event-handling system.
381  *
382  * \param ws Waitset
383  * \param retclosure Pointer to storage space for returned event closure
384  */
385 errval_t get_next_event(struct waitset *ws, struct event_closure *retclosure)
386 {
387     dispatcher_handle_t handle = disp_disable();
388     struct waitset_chanstate *channel;
389     errval_t err = get_next_event_disabled(ws, &channel, retclosure, NULL,
390                                             handle, false);
391     disp_enable(handle);
392     return err;
393 }
394
395
396
397 /**
398  * \brief Check if there is an event pending on given waitset
399  *
400  * This is essentially a non-blocking variant of get_next_event(). It should be
401  * used with great care, to avoid the creation of busy-waiting loops.
402  *
403  * \param ws Waitset
404  *
405  * \returns LIB_ERR_NO_EVENT if nothing is pending
406  */
407 static errval_t check_for_event_disabled(struct waitset *ws, dispatcher_handle_t handle)
408 {
409     struct waitset_chanstate *chan;
410
411     poll_channels_disabled(handle);
412     chan = get_pending_event_disabled(ws, NULL);
413     if (chan != NULL) {
414         return SYS_ERR_OK;
415     }
416     return LIB_ERR_NO_EVENT;
417 }
418
419 errval_t check_for_event(struct waitset *ws)
420 {
421     errval_t err;
422
423     assert(ws != NULL);
424     dispatcher_handle_t handle = disp_disable();
425     err = check_for_event_disabled(ws, handle);
426     disp_enable(handle);
427     return err;
428 }
429
430 /**
431  * \brief Wait for (block) and dispatch next event on given waitset
432  *
433  * Wait until something happens, either activity on some channel, or deferred
434  * call, and then call the corresponding closure.
435  *
436  * \param ws Waitset
437  */
438
439 errval_t event_dispatch(struct waitset *ws)
440 {
441     struct event_closure closure;
442     errval_t err = get_next_event(ws, &closure);
443     if (err_is_fail(err)) {
444         return err;
445     }
446
447     assert(closure.handler != NULL);
448     closure.handler(closure.arg);
449     return SYS_ERR_OK;
450 }
451
452 errval_t event_dispatch_debug(struct waitset *ws)
453 {
454     struct event_closure closure;
455     struct waitset_chanstate *channel;
456     dispatcher_handle_t handle = disp_disable();
457     errval_t err = get_next_event_disabled(ws, &channel, &closure, NULL, handle,
458                                             true);
459     disp_enable(handle);
460     if (err_is_fail(err)) {
461         return err;
462     }
463
464     assert(closure.handler != NULL);
465     closure.handler(closure.arg);
466     return SYS_ERR_OK;
467 }
468
469 /**
470  * \brief Dispatch events until a specific event is received
471  *
472  * Wait for events and dispatch them. If a specific event comes, don't call
473  * a closure, just return.
474  *
475  * \param ws Waitset
476  * \param waitfor Event, that we are waiting for
477  * \param error_var Error variable that can be changed by closures
478  */
479
480 errval_t wait_for_channel(struct waitset *ws, struct waitset_chanstate *waitfor,
481                             errval_t *error_var)
482 {
483     assert(waitfor->waitset == ws);
484     for (;;) {
485         struct event_closure closure;
486         struct waitset_chanstate *channel;
487
488         dispatcher_handle_t handle = disp_disable();
489         errval_t err = get_next_event_disabled(ws, &channel, &closure, waitfor,
490                                                 handle, false);
491         disp_enable(handle);
492         if (err_is_fail(err)) {
493             assert(0);
494             return err;
495         }
496         if (channel == waitfor) {
497             return SYS_ERR_OK;
498         }
499         assert(!channel->wait_for);
500         assert(closure.handler != NULL);
501         closure.handler(closure.arg);
502         if (err_is_fail(*error_var))
503             return *error_var;
504     }
505 }
506
507 /**
508  * \brief check and dispatch next event on given waitset
509  *
510  * Check if there is any pending activity on some channel, or deferred
511  * call, and then call the corresponding closure.
512  *
513  * Do not wait!  In case of no pending events, return err LIB_ERR_NO_EVENT.
514  *
515  * \param ws Waitset
516  */
517 errval_t event_dispatch_non_block(struct waitset *ws)
518 {
519     struct waitset_chanstate *channel;
520     struct event_closure closure;
521
522     assert(ws != NULL);
523
524     // are there any pending events on the waitset?
525     dispatcher_handle_t handle = disp_disable();
526     errval_t err = check_for_event_disabled(ws, handle);
527     if (err_is_fail(err)) {
528         disp_enable(handle);
529         return err;
530     }
531     err = get_next_event_disabled(ws, &channel, &closure, NULL, handle,
532                                             false);
533     if (err_is_fail(err))
534         return err;
535     disp_enable(handle);
536     assert(closure.handler != NULL);
537     closure.handler(closure.arg);
538     return SYS_ERR_OK;
539 }
540
541
542 /**
543  * \privatesection
544  * "Private" functions that are called only by the channel implementations
545  */
546
547 /**
548  * \brief Initialise per-channel waitset state
549  *
550  * \param chan Channel state
551  * \param chantype Channel type
552  */
553 void waitset_chanstate_init(struct waitset_chanstate *chan,
554                             enum ws_chantype chantype)
555 {
556     assert(chan != NULL);
557     chan->waitset = NULL;
558     chan->chantype = chantype;
559     chan->state = CHAN_UNREGISTERED;
560 #ifndef NDEBUG
561     chan->prev = chan->next = NULL;
562 #endif
563     chan->persistent = false;
564     chan->token = 0;
565     chan->wait_for = NULL;
566 }
567
568 /**
569  * \brief Destroy previously-initialised per-channel waitset state
570  * \param chan Channel state
571  */
572 void waitset_chanstate_destroy(struct waitset_chanstate *chan)
573 {
574     assert(chan != NULL);
575     if (chan->waitset != NULL) {
576         errval_t err = waitset_chan_deregister(chan);
577         assert(err_is_ok(err)); // can't fail if registered
578     }
579 }
580
581 /**
582  * \brief Register a closure to be called when a channel is triggered
583  *
584  * In the Future, call the closure on a thread associated with the waitset
585  * when the channel is triggered. Only one closure may be registered per
586  * channel state at any one time.
587  * This function must only be called when disabled.
588  *
589  * \param ws Waitset
590  * \param chan Waitset's per-channel state
591  * \param closure Event handler
592  */
593 errval_t waitset_chan_register_disabled(struct waitset *ws,
594                                         struct waitset_chanstate *chan,
595                                         struct event_closure closure)
596 {
597     if (chan->waitset != NULL) {
598         return LIB_ERR_CHAN_ALREADY_REGISTERED;
599     }
600
601     chan->waitset = ws;
602     chan->token = 0;
603
604     // channel must not already be registered!
605     assert_disabled(chan->next == NULL && chan->prev == NULL);
606     assert_disabled(chan->state == CHAN_UNREGISTERED);
607
608     // this is probably insane! :)
609     assert_disabled(closure.handler != NULL);
610
611     // store closure
612     chan->closure = closure;
613
614     // enqueue this channel on the waitset's queue of idle channels
615     enqueue(&ws->idle, chan);
616     chan->state = CHAN_IDLE;
617
618     return SYS_ERR_OK;
619 }
620
621 /**
622  * \brief Register a closure on a channel, and mark the channel as polled
623  *
624  * In the Future, call the closure on a thread associated with the waitset
625  * when the channel is triggered. Only one closure may be registered per
626  * channel state at any one time. Additionally, mark the channel as polled.
627  * This function must only be called when disabled.
628  *
629  * \param ws Waitset
630  * \param chan Waitset's per-channel state
631  * \param closure Event handler
632  * \param disp Current dispatcher pointer
633  */
634 errval_t waitset_chan_register_polled_disabled(struct waitset *ws,
635                                                struct waitset_chanstate *chan,
636                                                struct event_closure closure,
637                                                dispatcher_handle_t handle)
638 {
639     if (chan->waitset != NULL) {
640         return LIB_ERR_CHAN_ALREADY_REGISTERED;
641     }
642
643     chan->waitset = ws;
644     chan->token = 0;
645
646     // channel must not already be registered!
647     assert_disabled(chan->next == NULL && chan->prev == NULL);
648     assert_disabled(chan->state == CHAN_UNREGISTERED);
649
650     // store closure
651     chan->closure = closure;
652
653     // enqueue this channel on the waitset's queue of polled channels
654     enqueue(&ws->polled, chan);
655     chan->state = CHAN_POLLED;
656     enqueue_polled(&get_dispatcher_generic(handle)->polled_channels, chan);
657
658     return SYS_ERR_OK;
659 }
660
661 /**
662  * \brief Register a closure to be called when a channel is triggered
663  *
664  * In the Future, call the closure on a thread associated with the waitset
665  * when the channel is triggered. Only one closure may be registered per
666  * channel state at any one time.
667  * This function must only be called when enabled.
668  *
669  * \param ws Waitset
670  * \param chan Waitset's per-channel state
671  * \param closure Event handler
672  */
673 errval_t waitset_chan_register(struct waitset *ws, struct waitset_chanstate *chan,
674                                struct event_closure closure)
675 {
676     dispatcher_handle_t handle = disp_disable();
677     errval_t err = waitset_chan_register_disabled(ws, chan, closure);
678     disp_enable(handle);
679     return err;
680 }
681
682 /**
683  * \brief Register a closure on a channel, and mark the channel as polled
684  *
685  * In the Future, call the closure on a thread associated with the waitset
686  * when the channel is triggered. Only one closure may be registered per
687  * channel state at any one time. Additionally, mark the channel as polled.
688  * This function must only be called when enabled. It is equivalent to
689  * calling waitset_chan_register() followed by waitset_chan_start_polling().
690  *
691  * \param ws Waitset
692  * \param chan Waitset's per-channel state
693  * \param closure Event handler
694  */
695 errval_t waitset_chan_register_polled(struct waitset *ws,
696                                       struct waitset_chanstate *chan,
697                                       struct event_closure closure)
698 {
699     dispatcher_handle_t handle = disp_disable();
700     errval_t err = waitset_chan_register_polled_disabled(ws, chan, closure, handle);
701     disp_enable(handle);
702     return err;
703 }
704
705 /**
706  * \brief Cancel a previous callback registration
707  *
708  * Remove the registration for a callback on the given channel.
709  * This function must only be called when disabled.
710  *
711  * \param chan Waitset's per-channel state
712  */
713 errval_t waitset_chan_deregister_disabled(struct waitset_chanstate *chan,
714                                           dispatcher_handle_t handle)
715 {
716     assert_disabled(chan != NULL);
717     struct waitset *ws = chan->waitset;
718     if (ws == NULL) {
719         return LIB_ERR_CHAN_NOT_REGISTERED;
720     }
721
722     // remove this channel from the queue in which it is waiting
723     chan->waitset = NULL;
724     assert_disabled(chan->next != NULL && chan->prev != NULL);
725
726     switch (chan->state) {
727     case CHAN_IDLE:
728         dequeue(&ws->idle, chan);
729         break;
730
731     case CHAN_POLLED:
732         dequeue(&ws->polled, chan);
733         dequeue_polled(&get_dispatcher_generic(handle)->polled_channels, chan);
734         break;
735
736     case CHAN_PENDING:
737         dequeue(&ws->pending, chan);
738         break;
739
740     case CHAN_WAITING:
741         dequeue(&ws->waiting, chan);
742         break;
743
744     default:
745         assert_disabled(!"invalid channel state in deregister");
746     }
747     chan->state = CHAN_UNREGISTERED;
748     chan->wait_for = NULL;
749     return SYS_ERR_OK;
750 }
751
752 /**
753  * \brief Cancel a previous callback registration
754  *
755  * Remove the registration for a callback on the given channel.
756  * This function must only be called when enabled.
757  *
758  * \param chan Waitset's per-channel state
759  */
760 errval_t waitset_chan_deregister(struct waitset_chanstate *chan)
761 {
762     dispatcher_handle_t handle = disp_disable();
763     errval_t err = waitset_chan_deregister_disabled(chan, handle);
764     disp_enable(handle);
765     return err;
766 }
767
768 /**
769  * \brief Migrate callback registrations to a new waitset.
770  *
771  * \param chan Old waitset's per-channel state to migrate
772  * \param new_ws New waitset to migrate to
773  */
774 void waitset_chan_migrate(struct waitset_chanstate *chan,
775                           struct waitset *new_ws)
776 {
777     struct waitset *ws = chan->waitset;
778
779     // Only when registered
780     if(ws == NULL) {
781         return;
782     }
783
784     switch(chan->state) {
785     case CHAN_IDLE:
786         dequeue(&ws->idle, chan);
787         enqueue(&new_ws->idle, chan);
788         break;
789
790     case CHAN_POLLED:
791         dequeue(&ws->polled, chan);
792         enqueue(&new_ws->polled, chan);
793         break;
794
795     case CHAN_PENDING:
796         dequeue(&ws->pending, chan);
797         enqueue(&new_ws->pending, chan);
798         break;
799
800     case CHAN_WAITING:
801         dequeue(&ws->waiting, chan);
802         enqueue(&new_ws->waiting, chan);
803         break;
804
805     case CHAN_UNREGISTERED:
806         // Do nothing
807         break;
808     }
809
810     // Remember new waitset association
811     chan->waitset = new_ws;
812 }
813
814 /**
815  * \brief Trigger an event callback on a channel
816  *
817  * Marks the given channel as having a pending event, causing some future call
818  * to get_next_event() to return the registered closure.
819  * This function must only be called when disabled.
820  *
821  * \param chan Waitset's per-channel state
822  * \param disp Current dispatcher pointer
823  */
824 errval_t waitset_chan_trigger_disabled(struct waitset_chanstate *chan,
825                                        dispatcher_handle_t handle)
826 {
827     assert_disabled(chan != NULL);
828     struct waitset *ws = chan->waitset;
829     assert_disabled(ws != NULL);
830     assert_disabled(chan->prev != NULL && chan->next != NULL);
831
832     // no-op if already pending
833     if (chan->state == CHAN_PENDING) {
834         return SYS_ERR_OK;
835     }
836
837     // remove from previous queue (either idle or polled)
838     if (chan->state == CHAN_IDLE) {
839         dequeue(&ws->idle, chan);
840     } else {
841         assert_disabled(chan->state == CHAN_POLLED);
842         dequeue(&ws->polled, chan);
843         dequeue_polled(&get_dispatcher_generic(handle)->polled_channels, chan);
844     }
845
846     // else mark channel pending and move to end of pending event queue
847     enqueue(&ws->pending, chan);
848     chan->state = CHAN_PENDING;
849
850     // is there a thread blocked on this waitset? if so, awaken it with the event
851     struct thread *thread = find_recipient(ws, chan, thread_self());
852     if (thread) {
853         struct thread *t;
854         ws->waiting_threads = thread;
855         t = thread_unblock_one_disabled(handle, &ws->waiting_threads, chan);
856         assert_disabled(t == NULL);
857     }
858     return SYS_ERR_OK;
859 }
860
861 /**
862  * \brief Trigger an event callback on a channel
863  *
864  * Marks the given channel as having a pending event, causing some future call
865  * to get_next_event() to return the registered closure.
866  * This function must only be called when enabled.
867  *
868  * \param chan Waitset's per-channel state
869  * \param disp Current dispatcher pointer
870  */
871 errval_t waitset_chan_trigger(struct waitset_chanstate *chan)
872 {
873     dispatcher_handle_t handle = disp_disable();
874     errval_t err = waitset_chan_trigger_disabled(chan, handle);
875     disp_enable(handle);
876     return err;
877 }
878
879 /**
880  * \brief Trigger a specific event callback on an unregistered channel
881  *
882  * This function is equivalent to waitset_chan_register_disabled() immediately
883  * followed by waitset_chan_trigger_disabled(), but avoids unneccessary queue
884  * manipulation. This function must only be called when disabled.
885  *
886  * \param ws Waitset
887  * \param chan Waitset's per-channel state
888  * \param closure Event handler
889  * \param disp Current dispatcher pointer
890  */
891 errval_t waitset_chan_trigger_closure_disabled(struct waitset *ws,
892                                                struct waitset_chanstate *chan,
893                                                struct event_closure closure,
894                                                dispatcher_handle_t handle)
895 {
896     assert_disabled(chan != NULL);
897     assert_disabled(ws != NULL);
898
899     // check if already registered
900     if (chan->waitset != NULL || chan->state != CHAN_UNREGISTERED) {
901         return LIB_ERR_CHAN_ALREADY_REGISTERED;
902     }
903
904     assert_disabled(chan->prev == NULL && chan->next == NULL);
905
906     // set closure
907     chan->closure = closure;
908
909     // mark channel pending and place on end of pending event queue
910     chan->waitset = ws;
911     enqueue(&ws->pending, chan);
912     // if (first)
913     //     ws->pending = chan;
914     chan->state = CHAN_PENDING;
915
916     // is there a thread blocked on this waitset? if so, awaken it with the event
917     struct thread *thread = find_recipient(ws, chan, thread_self());
918     if (thread) {
919         struct thread *t;
920         ws->waiting_threads = thread;
921         t = thread_unblock_one_disabled(handle, &ws->waiting_threads, chan);
922         assert_disabled(t == NULL);
923     }
924     return SYS_ERR_OK;
925 }
926
927
928 /**
929  * \brief Trigger a specific event callback on an unregistered channel
930  *
931  * This function is equivalent to waitset_chan_register()
932  * followed by waitset_chan_trigger(), but avoids unneccessary queue
933  * manipulation. This function must only be called when enabled.
934  *
935  * \param ws Waitset
936  * \param chan Waitset's per-channel state
937  * \param closure Event handler
938  */
939 errval_t waitset_chan_trigger_closure(struct waitset *ws,
940                                       struct waitset_chanstate *chan,
941                                       struct event_closure closure)
942 {
943     dispatcher_handle_t disp = disp_disable();
944     errval_t err = waitset_chan_trigger_closure_disabled(ws, chan, closure, disp);
945     disp_enable(disp);
946     return err;
947 }