errval_t flounder_stub_send_cap(struct flounder_cap_state *s,
                                struct monitor_binding *mb,
                                uintptr_t monitor_id,
                                struct capref cap, bool give_away,
                                void (*cont)(void *st))
{
    errval_t err;

    s->cap_send_continuation = cont;

    if (give_away) {
        err = mb->tx_vtbl.cap_move_request(mb, MKCONT(cap_send_cont, s),
                                           monitor_id, cap, s->tx_capnum);
    }
    else {
        err = mb->tx_vtbl.cap_send_request(mb, MKCONT(cap_send_cont, s),
                                           monitor_id, cap, s->tx_capnum);
    }
    if (err_is_ok(err)) {
        s->tx_capnum++;
        return err;
    } else if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        // register to retry
        return mb->register_send(mb, mb->waitset, MKCONT(cap_send_cont, s));
    } else {
        return err_push(err, LIB_ERR_MONITOR_CAP_SEND);
    }
}

static errval_t spawn_reply(struct spawn_binding *b, errval_t rerr,
                            domainid_t domainid)
{
    errval_t err;

    err = b->tx_vtbl.spawn_domain_response(b, NOP_CONT, rerr, domainid);

    if (err_is_fail(err)) {
        DEBUG_ERR(err, "error sending spawn_domain reply\n");

        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            // this will be freed in the retry handler
            struct pending_spawn_response *sr =
                malloc(sizeof(struct pending_spawn_response));
            if (sr == NULL) {
                return LIB_ERR_MALLOC_FAIL;
            }
            sr->b = b;
            sr->err = rerr;
            sr->domainid = domainid;
            err = b->register_send(b, get_default_waitset(),
                                   MKCONT(retry_spawn_domain_response, sr));
            if (err_is_fail(err)) {
                // note that only one continuation may be registered at a time
                free(sr);
                DEBUG_ERR(err, "register_send failed!");
                return err;
            }
        }
    }

    return SYS_ERR_OK;
}

static void
new_monitor_binding_reply_cont(struct monitor_binding *b,
                               errval_t reterr, struct capref retcap,
                               uintptr_t st)
{
    errval_t err =
        b->tx_vtbl.new_monitor_binding_reply(b, NOP_CONT, reterr, retcap, st);

    if (err_is_fail(err)) {
        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            struct monitor_state *ms = b->st;
            struct new_monitor_binding_reply_state *me =
                malloc(sizeof(struct new_monitor_binding_reply_state));
            assert(me != NULL);
            me->args.err = reterr;
            me->args.ep = retcap;
            me->args.st = st;
            me->elem.cont = new_monitor_binding_reply_handler;
            err = monitor_enqueue_send(b, &ms->queue,
                                       get_default_waitset(), &me->elem.queue);
            if (err_is_fail(err)) {
                USER_PANIC_ERR(err, "monitor_enqueue_send failed");
            }
            return;
        }

        USER_PANIC_ERR(err, "failed to send new_monitor_binding_reply");
    }
}

/**
 * \brief polls the channels of the IOAT DMA device
 *
 * \param dev   IOAT DMA device
 *
 * \returns SYS_ERR_OK on success
 *          DMA_ERR_DEVICE_IDLE if there is nothing completed on the channels
 *          errval on error
 */
errval_t ioat_dma_device_poll_channels(struct dma_device *dev)
{
    errval_t err;

    uint8_t idle = 0x1;

    for (uint8_t i = 0; i < dev->channels.count; ++i) {
        err = ioat_dma_channel_poll(dev->channels.c[i]);
        switch (err_no(err)) {
            case DMA_ERR_CHAN_IDLE:
                break;
            case SYS_ERR_OK:
                idle = 0;
                break;
            default:
                return err;
        }
    }

    if (idle) {
        return DMA_ERR_DEVICE_IDLE;
    }

    return SYS_ERR_OK;
}

static void get_io_cap(struct monitor_blocking_binding *b)
{
    // XXX: We should not just hand out this cap to everyone
    // who requests it. There is currently no way to determine
    // if the client is a valid recipient
    errval_t err;
    struct capref src = {
        .cnode = cnode_task,
        .slot  = TASKCN_SLOT_IO
    };

    err = b->tx_vtbl.get_io_cap_response(b, NOP_CONT, src,
            SYS_ERR_OK);
    if (err_is_fail(err)) {
        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            err = b->register_send(b, get_default_waitset(),
                                   MKCONT((void (*)(void *))get_io_cap, b));
            if (err_is_fail(err)) {
                USER_PANIC_ERR(err, "register_send failed");
            }
        }

        USER_PANIC_ERR(err, "sending get_io_cap_response failed");
    }
}

static void ipi_alloc_notify_reply_cont(struct monitor_binding *b,
                                        uintptr_t state,
                                        struct capref notify_cap,
                                        errval_t reterr)
{
    errval_t err =
        b->tx_vtbl.ipi_alloc_notify_reply(b, NOP_CONT, state,
                                          notify_cap, reterr);

    if(err_is_fail(err)) {
        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            struct monitor_state *st = b->st;
            struct ipi_alloc_notify_reply_state *me =
                malloc(sizeof(struct ipi_alloc_notify_reply_state));
            assert(me != NULL);
            me->args.state = state;
            me->args.notify = notify_cap;
            me->args.err = reterr;
            me->elem.cont = ipi_alloc_notify_reply_handler;
            err = monitor_enqueue_send(b, &st->queue,
                                       get_default_waitset(), &me->elem.queue);
            if (err_is_fail(err)) {
                USER_PANIC_ERR(err, "monitor_enqueue_send failed");
            }
            return;
        }
        USER_PANIC_ERR(err, "sending reply");
    }
    assert(err_is_ok(err));
}

/**
 * \brief Non-blocking write to a terminal.
 *
 * \param client  Terminal client state.
 * \param data    Buffer holding characters to write.
 * \param length  The number of characters to write.
 * \param cont    Continuation invoked once the write completes.
 *
 * \return SYS_ERR_OK if successful.
 *         TERM_ERR_TX_BUSY if another message is buffered but not yet sent.
 *         TERM_ERR_IO if an I/O error occurred.
 */
errval_t term_client_write(struct term_client *client, const char *data,
                           size_t length, struct event_closure cont)
{
    errval_t err = SYS_ERR_OK;
    char *outdata = NULL;

    assert(client != NULL);
    assert(data != NULL);
    assert(length > 0);

    /* Make a copy of the data, since the output filters might modify them. */
    outdata = malloc(length);
    assert(outdata != NULL);
    memcpy(outdata, data, length);

    /* apply output filters */
    term_filter_apply(client->output_filters, &outdata, &length);

    /* try to send characters */
    err = client->out_binding->tx_vtbl.characters(client->out_binding, cont,
                                                  outdata, length);
    if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        err = err_push(err, TERM_ERR_TX_BUSY);
        goto out;
    } else if (err_is_fail(err)) {
        err = err_push(err, TERM_ERR_IO);
        goto out;
    }

out:
    /* free data */
    free(outdata);
    return err;
}

static void alloc_iref_reply_cont(struct monitor_binding *b,
                                    uintptr_t service_id,
                                    iref_t iref, errval_t reterr)
{
    errval_t err;

    err = b->tx_vtbl.alloc_iref_reply(b, NOP_CONT, service_id, iref, reterr);
    if (err_is_fail(err)) {
        if(err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            struct alloc_iref_reply_state *me =
                malloc(sizeof(struct alloc_iref_reply_state));
            assert(me != NULL);
            struct monitor_state *ist = b->st;
            assert(ist != NULL);
            me->args.service_id = service_id;
            me->args.iref = iref;
            me->args.err = reterr;
            me->b = b;
            me->elem.cont = alloc_iref_reply_handler;

            err = monitor_enqueue_send(b, &ist->queue,
                                       get_default_waitset(), &me->elem.queue);
            if (err_is_fail(err)) {
                USER_PANIC_ERR(err, "monitor_enqueue_send failed");
            }
            return;
        }

        USER_PANIC_ERR(err, "reply failed");
    }
}

static void
capsend_mc_send_cont(struct intermon_binding *b, struct intermon_msg_queue_elem *e)
{
    struct capsend_mc_msg_st *msg_st = (struct capsend_mc_msg_st*)e;
    struct capsend_mc_st *mc_st = msg_st->mc_st;
    errval_t err = SYS_ERR_OK;

    // if do_send is false, an error occured in the multicast setup, so do not
    // send anything
    if (mc_st->do_send) {
        err = mc_st->send_fn(b, &mc_st->caprep, mc_st);
    }

    if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        err = capsend_target(msg_st->dest, (struct msg_queue_elem*)msg_st);
    }

    if (err_is_fail(err)) {
        USER_PANIC_ERR(err, "sending dequeued capops message");
    }

    // decrement counter of number of queued messages
    if (!--mc_st->num_queued) {
        // if counter is zero, cleanup outgoing memory
        free(mc_st->msg_st_arr);
        mc_st->msg_st_arr = NULL;
        if (!mc_st->do_send || !mc_st->num_pending) {
            // if the send has been aborted, also cleanup cross-call state
            free(mc_st);
        }
    }
}

static void handle_notification(void *arg)
{
    struct lmp_endpoint *ep = arg;
    errval_t err;

    do { // consume messages
        struct lmp_recv_msg msg = LMP_RECV_MSG_INIT;
        err = lmp_endpoint_recv(ep, &msg.buf, NULL);

        if (err_is_ok(err)) {
            if(msg.buf.msglen == 1) {
                domainid_t domid = msg.words[0];

                // XXX: This is done by spawnd now
                if (domid != 0) {
                    debug_printf("Dispatcher with domain ID %"PRIuDOMAINID" exited\n",
                                 domid);
                }
            } else if(msg.buf.msglen == sizeof(struct RAM) / sizeof(uintptr_t) + 1) {
#ifndef __arm__
                //defined(__x86_64__) || defined(__i386__)
                union rammsg {
                    uintptr_t msgwords[LMP_MSG_LENGTH];
                    struct RAM ram;
                } *u;
                u = (union rammsg *)&msg.words;

                /* printf("%s.%d: RAM cap deleted, base = %" PRIxGENPADDR ", bits = %u\n", */
                /*        disp_name(), disp_get_core_id(), ram->base, ram->bits); */

                err = reclaim_memory(u->ram.base, u->ram.bits);
                if(err_is_fail(err)) {
                    DEBUG_ERR(err, "reclaim_memory");
                }
#else
                /* XXX: Disabling memory reclamation on ARM. I
                 * couldn't get the compiler to accept the above code
                 * due to strict aliasing restrictions. I do believe
                 * though that the above is according to the C99
                 * spec. Please help fix it, so that it can be
                 * enabled.
                 */
#endif
            } else {
                printf("%s: Unknown kernel notification of length %zu received\n",
                       disp_name(), msg.buf.msglen);
            }
        } else if (err_no(err) != LIB_ERR_NO_LMP_MSG) {
            DEBUG_ERR(err, "unexpected error from lmp_endpoint_recv");
        }
    } while(err_is_ok(err));

    // re-register
    struct event_closure cl = {
        .handler = handle_notification,
        .arg = arg,
    };
    err = lmp_endpoint_register(ep, get_default_waitset(), cl);
    assert(err_is_ok(err));
}

/**
 * \brief Wakeup a thread on a foreign dispatcher while disabled.
 *
 * \param core_id       Core ID to wakeup on
 * \param thread        Pointer to thread to wakeup
 * \param mydisp        Dispatcher this function is running on
 *
 * \return SYS_ERR_OK on success.
 */
static errval_t domain_wakeup_on_coreid_disabled(coreid_t core_id,
                                                 struct thread *thread,
                                                 dispatcher_handle_t mydisp)
{
    struct domain_state *ds = get_domain_state();

    // XXX: Ugly hack to allow waking up on a core id we don't have a
    // dispatcher handler for
    thread->coreid = core_id;

    // Catch this early
    assert_disabled(ds != NULL);
    if (ds->b[core_id] == NULL) {
        return LIB_ERR_NO_SPANNED_DISP;
    }

    thread_enqueue(thread, &ds->remote_wakeup_queue);

    // Signal the inter-disp waitset of this event
    struct event_closure closure = {
        .handler = handle_wakeup_on
    };
    errval_t err =
        waitset_chan_trigger_closure_disabled(&ds->interdisp_ws,
                                              &ds->remote_wakeup_event,
                                              closure,
                                              mydisp);
    assert_disabled(err_is_ok(err) ||
                    err_no(err) == LIB_ERR_CHAN_ALREADY_REGISTERED);

    return SYS_ERR_OK;
}

/**
 * \brief Continuation function for binding. This function
 *        send the bind request to the monitor.
 * \param pointer to the multihop_chan
 */
static void multihop_chan_bind_cont(void *st)
{

    errval_t err;
    struct multihop_chan *mc = st;
    struct monitor_binding *monitor_binding = mc->monitor_binding;

    // send bind request to the monitor
    // we do not get a lock on the monitor binding, as we did not expose it to the application
    MULTIHOP_DEBUG("sending bind request to monitor...\n");
    err = monitor_binding->tx_vtbl.multihop_bind_client_request(monitor_binding,
            NOP_CONT, mc->iref, mc->my_vci);

    if (err_is_ok(err)) {
        // request was successfully sent
    } else if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        // register to retry
        err = monitor_binding->register_send(monitor_binding,
                monitor_binding->waitset, MKCONT(multihop_chan_bind_cont, st));
        assert(err_is_ok(err));
    } else { // permanent failure sending message
        mc->bind_continuation.handler(mc->bind_continuation.st,
                err_push(err, LIB_ERR_BIND_MULTIHOP_REQ), NULL);
        //TODO destroy channel state?
    }
}

static void send_bind_reply(void *arg)
{
    struct bind_lmp_reply_state *st = arg;
    struct monitor_binding *b = st->b;
    errval_t err;

    err = st->b->tx_vtbl.bind_lmp_reply_monitor(st->b, NOP_CONT, st->args.err,
                                                st->args.mon_id, st->args.conn_id, 
                                                st->args.ep);
    if (err_is_ok(err)) {
        event_mutex_unlock(&b->mutex);
        free(st);
    } else if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        err = st->b->register_send(st->b, st->b->waitset,
                                   MKCONT(send_bind_reply,st));
        assert(err_is_ok(err)); // shouldn't fail, as we have the mutex
    } else {
        event_mutex_unlock(&b->mutex);
        USER_PANIC_ERR(err, "failed sending back reply to LMP bind request;"
                       " request dropped!");
        if (st->lc != NULL) {
            lmp_chan_destroy(st->lc);
            // FIXME: how do we tell the binding about this!?
        }
        free(st);
    }
}

void
update_owner__rx_handler(struct intermon_binding *b, intermon_caprep_t caprep, genvaddr_t st)
{
    errval_t err;
    struct intermon_state *inter_st = (struct intermon_state*)b->st;
    coreid_t from = inter_st->core_id;
    struct capref capref;
    struct capability cap;
    caprep_to_capability(&caprep, &cap);

    err = slot_alloc(&capref);
    if (err_is_fail(err)) {
        USER_PANIC_ERR(err, "failed to allocate slot for owner update");
    }

    err = monitor_copy_if_exists(&cap, capref);
    if (err_is_ok(err)) {
        err = monitor_set_cap_owner(cap_root, get_cap_addr(capref),
                                    get_cap_valid_bits(capref), from);
    }
    if (err_no(err) == SYS_ERR_CAP_NOT_FOUND) {
        err = SYS_ERR_OK;
    }

    if (err_is_fail(err)) {
        USER_PANIC_ERR(err, "failed to update cap ownership");
    }

    cap_destroy(capref);

    err = owner_updated(from, st);
    if (err_is_fail(err)) {
        USER_PANIC_ERR(err, "failed to send ownership update response");
    }
}

static errval_t mymm_free(struct capref ramcap, genpaddr_t base, uint8_t bits)
{
    errval_t ret;
    genpaddr_t mem_to_add;

    mem_to_add = (genpaddr_t)1 << bits;

    ret = mm_free(&mm_ram, ramcap, base, bits);
    if (err_is_fail(ret)) {
        if (err_no(ret) == MM_ERR_NOT_FOUND) {
            // memory wasn't there initially, add it
            ret = mm_add(&mm_ram, ramcap, bits, base);
            if (err_is_fail(ret)) {
                /* DEBUG_ERR(ret, "failed to add RAM to allocator"); */
                return ret;
            }
            mem_total += mem_to_add;
        } else {
            /* DEBUG_ERR(ret, "failed to free RAM in allocator"); */
            return ret;
        }
    }

    mem_avail += mem_to_add;

    return SYS_ERR_OK;
}

void
find_cap_result__rx_handler(struct intermon_binding *b, errval_t result, genvaddr_t st)
{
    // if we receive a positive result, immediately forward to caller
    lvaddr_t lst = (lvaddr_t)st;
    struct find_cap_broadcast_st *fc_bc_st = (struct find_cap_broadcast_st*)lst;
    if (err_is_ok(result)) {
        if (!fc_bc_st->found) {
            fc_bc_st->found = true;
            struct intermon_state *inter_st = (struct intermon_state*)b->st;
            coreid_t from = inter_st->core_id;
            fc_bc_st->result_handler(SYS_ERR_OK, from, fc_bc_st->st);
        }
    }
    else if (err_no(result) != SYS_ERR_CAP_NOT_FOUND) {
        DEBUG_ERR(result, "ignoring bad find_cap_result");
    }

    // check to see if broadcast is complete
    if (capsend_handle_mc_reply(&fc_bc_st->bc)) {
        if (!fc_bc_st->found) {
            // broadcast did not find a core, report notfound to caller
            fc_bc_st->result_handler(SYS_ERR_CAP_NOT_FOUND, 0, fc_bc_st->st);
        }
        free(fc_bc_st);
    }
}

static void send_myrpc_response(void *a)
{
    errval_t err;
    struct server_state *st = (struct server_state*)a;

    debug_printf("server: sending myresponse\n");

    struct event_closure txcont = MKCONT(send_myrpc_response_cb, st);
    err = xmplrpc_myrpc_response__tx(st->b, txcont, st->s);

    if (err_is_fail(err)) {
        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            debug_printf("server: re-sending myresponse\n");
            struct waitset *ws = get_default_waitset();
            txcont = MKCONT(send_myrpc_response, st);
            err = st->b->register_send(st->b, ws, txcont);
            if (err_is_fail(err)) {
                // note that only one continuation may be registered at a time
                DEBUG_ERR(err, "register_send on binding failed!");
                free_st(st);
            }
        } else {
            DEBUG_ERR(err, "error sending mycall message\n");
            free_st(st);
        }
    }
}

static void send_myrpc_call(void *a)
{
    errval_t err;

    debug_printf("client: sending mycall\n");

    struct xmplrpc_binding *b = (struct xmplrpc_binding *)a;

    struct event_closure txcont = MKCONT(send_myrpc_call_cb, b);

    err = xmplrpc_myrpc_call__tx(b, txcont, 42);

    if (err_is_fail(err)) {
        if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
            debug_printf("client: re-sending mycall\n");
            struct waitset *ws = get_default_waitset();
            txcont = MKCONT(send_myrpc_call, b);
            err = b->register_send(b, ws, txcont);
            if (err_is_fail(err)) {
                // note that only one continuation may be registered at a time
                DEBUG_ERR(err, "register_send on binding failed!");
            }
        } else {
            DEBUG_ERR(err, "error sending mycall message\n");
        }
    }
}

/**
 * \brief Send a multi-hop message that contains no payload.
 * 		  It is used to acknowledge received messages.
 *
 * \param mc pointer to the multi-hop channel
 */
static void multihop_send_dummy_message(struct multihop_chan *mc)
{
    assert(mc->connstate == MULTIHOP_CONNECTED);

#if MULTIHOP_FLOW_CONTROL

    MULTIHOP_DEBUG("sending dummy message, ack %d...\n", mc->unacked_received);

    errval_t err;
    struct monitor_binding *monitor_binding = mc->monitor_binding;

    // send message
    err = monitor_binding->tx_vtbl.multihop_message(monitor_binding, NOP_CONT,
            mc->vci, mc->direction, MULTIHOP_MESSAGE_FLAG_DUMMY,
            mc->unacked_received, (uint8_t *) mc, 1);

    if (err_is_ok(err)) {
        // we have just acknowledged all received messages
        mc->unacked_received = 0;
    } else if (err_no(err) != FLOUNDER_ERR_TX_BUSY) {
        USER_PANIC_ERR(err,
                "Could not send dummy message over multi-hop channel\n");

    }

#endif // MULTIHOP_FLOW_CONTROL
}

/**
 * \brief Send a capability over the multi-hop channel
 *
 * \param mc pointer to the multi-hop channel
 * \param _continuation callback to be executed after the message is sent
 * \param cap_state pointer to the cap state of the channel
 * \param cap the capability to send
 */
errval_t multihop_send_capability(struct multihop_chan *mc,
        struct event_closure _continuation,
        struct flounder_cap_state *cap_state, struct capref cap)
{

    errval_t err;
    assert(mc->connstate == MULTIHOP_CONNECTED);
    struct monitor_binding *mon_binding = mc->monitor_binding;

    // send the message
    err = mon_binding->tx_vtbl.multihop_cap_send(mon_binding, _continuation,
                                                 mc->vci, mc->direction,
                                                 SYS_ERR_OK, cap,
                                                 cap_state->tx_capnum);

    if (err_is_ok(err)) {
        // increase capability number
        cap_state->tx_capnum++;
        return err;
    } else if (err_no(err) == FLOUNDER_ERR_TX_BUSY) {
        return err;
    } else {
        return err_push(err, LIB_ERR_MONITOR_CAP_SEND);
    }
}