12 #include <xseg/xseg.h>
16 printf("xseg <spec> [[[<src_port>]:[<dst_port>]] [<command> <arg>]* ]*\n"
18 " <type:name:nr_ports:nr_requests:request_size:extra_size:page_shift>\n"
25 " bridge <portno1> <portno2> <logfile> {full|summary|stats}\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> <namesize> <datasize> <objectsize>\n"
36 " rndread <nr_loops> <seed> <namesize> <datasize> <objectsize>\n"
38 " read <name> <offset> <size>\n"
39 " write <name> <offset> < data\n"
40 " truncate <name> <size>\n"
45 " clone <src> <dst>\n"
52 struct xseg_config cfg;
54 uint32_t srcport, dstport;
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
62 void mkname_heavy(char *name, uint32_t namesize, uint32_t seed)
66 for (i = 0; i < namesize; i += 1) {
67 c = seed + (seed >> 8) + (seed >> 16) + (seed >> 24);
68 c = '0' + ((c + (c >> 4)) & 0xf);
72 seed *= ((seed % 137911) | 1) * 137911;
76 void mkname_light(char *name, uint32_t namesize, uint32_t seed)
80 for (i = 0; i < namesize; i += 1) {
82 name[i] = 'A' + (c & 0xf);
87 uint64_t pick(uint64_t size)
89 return (uint64_t)((double)(RAND_MAX) / random());
92 void mkchunk( char *chunk, uint32_t datasize,
93 char *name, uint32_t namesize, uint64_t offset)
95 long i, r, bufsize = namesize + 16;
97 r = datasize % bufsize;
98 snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, name);
100 for (i = 0; i <= (long)datasize - bufsize; i += bufsize)
101 memcpy(chunk + i, buf, bufsize);
103 memcpy(chunk + datasize - r, buf, r);
106 int chkchunk( char *chunk, uint32_t datasize,
107 char *name, uint32_t namesize, uint64_t offset)
110 int bufsize = namesize + 16;
112 r = datasize % namesize;
113 snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, name);
115 for (i = 0; i <= (long)datasize - bufsize; i += bufsize)
116 if (memcmp(chunk + i, buf, bufsize)) {
117 /*printf("mismatch: '%*s'* vs '%*s'\n",
118 bufsize, buf, datasize, chunk);
123 if (memcmp(chunk + datasize - r, buf, r))
130 #define ALLOC_MIN 4096
131 #define ALLOC_MAX 1048576
133 void inputbuf(FILE *fp, char **retbuf, uint64_t *retsize)
135 static uint64_t alloc_size;
141 if (alloc_size < ALLOC_MIN)
142 alloc_size = ALLOC_MIN;
144 if (alloc_size > ALLOC_MAX)
145 alloc_size = ALLOC_MAX;
147 p = realloc(buf, alloc_size);
158 r = fread(buf + size, 1, alloc_size - size, fp);
162 if (size >= alloc_size) {
163 p = realloc(buf, alloc_size * 2);
181 void report_request(struct xseg_request *req)
183 uint32_t max = req->datasize;
186 req->data[max-1] = 0;
187 printf("request %llu state %u\n", (unsigned long long)req->serial, req->state);
188 printf("data: %s\n", req->data);
191 int cmd_info(char *name)
196 int cmd_read(char *name, uint64_t offset, uint64_t size)
198 uint32_t namesize = strlen(name);
201 struct xseg_request *req = xseg_get_request(xseg, srcport);
203 printf("No request\n");
207 r = xseg_prep_request(req, namesize, size);
209 printf("Cannot prepare request! (%lu, %llu)\n",
210 (unsigned long)namesize, (unsigned long long)size);
211 xseg_put_request(xseg, srcport, req);
215 strncpy(req->name, name, namesize);
216 req->offset = offset;
220 srl = xseg_submit(xseg, dstport, req);
224 xseg_signal(xseg, dstport);
228 int cmd_write(char *name, uint64_t offset)
234 uint32_t namesize = strlen(name);
235 struct xseg_request *req;
237 inputbuf(stdin, &buf, &size);
239 printf("No input\n");
243 req = xseg_get_request(xseg, srcport);
245 printf("No request\n");
249 r = xseg_prep_request(req, namesize, size);
251 printf("Cannot prepare request! (%lu, %llu)\n",
252 (unsigned long)namesize, (unsigned long long)size);
253 xseg_put_request(xseg, srcport, req);
257 strncpy(req->name, name, namesize);
258 memcpy(req->buffer, buf, size);
259 req->offset = offset;
263 srl = xseg_submit(xseg, dstport, req);
265 printf("Cannot submit\n");
272 int cmd_truncate(char *name, uint64_t offset)
277 int cmd_delete(char *name)
282 int cmd_acquire(char *name)
287 int cmd_release(char *name)
292 int cmd_copy(char *src, char *dst)
297 int cmd_clone(char *src, char *dst)
302 void log_req( uint32_t portno2, uint32_t portno1, int op, int method,
303 struct xseg_request *req)
309 #define LOG_RECEIVE 1
311 int cmd_bridge(uint32_t portno1, uint32_t portno2, char *logfile, char *how)
313 struct xseg_request *req;
315 if (!strcmp(logfile, "-"))
318 logfd = open(logfile, O_WRONLY|O_APPEND|O_CREAT, 0600);
325 if (!strcmp(how, "full"))
327 else if (!strcmp(how, "full"))
333 int reloop = 0, active;
334 xseg_prepare_wait(xseg, portno1);
335 xseg_prepare_wait(xseg, portno2);
341 req = xseg_accept(xseg, portno1);
343 xseg_submit(xseg, portno2, req);
344 log_req(portno1, portno2, LOG_ACCEPT, method, req);
348 req = xseg_accept(xseg, portno2);
350 xseg_submit(xseg, portno1, req);
351 log_req(portno2, portno1, LOG_ACCEPT, method, req);
355 req = xseg_receive(xseg, portno1);
357 xseg_respond(xseg, portno2, req);
358 log_req(portno1, portno2, LOG_RECEIVE, method, req);
362 req = xseg_receive(xseg, portno2);
364 xseg_respond(xseg, portno1, req);
365 log_req(portno2, portno1, LOG_RECEIVE, method, req);
372 /* wait on multiple queues? */
373 xseg_wait_signal(xseg, portno1, 100000);
376 xseg_cancel_wait(xseg, portno1);
377 xseg_cancel_wait(xseg, portno2);
386 int cmd_rndwrite(long loops, int32_t seed, uint32_t namesize, uint32_t chunksize, uint64_t size)
391 if (namesize >= chunksize) {
392 printf("namesize >= chunksize\n");
396 char *p = realloc(namebuf, namesize+1);
398 printf("Cannot allocate memory\n");
403 p = realloc(chunk, chunksize);
405 printf("Cannot allocate memory\n");
409 memset(chunk, 0, chunksize);
413 struct xseg_request *submitted = NULL, *received;
414 long nr_submitted = 0, nr_received = 0, nr_failed = 0;
420 xseg_prepare_wait(xseg, srcport);
421 if (nr_submitted < loops &&
422 (submitted = xseg_get_request(xseg, srcport))) {
423 xseg_cancel_wait(xseg, srcport);
424 r = xseg_prep_request(submitted, namesize, chunksize);
426 printf("Cannot prepare request! (%u, %u)\n",
427 namesize, chunksize);
428 xseg_put_request(xseg, submitted->portno, submitted);
435 mkname(namebuf, namesize, seed);
436 namebuf[namesize] = 0;
437 //printf("%ld: %s\n", nr_submitted, namebuf);
438 strncpy(submitted->name, namebuf, namesize);
439 offset = 0;// pick(size);
440 mkchunk(submitted->buffer, chunksize, namebuf, namesize, offset);
442 submitted->offset = offset;
443 submitted->size = chunksize;
444 submitted->op = X_WRITE;
445 submitted->flags |= XF_NOSYNC;
447 srl = xseg_submit(xseg, dstport, submitted);
449 xseg_signal(xseg, dstport);
452 received = xseg_receive(xseg, srcport);
454 xseg_cancel_wait(xseg, srcport);
456 if (!(received->state & XS_SERVED)) {
458 report_request(received);
460 if (xseg_put_request(xseg, received->portno, received))
461 printf("Cannot put request at port %u\n", received->portno);
464 if (!submitted && !received)
465 xseg_wait_signal(xseg, srcport, 1000000);
467 if (nr_submitted % 1000 == 0 && !reported) {
469 printf("submitted %ld, received %ld, failed %ld\n",
470 nr_submitted, nr_received, nr_failed);
473 if (nr_received >= loops)
477 printf("submitted %ld, received %ld, failed %ld\n",
478 nr_submitted, nr_received, nr_failed);
483 * prepare/wait rhythm,
484 * files are converted to independent chunk access patterns,
487 int cmd_rndread(long loops, int32_t seed, uint32_t namesize, uint32_t chunksize, uint64_t size)
492 if (namesize >= chunksize) {
493 printf("namesize >= chunksize\n");
497 char *p = realloc(namebuf, namesize+1);
499 printf("Cannot allocate memory\n");
504 p = realloc(chunk, chunksize);
506 printf("Cannot allocate memory\n");
510 memset(chunk, 0, chunksize);
514 struct xseg_request *submitted = NULL, *received;
515 long nr_submitted = 0, nr_received = 0, nr_failed = 0, nr_mismatch = 0;
522 xseg_prepare_wait(xseg, srcport);
523 if (nr_submitted < loops &&
524 (submitted = xseg_get_request(xseg, srcport))) {
525 xseg_cancel_wait(xseg, srcport);
526 r = xseg_prep_request(submitted, namesize, chunksize);
528 printf("Cannot prepare request! (%u, %u)\n",
529 namesize, chunksize);
530 xseg_put_request(xseg, submitted->portno, submitted);
537 mkname(namebuf, namesize, seed);
538 namebuf[namesize] = 0;
539 //printf("%ld: %s\n", nr_submitted, namebuf);
540 offset = 0;//pick(size);
542 strncpy(submitted->name, namebuf, namesize);
543 submitted->offset = offset;
544 submitted->size = chunksize;
545 submitted->op = X_READ;
547 srl = xseg_submit(xseg, dstport, submitted);
549 xseg_signal(xseg, dstport);
552 received = xseg_receive(xseg, srcport);
554 xseg_cancel_wait(xseg, srcport);
556 if (!(received->state & XS_SERVED)) {
558 report_request(received);
559 } else if (!chkchunk(received->data, received->datasize,
560 received->name, received->namesize, received->offset)) {
564 if (xseg_put_request(xseg, received->portno, received))
565 printf("Cannot put request at port %u\n", received->portno);
568 if (!submitted && !received)
569 xseg_wait_signal(xseg, srcport, 1000000);
571 if (nr_submitted % 1000 == 0 && !reported) {
573 printf("submitted %ld, received %ld, failed %ld, mismatched %ld\n",
574 nr_submitted, nr_received, nr_failed, nr_mismatch);
577 if (nr_received >= loops)
581 printf("submitted %ld, received %ld, failed %ld, mismatched %ld\n",
582 nr_submitted, nr_received, nr_failed, nr_mismatch);
586 int cmd_report(uint32_t port)
588 struct xq *fq, *rq, *pq;
589 fq = &xseg->ports[port].free_queue;
590 rq = &xseg->ports[port].request_queue;
591 pq = &xseg->ports[port].reply_queue;
593 " free_queue [%p] count : %u\n"
594 " request_queue [%p] count : %u\n"
595 " reply_queue [%p] count : %u\n",
597 (void *)fq, xq_count(fq),
598 (void *)rq, xq_count(rq),
599 (void *)pq, xq_count(pq));
608 xseg = xseg_join(cfg.type, cfg.name);
610 printf("cannot join segment!\n");
616 int cmd_reportall(void)
623 printf("global free requests: %u\n", xq_count(xseg->free_requests));
624 for (t = 0; t < xseg->config.nr_ports; t++)
632 int r = xseg_create(&cfg);
634 printf("cannot create segment!\n");
638 printf("Segment initialized.\n");
642 int cmd_destroy(void)
644 if (!xseg && cmd_join())
648 printf("Segment destroyed.\n");
652 int cmd_alloc_requests(unsigned long nr)
654 return xseg_alloc_requests(xseg, srcport, nr);
657 int cmd_free_requests(unsigned long nr)
659 return xseg_free_requests(xseg, srcport, nr);
662 int cmd_put_requests(void)
664 struct xseg_request *req;
667 req = xseg_accept(xseg, dstport);
670 if (xseg_put_request(xseg, req->portno, req))
671 printf("Cannot put request at port %u\n", req->portno);
677 int cmd_finish(unsigned long nr, int fail)
679 struct xseg_request *req;
682 req = xseg_accept(xseg, srcport);
686 req->state &= ~XS_SERVED;
688 req->state |= XS_SERVED;
689 xseg_respond(xseg, dstport, req);
690 xseg_signal(xseg, dstport);
696 void handle_reply(struct xseg_request *req)
698 if (!(req->state & XS_SERVED)) {
705 fwrite(req->data, 1, req->datasize, stdout);
719 if (xseg_put_request(xseg, req->portno, req))
720 fprintf(stderr, "Cannot put reply at port %u\n", req->portno);
723 int cmd_wait(uint32_t nr)
725 struct xseg_request *req;
729 req = xseg_receive(xseg, srcport);
738 ret = xseg_prepare_wait(xseg, srcport);
742 ret = xseg_wait_signal(xseg, srcport, 1000000);
743 ret = xseg_cancel_wait(xseg, srcport);
751 int cmd_put_replies(void)
753 struct xseg_request *req;
756 req = xseg_receive(xseg, dstport);
759 printf("request: %08llx%08llx\n"
762 0LL, (unsigned long long)req->serial,
767 //fwrite(req->buffer, 1, req->buffersize, stdout);
769 if (xseg_put_request(xseg, req->portno, req))
770 fprintf(stderr, "Cannot put reply\n");
776 int cmd_bind(long portno)
778 struct xseg_port *port = xseg_bind_port(xseg, portno);
780 printf("failed to bind port %ld\n", portno);
784 printf("bound port %u\n", xseg_portno(xseg, port));
788 int cmd_signal(uint32_t portno)
790 return xseg_signal(xseg, portno);
793 int parse_ports(char *str)
804 if ((s > str) && isdigit(str[0])) {
818 if ((s > str) && isdigit(str[0])) {
830 int main(int argc, char **argv) {
842 if (xseg_parse_spec(spec, &cfg)) {
843 printf("Cannot parse spec\n");
847 if (xseg_initialize("posix")) {
848 printf("cannot initialize!\n");
852 for (i = 2; i < argc; i++) {
854 if (!strcmp(argv[i], "create")) {
859 if (!strcmp(argv[i], "join")) {
862 printf("Segment joined.\n");
866 if (!strcmp(argv[i], "destroy")) {
874 if (!strcmp(argv[i], "reportall")) {
875 ret = cmd_reportall();
879 if (!strcmp(argv[i], "bind") && (i + 1 < argc)) {
880 ret = cmd_bind(atol(argv[i+1]));
885 if (!strcmp(argv[i], "signal") && (i + 1 < argc)) {
886 ret = cmd_signal(atol(argv[i+1]));
891 if (!strcmp(argv[i], "bridge") && (i + 4 < argc)) {
892 ret = cmd_bridge(atol(argv[i+1]),
901 if (!parse_ports(argv[i]))
902 printf("source port undefined: %s\n", argv[i]);
907 if (!parse_ports(argv[i]))
908 printf("destination port undefined: %s\n", argv[i]);
912 if (!strcmp(argv[i], "report")) {
913 ret = cmd_report(dstport);
917 if (!strcmp(argv[i], "alloc_requests") && (i + 1 < argc)) {
918 ret = cmd_alloc_requests(atol(argv[i+1]));
923 if (!strcmp(argv[i], "free_requests") && (i + 1 < argc)) {
924 ret = cmd_free_requests(atol(argv[i+1]));
929 if (!strcmp(argv[i], "put_requests")) {
930 ret = cmd_put_requests();
934 if (!strcmp(argv[i], "put_replies")) {
935 ret = cmd_put_replies();
939 if (!strcmp(argv[i], "complete") && (i + 1 < argc)) {
940 ret = cmd_finish(atol(argv[i+1]), 0);
945 if (!strcmp(argv[i], "fail") && (i + 1 < argc)) {
946 ret = cmd_finish(atol(argv[i+1]), 1);
951 if (!strcmp(argv[i], "wait") && (i + 1 < argc)) {
952 ret = cmd_wait(atol(argv[i+1]));
957 if (!strcmp(argv[i], "rndwrite") && (i + 5 < argc)) {
958 long nr_loops = atol(argv[i+1]);
959 unsigned int seed = atoi(argv[i+2]);
960 unsigned int namesize = atoi(argv[i+3]);
961 unsigned int chunksize = atoi(argv[i+4]);
962 unsigned long objectsize = atol(argv[i+5]);
963 ret = cmd_rndwrite(nr_loops, seed, namesize, chunksize, objectsize);
968 if (!strcmp(argv[i], "rndread") && (i + 5 < argc)) {
969 long nr_loops = atol(argv[i+1]);
970 unsigned int seed = atoi(argv[i+2]);
971 unsigned int namesize = atoi(argv[i+3]);
972 unsigned int chunksize = atoi(argv[i+4]);
973 unsigned long objectsize = atol(argv[i+5]);
974 ret = cmd_rndread(nr_loops, seed, namesize, chunksize, objectsize);
979 if (!strcmp(argv[i], "read") && (i + 3 < argc)) {
980 char *name = argv[i+1];
981 uint64_t offset = atol(argv[i+2]);
982 uint64_t size = atol(argv[i+3]);
983 ret = cmd_read(name, offset, size);
988 if (!strcmp(argv[i], "write") && (i + 2 < argc)) {
989 char *name = argv[i+1];
990 uint64_t offset = atol(argv[i+2]);
991 ret = cmd_write(name, offset);
996 if (!strcmp(argv[i], "truncate") && (i + 2 < argc)) {
997 char *name = argv[i+1];
998 uint64_t offset = atol(argv[i+2]);
999 ret = cmd_truncate(name, offset);
1004 if (!strcmp(argv[i], "delete") && (i + 1 < argc)) {
1005 char *name = argv[i+1];
1006 ret = cmd_delete(name);
1011 if (!strcmp(argv[i], "acquire") && (i + 1 < argc)) {
1012 char *name = argv[i+1];
1013 ret = cmd_acquire(name);
1018 if (!strcmp(argv[i], "release") && (i + 1 < argc)) {
1019 char *name = argv[i+1];
1020 ret = cmd_release(name);
1025 if (!strcmp(argv[i], "copy") && (i + 2) < argc) {
1026 char *src = argv[i+1];
1027 char *dst = argv[i+2];
1028 ret = cmd_copy(src, dst);
1033 if (!strcmp(argv[i], "clone") && (i + 2 < argc)) {
1034 char *src = argv[i+1];
1035 char *dst = argv[i+2];
1036 ret = cmd_clone(src, dst);
1041 if (!parse_ports(argv[i]))
1042 printf("invalid argument: %s\n", argv[i]);