/* -*- mode:c; c-file-style:"k&r"; c-basic-offset: 4; tab-width:4; indent-tabs-mode:nil; mode:auto-fill; fill-column:78; -*- */
/* vim: set ts=4 sw=4 et tw=78 fo=cqt wm=0: */

/* Copyright (C) 2014 Stony Brook University
   This file is part of Graphene Library OS.

   Graphene Library OS is free software: you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public License
   as published by the Free Software Foundation, either version 3 of the
   License, or (at your option) any later version.

   Graphene Library OS is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.  */

/*
 * shim_ipc_pid.c
 *
 * This file contains functions and callbacks to handle IPC of SYSV namespace.
 */

#include <shim_internal.h>
#include <shim_thread.h>
#include <shim_ipc.h>
#include <shim_checkpoint.h>
#include <shim_sysv.h>

#include <pal.h>
#include <pal_error.h>

#include <errno.h>

#define SYSV_RANGE_SIZE     128
#define SYSV_LEASE_TIME     1000

#define KEY_HASH(k)       ((k)->key)
#define KEY_COMP(k1, k2)    \
        ((k1)->key != (k2)->key || (k1)->type != (k2)->type)
#define KEY_COPY(k1, k2)    \
        do { (k1)->key = (k2)->key; (k1)->type = (k2)->type; } while (0)

#define NS     sysv
#define NS_CAP SYSV
#define NS_KEY struct sysv_key

#include "shim_ipc_nsimpl.h"

int init_ns_sysv (void)
{
    init_namespace();
    return 0;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_delres_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_delres_callback, ipc);

int ipc_sysv_delres_send (struct shim_ipc_port * port, IDTYPE dest,
                          IDTYPE resid, enum sysv_type type)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    bool owned = false;

    if (!port) {
        if ((ret = connect_owner(resid, &port, &dest)) < 0)
            goto out;

        owned = true;
    }

    if (!owned) {
        struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                        IPC_SYSV_DELRES,
                        sizeof(struct shim_ipc_sysv_delres),
                        dest);
        struct shim_ipc_sysv_delres * msgin = (struct shim_ipc_sysv_delres *)
            &msg->msg;
        msgin->resid = resid;
        msgin->type  = type;

        debug("ipc send to %u: IPC_SYSV_DELRES(%u, %s)\n", dest, resid,
              SYSV_TYPE_STR(type));

        ret = send_ipc_message(msg, port);
        goto out;
    }

    struct shim_ipc_msg_obj * msg = create_ipc_msg_duplex_on_stack(
                                        IPC_SYSV_DELRES,
                                        sizeof(struct shim_ipc_sysv_delres),
                                        dest);
    struct shim_ipc_sysv_delres * msgin = (struct shim_ipc_sysv_delres *)
                                          &msg->msg.msg;
    msgin->resid = resid;
    msgin->type  = type;

    debug("ipc send to %u: IPC_SYSV_DELRES(%u, %s)\n", dest, resid,
          SYSV_TYPE_STR(type));

    ret = do_ipc_duplex(msg, port, NULL, NULL);
    put_ipc_port(port);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_delres_send);
    return ret;
}

int ipc_sysv_delres_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_delres * msgin  =
                       (struct shim_ipc_sysv_delres *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_DELRES(%u, %s)\n", msg->src,
          msgin->resid, SYSV_TYPE_STR(msgin->type));

    bool owned = false;
    ret = -ENOENT;
    switch(msgin->type) {
        case SYSV_MSGQ: {
            struct shim_msg_handle * msgq = get_msg_handle_by_id(msgin->resid);
            if (!msgq)
                goto out;
            owned = msgq->owned;
            ret = del_msg_handle(msgq);
            break;
        }
        case SYSV_SEM: {
            struct shim_sem_handle * sem = get_sem_handle_by_id(msgin->resid);
            if (!sem)
                goto out;
            owned = sem->owned;
            ret = del_sem_handle(sem);
            break;
        }
        default:
            ret = -ENOSYS;
            break;
    }

    if (!ret)
        ret = owned ? RESPONSE_CALLBACK : 0;
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_delres_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_movres_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_movres_callback, ipc);

int ipc_sysv_movres_send (struct sysv_client * client, IDTYPE owner,
                          const char * uri, LEASETYPE lease, IDTYPE resid,
                          enum sysv_type type)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    int len = strlen(uri);

    struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                                        IPC_SYSV_MOVRES,
                                        sizeof(struct shim_ipc_sysv_movres) +
                                        len, client->vmid);
    struct shim_ipc_sysv_movres * msgin = (struct shim_ipc_sysv_movres *)
                                          &msg->msg;
    msgin->resid = resid;
    msgin->type  = type;
    msgin->owner = owner;
    msgin->lease = lease;
    memcpy(msgin->uri, uri, len + 1);
    msg->seq = client->seq;

    debug("ipc send to %u: IPC_SYSV_MOVRES(%u, %s, %u, %s)\n", client->vmid,
          resid, SYSV_TYPE_STR(type), owner, uri);

    ret = send_ipc_message(msg, client->port);
    SAVE_PROFILE_INTERVAL(ipc_sysv_movres_send);
    return ret;
}

int ipc_sysv_movres_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_movres * msgin  =
                       (struct shim_ipc_sysv_movres *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_MOVRES(%u, %s, %u, %s)\n", msg->src,
          msgin->resid, SYSV_TYPE_STR(msgin->type), msgin->owner, msgin->uri);

    struct shim_ipc_msg_obj * obj = find_ipc_msg_duplex(port, msg->seq);
    if (!obj)
        goto out;

    switch(msgin->type) {
        case SYSV_MSGQ:
        case SYSV_SEM:
            obj->retval = -EAGAIN;
            break;
        default:
            ret = -ENOSYS;
            goto out;
    }

    add_sysv_subrange(msgin->resid, msgin->owner, msgin->uri, &msgin->lease);

    if (obj->thread)
        thread_wakeup(obj->thread);

out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_movres_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_msgsnd_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_msgsnd_callback, ipc);

int ipc_sysv_msgsnd_send (struct shim_ipc_port * port, IDTYPE dest,
                          IDTYPE msgid, long msgtype,
                          const void * buf, size_t size, unsigned long seq)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    bool owned = true;

    if (!dest) {
        if ((ret = connect_owner(msgid, &port, &dest)) < 0)
            goto out;

        owned = false;
    }

    struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                                    IPC_SYSV_MSGSND,
                                    sizeof(struct shim_ipc_sysv_msgsnd) +
                                    size, dest);
    struct shim_ipc_sysv_msgsnd * msgin =
                               (struct shim_ipc_sysv_msgsnd *) &msg->msg;
    msgin->msgid = msgid;
    msgin->msgtype = msgtype;
    memcpy(msgin->msg, buf, size);
    msg->seq = seq;

    debug("ipc send to %u: IPC_SYSV_MSGSND(%u, %ld)\n", dest,
          msgid, msgtype);

    ret = send_ipc_message(msg, port);

    if (!owned)
        put_ipc_port(port);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgsnd_send);
    return ret;
}

int ipc_sysv_msgsnd_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_msgsnd * msgin =
                        (struct shim_ipc_sysv_msgsnd *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_MSGSND(%u, %ld)\n", msg->src,
          msgin->msgid, msgin->msgtype);

    size_t size = msg->size - sizeof(*msg) - sizeof(*msgin);

    if (msg->seq) {
        struct shim_ipc_msg_obj * obj = find_ipc_msg_duplex(port, msg->seq);
        void * priv = obj ? obj->private : NULL;

        if (priv) {
            struct shim_ipc_sysv_msgrcv * rcv =
                        (struct shim_ipc_sysv_msgrcv *) obj->msg.msg;

            if (size > rcv->size)
                size = rcv->size;

            memcpy(priv, msgin->msg, size);
            obj->retval = size;
            if (obj->thread)
                thread_wakeup(obj->thread);
            goto out;
        }
    }

    struct shim_msg_handle * msgq = get_msg_handle_by_id(msgin->msgid);
    if (!msgq) {
        ret = -ENOENT;
        goto out;
    }

    if (msg->seq) {
        ret = add_sysv_msg(msgq, msgin->msgtype, size, msgin->msg, NULL);
    } else {
        struct sysv_client src;
        src.port = port;
        src.vmid = msg->src;
        src.seq  = msg->seq;
        ret = add_sysv_msg(msgq, msgin->msgtype, size, msgin->msg, &src);
    }

out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgsnd_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_msgrcv_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_msgrcv_callback, ipc);

int ipc_sysv_msgrcv_send (IDTYPE msgid, long msgtype, int flags, void * buf,
                          size_t size)
{
    BEGIN_PROFILE_INTERVAL();
    IDTYPE owner;
    struct shim_ipc_port * port = NULL;
    int ret = 0;

    if ((ret = connect_owner(msgid, &port, &owner)) < 0)
        goto out;

    if (owner == cur_process.vmid) {
        ret = -EAGAIN;
        goto out;
    }

    assert(port);

    struct shim_ipc_msg_obj * msg = create_ipc_msg_duplex_on_stack(
                                        IPC_SYSV_MSGRCV,
                                        sizeof(struct shim_ipc_sysv_msgrcv),
                                        true);
    struct shim_ipc_sysv_msgrcv * msgin =
                (struct shim_ipc_sysv_msgrcv *) &msg->msg.msg;
    msgin->msgid = msgid;
    msgin->msgtype = msgtype;
    msgin->size = size;
    msgin->flags = flags;

    debug("ipc send to %u: IPC_SYSV_MSGRCV(%u, %ld)\n", owner,
          msgid, msgtype);

    ret = do_ipc_duplex(msg, port, NULL, buf);
    put_ipc_port(port);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgrcv_send);
    return ret;
}

int ipc_sysv_msgrcv_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_msgrcv * msgin =
                (struct shim_ipc_sysv_msgrcv *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_MSGRCV(%u, %ld)\n", msg->src,
          msgin->msgid, msgin->msgtype);

    struct shim_msg_handle * msgq = get_msg_handle_by_id(msgin->msgid);

    if (!msgq) {
        ret = -ENOENT;
        goto out;
    }

    void * buf = __alloca(msgin->size);
    struct sysv_client src;
    src.port = port;
    src.vmid = msg->src;
    src.seq  = msg->seq;

    ret = get_sysv_msg(msgq, msgin->msgtype, msgin->size, buf, msgin->flags,
                       &src);

    if (ret > 0) {
        size_t size = ret;
        ret = ipc_sysv_msgsnd_send(port, msg->src, msgin->msgid, msgin->msgtype,
                                   buf, size, msg->seq);
    }

    put_msg_handle(msgq);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgrcv_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_msgmov_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_msgmov_callback, ipc);

int ipc_sysv_msgmov_send (struct shim_ipc_port * port, IDTYPE dest,
                          IDTYPE msgid, LEASETYPE lease,
                          struct sysv_score * scores, int nscores)
{
    BEGIN_PROFILE_INTERVAL();
    struct shim_ipc_msg * msg =
            create_ipc_msg_on_stack(IPC_SYSV_MSGMOV,
                                    sizeof(struct shim_ipc_sysv_msgmov) +
                                    sizeof(struct sysv_score) * nscores,
                                    dest);
    struct shim_ipc_sysv_msgmov * msgin =
                            (struct shim_ipc_sysv_msgmov *) &msg->msg;

    msgin->msgid   = msgid;
    msgin->lease   = lease;
    msgin->nscores = nscores;
    if (nscores)
        memcpy(msgin->scores, scores, sizeof(struct sysv_score) * nscores);

    debug("ipc send to %u: IPC_SYSV_MSGMOV(%ld)\n", dest, msgid);
    int ret = send_ipc_message(msg, port);
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgmov_send);
    return ret;
}

int ipc_sysv_msgmov_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_msgmov * msgin =
                (struct shim_ipc_sysv_msgmov *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_MSGMOV(%ld)\n", msg->src,
          msgin->msgid);

    struct shim_msg_handle * msgq = get_msg_handle_by_id(msgin->msgid);
    if (!msgq) {
        ret = -ENOENT;
        goto out;
    }

    struct shim_handle * hdl = container_of(msgq, struct shim_handle,
                                            info.msg);

    lock(hdl->lock);
    int nscores = (msgin->nscores > MAX_SYSV_CLIENTS) ?
                  MAX_SYSV_CLIENTS : msgin->nscores;
    if (nscores)
        memcpy(msgq->scores, msgin->scores, nscores);
    if (nscores < MAX_SYSV_CLIENTS)
        memset(msgq->scores + nscores, 0,
               sizeof(struct sysv_score) * (MAX_SYSV_CLIENTS - nscores));
    unlock(hdl->lock);

    ret = recover_msg_ownership(msgq);

    struct shim_ipc_info * info;
    if (!create_ipc_location(&info)) {
        add_sysv_subrange(msgin->msgid, info->vmid, qstrgetstr(&info->uri),
                          &msgin->lease);
        put_ipc_info(info);
    }

    put_msg_handle(msgq);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_msgmov_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_semop_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semop_callback, ipc);

int ipc_sysv_semop_send (IDTYPE semid, struct sembuf * sops, int nsops,
                         unsigned long timeout, unsigned long * seq)
{
    BEGIN_PROFILE_INTERVAL();
    IDTYPE owner;
    struct shim_ipc_port * port = NULL;
    int ret = 0;
    bool waitforreply = false;

    for (int i = 0 ; i < nsops ; i++)
        if (sops[i].sem_op <= 0) {
            waitforreply = true;
            break;
        }

    if ((ret = connect_owner(semid, &port, &owner)) < 0)
        goto out;

    if (owner == cur_process.vmid) {
        ret = -EAGAIN;
        goto out;
    }

    assert(port);

    if (!waitforreply) {
        struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                                        IPC_SYSV_SEMOP,
                                        sizeof(struct shim_ipc_sysv_semop) +
                                        sizeof(struct sembuf) * nsops,
                                        owner);
        struct shim_ipc_sysv_semop * msgin =
                (struct shim_ipc_sysv_semop *) &msg->msg;

        msgin->semid   = semid;
        msgin->timeout = timeout;
        msgin->nsops   = nsops;
        memcpy(msgin->sops, sops, sizeof(struct sembuf) * nsops);
        msg->seq = *seq;


        debug("ipc send to %u: IPC_SYSV_SEMOP(%u, %ld, %u)\n", owner, semid,
              timeout, nsops);

        ret = send_ipc_message(msg, port);
        put_ipc_port(port);
        goto out;
    }

    struct shim_ipc_msg_obj * msg = create_ipc_msg_duplex_on_stack(
                                        IPC_SYSV_SEMOP,
                                        sizeof(struct shim_ipc_sysv_semop) +
                                        sizeof(struct sembuf) * nsops,
                                        owner);
    struct shim_ipc_sysv_semop * msgin =
            (struct shim_ipc_sysv_semop *) &msg->msg.msg;
    msgin->semid   = semid;
    msgin->timeout = timeout;
    msgin->nsops   = nsops;
    memcpy(msgin->sops, sops, sizeof(struct sembuf) * nsops);
    msg->msg.seq   = *seq;

    debug("ipc send to %u: IPC_SYSV_SEMOP(%u, %ld, %u)\n", owner, semid,
          timeout, nsops);

    ret = do_ipc_duplex(msg, port, seq, NULL);
    put_ipc_port(port);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semop_send);
    return ret;
}

int ipc_sysv_semop_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_semop * msgin =
                (struct shim_ipc_sysv_semop *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMOP(%u, %ld, %u)\n", msg->src,
          msgin->semid, msgin->timeout, msgin->nsops);

    struct shim_sem_handle * sem = get_sem_handle_by_id(msgin->semid);
    if (!sem) {
        ret = -ENOENT;
        goto out;
    }

    struct sysv_client client;
    client.port = port;
    client.vmid = msg->src;
    client.seq  = msg->seq;
    ret = submit_sysv_sem(sem, msgin->sops, msgin->nsops, msgin->timeout,
                          &client);
    put_sem_handle(sem);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semop_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_semctl_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semctl_callback, ipc);

int ipc_sysv_semctl_send (IDTYPE semid, int semnum, int cmd, void * vals,
                          int valsize)
{
    BEGIN_PROFILE_INTERVAL();
    IDTYPE owner;
    struct shim_ipc_port * port = NULL;
    int ret = 0;

    if ((ret = connect_owner(semid, &port, &owner)) < 0)
        goto out;

    int ctlvalsize = (cmd == SETALL || cmd == SETVAL) ? valsize : 0;

    struct shim_ipc_msg_obj * msg = create_ipc_msg_duplex_on_stack(
                                        IPC_SYSV_SEMCTL,
                                        sizeof(struct shim_ipc_sysv_semctl) +
                                        ctlvalsize,
                                        owner);
    struct shim_ipc_sysv_semctl * msgin =
                (struct shim_ipc_sysv_semctl *) &msg->msg.msg;

    msgin->semid   = semid;
    msgin->semnum  = semnum;
    msgin->cmd     = cmd;
    msgin->valsize = ctlvalsize;
    if (ctlvalsize)
        memcpy(msgin->vals, vals, ctlvalsize);

    debug("ipc send to %u: IPC_SYSV_SEMCTL(%u, %d, %d)\n", owner, semid,
          semnum, cmd);

    ret = do_ipc_duplex(msg, port, NULL, vals);
    put_ipc_port(port);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semctl_send);
    return ret;
}

int ipc_sysv_semctl_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_semctl * msgin =
                (struct shim_ipc_sysv_semctl *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMCTL(%u, %d, %d)\n", msg->src,
          msgin->semid, msgin->semnum, msgin->cmd);

    struct shim_sem_handle * sem = get_sem_handle_by_id(msgin->semid);
    if (!sem) {
        ret = -ENOENT;
        goto out;
    }

    void * vals = NULL;
    int valsize;
    switch(msgin->cmd) {
        case GETALL: {
            unsigned short * allsems = __alloca(sizeof(unsigned short) *
                                                sem->nsems);
            for (int i = 0 ; i < sem->nsems ; i++)
                allsems[i] = sem->sems[i].val;

            vals = allsems;
            valsize = sizeof(unsigned short) * sem->nsems;
            goto semret;
        }

        case GETNCNT:
            vals = &sem->sems[msgin->semnum].ncnt;
            valsize = sizeof(unsigned short);
            goto semret;

        case GETPID:
            vals = &sem->sems[msgin->semnum].pid;
            valsize = sizeof(IDTYPE);
            goto semret;

        case GETVAL:
            vals = &sem->sems[msgin->semnum].val;
            valsize = sizeof(unsigned short);
            goto semret;

        case GETZCNT:
            vals = &sem->sems[msgin->semnum].zcnt;
            valsize = sizeof(unsigned short);
            break;

        case SETALL: {
            if (msgin->valsize != sizeof(unsigned short) * sem->nsems) {
                ret = -EINVAL;
                break;
            }

            unsigned short * vals = (void *) msgin->vals;

            for (int i = 0 ; i < sem->nsems ; i++)
                sem->sems[i].val = vals[i];

            ret = RESPONSE_CALLBACK;
            break;
        }

        case SETVAL: {
            ret = -EINVAL;
            if (msgin->valsize != sizeof(unsigned short))
                break;
            if (msgin->semnum >= sem->nsems)
                break;

            sem->sems[msgin->semnum].val = *((int *) msgin->vals);
            ret = RESPONSE_CALLBACK;
            break;
        }

        default:
            ret = -ENOSYS;
            break;
    }

    put_sem_handle(sem);
    goto out;
semret:
    ret = ipc_sysv_semret_send(port, msg->src, vals, valsize,
                               msg->seq);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semctl_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_semret_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semret_callback, ipc);

int ipc_sysv_semret_send (struct shim_ipc_port * port, IDTYPE dest, void * vals,
                          int valsize, unsigned long seq)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                                        IPC_SYSV_SEMRET,
                                        sizeof(struct shim_ipc_sysv_semret) +
                                        valsize,
                                        dest);
    struct shim_ipc_sysv_semret * msgin =
                (struct shim_ipc_sysv_semret *) &msg->msg;
    msgin->valsize = valsize;
    memcpy(msgin->vals, vals, valsize);
    msg->seq = seq;

    debug("ipc send to %u: IPC_SYSV_SEMRET\n", dest);

    ret = send_ipc_message(msg, port);
    SAVE_PROFILE_INTERVAL(ipc_sysv_semret_send);
    return ret;
}

int ipc_sysv_semret_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    struct shim_ipc_sysv_semret * semret =
                                (struct shim_ipc_sysv_semret *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMRET\n", msg->src);

    struct shim_ipc_msg_obj * obj = find_ipc_msg_duplex(port, msg->seq);
    if (obj) {
        struct shim_ipc_sysv_semctl * semctl =
                                (struct shim_ipc_sysv_semctl *) &obj->msg.msg;

        void * vals = obj->private;

        if (vals) {
            switch(semctl->cmd) {
                case GETALL:
                case GETNCNT:
                case GETPID:
                case GETVAL:
                case GETZCNT: {
                    int retvalsize = semret->valsize;
                    if (retvalsize > semctl->valsize)
                        retvalsize = semctl->valsize;
                    memcpy(vals, semret->vals, retvalsize);
                    break;
                }
            }
        }

        if (obj->thread)
            thread_wakeup(obj->thread);
    }

    SAVE_PROFILE_INTERVAL(ipc_sysv_semret_callback);
    return 0;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_semmov_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semmov_callback, ipc);

int ipc_sysv_semmov_send (struct shim_ipc_port * port, IDTYPE dest,
                          IDTYPE semid, LEASETYPE lease,
                          struct sem_backup * sems, int nsems,
                          struct sem_client_backup * srcs, int nsrcs,
                          struct sysv_score * scores, int nscores)
{
    BEGIN_PROFILE_INTERVAL();
    struct shim_ipc_msg * msg =
            create_ipc_msg_on_stack(IPC_SYSV_SEMMOV,
                                    sizeof(struct shim_ipc_sysv_semmov) +
                                    sizeof(struct sem_backup) * nsems +
                                    sizeof(struct sem_client_backup) * nsrcs +
                                    sizeof(struct sysv_score) * nscores,
                                    dest);
    struct shim_ipc_sysv_semmov * msgin =
                            (struct shim_ipc_sysv_semmov *) &msg->msg;
    msgin->semid   = semid;
    msgin->lease   = lease;
    msgin->nsems   = nsems;
    msgin->nsrcs   = nsrcs;
    msgin->nscores = nscores;

    memcpy(&msgin->sems, sems, sizeof(struct sem_backup) * nsems);
    memcpy((void *) msgin->sems + sizeof(struct sem_backup) * nsems,
           srcs, sizeof(struct sem_client_backup) * nsrcs);
    memcpy((void *) msgin->sems + sizeof(struct sem_backup) * nsems +
           sizeof(struct sem_client_backup) * nsrcs,
           scores, sizeof(struct sysv_score) * nscores);

    debug("ipc send to %u: IPC_SYSV_SEMMOV(%ld)\n", semid);

    int ret = send_ipc_message(msg, port);
    SAVE_PROFILE_INTERVAL(ipc_sysv_semmov_send);
    return ret;
}

int ipc_sysv_semmov_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_semmov * msgin =
                (struct shim_ipc_sysv_semmov *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMMOV(%ld)\n", msg->src,
          msgin->semid);

    struct sem_backup * sems = msgin->sems;
    struct sem_client_backup * clients =
            (struct sem_client_backup *) (sems + msgin->nsems);
    struct sysv_score * scores =
            (struct sysv_score *) (clients + msgin->nsrcs);

    struct shim_sem_handle * sem = get_sem_handle_by_id(msgin->semid);
    if (!sem) {
        ret = -ENOENT;
        goto out;
    }

    struct shim_handle * hdl = container_of(sem, struct shim_handle,
                                            info.sem);

    lock(hdl->lock);
    int nscores = (msgin->nscores > MAX_SYSV_CLIENTS) ?
                  MAX_SYSV_CLIENTS : msgin->nscores;
    if (nscores)
        memcpy(sem->scores, scores, nscores);
    if (nscores < MAX_SYSV_CLIENTS)
        memset(sem->scores + nscores, 0,
               sizeof(struct sysv_score) * (MAX_SYSV_CLIENTS - nscores));
    unlock(hdl->lock);

    ret = recover_sem_ownership(sem, sems, msgin->nsems, clients,
                                msgin->nsrcs);

    struct shim_ipc_info * info;
    if (!create_ipc_location(&info)) {
        add_sysv_subrange(msgin->semid, info->vmid, qstrgetstr(&info->uri),
                          &msgin->lease);
        put_ipc_info(info);
    }

    put_sem_handle(sem);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semmov_callback);
    return ret;
}

#ifdef USE_SHARED_SEMAPHORE

DEFINE_PROFILE_INTERVAL(ipc_sysv_semquery_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semquery_callback, ipc);

int ipc_sysv_semquery_send (IDTYPE semid, int * nsems,
                            PAL_NUM ** host_sem_ids)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    IDTYPE dest;
    struct shim_ipc_port * port = NULL;

    if ((ret = connect_owner(semid, &port, &dest)) < 0)
        goto out;

    if (dest == cur_process.vmid) {
        ret = -EAGAIN;
        goto out;
    }

    assert(port);
    struct shim_ipc_msg_obj * msg = create_ipc_msg_duplex_on_stack(
                                        IPC_SYSV_SEMQUERY,
                                        sizeof(struct shim_ipc_sysv_semquery),
                                        dest);
    struct shim_ipc_sysv_semquery * msgin =
                (struct shim_ipc_sysv_semquery *) &msg->msg.msg;
    msgin->semid = semid;

    debug("ipc send to %u: IPC_SYSV_SEMQUERY(%u)\n", dest, semid);

    ret = do_ipc_duplex(msg, port, NULL, host_sem_ids);
    put_ipc_port(port);
    if (ret >= 0) {
        *nsems = ret;
        ret = 0;
    }
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semquery_send);
    return ret;
}

int ipc_sysv_semquery_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_semquery * msgin =
                (struct shim_ipc_sysv_semquery *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMQUERY(%u)\n", msg->src,
          msgin->semid);

    struct shim_sem_handle * sem = get_sem_handle_by_id(msgin->semid);
    if (!sem) {
        ret = -ENOENT;
        goto out;
    }

    ret = send_sem_host_ids(sem, port, msg->src, msg->seq);
    put_sem_handle(sem);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semreply_callback);
    return ret;
}

DEFINE_PROFILE_INTERVAL(ipc_sysv_semreply_send, ipc);
DEFINE_PROFILE_INTERVAL(ipc_sysv_semreply_callback, ipc);

int ipc_sysv_semreply_send (struct shim_ipc_port * port, IDTYPE dest,
                            IDTYPE semid, int nsems, PAL_NUM * host_sem_ids,
                            unsigned long seq)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_msg * msg = create_ipc_msg_on_stack(
                                        IPC_SYSV_SEMREPLY,
                                        sizeof(struct shim_ipc_sysv_semreply)
                                        + sizeof(PAL_NUM) * nsems,
                                        dest);
    struct shim_ipc_sysv_semreply * msgin =
                    (struct shim_ipc_sysv_semreply *) &msg->msg;
    msgin->semid = semid;
    msgin->nsems = nsems;
    if (nsems)
        memcpy(msgin->host_sem_ids, host_sem_ids, sizeof(PAL_NUM) * nsems);
    msg->seq = seq;

    debug("ipc send to %u: IPC_SYSV_SEMREPLY(%u, %d)\n", dest, semid, nsems);

    ret = send_ipc_message(msg, port);
    SAVE_PROFILE_INTERVAL(ipc_sysv_semreply_send);
    return ret;
}

int ipc_sysv_semreply_callback (IPC_CALLBACK_ARGS)
{
    BEGIN_PROFILE_INTERVAL();
    int ret = 0;
    struct shim_ipc_sysv_semreply * msgin =
                    (struct shim_ipc_sysv_semreply *) &msg->msg;

    debug("ipc callback from %u: IPC_SYSV_SEMREPLY(%u, %d)\n", msg->src,
          msgin->semid, msgin->nsems);

    struct shim_ipc_msg_obj * obj = find_ipc_msg_duplex(port, msg->seq);
    if (!obj)
        goto out;

    PAL_NUM ** semids = obj->private;
    if (semids)
        *semids = remalloc(msgin->host_sem_ids, sizeof(PAL_NUM) * msgin->nsems);
    obj->retval = msgin->nsems;

    if (obj->thread)
        thread_wakeup(obj->thread);
out:
    SAVE_PROFILE_INTERVAL(ipc_sysv_semreply_callback);
    return ret;
}

#endif /* USE_SHARED_SEMAPHORE */

int __balance_sysv_score (struct sysv_balance_policy * policy,
                          struct shim_handle * hdl,
                          struct sysv_score * scores, int nscores,
                          struct sysv_client * src, long score)
{
    struct sysv_score * s       = scores;
    struct sysv_score * last    = scores + nscores;

    for ( ; s < last && !s->vmid ; s++);

    struct sysv_score * free    = s > scores ? scores : NULL;
    struct sysv_score * highest = s < last ? s : NULL;
    struct sysv_score * lowest  = highest;
    struct sysv_score * owner   = NULL;
    struct sysv_score * chosen  = NULL;

    for ( ; s < last ; s++) {
        if (!s->vmid) {
            if (!free)
                free = s;
            continue;
        }

        if (s->score >= highest->score)
            highest = s;
        if (s->score < lowest->score)
            lowest = s;

        if (src) {
            if (s->vmid == cur_process.vmid)
                owner = s;
            if (s->vmid == src->vmid) {
                chosen = s;
                continue;
            }
        } else {
            if (s->vmid == cur_process.vmid) {
                owner = chosen = s;
                continue;
            }
        }

        s->score = (s->score >= policy->score_decay) ?
                    s->score - policy->score_decay : 0;
        debug("balance: %u => %d\n", s->vmid, s->score);
    }

    if (!chosen) {
        chosen = free ? : lowest;
        chosen->vmid  = src ? src->vmid : cur_process.vmid;
        chosen->score = 0;
    }

    chosen->score += score;
    if (chosen->score > policy->score_max)
        chosen->score = policy->score_max;

    debug("balance: %u => %d\n", chosen->vmid, chosen->score);

    if (!src || chosen != highest ||
        chosen->score < (owner ? owner->score : 0) + policy->balance_threshold)
        return 0;

    return policy->migrate(hdl, src);
}