Archipelago

/*

 * Copyright 2012 GRNET S.A. All rights reserved.

 *

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

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *   2. 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.

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``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 GRNET S.A 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.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 */

#define _GNU_SOURCE

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <sys/syscall.h>

#include <sys/types.h>

#include <pthread.h>

#include <xseg/xseg.h>

#include <peer.h>

#include <time.h>

#include <sys/util.h>

#include <signal.h>

#include <bench-xseg.h>

#include <bench-lfsr.h>

#include <limits.h>

char global_id[IDLEN];

/*

 * This macro checks two things:

 * a) If in-flight requests are less than given iodepth

 * b) If we have submitted all of the requests

 */

#define CAN_SEND_REQUEST(__p)                                                                                        \

        ((__p->status->submitted - __p->status->received < __p->iodepth) && \

        (__p->status->submitted < __p->status->max) &&                                                \

        !isTerminate())

#define CAN_VERIFY(__p)                                                                                                        \

        ((GET_FLAG(VERIFY, __p->flags) != VERIFY_NO) && __p->op == X_READ)

#define CAN_PRINT_PROGRESS(__p, __q)                                                                        \

        ((GET_FLAG(PROGRESS, __p->flags) == PROGRESS_YES) &&                                \

        (__p->status->received == __q))

void custom_peer_usage()

{

        fprintf(stderr, "Custom peer options: \n"

                        "  --------------------------------------------\n"

                        "    -op       | None    | XSEG operation [read|write|info|delete]\n"

                        "    --pattern | None    | I/O pattern [seq|rand]\n"

                        "    --verify  | no      | Verify written requests [no|meta|full]\n"

                        "    -rc       | None    | Request cap\n"

                        "    -to       | None    | Total objects\n"

                        "    -ts       | None    | Total I/O size\n"

                        "    -os       | 4M      | Object size\n"

                        "    -bs       | 4k      | Block size\n"

                        "    -tp       | None    | Target port\n"

                        "    --iodepth | 1       | Number of in-flight I/O requests\n"

                        "    --seed    | None    | Initialize LFSR and target names\n"

                        "    --insanity| sane    | Adjust insanity level of benchmark:\n"

                        "              |         |     [sane|eccentric|manic|paranoid]\n"

                        "    --progress| yes     | Show progress of requests\n"

                        "\n"

                        "Additional information:\n"

                        "  --------------------------------------------\n"

                        "  The -to and -ts options are mutually exclusive\n"

                        "\n");

}

int custom_peer_init(struct peerd *peer, int argc, char *argv[])

{

        struct bench *prefs;

        char request_cap[MAX_ARG_LEN + 1];

        char total_objects[MAX_ARG_LEN + 1];

        char total_size[MAX_ARG_LEN + 1];

        char object_size[MAX_ARG_LEN + 1];

        char block_size[MAX_ARG_LEN + 1];

        char op[MAX_ARG_LEN + 1];

        char pattern[MAX_ARG_LEN + 1];

        char insanity[MAX_ARG_LEN + 1];

        char verify[MAX_ARG_LEN + 1];

        char progress[MAX_ARG_LEN + 1];

        struct xseg *xseg = peer->xseg;

        unsigned int xseg_page_size = 1 << xseg->config.page_shift;

        long iodepth = -1;

        long dst_port = -1;

        unsigned long seed = -1;

        uint64_t rc;

        struct timespec timer_seed;

        struct timespec *ts;

        int set_by_hand = 0;

        int i, r;

        op[0] = 0;

        pattern[0] = 0;

        total_objects[0] = 0;

        total_size[0] = 0;

        block_size[0] = 0;

        object_size[0] = 0;

        insanity[0] = 0;

        verify[0] = 0;

        request_cap[0] = 0;

        progress[0] = 0;

#ifdef MT

        for (i = 0; i < nr_threads; i++) {

                prefs = peer->thread[i]->priv;

                prefs = malloc(sizeof(struct bench));

                if (!prefs) {

                        perror("malloc");

                        return -1;

                }

        }

#endif

        prefs = malloc(sizeof(struct bench));

        if (!prefs) {

                perror("malloc");

                return -1;

        }

        memset(prefs, 0, sizeof(struct bench));

        prefs->status = malloc(sizeof(struct req_status));

        if (!prefs->status) {

                perror("malloc");

                return -1;

        }

        memset(prefs->status, 0, sizeof(struct req_status));

        //Begin reading the benchmark-specific arguments

        BEGIN_READ_ARGS(argc, argv);

        READ_ARG_STRING("-rc", request_cap, MAX_ARG_LEN);

        READ_ARG_STRING("-op", op, MAX_ARG_LEN);

        READ_ARG_STRING("--pattern", pattern, MAX_ARG_LEN);

        READ_ARG_STRING("-to", total_objects, MAX_ARG_LEN);

        READ_ARG_STRING("-ts", total_size, MAX_ARG_LEN);

        READ_ARG_STRING("-os", object_size, MAX_ARG_LEN);

        READ_ARG_STRING("-bs", block_size, MAX_ARG_LEN);

        READ_ARG_ULONG("--iodepth", iodepth);

        READ_ARG_ULONG("-tp", dst_port);

        READ_ARG_ULONG("--seed", seed);

        READ_ARG_STRING("--insanity", insanity, MAX_ARG_LEN);

        READ_ARG_STRING("--verify", verify, MAX_ARG_LEN);

        READ_ARG_STRING("--progress", progress, MAX_ARG_LEN);

        END_READ_ARGS();

        /*****************************\

         * Check I/O type parameters *

        \*****************************/

        //We support 4 xseg operations: X_READ, X_WRITE, X_DELETE, X_INFO

        //The I/O pattern of these operations can be either sequential (seq) or

        //random (rand)

        if (!op[0]) {

                XSEGLOG2(&lc, E, "xseg operation needs to be supplied\n");

                goto arg_fail;

        }

        r = read_op(op);

        if (r < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: -op %s\n", op);

                goto arg_fail;

        }

        prefs->op = r;

        if (!pattern[0]) {

                XSEGLOG2(&lc, E, "I/O pattern needs to be supplied\n");

                goto arg_fail;

        }

        r = read_pattern(pattern);

        if (r < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: --pattern %s\n", pattern);

                goto arg_fail;

        }

        SET_FLAG(PATTERN, prefs->flags, r);

        if (!verify[0])

                strcpy(verify, "no");

        r = read_verify(verify);

        if (r < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: --verify %s\n", verify);

                goto arg_fail;

        }

        SET_FLAG(VERIFY, prefs->flags, r);

        //Default iodepth value is 1

        if (iodepth < 0)

                prefs->iodepth = 1;

        else

                prefs->iodepth = iodepth;

        /**************************\

         * Check timer parameters *

        \**************************/

        //Most of the times, not all timers need to be used.

        //We can choose which timers will be used by adjusting the "insanity"

        //level of the benchmark i.e. the obscurity of code paths (get request,

        //submit request) that will be timed.

        if (!insanity[0])

                strcpy(insanity, "sane");

        r = read_insanity(insanity);

        if (r < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: --insanity %s\n", insanity);

                goto arg_fail;

        }

        SET_FLAG(INSANITY, prefs->flags, r);

        /*****************************\

         * Check I/O size parameters *

        \*****************************/

        //Block size (bs): Defaults to 4K.

        //It must be a number followed by one of these characters:

        //                                                [k|K|m|M|g|G]

        //If not, it will be considered as size in bytes.

        //Must be integer multiple of segment's page size (typically 4k).

        if (!block_size[0])

                strcpy(block_size,"4k");

        prefs->bs = str2num(block_size);

        if (!prefs->bs) {

                XSEGLOG2(&lc, E, "Invalid syntax: -bs %s\n", block_size);

                goto arg_fail;

        } else if (prefs->bs % xseg_page_size) {

                XSEGLOG2(&lc, E, "Misaligned block size: %s\n", block_size);

                goto arg_fail;

        }

        //Total objects (to) or total I/O size (ts).

        //Must have the same format as "block size"

        //They are mutually exclusive

        if (total_objects[0] && total_size[0]) {

                XSEGLOG2(&lc, E, "Total objects and total size are mutually exclusive\n");

                goto arg_fail;

        } else if (total_objects[0]) {

                prefs->to = str2num(total_objects);

                if (!prefs->to) {

                        XSEGLOG2(&lc, E, "Invalid syntax: -to %s\n", total_objects);

                        goto arg_fail;

                }

                //In this case, the maximum number of requests is the total number of

                //objects we will handle

                prefs->status->max = prefs->to;

        } else if (total_size[0]) {

                if (prefs->op != X_READ && prefs->op != X_WRITE) {

                        XSEGLOG2(&lc, E,

                                        "Total objects must be supplied (required by -op %s)\n", op);

                        goto arg_fail;

                }

                prefs->ts = str2num(total_size);

                if (!prefs->ts) {

                        XSEGLOG2(&lc, E, "Invalid syntax: -ts %s\n", total_size);

                        goto arg_fail;

                } else if (prefs->ts % prefs->bs) {

                        XSEGLOG2(&lc, E, "Misaligned total I/O size: %s\n", total_size);

                        goto arg_fail;

                }

                //In this case, the maximum number of requests is the number of blocks

                //we need to cover the total I/O size

                prefs->status->max = prefs->ts / prefs->bs;

        } else {

                XSEGLOG2(&lc, E, "Total objects or total size must be supplied\n");

                goto arg_fail;

        }

        //Object size (os): Defaults to 4M.

        //Must have the same format as "block size"

        //Must be integer multiple of "block size"

        if (!object_size[0])

                strcpy(object_size,"4M");

        prefs->os = str2num(object_size);

        if (!prefs->os) {

                XSEGLOG2(&lc, E, "Invalid syntax: -os %s\n", object_size);

                goto arg_fail;

        } else if (prefs->os % prefs->bs) {

                XSEGLOG2(&lc, E, "Misaligned object size: %s\n", object_size);

                goto arg_fail;

        }

        /*************************\

         * Check port parameters *

        \*************************/

        if (dst_port < 0){

                XSEGLOG2(&lc, E, "Target port must be supplied\n");

                goto arg_fail;

        }

        prefs->src_port = peer->portno_start; //TODO: allow user to change this

        prefs->dst_port = (xport) dst_port;

        /*********************************\

         * Create timers for all metrics *

        \*********************************/

        if (init_timer(&prefs->total_tm, INSANITY_SANE))

                goto tm_fail;

        if (init_timer(&prefs->sub_tm, INSANITY_MANIC))

                goto tm_fail;

        if (init_timer(&prefs->get_tm, INSANITY_PARANOID))

                goto tm_fail;

        if (init_timer(&prefs->rec_tm, INSANITY_ECCENTRIC))

                goto tm_fail;

        if (prefs->rec_tm->insanity <= GET_FLAG(INSANITY, prefs->flags)) {

                for (i = 0; i < peer->nr_ops; i++) {

                        ts = malloc(sizeof(struct timespec));

                        if (!ts) {

                                XSEGLOG2(&lc, E, "Timespec allocation failed\n");

                                goto tm_fail;

                        }

                        peer->peer_reqs[i].priv = ts;

                }

        }

        /*************************************\

         * Initialize the LFSR and global_id *

        \*************************************/

reseed:

        //We proceed to initialise the global_id, and seed variables.

        if (seed == -1) {

                clock_gettime(CLOCK_BENCH, &timer_seed);

                seed = timer_seed.tv_nsec;

        } else {

                set_by_hand = 1;

        }

        create_id(seed);

        if (prefs->status->max == 1)

                SET_FLAG(PATTERN, prefs->flags, PATTERN_SEQ);

        if (GET_FLAG(PATTERN, prefs->flags) == PATTERN_RAND) {

                prefs->lfsr = malloc(sizeof(struct bench_lfsr));

                if (!prefs->lfsr) {

                        perror("malloc");

                        goto lfsr_fail;

                }

                r = lfsr_init(prefs->lfsr, prefs->status->max, seed, seed & 0xF);

                if (r && set_by_hand) {

                        XSEGLOG2(&lc, E, "LFSR could not be initialized.\n");

                        goto lfsr_fail;

                } else if (r) {

                        seed = -1;

                        goto reseed;

                }

        }

        /*********************************\

         * Miscellaneous initializations *

        \*********************************/

        /* The request cap must be enforced only after the LFSR is initialized */

        if (request_cap[0]) {

                rc = str2num(request_cap);

                if (!rc) {

                        XSEGLOG2(&lc, E, "Invalid syntax: -rc %s\n", request_cap);

                        goto arg_fail;

                } else if (rc > prefs->status->max) {

                        XSEGLOG2(&lc, E, "Request cap exceeds current request total.\n");

                        goto arg_fail;

                }

                prefs->status->max = rc;

        }

        /* Benchmarking progress printing is on by default */

        if (!progress[0])

                strcpy(progress, "yes");

        r = read_progress(progress);

        if (r < 0) {

                XSEGLOG2(&lc, E, "Invalid syntax: --progress %s\n", progress);

                goto arg_fail;

        }

        SET_FLAG(PROGRESS, prefs->flags, r);

        prefs->peer = peer;

        peer->peerd_loop = bench_peerd_loop;

        peer->priv = (void *) prefs;

        XSEGLOG2(&lc, I, "Global ID is %s\n", global_id);

        return 0;

arg_fail:

        custom_peer_usage();

lfsr_fail:

        free(prefs->lfsr);

tm_fail:

        free(prefs->total_tm);

        free(prefs->sub_tm);

        free(prefs->get_tm);

        free(prefs->rec_tm);

        free(prefs);

        return -1;

}

static int send_request(struct peerd *peer, struct bench *prefs)

{

        struct xseg_request *req;

        struct xseg *xseg = peer->xseg;

        struct peer_req *pr;

        xport srcport = prefs->src_port;

        xport dstport = prefs->dst_port;

        xport p;

        int r;

        uint64_t new;

        uint64_t size = prefs->bs;

        struct timespec *ts;

        //srcport and dstport must already be provided by the user.

        //returns struct xseg_request with basic initializations

        XSEGLOG2(&lc, D, "Get new request\n");

        timer_start(prefs, prefs->get_tm);

        req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);

        if (!req) {

                XSEGLOG2(&lc, W, "Cannot get request\n");

                return -1;

        }

        timer_stop(prefs, prefs->get_tm, NULL);

        //Allocate enough space for the data and the target's name

        XSEGLOG2(&lc, D, "Prepare new request\n");

        r = xseg_prep_request(xseg, req, TARGETLEN, size);

        if (r < 0) {

                XSEGLOG2(&lc, W, "Cannot prepare request! (%lu, %llu)\n",

                                TARGETLEN, (unsigned long long)size);

                goto put_xseg_request;

        }

        //Determine what the next target/chunk will be, based on I/O pattern

        new = determine_next(prefs);

        req->op = prefs->op;

        XSEGLOG2(&lc, I, "Our new request is %lu\n", new);

        //Create a target of this format: "bench-<global_id>-<obj_no>"

        create_target(prefs, req, new);

        if (prefs->op == X_WRITE || prefs->op == X_READ) {

                req->size = size;

                //Calculate the chunk's offset inside the object

                req->offset = calculate_offset(prefs, new);

                XSEGLOG2(&lc, D, "Offset of request %lu is %lu\n", new, req->offset);

                if (prefs->op == X_WRITE)

                        create_chunk(prefs, req, new);

        }

        XSEGLOG2(&lc, D, "Allocate peer request\n");

        pr = alloc_peer_req(peer);

        if (!pr) {

                XSEGLOG2(&lc, W, "Cannot allocate peer request (%ld remaining)\n",

                                peer->nr_ops - xq_count(&peer->free_reqs));

                goto put_xseg_request;

        }

        pr->peer = peer;

        pr->portno = srcport;

        pr->req = req;

        //XSEGLOG2(&lc, D, "Set request data\n");

        r = xseg_set_req_data(xseg, req, pr);

        if (r < 0) {

                XSEGLOG2(&lc, W, "Cannot set request data\n");

                goto put_peer_request;

        }

        /*

         * Start measuring receive time.

         * When we receive a request, we need to have its submission time to

         * measure elapsed time. Thus, we copy its submission time to pr->priv.

         * QUESTION: Is this the fastest way?

         */

        timer_start(prefs, prefs->rec_tm);

        if (prefs->rec_tm->insanity <= GET_FLAG(INSANITY, prefs->flags)) {

                ts = (struct timespec *)pr->priv;

                ts->tv_sec = prefs->rec_tm->start_time.tv_sec;

                ts->tv_nsec = prefs->rec_tm->start_time.tv_nsec;

        }

        //Submit the request from the source port to the target port

        XSEGLOG2(&lc, D, "Submit request %lu\n", new);

        timer_start(prefs, prefs->sub_tm);

        p = xseg_submit(xseg, req, srcport, X_ALLOC);

        if (p == NoPort) {

                XSEGLOG2(&lc, W, "Cannot submit request\n");

                goto put_peer_request;

        }

        prefs->status->submitted++;

        timer_stop(prefs, prefs->sub_tm, NULL);

        //Send SIGIO to the process that has bound this port to inform that

        //IO is possible

        r = xseg_signal(xseg, p);

        //if (r < 0)

        //        XSEGLOG2(&lc, W, "Cannot signal destination peer (reason %d)\n", r);

        return 0;

put_peer_request:

        free(pr->priv);

        free_peer_req(peer, pr);

put_xseg_request:

        if (xseg_put_request(xseg, req, srcport))

                XSEGLOG2(&lc, W, "Cannot put request\n");

        return -1;

}

/*

 * This function substitutes the default generic_peerd_loop of peer.c.

 * It's plugged to struct peerd at custom peer's initialisation

 */

int bench_peerd_loop(void *arg)

{

#ifdef MT

        struct thread *t = (struct thread *) arg;

        struct peerd *peer = t->peer;

        char *id = t->arg;

#else

        struct peerd *peer = (struct peerd *) arg;

        char id[4] = {'P','e','e','r'};

#endif

        struct xseg *xseg = peer->xseg;

        struct bench *prefs = peer->priv;

        xport portno_start = peer->portno_start;

        xport portno_end = peer->portno_end;

        pid_t pid = syscall(SYS_gettid);

        uint64_t threshold=1000/(1 + portno_end - portno_start);

        uint64_t cached_prog_quantum = 0;

        uint64_t prog_quantum = 0;

        int r;

        uint64_t loops;

        if (GET_FLAG(PROGRESS, prefs->flags) == PROGRESS_YES) {

                prog_quantum = calculate_prog_quantum(prefs);

                cached_prog_quantum = prog_quantum;

                print_stats(prefs);

        }

        XSEGLOG2(&lc, I, "%s has tid %u.\n",id, pid);

        xseg_init_local_signal(xseg, peer->portno_start);

        timer_start(prefs, prefs->total_tm);

send_request:

        while (!(isTerminate() && all_peer_reqs_free(peer))) {

                while (CAN_SEND_REQUEST(prefs)) {

                        xseg_cancel_wait(xseg, peer->portno_start);

                        XSEGLOG2(&lc, D, "...because %lu < %lu && %lu < %lu\n",

                                        prefs->status->submitted - prefs->status->received,

                                        prefs->iodepth, prefs->status->received,

                                        prefs->status->max);

                        XSEGLOG2(&lc, D, "Start sending new request\n");

                        r = send_request(peer, prefs);

                        if (r < 0)

                                break;

                }

                //Heart of peerd_loop. This loop is common for everyone.

                for (loops = threshold; loops > 0; loops--) {

                        if (loops == 1)

                                xseg_prepare_wait(xseg, peer->portno_start);

                        if (UNLIKELY(CAN_PRINT_PROGRESS(prefs, prog_quantum))) {

                                prog_quantum += cached_prog_quantum;

                                print_progress(prefs);

                        }

                        if (check_ports(peer)) {

                                //If an old request has just been acked, the most sensible

                                //thing to do is to immediately send a new one

                                if (prefs->status->received < prefs->status->max)

                                        goto send_request;

                                else

                                        return 0;

                        }

                }

                //struct xseg_port *port = xseg_get_port(xseg, portno_start);

                //struct xq *q;

                //q = XPTR_TAKE(port->request_queue, xseg->segment);

                //XSEGLOG2(&lc, I, "%s goes to sleep with %u requests pending\n",

                //                id, xq_count(q));

                XSEGLOG2(&lc, I, "%s goes to sleep\n", id);

                xseg_wait_signal(xseg, 10000000UL);

                xseg_cancel_wait(xseg, peer->portno_start);

                XSEGLOG2(&lc, I, "%s woke up\n", id);

        }

        XSEGLOG2(&lc, I, "peer->free_reqs = %d, peer->nr_ops = %d\n",

                        xq_count(&peer->free_reqs), peer->nr_ops);

        return 0;

}

void custom_peer_finalize(struct peerd *peer)

{

        struct bench *prefs = peer->priv;

        //TODO: Measure mean time, standard variation

        if (!prefs->total_tm->completed)

                timer_stop(prefs, prefs->total_tm, NULL);

        if (GET_FLAG(PROGRESS, prefs->flags) == PROGRESS_YES)

                print_progress(prefs);

        else

                print_stats(prefs);

        print_remaining(prefs);

        print_res(prefs);

        return;

}

/*

 * handle_received: +1 to our received requests.

 * Do some sanity checks and then check if request is failed.

 * If not try to verify the request if asked.

 */

static void handle_received(struct peerd *peer, struct peer_req *pr)

{

        //FIXME: handle null pointer

        struct bench *prefs = peer->priv;

        struct timer *rec = prefs->rec_tm;

        prefs->status->received++;

        if (!pr->req) {

                //This is a serious error, so we must stop

                XSEGLOG2(&lc, E, "Received peer request with no xseg request");

                terminated++;

                return;

        }

        if ((GET_FLAG(INSANITY, prefs->flags) < rec->insanity) && !pr->priv) {

                XSEGLOG2(&lc, W, "Cannot find submission time of request");

                return;

        }

        timer_stop(prefs, rec, (struct timespec *)pr->priv);

        if (!(pr->req->state & XS_SERVED))

                prefs->status->failed++;

        else if (CAN_VERIFY(prefs) && read_chunk(prefs, pr->req))

                prefs->status->corrupted++;

        if (xseg_put_request(peer->xseg, pr->req, pr->portno))

                XSEGLOG2(&lc, W, "Cannot put xseg request\n");

        //QUESTION, can't we just keep the malloced memory for future use?

        free(pr->priv);

        free_peer_req(peer, pr);

}

int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req,

                enum dispatch_reason reason)

{

        switch (reason) {

                case dispatch_accept:

                        //This is wrong, benchmarking peer should not accept requests,

                        //only receive them.

                        XSEGLOG2(&lc, W, "Bench peer should not accept requests\n");

                        complete(peer, pr);

                        break;

                case dispatch_receive:

                        handle_received(peer, pr);

                        break;

                default:

                        fail(peer, pr);

        }

        return 0;

}