code.grnet.gr Git - archipelago/blob - xseg/peers/user/xseg-tool.c

   1 #define _GNU_SOURCE
   2 #include <stdio.h>
   3 #include <stdlib.h>
   4 #include <string.h>
   5 #include <stdint.h>
   6 #include <ctype.h>
   7 #include <sys/stat.h>
   8 #include <sys/types.h>
   9 #include <fcntl.h>
  10 #include <unistd.h>
  11
  12 #include <xseg/xseg.h>
  13 #include <xseg/protocol.h>
  14 int help(void)
  15 {
  16         printf("xseg <spec> [[[<src_port>]:[<dst_port>]] [<command> <arg>*] ]*\n"
  17                 "spec:\n"
  18                 "    <type:name:nr_ports:nr_requests:request_size:extra_size:page_shift>\n"
  19                 "global commands:\n"
  20                 "    reportall\n"
  21                 "    create\n"
  22                 "    destroy\n"
  23                 "    bind <portno>\n"
  24                 "    signal <portno>\n"
  25                 "    bridge <portno1> <portno2> <logfile> {full|summary|stats}\n"
  26                 "port commands:\n"
  27                 "    report\n"
  28                 "    alloc_requests (to source) <nr>\n"
  29                 "    free_requests (from source) <nr>\n"
  30                 "    put_requests (all from dest)\n"
  31                 "    put_replies (all from dest)\n"
  32                 "    wait        <nr_replies>\n"
  33                 "    complete    <nr_requests>\n"
  34                 "    fail        <nr_requests>\n"
  35                 "    rndwrite    <nr_loops> <seed> <targetlen> <datalen> <objectsize>\n"
  36                 "    rndread     <nr_loops> <seed> <targetlen> <datalen> <objectsize>\n"
  37                 "    submit_reqs <nr_loops> <concurrent_reqs>\n"
  38                 "    info        <target>\n"
  39                 "    read        <target> <offset> <size>\n"
  40                 "    write       <target> <offset> < data\n"
  41                 "    truncate    <target> <size>\n"
  42                 "    delete      <target>\n"
  43                 "    acquire     <target>\n"
  44                 "    release     <target>\n"
  45                 "    copy        <src>  <dst>\n"
  46                 "    clone       <src>  <dst>\n"
  47         );
  48         return 1;
  49 }
  50
  51 char *namebuf;
  52 char *chunk;
  53 struct xseg_config cfg;
  54 struct xseg *xseg;
  55 uint32_t srcport, dstport;
  56 uint64_t reqs;
  57 #define mkname mkname_heavy
  58 /* heavy distributes duplicates much more widely than light
  59  * ./xseg-tool random 100000 | cut -d' ' -f2- | sort | uniq -d -c |wc -l
  60  */
  61
  62 xport sport = NoPort;
  63 static void init_local_signal()
  64 {
  65         if (xseg && sport != srcport){
  66                 xseg_init_local_signal(xseg, srcport);
  67                 sport = srcport;
  68         }
  69 }
  70
  71 void mkname_heavy(char *name, uint32_t namelen, uint32_t seed)
  72 {
  73         int i;
  74         char c;
  75         for (i = 0; i < namelen; i += 1) {
  76                 c = seed + (seed >> 8) + (seed >> 16) + (seed >> 24);
  77                 c = '0' + ((c + (c >> 4)) & 0xf);
  78                 if (c > '9')
  79                         c += 'a'-'0'-10;
  80                 name[i] = c;
  81                 seed *= ((seed % 137911) | 1) * 137911;
  82         }
  83 }
  84
  85 void mkname_light(char *name, uint32_t namelen, uint32_t seed)
  86 {
  87         int i;
  88         char c;
  89         for (i = 0; i < namelen; i += 1) {
  90                 c = seed;
  91                 name[i] = 'A' + (c & 0xf);
  92                 seed += 1;
  93         }
  94 }
  95
  96 uint64_t pick(uint64_t size)
  97 {
  98         return (uint64_t)((double)(RAND_MAX) / random());
  99 }
 100
 101 void mkchunk(   char *chunk, uint32_t datalen,
 102                 char *target, uint32_t targetlen, uint64_t offset)
 103 {
 104         long i, r, bufsize = targetlen + 16;
 105         char buf[bufsize];
 106         r = datalen % bufsize;
 107         snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, target);
 108
 109         for (i = 0; i <= (long)datalen - bufsize; i += bufsize)
 110                 memcpy(chunk + i, buf, bufsize);
 111
 112         memcpy(chunk + datalen - r, buf, r);
 113 }
 114
 115 int chkchunk(   char *chunk, uint32_t datalen,
 116                 char *target, uint32_t targetlen, uint64_t offset)
 117 {
 118         long i, r;
 119         int bufsize = targetlen + 16;
 120         char buf[bufsize];
 121         r = datalen % targetlen;
 122         snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, target);
 123
 124         for (i = 0; i <= (long)datalen - bufsize; i += bufsize)
 125                 if (memcmp(chunk + i, buf, bufsize)) {
 126                         /*printf("mismatch: '%*s'* vs '%*s'\n",
 127                                 bufsize, buf, datalen, chunk);
 128                         */
 129                         return 0;
 130                 }
 131
 132         if (memcmp(chunk + datalen - r, buf, r))
 133                 return 0;
 134
 135         return 1;
 136 }
 137
 138
 139 #define ALLOC_MIN 4096
 140 #define ALLOC_MAX 1048576
 141
 142 void inputbuf(FILE *fp, char **retbuf, uint64_t *retsize)
 143 {
 144         static uint64_t alloc_size;
 145         static char *buf;
 146         uint64_t size = 0;
 147         char *p;
 148         size_t r;
 149
 150         if (alloc_size < ALLOC_MIN)
 151                 alloc_size = ALLOC_MIN;
 152
 153         if (alloc_size > ALLOC_MAX)
 154                 alloc_size = ALLOC_MAX;
 155
 156         p = realloc(buf, alloc_size);
 157         if (!p) {
 158                 if (buf)
 159                         free(buf);
 160                 buf = NULL;
 161                 goto out;
 162         }
 163
 164         buf = p;
 165
 166         while (!feof(fp)) {
 167                 r = fread(buf + size, 1, alloc_size - size, fp);
 168                 if (!r)
 169                         break;
 170                 size += r;
 171                 if (size >= alloc_size) {
 172                         p = realloc(buf, alloc_size * 2);
 173                         if (!p) {
 174                                 if (buf)
 175                                         free(buf);
 176                                 buf = NULL;
 177                                 size = 0;
 178                                 goto out;
 179                         }
 180                         buf = p;
 181                         alloc_size *= 2;
 182                 }
 183         }
 184
 185 out:
 186         *retbuf = buf;
 187         *retsize = size;
 188 }
 189
 190 void report_request(struct xseg_request *req)
 191 {
 192         //uint32_t max = req->datalen;
 193         //char *data = xseg_get_data(xseg, req);
 194         //if (max > 128)
 195         //      max = 128;
 196         //data[max-1] = 0;
 197         fprintf(stderr, "request %llu state %u\n", (unsigned long long)req->serial, req->state);
 198 }
 199
 200 int cmd_info(char *target)
 201 {
 202         uint32_t targetlen = strlen(target);
 203         size_t size = sizeof(uint64_t);
 204         int r;
 205         xport p;
 206         struct xseg_request *req;
 207         char *req_target;
 208
 209         req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
 210         if (!req) {
 211                 fprintf(stderr, "No request!\n");
 212                 return -1;
 213         }
 214
 215         r = xseg_prep_request(xseg, req, targetlen, size);
 216         if (r < 0) {
 217                 fprintf(stderr, "Cannot prepare request! (%lu, %lu)\n",
 218                         (unsigned long) targetlen, (unsigned long) size);
 219                 xseg_put_request(xseg, req, srcport);
 220                 return -1;
 221         }
 222
 223         req_target = xseg_get_target(xseg, req);
 224         strncpy(req_target, target, targetlen);
 225         req->offset = 0;
 226         req->size = size;
 227         req->op = X_INFO;
 228
 229         p = xseg_submit(xseg, req, srcport, X_ALLOC);
 230         if (p == NoPort)
 231                 return -1;
 232
 233         xseg_signal(xseg, p);
 234
 235         return 0;
 236 }
 237
 238 int cmd_read(char *target, uint64_t offset, uint64_t size)
 239 {
 240         uint32_t targetlen = strlen(target);
 241         int r;
 242         xport p;
 243         char *req_target;
 244         struct xseg_request *req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
 245         printf("%x\n", req);
 246         if (!req) {
 247                 fprintf(stderr, "No request\n");
 248                 return -1;
 249         }
 250
 251         r = xseg_prep_request(xseg, req, targetlen, size);
 252         if (r < 0) {
 253                 fprintf(stderr, "Cannot prepare request! (%lu, %llu)\n",
 254                         (unsigned long)targetlen, (unsigned long long)size);
 255                 xseg_put_request(xseg, req, srcport);
 256                 return -1;
 257         }
 258
 259         req_target = xseg_get_target(xseg, req);
 260         strncpy(req_target, target, targetlen);
 261         req->offset = offset;
 262         req->size = size;
 263         req->op = X_READ;
 264         report_request(req);
 265         p = xseg_submit(xseg, req, srcport, X_ALLOC);
 266         if (p == NoPort)
 267                 return -1;
 268
 269         xseg_signal(xseg, p);
 270         return 0;
 271 }
 272
 273 int cmd_write(char *target, uint64_t offset)
 274 {
 275         char *buf = NULL;
 276         int r;
 277         xport p;
 278         uint64_t size = 0;
 279         char *req_target, *req_data;
 280         uint32_t targetlen = strlen(target);
 281         struct xseg_request *req;
 282
 283         inputbuf(stdin, &buf, &size);
 284         if (!size) {
 285                 fprintf(stderr, "No input\n");
 286                 return -1;
 287         }
 288
 289         req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
 290         if (!req) {
 291                 fprintf(stderr, "No request\n");
 292                 return -1;
 293         }
 294
 295         r = xseg_prep_request(xseg, req, targetlen, size);
 296         if (r < 0) {
 297                 fprintf(stderr, "Cannot prepare request! (%lu, %llu)\n",
 298                         (unsigned long)targetlen, (unsigned long long)size);
 299                 xseg_put_request(xseg, req, srcport);
 300                 return -1;
 301         }
 302
 303         req_target = xseg_get_target(xseg, req);
 304         strncpy(req_target, target, targetlen);
 305
 306         req_data = xseg_get_data(xseg, req);
 307         memcpy(req_data, buf, size);
 308         req->offset = offset;
 309         req->size = size;
 310         req->op = X_WRITE;
 311
 312         p = xseg_submit(xseg, req, srcport, X_ALLOC);
 313         if (p == NoPort) {
 314                 fprintf(stderr, "Cannot submit\n");
 315                 return -1;
 316         }
 317         xseg_signal(xseg, p);
 318
 319         return 0;
 320 }
 321
 322 int cmd_truncate(char *target, uint64_t offset)
 323 {
 324         return 0;
 325 }
 326
 327 int cmd_delete(char *target)
 328 {
 329         uint32_t targetlen = strlen(target);
 330         int r;
 331         struct xseg_request *req;
 332         init_local_signal();
 333         xseg_bind_port(xseg, srcport, NULL);
 334
 335         req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
 336         if (!req) {
 337                 fprintf(stderr, "No request!\n");
 338                 return -1;
 339         }
 340
 341         r = xseg_prep_request(xseg, req, targetlen, 0);
 342         if (r < 0) {
 343                 fprintf(stderr, "Cannot prepare request! (%lu, %lu)\n",
 344                         (unsigned long) targetlen, (unsigned long) req->bufferlen - targetlen);
 345                 xseg_put_request(xseg, req, srcport);
 346                 return -1;
 347         }
 348
 349         char *reqtarget = xseg_get_target(xseg, req);
 350         strncpy(reqtarget, target, targetlen);
 351         req->op = X_DELETE;
 352
 353         xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
 354         if (p == NoPort){
 355                 fprintf(stderr, "Couldn't submit request\n");
 356                 xseg_put_request(xseg, req, srcport);
 357                 return -1;
 358         }
 359
 360         xseg_signal(xseg, p);
 361
 362         return 0;
 363 }
 364
 365 int cmd_acquire(char *target)
 366 {
 367         return 0;
 368 }
 369
 370 int cmd_release(char *target)
 371 {
 372         return 0;
 373 }
 374
 375 int cmd_copy(char *src, char *dst)
 376 {
 377         return 0;
 378 }
 379
 380 int cmd_clone(char *src, char *dst)
 381 {
 382
 383         uint32_t targetlen = strlen(dst);
 384         uint32_t parentlen = strlen(src);
 385         struct xseg_request *req;
 386         struct xseg_request_clone *xclone;
 387         xseg_bind_port(xseg, srcport, NULL);
 388         req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
 389         if (!req) {
 390                 fprintf(stderr, "No request\n");
 391                 return -1;
 392         }
 393
 394         int r = xseg_prep_request(xseg, req, targetlen, sizeof(struct xseg_request_clone));
 395         if (r < 0) {
 396                 fprintf(stderr, "Cannot prepare request!\n");
 397                 xseg_put_request(xseg, req, srcport);
 398                 return -1;
 399         }
 400
 401         char *target = xseg_get_target(xseg, req);
 402         char *data = xseg_get_data(xseg, req);
 403
 404         strncpy(target, dst, targetlen);
 405         xclone = (struct xseg_request_clone *) data;
 406         strncpy(xclone->target, src, parentlen);
 407         xclone->targetlen = parentlen;
 408         xclone->size = -1;
 409         req->offset = 0;
 410         req->size = sizeof(struct xseg_request_clone);
 411         req->op = X_CLONE;
 412
 413         xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
 414         if (p == NoPort){
 415                 fprintf(stderr, "Cannot submit request\n");
 416                 return -1;
 417         }
 418         xseg_signal(xseg, p);
 419
 420         return 0;
 421 }
 422
 423 void log_req(int logfd, uint32_t portno2, uint32_t portno1, int op, int method,
 424                 struct xseg_request *req)
 425 {
 426         FILE *logfp;
 427         char target[64], data[64];
 428         char *req_target, *req_data;
 429         /* null terminate name in case of req->target is less than 63 characters,
 430          * and next character after name (aka first byte of next buffer) is not
 431          * null
 432          */
 433         unsigned int end = (req->targetlen > 63) ? 63 : req->targetlen;
 434
 435         req_target = xseg_get_target(xseg, req);
 436         req_data = xseg_get_data(xseg, req);
 437
 438         logfp = fdopen(logfd, "a");
 439         if (!logfp)
 440                 return;
 441
 442         switch(method) {
 443         case 0:
 444                 strncpy(target, req_target, end);
 445                 target[end] = 0;
 446                 strncpy(data, req_data, 63);
 447                 data[63] = 0;
 448
 449                 fprintf(logfp,
 450                         "src port: %u, dst port: %u,  op:%u offset: %llu size: %lu, reqstate: %u\n"
 451                         "target[%u]: '%s', data[%llu]:\n%s------------------\n\n",
 452                         (unsigned int)portno1,
 453                         (unsigned int)portno2,
 454                         (unsigned int)req->op,
 455                         (unsigned long long)req->offset,
 456                         (unsigned long)req->size,
 457                         (unsigned int)req->state,
 458                         (unsigned int)req->targetlen, target,
 459                         (unsigned long long)req->datalen, data);
 460                 break;
 461         case 1:
 462                 fprintf(logfp,
 463                         "src port: %u, dst port: %u, op: %u\n",
 464                         (unsigned int)portno1,
 465                         (unsigned int)portno2,
 466                         (unsigned int)req->op);
 467                 break;
 468         case 2:
 469                 fprintf(logfp, "src port: %u, dst port: %u, reqs: %llu\n",
 470                         (unsigned int)portno1,
 471                         (unsigned int)portno2,
 472                         (unsigned long long)++reqs);
 473         }
 474
 475         fclose(logfp);
 476         return;
 477 }
 478
 479 #define LOG_ACCEPT  0
 480 #define LOG_RECEIVE 1
 481
 482 int cmd_bridge(uint32_t portno1, uint32_t portno2, char *logfile, char *how)
 483 {
 484         struct xseg_request *req;
 485         int logfd, method;
 486         if (!strcmp(logfile, "-"))
 487                 logfd = 1;
 488         else {
 489                 logfd = open(logfile, O_WRONLY|O_APPEND|O_CREAT, 0600);
 490                 if (logfd < 0) {
 491                         perror(logfile);
 492                         return -1;
 493                 }
 494         }
 495
 496         if (!strcmp(how, "full"))
 497                 method = 0;
 498         else if (!strcmp(how, "summary"))
 499                 method = 1;
 500         else
 501                 method = 2;
 502
 503         for (;;) {
 504                 int reloop = 0, active;
 505                 xseg_prepare_wait(xseg, portno1);
 506                 xseg_prepare_wait(xseg, portno2);
 507                 req = NULL;
 508
 509                 for (;;) {
 510                         active = 0;
 511
 512                         //FIXME
 513                         req = xseg_accept(xseg, portno1);
 514                         if (req) {
 515                                 xseg_submit(xseg, req, portno2, X_ALLOC);
 516                                 log_req(logfd, portno1, portno2, LOG_ACCEPT, method, req);
 517                                 active += 1;
 518                         }
 519
 520                         req = xseg_accept(xseg, portno2);
 521                         if (req) {
 522                                 xseg_submit(xseg, req, portno1, X_ALLOC);
 523                                 log_req(logfd, portno2, portno1, LOG_ACCEPT, method, req);
 524                                 active += 1;
 525                         }
 526
 527                         req = xseg_receive(xseg, portno1);
 528                         if (req) {
 529                                 xseg_respond(xseg, req, portno2, X_ALLOC);
 530                                 log_req(logfd, portno1, portno2, LOG_RECEIVE, method, req);
 531                                 active += 1;
 532                         }
 533
 534                         req = xseg_receive(xseg, portno2);
 535                         if (req) {
 536                                 xseg_respond(xseg, req, portno1, X_ALLOC);
 537                                 log_req(logfd, portno2, portno1, LOG_RECEIVE, method, req);
 538                                 active += 1;
 539                         }
 540
 541                         if (active == 0) {
 542                                 if (reloop)
 543                                         break;
 544                                 /* wait on multiple queues? */
 545                                 xseg_wait_signal(xseg, 100000);
 546                                 break;
 547                         } else {
 548                                 xseg_cancel_wait(xseg, portno1);
 549                                 xseg_cancel_wait(xseg, portno2);
 550                                 reloop = 1;
 551                         }
 552                 }
 553         }
 554
 555         close(logfd);
 556
 557         return 0;
 558 }
 559
 560 int cmd_rndwrite(long loops, int32_t seed, uint32_t targetlen, uint32_t chunksize, uint64_t size)
 561 {
 562         if (loops < 0)
 563                 return help();
 564
 565         if (targetlen >= chunksize) {
 566                 fprintf(stderr, "targetlen >= chunksize\n");
 567                 return -1;
 568         }
 569
 570         char *p = realloc(namebuf, targetlen+1);
 571         if (!p) {
 572                 fprintf(stderr, "Cannot allocate memory\n");
 573                 return -1;
 574         }
 575         namebuf = p;
 576
 577         p = realloc(chunk, chunksize);
 578         if (!p) {
 579                 fprintf(stderr, "Cannot allocate memory\n");
 580                 return -1;
 581         }
 582         chunk = p;
 583         memset(chunk, 0, chunksize);
 584
 585         srandom(seed);
 586
 587         struct xseg_request *submitted = NULL, *received;
 588         long nr_submitted = 0, nr_received = 0, nr_failed = 0;
 589         int reported = 0, r;
 590         uint64_t offset;
 591         xport port;
 592         char *req_data, *req_target;
 593         seed = random();
 594         init_local_signal();
 595
 596         for (;;) {
 597                 xseg_prepare_wait(xseg, srcport);
 598                 if (nr_submitted < loops &&
 599                     (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
 600                         xseg_cancel_wait(xseg, srcport);
 601                         r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
 602                         if (r < 0) {
 603                                 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
 604                                         targetlen, chunksize);
 605                                 xseg_put_request(xseg, submitted, srcport);
 606                                 return -1;
 607                         }
 608
 609                         req_target = xseg_get_target(xseg, submitted);
 610                         req_data = xseg_get_data(xseg, submitted);
 611
 612                         reported = 0;
 613                         mkname(namebuf, targetlen, seed);
 614                         namebuf[targetlen] = 0;
 615                         //printf("%ld: %s\n", nr_submitted, namebuf);
 616                         strncpy(req_target, namebuf, targetlen);
 617                         offset = 0;// pick(size);
 618                         mkchunk(req_data, chunksize, namebuf, targetlen, offset);
 619
 620                         submitted->offset = offset;
 621                         submitted->size = chunksize;
 622                         submitted->op = X_WRITE;
 623                         submitted->flags |= XF_NOSYNC;
 624
 625                         port =  xseg_submit(xseg, submitted, srcport, X_ALLOC);
 626                         if (port == NoPort) {
 627                                 xseg_put_request(xseg, submitted, srcport);
 628                         } else {
 629                                 seed = random();
 630                                 nr_submitted += 1;
 631                                 xseg_signal(xseg, port);
 632                         }
 633                 }
 634
 635                 received = xseg_receive(xseg, srcport);
 636                 if (received) {
 637                         xseg_cancel_wait(xseg, srcport);
 638                         nr_received += 1;
 639                         if (!(received->state & XS_SERVED)) {
 640                                 nr_failed += 1;
 641                                 report_request(received);
 642                         }
 643                         if (xseg_put_request(xseg, received, srcport))
 644                                 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
 645                 }
 646
 647                 if (!submitted && !received)
 648                         xseg_wait_signal(xseg, 1000000);
 649
 650                         if (nr_submitted % 1000 == 0 && !reported) {
 651                                 reported = 1;
 652                                 fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
 653                                         nr_submitted, nr_received, nr_failed);
 654                         }
 655
 656                         if (nr_received >= loops)
 657                                 break;
 658         }
 659
 660         fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
 661                 nr_submitted, nr_received, nr_failed);
 662         return 0;
 663 }
 664
 665 /* note:
 666  * prepare/wait rhythm,
 667  * files are converted to independent chunk access patterns,
 668 */
 669
 670 int cmd_rndread(long loops, int32_t seed, uint32_t targetlen, uint32_t chunksize, uint64_t size)
 671 {
 672         if (loops < 0)
 673                 return help();
 674
 675         if (targetlen >= chunksize) {
 676                 fprintf(stderr, "targetlen >= chunksize\n");
 677                 return -1;
 678         }
 679
 680         char *p = realloc(namebuf, targetlen+1);
 681         if (!p) {
 682                 fprintf(stderr, "Cannot allocate memory\n");
 683                 return -1;
 684         }
 685         namebuf = p;
 686
 687         p = realloc(chunk, chunksize);
 688         if (!p) {
 689                 fprintf(stderr, "Cannot allocate memory\n");
 690                 return -1;
 691         }
 692         chunk = p;
 693         memset(chunk, 0, chunksize);
 694
 695         srandom(seed);
 696
 697         struct xseg_request *submitted = NULL, *received;
 698         long nr_submitted = 0, nr_received = 0, nr_failed = 0, nr_mismatch = 0;
 699         int reported = 0, r;
 700         uint64_t offset;
 701         xport port;
 702         char *req_data, *req_target;
 703         init_local_signal();
 704
 705         seed = random();
 706         for (;;) {
 707                 submitted = NULL;
 708                 xseg_prepare_wait(xseg, srcport);
 709                 if (nr_submitted < loops &&
 710                     (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
 711                         xseg_cancel_wait(xseg, srcport);
 712                         r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
 713                         if (r < 0) {
 714                                 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
 715                                         targetlen, chunksize);
 716                                 xseg_put_request(xseg, submitted, srcport);
 717                                 return -1;
 718                         }
 719
 720                         req_target = xseg_get_target(xseg, submitted);
 721                         reported = 0;
 722                         mkname(namebuf, targetlen, seed);
 723                         namebuf[targetlen] = 0;
 724                         //printf("%ld: %s\n", nr_submitted, namebuf);
 725                         offset = 0;//pick(size);
 726
 727                         strncpy(req_target, namebuf, targetlen);
 728                         submitted->offset = offset;
 729                         submitted->size = chunksize;
 730                         submitted->op = X_READ;
 731                         port = xseg_submit(xseg, submitted, srcport, X_ALLOC);
 732                         if (port == NoPort) {
 733                                 xseg_put_request(xseg, submitted, srcport);
 734                         } else {
 735                                 seed = random();
 736                                 nr_submitted += 1;
 737                                 xseg_signal(xseg, port);
 738                         }
 739                 }
 740
 741                 received = xseg_receive(xseg, srcport);
 742                 if (received) {
 743                         xseg_cancel_wait(xseg, srcport);
 744                         nr_received += 1;
 745                         req_target = xseg_get_target(xseg, received);
 746                         req_data = xseg_get_data(xseg, received);
 747                         if (!(received->state & XS_SERVED)) {
 748                                 nr_failed += 1;
 749                                 report_request(received);
 750                         } else if (!chkchunk(req_data, received->datalen,
 751                                         req_target, received->targetlen, received->offset)) {
 752         //                      report_request(received);
 753                                 nr_mismatch += 1;
 754                         }
 755
 756                         if (xseg_put_request(xseg, received, srcport))
 757                                 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
 758                 }
 759
 760                 if (!submitted && !received)
 761                         xseg_wait_signal(xseg, 1000000);
 762
 763                 if (nr_submitted % 1000 == 0 && !reported) {
 764                         reported = 1;
 765                         fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
 766                         nr_submitted, nr_received, nr_failed, nr_mismatch);
 767                 }
 768
 769                 if (nr_received >= loops)
 770                         break;
 771         }
 772
 773         fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
 774                 nr_submitted, nr_received, nr_failed, nr_mismatch);
 775         return 0;
 776 }
 777
 778 int cmd_submit_reqs(long loops, long concurrent_reqs, int op)
 779 {
 780         if (loops < 0)
 781                 return help();
 782
 783         struct xseg_request *submitted = NULL, *received;
 784         long nr_submitted = 0, nr_received = 0, nr_failed = 0, nr_mismatch = 0, nr_flying = 0;
 785         int reported = 0, r;
 786         uint64_t offset;
 787         uint32_t targetlen = 10, chunksize = 4096;
 788         struct timeval tv1, tv2;
 789         xport p;
 790         char *req_data, *req_target;
 791
 792         xseg_bind_port(xseg, srcport, NULL);
 793
 794         gettimeofday(&tv1, NULL);
 795         for (;;) {
 796                 submitted = NULL;
 797                 xseg_prepare_wait(xseg, srcport);
 798                 if (nr_submitted < loops &&  nr_flying < concurrent_reqs &&
 799                     (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
 800                         xseg_cancel_wait(xseg, srcport);
 801                         r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
 802                         if (r < 0) {
 803                                 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
 804                                         targetlen, chunksize);
 805                                 xseg_put_request(xseg, submitted, srcport);
 806                                 return -1;
 807                         }
 808
 809                         //FIXME
 810                         ++nr_flying;
 811                         nr_submitted += 1;
 812                         reported = 0;
 813                         offset = 0;//pick(size);
 814
 815                         submitted->offset = offset;
 816                         submitted->size = chunksize;
 817                         req_target = xseg_get_target(xseg, submitted);
 818                         req_data = xseg_get_data(xseg, submitted);
 819
 820                         if (op == 0)
 821                                 submitted->op = X_INFO;
 822                         else if (op == 1)
 823                                 submitted->op = X_READ;
 824                         else if (op == 2) {
 825                                 submitted->op = X_WRITE;
 826                                 mkchunk(req_data, submitted->datalen, req_target, submitted->targetlen, submitted->offset);
 827                         }
 828
 829                         p = xseg_submit(xseg, submitted, srcport, X_ALLOC);
 830                         if ( p != NoPort){
 831                                 if (xseg_signal(xseg, p) < 0)
 832                                         perror("Cannot signal peer");
 833                         }
 834                 }
 835                 received = xseg_receive(xseg, srcport);
 836                 if (received) {
 837                         xseg_cancel_wait(xseg, srcport);
 838                         --nr_flying;
 839                         if (nr_received == 0)
 840                                 fprintf(stderr, "latency (time for the first req to complete): %llu usecs\n",
 841                                         (unsigned long long)received->elapsed);
 842                         nr_received += 1;
 843                         if (!(received->state & XS_SERVED)) {
 844                                 nr_failed += 1;
 845                                 //report_request(received);
 846                         }
 847
 848                         if (xseg_put_request(xseg, received, srcport))
 849                                 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
 850                 }
 851
 852                 if (!submitted && !received)
 853                         xseg_wait_signal(xseg, 10000000L);
 854
 855                 if (nr_received >= loops)
 856                         break;
 857         }
 858         gettimeofday(&tv2, NULL);
 859
 860         fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
 861                 nr_submitted, nr_received, nr_failed, nr_mismatch);
 862         long t = (tv2.tv_sec - tv1.tv_sec)*1000000 + (tv2.tv_usec - tv1.tv_usec);
 863         fprintf(stderr, "elpased time: %lf secs, throughput: %lf reqs/sec\n", (double) t / 1000000.0, (double) nr_submitted / (t / 1000000.0));
 864
 865         return 0;
 866 }
 867
 868 int cmd_report(uint32_t portno)
 869 {
 870         struct xseg_port *port = xseg_get_port(xseg, portno);
 871         if (!port) {
 872                 printf("port %u is not assigned\n", portno);
 873                 return 0;
 874         }
 875         struct xq *fq, *rq, *pq;
 876         fq = xseg_get_queue(xseg, port, free_queue);
 877         rq = xseg_get_queue(xseg, port, request_queue);
 878         pq = xseg_get_queue(xseg, port, reply_queue);
 879         fprintf(stderr, "port %u:\n"
 880                 "   requests: %llu/%llu  src gw: %u  dst gw: %u\n"
 881                 "       free_queue [%p] count : %u\n"
 882                 "    request_queue [%p] count : %u\n"
 883                 "      reply_queue [%p] count : %u\n",
 884                 portno, port->alloc_reqs, port->max_alloc_reqs,
 885                 xseg->src_gw[portno], xseg->dst_gw[portno],
 886                 (void *)fq, xq_count(fq),
 887                 (void *)rq, xq_count(rq),
 888                 (void *)pq, xq_count(pq));
 889         return 0;
 890 }
 891
 892 int cmd_join(void)
 893 {
 894         if (xseg)
 895                 return 0;
 896
 897         xseg = xseg_join(cfg.type, cfg.name, "posix", NULL);
 898         if (!xseg) {
 899                 fprintf(stderr, "cannot join segment!\n");
 900                 return -1;
 901         }
 902         return 0;
 903 }
 904
 905 int cmd_reportall(void)
 906 {
 907         uint32_t t;
 908
 909         if (cmd_join())
 910                 return -1;
 911
 912
 913         fprintf(stderr, "Heap usage: %llu / %llu\n", xseg->heap->cur, xseg->config.heap_size);
 914         for (t = 0; t < xseg->config.nr_ports; t++)
 915                 cmd_report(t);
 916
 917         return 0;
 918 }
 919
 920 int cmd_create(void)
 921 {
 922         int r = xseg_create(&cfg);
 923         if (r) {
 924                 fprintf(stderr, "cannot create segment!\n");
 925                 return -1;
 926         }
 927
 928         fprintf(stderr, "Segment initialized.\n");
 929         return 0;
 930 }
 931
 932 int cmd_destroy(void)
 933 {
 934         if (!xseg && cmd_join())
 935                 return -1;
 936         xseg_leave(xseg);
 937         xseg_destroy(xseg);
 938         xseg = NULL;
 939         fprintf(stderr, "Segment destroyed.\n");
 940         return 0;
 941 }
 942
 943 int cmd_alloc_requests(unsigned long nr)
 944 {
 945         return xseg_alloc_requests(xseg, srcport, nr);
 946 }
 947
 948 int cmd_free_requests(unsigned long nr)
 949 {
 950         return xseg_free_requests(xseg, srcport, nr);
 951 }
 952
 953 int cmd_put_requests(void)
 954 {
 955         struct xseg_request *req;
 956
 957         for (;;) {
 958                 req = xseg_accept(xseg, dstport);
 959                 if (!req)
 960                         break;
 961                 if (xseg_put_request(xseg, req, srcport))
 962                         fprintf(stderr, "Cannot put request at port %u\n", req->src_portno);
 963         }
 964
 965         return 0;
 966 }
 967
 968 int cmd_finish(unsigned long nr, int fail)
 969 {
 970         struct xseg_request *req;
 971         char *buf = malloc(sizeof(char) * 8128);
 972         char *req_target, *req_data;
 973         xseg_bind_port(xseg, srcport, NULL);
 974         xport p;
 975
 976         for (; nr--;) {
 977                 xseg_prepare_wait(xseg, srcport);
 978                 req = xseg_accept(xseg, srcport);
 979                 if (req) {
 980                         req_target = xseg_get_target(xseg, req);
 981                         req_data = xseg_get_data(xseg, req);
 982                         xseg_cancel_wait(xseg, srcport);
 983                         if (fail == 1)
 984                                 req->state &= ~XS_SERVED;
 985                         else {
 986                                 if (req->op == X_READ)
 987                                         mkchunk(req_data, req->datalen, req_target, req->targetlen, req->offset);
 988                                 else if (req->op == X_WRITE)
 989                                         memcpy(buf, req_data, (sizeof(*buf) > req->datalen) ? req->datalen : sizeof(*buf));
 990                                 else if (req->op == X_INFO)
 991                                         *((uint64_t *) req->data) = 4294967296;
 992
 993                                 req->state |= XS_SERVED;
 994                                 req->serviced = req->size;
 995                         }
 996
 997                         p = xseg_respond(xseg, req, srcport, X_ALLOC);
 998                         xseg_signal(xseg, p);
 999                         continue;
1000                 }
1001                 ++nr;
1002                 xseg_wait_signal(xseg, 10000000L);
1003         }
1004
1005         free(buf);
1006
1007         return 0;
1008 }
1009
1010 void handle_reply(struct xseg_request *req)
1011 {
1012         char *req_data = xseg_get_data(xseg, req);
1013         char *req_target = xseg_get_target(xseg, req);
1014         if (!(req->state & XS_SERVED)) {
1015                 report_request(req);
1016                 goto put;
1017         }
1018
1019         switch (req->op) {
1020         case X_READ:
1021                 fwrite(req_data, 1, req->datalen, stdout);
1022                 break;
1023
1024         case X_WRITE:
1025                 fprintf(stdout, "wrote: ");
1026                 fwrite(req_data, 1, req->datalen, stdout);
1027                 break;
1028         case X_SYNC:
1029         case X_DELETE:
1030                 fprintf(stderr, "deleted %s\n", req_target);
1031                 break;
1032         case X_TRUNCATE:
1033         case X_COMMIT:
1034         case X_CLONE:
1035                 fprintf(stderr, "cloned %s\n", ((struct xseg_request_clone *)req_data)->target);
1036                 break;
1037         case X_INFO:
1038                 fprintf(stderr, "size: %llu\n", (unsigned long long)*((uint64_t *)req_data));
1039                 break;
1040
1041         default:
1042                 break;
1043         }
1044
1045 put:
1046         if (xseg_put_request(xseg, req, srcport))
1047                 fprintf(stderr, "Cannot put reply at port %u\n", req->src_portno);
1048 }
1049
1050 int cmd_wait(uint32_t nr)
1051 {
1052         struct xseg_request *req;
1053         long ret;
1054         init_local_signal();
1055
1056         for (;;) {
1057                 req = xseg_receive(xseg, srcport);
1058                 if (req) {
1059                         handle_reply(req);
1060                         nr--;
1061                         if (nr == 0)
1062                                 break;
1063                         continue;
1064                 }
1065
1066                 ret = xseg_prepare_wait(xseg, srcport);
1067                 if (ret)
1068                         return -1;
1069
1070                 ret = xseg_wait_signal(xseg, 1000000);
1071                 ret = xseg_cancel_wait(xseg, srcport);
1072                 if (ret)
1073                         return -1;
1074         }
1075
1076         return 0;
1077 }
1078
1079 int cmd_put_replies(void)
1080 {
1081         struct xseg_request *req;
1082
1083         for (;;) {
1084                 req = xseg_receive(xseg, dstport);
1085                 if (!req)
1086                         break;
1087                 fprintf(stderr, "request: %08llx%08llx\n"
1088                         "     op: %u\n"
1089                         "  state: %u\n",
1090                         0LL, (unsigned long long)req->serial,
1091                         req->op,
1092                         req->state);
1093                 report_request(req);
1094
1095                 //fwrite(req->buffer, 1, req->bufferlen, stdout);
1096
1097                 if (xseg_put_request(xseg, req, srcport))
1098                         fprintf(stderr, "Cannot put reply\n");
1099         }
1100
1101         return 0;
1102 }
1103
1104 int cmd_bind(long portno)
1105 {
1106         struct xseg_port *port = xseg_bind_port(xseg, portno, NULL);
1107         if (!port) {
1108                 fprintf(stderr, "failed to bind port %ld\n", portno);
1109                 return 1;
1110         }
1111
1112         fprintf(stderr, "bound port %u\n", xseg_portno(xseg, port));
1113         return 0;
1114 }
1115
1116 int cmd_signal(uint32_t portno)
1117 {
1118         return xseg_signal(xseg, portno);
1119 }
1120
1121 int parse_ports(char *str)
1122 {
1123         int ret = 0;
1124         char *s = str;
1125
1126         for (;;) {
1127                 if (*s == 0)
1128                         return 0;
1129
1130                 if (*s == ':') {
1131                         *s = 0;
1132                         if ((s > str) && isdigit(str[0])) {
1133                                 srcport = atol(str);
1134                                 ret ++;
1135                         }
1136                         break;
1137                 }
1138                 s ++;
1139         }
1140
1141         s += 1;
1142         str = s;
1143
1144         for (;;) {
1145                 if (*s == 0) {
1146                         if ((s > str) && isdigit(str[0])) {
1147                                 dstport = atol(str);
1148                                 ret ++;
1149                         }
1150                         break;
1151                 }
1152                 s ++;
1153         }
1154
1155         return ret;
1156 }
1157
1158 int main(int argc, char **argv)
1159 {
1160         int i, ret = 0;
1161         char *spec;
1162
1163         if (argc < 3)
1164                 return help();
1165
1166         srcport = -1;
1167         dstport = -1;
1168         spec = argv[1];
1169
1170         if (xseg_parse_spec(spec, &cfg)) {
1171                 fprintf(stderr, "Cannot parse spec\n");
1172                 return -1;
1173         }
1174
1175         if (xseg_initialize()) {
1176                 fprintf(stderr, "cannot initialize!\n");
1177                 return -1;
1178         }
1179
1180         for (i = 2; i < argc; i++) {
1181
1182                 if (!strcmp(argv[i], "create")) {
1183                         ret = cmd_create();
1184                         continue;
1185                 }
1186
1187                 if (!strcmp(argv[i], "join")) {
1188                         ret = cmd_join();
1189                         if (!ret)
1190                                 fprintf(stderr, "Segment joined.\n");
1191                         continue;
1192                 }
1193
1194                 if (!strcmp(argv[i], "destroy")) {
1195                         ret = cmd_destroy();
1196                         continue;
1197                 }
1198
1199                 if (cmd_join())
1200                         return -1;
1201
1202                 if (!strcmp(argv[i], "reportall")) {
1203                         ret = cmd_reportall();
1204                         continue;
1205                 }
1206
1207                 if (!strcmp(argv[i], "bind") && (i + 1 < argc)) {
1208                         ret = cmd_bind(atol(argv[i+1]));
1209                         i += 1;
1210                         continue;
1211                 }
1212
1213                 if (!strcmp(argv[i], "signal") && (i + 1 < argc)) {
1214                         ret = cmd_signal(atol(argv[i+1]));
1215                         i += 1;
1216                         continue;
1217                 }
1218
1219                 if (!strcmp(argv[i], "bridge") && (i + 4 < argc)) {
1220                         ret = cmd_bridge(atol(argv[i+1]),
1221                                          atol(argv[i+2]),
1222                                          argv[i+3],
1223                                          argv[i+4]);
1224                         i += 4;
1225                         continue;
1226                 }
1227
1228                 if (srcport == -1) {
1229                         if (!parse_ports(argv[i]))
1230                                 fprintf(stderr, "source port undefined: %s\n", argv[i]);
1231                         continue;
1232                 }
1233
1234                 if (dstport == -1) {
1235                         if (!parse_ports(argv[i]))
1236                                 fprintf(stderr, "destination port undefined: %s\n", argv[i]);
1237                         continue;
1238                 }
1239
1240                 if (!strcmp(argv[i], "report")) {
1241                         ret = cmd_report(dstport);
1242                         continue;
1243                 }
1244
1245                 if (!strcmp(argv[i], "alloc_requests") && (i + 1 < argc)) {
1246                         ret = cmd_alloc_requests(atol(argv[i+1]));
1247                         i += 1;
1248                         continue;
1249                 }
1250
1251                 if (!strcmp(argv[i], "free_requests") && (i + 1 < argc)) {
1252                         ret = cmd_free_requests(atol(argv[i+1]));
1253                         i += 1;
1254                         continue;
1255                 }
1256
1257                 if (!strcmp(argv[i], "put_requests")) {
1258                         ret = cmd_put_requests();
1259                         continue;
1260                 }
1261
1262                 if (!strcmp(argv[i], "put_replies")) {
1263                         ret = cmd_put_replies();
1264                         continue;
1265                 }
1266
1267                 if (!strcmp(argv[i], "complete") && (i + 1 < argc)) {
1268                         ret = cmd_finish(atol(argv[i+1]), 0);
1269                         i += 1;
1270                         continue;
1271                 }
1272
1273                 if (!strcmp(argv[i], "fail") && (i + 1 < argc)) {
1274                         ret = cmd_finish(atol(argv[i+1]), 1);
1275                         i += 1;
1276                         continue;
1277                 }
1278
1279                 if (!strcmp(argv[i], "wait") && (i + 1 < argc)) {
1280                         ret = cmd_wait(atol(argv[i+1]));
1281                         i += 1;
1282                         continue;
1283                 }
1284
1285                 if (!strcmp(argv[i], "rndwrite") && (i + 5 < argc)) {
1286                         long nr_loops = atol(argv[i+1]);
1287                         unsigned int seed = atoi(argv[i+2]);
1288                         unsigned int targetlen = atoi(argv[i+3]);
1289                         unsigned int chunksize = atoi(argv[i+4]);
1290                         unsigned long objectsize = atol(argv[i+5]);
1291                         ret = cmd_rndwrite(nr_loops, seed, targetlen, chunksize, objectsize);
1292                         i += 5;
1293                         continue;
1294                 }
1295
1296                 if (!strcmp(argv[i], "rndread") && (i + 5 < argc)) {
1297                         long nr_loops = atol(argv[i+1]);
1298                         unsigned int seed = atoi(argv[i+2]);
1299                         unsigned int targetlen = atoi(argv[i+3]);
1300                         unsigned int chunksize = atoi(argv[i+4]);
1301                         unsigned long objectsize = atol(argv[i+5]);
1302                         ret = cmd_rndread(nr_loops, seed, targetlen, chunksize, objectsize);
1303                         i += 5;
1304                         continue;
1305                 }
1306
1307                 if (!strcmp(argv[i], "submit_reqs") && (i + 3 < argc)) {
1308                         long nr_loops = atol(argv[i+1]);
1309                         long concurrent_reqs = atol(argv[i+2]);
1310                         int op = atoi(argv[i+3]);
1311                         ret = cmd_submit_reqs(nr_loops, concurrent_reqs, op);
1312                         i += 3;
1313                         continue;
1314                 }
1315
1316                 if (!strcmp(argv[i], "read") && (i + 3 < argc)) {
1317                         char *target = argv[i+1];
1318                         uint64_t offset = atol(argv[i+2]);
1319                         uint64_t size   = atol(argv[i+3]);
1320                         ret = cmd_read(target, offset, size);
1321                         i += 3;
1322                         continue;
1323                 }
1324
1325                 if (!strcmp(argv[i], "write") && (i + 2 < argc)) {
1326                         char *target = argv[i+1];
1327                         uint64_t offset = atol(argv[i+2]);
1328                         ret = cmd_write(target, offset);
1329                         i += 2;
1330                         continue;
1331                 }
1332
1333                 if (!strcmp(argv[i], "truncate") && (i + 2 < argc)) {
1334                         char *target = argv[i+1];
1335                         uint64_t offset = atol(argv[i+2]);
1336                         ret = cmd_truncate(target, offset);
1337                         i += 2;
1338                         continue;
1339                 }
1340
1341                 if (!strcmp(argv[i], "delete") && (i + 1 < argc)) {
1342                         char *target = argv[i+1];
1343                         ret = cmd_delete(target);
1344                         i += 1;
1345                         continue;
1346                 }
1347
1348                 if (!strcmp(argv[i], "acquire") && (i + 1 < argc)) {
1349                         char *target = argv[i+1];
1350                         ret = cmd_acquire(target);
1351                         i += 1;
1352                         continue;
1353                 }
1354
1355                 if (!strcmp(argv[i], "release") && (i + 1 < argc)) {
1356                         char *target = argv[i+1];
1357                         ret = cmd_release(target);
1358                         i += 1;
1359                         continue;
1360                 }
1361
1362                 if (!strcmp(argv[i], "copy") && (i + 2) < argc) {
1363                         char *src = argv[i+1];
1364                         char *dst = argv[i+2];
1365                         ret = cmd_copy(src, dst);
1366                         i += 2;
1367                         continue;
1368                 }
1369
1370                 if (!strcmp(argv[i], "clone") && (i + 2 < argc)) {
1371                         char *src = argv[i+1];
1372                         char *dst = argv[i+2];
1373                         ret = cmd_clone(src, dst);
1374                         i += 2;
1375                         continue;
1376                 }
1377
1378                 if (!strcmp(argv[i], "info") && (i + 1 < argc)) {
1379                         char *target = argv[i+1];
1380                         ret = cmd_info(target);
1381                         i += 1;
1382                         continue;
1383                 }
1384
1385
1386                 if (!parse_ports(argv[i]))
1387                         fprintf(stderr, "invalid argument: %s\n", argv[i]);
1388         }
1389
1390         /* xseg_leave(); */
1391         return ret;
1392 }