Archipelago » blktap-utils

/*

 * Copyright (c) 2010, Citrix Systems, Inc.

 * All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or without

 * modification, are permitted provided that the following conditions are met:

 *     * Redistributions of source code must retain the above copyright

 *       notice, this list of conditions and the following disclaimer.

 *     * Redistributions in binary form must reproduce the above copyright

 *       notice, this list of conditions and the following disclaimer in the

 *       documentation and/or other materials provided with the distribution.

 *     * Neither the name of XenSource Inc. nor the names of its contributors

 *       may be used to endorse or promote products derived from this software

 *       without specific prior written permission.

 *

 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER

 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,

 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR

 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF

 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING

 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include <stdlib.h>

#include <stdio.h>

#include <string.h>

#include <unistd.h>

#include <errno.h>

#include <sys/socket.h>

#include <sys/un.h>

#include "tapdisk.h"

#include "tapdisk-driver.h"

#include "tapdisk-server.h"

#include "tapdisk-interface.h"

#include "block-valve.h"

typedef struct td_valve td_valve_t;

typedef struct td_valve_request td_valve_request_t;

struct td_valve_request {

        td_request_t            treq;

        int                     secs;

        struct list_head        entry;

        td_valve_t             *valve;

};

struct td_valve_stats {

        unsigned long long      stor;

        unsigned long long      forw;

};

struct td_valve {

        char                   *brname;

        unsigned long           flags;

        int                     sock;

        event_id_t              sock_id;

        event_id_t              sched_id;

        event_id_t              retry_id;

        unsigned int            cred;

        unsigned int            need;

        unsigned int            done;

        struct list_head        stor;

        struct list_head        forw;

        td_valve_request_t      reqv[MAX_REQUESTS];

        td_valve_request_t     *free[MAX_REQUESTS];

        int                     n_free;

        struct td_valve_stats   stats;

};

#define td_valve_for_each_stored_request(_req, _next, _valve)                \

        list_for_each_entry_safe(_req, _next, &(_valve)->stor, entry)

#define td_valve_for_each_forwarded_request(_req, _next, _valve)        \

        list_for_each_entry_safe(_req, _next, &(_valve)->forw, entry)

#define TD_VALVE_CONNECT_INTERVAL 2 /* s */

#define TD_VALVE_RDLIMIT  (1<<0)

#define TD_VALVE_WRLIMIT  (1<<1)

#define TD_VALVE_KILLED   (1<<31)

static void valve_schedule_retry(td_valve_t *);

static void valve_conn_receive(td_valve_t *);

static void valve_conn_request(td_valve_t *, unsigned long);

static void valve_forward_stored_requests(td_valve_t *);

static void valve_kill(td_valve_t *);

#define DBG(_f, _a...)    if (1) { tlog_syslog(TLOG_DBG, _f, ##_a); }

#define INFO(_f, _a...)   tlog_syslog(TLOG_INFO, "valve: " _f, ##_a)

#define WARN(_f, _a...)   tlog_syslog(TLOG_WARN, "WARNING: "_f " in %s:%d", \

                                      ##_a, __func__, __LINE__)

#define ERR(_f, _a...)    tlog_syslog(TLOG_WARN, "ERROR: " _f " in %s:%d", \

                                      ##_a, __func__, __LINE__)

#define VERR(_err, _f, _a...) tlog_syslog(TLOG_WARN,                         \

                                          "ERROR: err=%d (%s), " _f ".", \

                                          _err, strerror(-(_err)), ##_a)

#undef  PERROR

#define PERROR(_f, _a...) VERR(-errno, _f, ##_a)

#define BUG() do {                                                \

                ERR("Aborting");                                \

                td_panic();                                        \

        } while (0)

#define BUG_ON(_cond)                                                \

        if (unlikely(_cond)) {                                        \

                ERR("(%s) = %ld", #_cond, (long)(_cond));        \

                BUG();                                                \

        }

#define WARN_ON(_cond) ({                                        \

        int __cond = _cond;                                        \

        if (unlikely(__cond))                                        \

                WARN("(%s) = %ld", #_cond, (long)(_cond));        \

        __cond;                                                \

})

#define ARRAY_SIZE(_a)   (sizeof(_a)/sizeof((_a)[0]))

#define TREQ_SIZE(_treq) ((unsigned int)(_treq.secs) << 9)

static td_valve_request_t *

valve_alloc_request(td_valve_t *valve)

{

        td_valve_request_t *req = NULL;

        if (valve->n_free)

                req = valve->free[--valve->n_free];

        return req;

}

static void

valve_free_request(td_valve_t *valve, td_valve_request_t *req)

{

        BUG_ON(valve->n_free >= ARRAY_SIZE(valve->free));

        list_del_init(&req->entry);

        valve->free[valve->n_free++] = req;

}

static void

__valve_sock_event(event_id_t id, char mode, void *private)

{

        td_valve_t *valve = private;

        valve_conn_receive(valve);

        valve_forward_stored_requests(valve);

}

static void

valve_set_done_pending(td_valve_t *valve)

{

        WARN_ON(valve->done == 0);

        tapdisk_server_mask_event(valve->sched_id, 0);

}

static void

valve_clear_done_pending(td_valve_t *valve)

{

        WARN_ON(valve->done != 0);

        tapdisk_server_mask_event(valve->sched_id, 1);

}

static void

__valve_sched_event(event_id_t id, char mode, void *private)

{

        td_valve_t *valve = private;

        if (likely(valve->done > 0))

                /* flush valve->done */

                valve_conn_request(valve, 0);

}

static void

valve_sock_close(td_valve_t *valve)

{

        if (valve->sock >= 0) {

                close(valve->sock);

                valve->sock = -1;

        }

        if (valve->sock_id >= 0) {

                tapdisk_server_unregister_event(valve->sock_id);

                valve->sock_id = -1;

        }

        if (valve->sched_id >= 0) {

                tapdisk_server_unregister_event(valve->sched_id);

                valve->sched_id = -1;

        }

}

static int

valve_sock_open(td_valve_t *valve)

{

        struct sockaddr_un addr = { .sun_family = AF_UNIX };

        int s, id, err;

        s = socket(AF_UNIX, SOCK_STREAM, 0);

        if (s < 0) {

                PERROR("socket");

                err = -errno;

                goto fail;

        }

        valve->sock = s;

        if (valve->brname[0] == '/')

                strncpy(addr.sun_path, valve->brname,

                        sizeof(addr.sun_path));

        else

                snprintf(addr.sun_path, sizeof(addr.sun_path),

                         "%s/%s", TD_VALVE_SOCKDIR, valve->brname);

        err = connect(valve->sock, &addr, sizeof(addr));

        if (err) {

                err = -errno;

                goto fail;

        }

        id = tapdisk_server_register_event(SCHEDULER_POLL_READ_FD,

                                           valve->sock, 0,

                                           __valve_sock_event,

                                           valve);

        if (id < 0) {

                err = id;

                goto fail;

        }

        valve->sock_id = id;

        id = tapdisk_server_register_event(SCHEDULER_POLL_TIMEOUT,

                                           -1, 0,

                                           __valve_sched_event,

                                           valve);

        if (id < 0) {

                err = id;

                goto fail;

        }

        valve->sched_id = id;

        INFO("Connected to %s", addr.sun_path);

        valve->cred = 0;

        valve->need = 0;

        valve->done = 0;

        valve_clear_done_pending(valve);

        return 0;

fail:

        valve_sock_close(valve);

        return err;

}

static int

valve_sock_send(td_valve_t *valve, const void *msg, size_t size)

{

        ssize_t n;

        n = send(valve->sock, msg, size, MSG_DONTWAIT);

        if (n < 0)

                return -errno;

        if (n != size)

                return -EPROTO;

        return 0;

}

static int

valve_sock_recv(td_valve_t *valve, void *msg, size_t size)

{

        ssize_t n;

        n = recv(valve->sock, msg, size, MSG_DONTWAIT);

        if (n < 0)

                return -errno;

        return n;

}

static void

__valve_retry_timeout(event_id_t id, char mode, void *private)

{

        td_valve_t *valve = private;

        int err;

        err = valve_sock_open(valve);

        if (!err)

                tapdisk_server_unregister_event(valve->retry_id);

}

static void

valve_schedule_retry(td_valve_t *valve)

{

        int id;

        BUG_ON(valve->sock_id >= 0);

        id = tapdisk_server_register_event(SCHEDULER_POLL_TIMEOUT,

                                           -1, TD_VALVE_CONNECT_INTERVAL,

                                           __valve_retry_timeout,

                                           valve);

        BUG_ON(id < 0);

        valve->retry_id = id;

}

static void

valve_conn_open(td_valve_t *valve)

{

        int err;

        BUG_ON(valve->flags & TD_VALVE_KILLED);

        err = valve_sock_open(valve);

        if (err) {

                WARN("%s: %s", valve->brname, strerror(-err));

                valve_schedule_retry(valve);

        }

}

static void

valve_conn_close(td_valve_t *valve, int reset)

{

        td_valve_request_t *req, *next;

        valve_sock_close(valve);

        if (reset)

                td_valve_for_each_stored_request(req, next, valve) {

                        td_forward_request(req->treq);

                        valve->stats.forw++;

                        valve_free_request(valve, req);

                }

        WARN_ON(!list_empty(&valve->stor));

}

static void

valve_conn_reset(td_valve_t *valve)

{

        valve_conn_close(valve, 1);

        valve_conn_open(valve);

}

void

valve_conn_receive(td_valve_t *valve)

{

        unsigned long buf[32], cred = 0;

        ssize_t n;

        int i, err;

        n = valve_sock_recv(valve, buf, sizeof(buf));

        if (!n) {

                err = -ECONNRESET;

                goto reset;

        }

        if (n < 0) {

                err = n;

                if (err != -EAGAIN)

                        goto reset;

        }

        for (i = 0; i < n / sizeof(buf[0]); i++) {

                err = WARN_ON(buf[i] >= TD_RLB_REQUEST_MAX);

                if (err)

                        goto kill;

                cred += buf[i];

        }

        if (cred > valve->need) {

                err = -EINVAL;

                goto reset;

        }

        valve->cred += cred;

        valve->need -= cred;

        return;

reset:

        VERR(err, "resetting connection");

        valve_conn_reset(valve);

        return;

kill:

        ERR("Killing valve.");

        valve_kill(valve);

}

static void

valve_conn_request(td_valve_t *valve, unsigned long size)

{

        struct td_valve_req _req;

        int err;

        _req.need    = size;

        _req.done    = valve->done;

        valve->need += size;

        valve->done  = 0;

        valve_clear_done_pending(valve);

        err = valve_sock_send(valve, &_req, sizeof(_req));

        if (!err)

                return;

        VERR(err, "resetting connection");

        valve_conn_reset(valve);

}

static int

valve_expend_request(td_valve_t *valve, const td_request_t treq)

{

        if (valve->flags & TD_VALVE_KILLED)

                return 0;

        if (valve->sock < 0)

                return 0;

        if (valve->cred < TREQ_SIZE(treq))

                return -EAGAIN;

        valve->cred -= TREQ_SIZE(treq);

        return 0;

}

static void

__valve_complete_treq(td_request_t treq, int error)

{

        td_valve_request_t *req = treq.cb_data;

        td_valve_t *valve = req->valve;

        BUG_ON(req->secs < treq.secs);

        req->secs -= treq.secs;

        valve->done += TREQ_SIZE(treq);

        valve_set_done_pending(valve);

        if (!req->secs) {

                td_complete_request(req->treq, error);

                valve_free_request(valve, req);

        }

}

static void

valve_forward_stored_requests(td_valve_t *valve)

{

        td_valve_request_t *req, *next;

        td_request_t clone;

        int err;

        td_valve_for_each_stored_request(req, next, valve) {

                err = valve_expend_request(valve, req->treq);

                if (err)

                        break;

                clone         = req->treq;

                clone.cb      = __valve_complete_treq;

                clone.cb_data = req;

                td_forward_request(clone);

                valve->stats.forw++;

                list_move(&req->entry, &valve->forw);

        }

}

static int

valve_store_request(td_valve_t *valve, td_request_t treq)

{

        td_valve_request_t *req;

        req = valve_alloc_request(valve);

        if (!req)

                return -EBUSY;

        valve_conn_request(valve, TREQ_SIZE(treq));

        req->treq = treq;

        req->secs = treq.secs;

        list_add_tail(&req->entry, &valve->stor);

        valve->stats.stor++;

        return 0;

}

static void

valve_kill(td_valve_t *valve)

{

        valve->flags |= TD_VALVE_KILLED;

        valve_conn_close(valve, 1);

}

static void

valve_init(td_valve_t *valve, unsigned long flags)

{

        int i;

        memset(valve, 0, sizeof(*valve));

        INIT_LIST_HEAD(&valve->stor);

        INIT_LIST_HEAD(&valve->forw);

        valve->sock     = -1;

        valve->sock_id  = -1;

        valve->retry_id = -1;

        valve->sched_id = -1;

        valve->flags    = flags;

        for (i = ARRAY_SIZE(valve->reqv) - 1; i >= 0; i--) {

                td_valve_request_t *req = &valve->reqv[i];

                req->valve = valve;

                INIT_LIST_HEAD(&req->entry);

                valve_free_request(valve, req);

        }

}

static int

td_valve_close(td_driver_t *driver)

{

        td_valve_t *valve = driver->data;

        WARN_ON(!list_empty(&valve->stor));

        WARN_ON(!list_empty(&valve->forw));

        valve_conn_close(valve, 0);

        if (valve->brname) {

                free(valve->brname);

                valve->brname = NULL;

        }

        return 0;

}

static int

td_valve_open(td_driver_t *driver,

              const char *name, td_flag_t flags)

{

        td_valve_t *valve = driver->data;

        int err;

        valve_init(valve, TD_VALVE_WRLIMIT);

        valve->brname = strdup(name);

        if (!valve->brname) {

                err = -errno;

                goto fail;

        }

        valve_conn_open(valve);

        return 0;

fail:

        td_valve_close(driver);

        return err;

}

static void

td_valve_queue_request(td_driver_t *driver, td_request_t treq)

{

        td_valve_t *valve = driver->data;

        int err;

        switch (treq.op) {

        case TD_OP_READ:

                if (valve->flags & TD_VALVE_RDLIMIT)

                        break;

                goto forward;

        case TD_OP_WRITE:

                if (valve->flags & TD_VALVE_WRLIMIT)

                        break;

                goto forward;

        default:

                BUG();

        }

        err = valve_expend_request(valve, treq);

        if (!err)

                goto forward;

        err = valve_store_request(valve, treq);

        if (err)

                td_complete_request(treq, -EBUSY);

        return;

forward:

        td_forward_request(treq);

        valve->stats.forw++;

}

static int

td_valve_get_parent_id(td_driver_t *driver, td_disk_id_t *id)

{

        return -EINVAL;

}

static int

td_valve_validate_parent(td_driver_t *driver,

                         td_driver_t *parent_driver, td_flag_t flags)

{

        return -EINVAL;

}

static void

td_valve_stats(td_driver_t *driver, td_stats_t *st)

{

        td_valve_t *valve = driver->data;

        td_valve_request_t *req, *next;

        int n_reqs;

        tapdisk_stats_field(st, "bridge", "d", valve->brname);

        tapdisk_stats_field(st, "flags", "#x", valve->flags);

        tapdisk_stats_field(st, "cred", "d", valve->cred);

        tapdisk_stats_field(st, "need", "d", valve->need);

        tapdisk_stats_field(st, "done", "d", valve->done);

        /*

         * stored is [ waiting, total-waits ]

         */

        n_reqs = 0;

        td_valve_for_each_stored_request(req, next, valve)

                n_reqs++;

        tapdisk_stats_field(st, "stor", "[");

        tapdisk_stats_val(st, "d", n_reqs);

        tapdisk_stats_val(st, "llu", valve->stats.stor);

        tapdisk_stats_leave(st, ']');

        /*

         * forwarded is [ in-flight, total-requests ]

         */

        n_reqs = 0;

        td_valve_for_each_forwarded_request(req, next, valve)

                n_reqs++;

        tapdisk_stats_field(st, "forw", "[");

        tapdisk_stats_val(st, "d", n_reqs);

        tapdisk_stats_val(st, "llu", valve->stats.forw);

        tapdisk_stats_leave(st, ']');

}

struct tap_disk tapdisk_valve = {

        .disk_type                  = "tapdisk_valve",

        .flags                      = 0,

        .private_data_size          = sizeof(td_valve_t),

        .td_open                    = td_valve_open,

        .td_close                   = td_valve_close,

        .td_queue_read              = td_valve_queue_request,

        .td_queue_write             = td_valve_queue_request,

        .td_get_parent_id           = td_valve_get_parent_id,

        .td_validate_parent         = td_valve_validate_parent,

        .td_stats                   = td_valve_stats,

};