12 #include <xtypes/xhash.h>
13 #include <xtypes/xobj.h>
14 #include <xseg/xseg.h>
15 #include <xseg/protocol.h>
18 printf("xseg <spec> [[[<src_port>]:[<dst_port>]] [<command> <arg>*] ]*\n"
20 " <type:name:nr_ports:nr_requests:request_size:extra_size:page_shift>\n"
27 " bridge <portno1> <portno2> <logfile> {full|summary|stats}\n"
30 " alloc_requests (to source) <nr>\n"
31 " free_requests (from source) <nr>\n"
32 " put_requests (all from dest)\n"
33 " put_replies (all from dest)\n"
34 " wait <nr_replies>\n"
35 " complete <nr_requests>\n"
36 " fail <nr_requests>\n"
37 " rndwrite <nr_loops> <seed> <targetlen> <datalen> <objectsize>\n"
38 " rndread <nr_loops> <seed> <targetlen> <datalen> <objectsize>\n"
39 " submit_reqs <nr_loops> <concurrent_reqs>\n"
41 " read <target> <offset> <size>\n"
42 " write <target> <offset> < data\n"
43 " truncate <target> <size>\n"
48 " clone <src> <dst>\n"
68 struct xseg_config cfg;
70 uint32_t srcport, dstport;
72 #define mkname mkname_heavy
73 /* heavy distributes duplicates much more widely than light
74 * ./xseg-tool random 100000 | cut -d' ' -f2- | sort | uniq -d -c |wc -l
78 static void init_local_signal()
80 if (xseg && sport != srcport){
81 xseg_init_local_signal(xseg, srcport);
86 void mkname_heavy(char *name, uint32_t namelen, uint32_t seed)
90 for (i = 0; i < namelen; i += 1) {
91 c = seed + (seed >> 8) + (seed >> 16) + (seed >> 24);
92 c = '0' + ((c + (c >> 4)) & 0xf);
96 seed *= ((seed % 137911) | 1) * 137911;
100 void mkname_light(char *name, uint32_t namelen, uint32_t seed)
104 for (i = 0; i < namelen; i += 1) {
106 name[i] = 'A' + (c & 0xf);
111 uint64_t pick(uint64_t size)
113 return (uint64_t)((double)(RAND_MAX) / random());
116 void mkchunk( char *chunk, uint32_t datalen,
117 char *target, uint32_t targetlen, uint64_t offset)
119 long i, r, bufsize = targetlen + 16;
121 r = datalen % bufsize;
122 snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, target);
124 for (i = 0; i <= (long)datalen - bufsize; i += bufsize)
125 memcpy(chunk + i, buf, bufsize);
127 memcpy(chunk + datalen - r, buf, r);
130 int chkchunk( char *chunk, uint32_t datalen,
131 char *target, uint32_t targetlen, uint64_t offset)
134 int bufsize = targetlen + 16;
136 r = datalen % targetlen;
137 snprintf(buf, bufsize, "%016llx%s", (unsigned long long)offset, target);
139 for (i = 0; i <= (long)datalen - bufsize; i += bufsize)
140 if (memcmp(chunk + i, buf, bufsize)) {
141 /*printf("mismatch: '%*s'* vs '%*s'\n",
142 bufsize, buf, datalen, chunk);
147 if (memcmp(chunk + datalen - r, buf, r))
154 #define ALLOC_MIN 4096
155 #define ALLOC_MAX 1048576
157 void inputbuf(FILE *fp, char **retbuf, uint64_t *retsize)
159 static uint64_t alloc_size;
165 if (alloc_size < ALLOC_MIN)
166 alloc_size = ALLOC_MIN;
168 if (alloc_size > ALLOC_MAX)
169 alloc_size = ALLOC_MAX;
171 p = realloc(buf, alloc_size);
182 r = fread(buf + size, 1, alloc_size - size, fp);
186 if (size >= alloc_size) {
187 p = realloc(buf, alloc_size * 2);
205 void report_request(struct xseg_request *req)
207 char target[64], data[64];
208 char *req_target, *req_data;
209 unsigned int end = (req->targetlen> 63) ? 63 : req->targetlen;
210 req_target = xseg_get_target(xseg, req);
211 req_data = xseg_get_data(xseg, req);
213 strncpy(target, req_target, end);
215 strncpy(data, req_data, 63);
218 "Request %lx: target[%u](xptr: %llu): %s, data[%llu](xptr: %llu): %s \n\t"
219 "offset: %llu, size: %llu, serviced; %llu, op: %u, state: %u, flags: %u \n\t"
220 "src: %u, src_transit: %u, dst: %u, dst_transit: %u\n",
221 (unsigned long) req, req->targetlen, (unsigned long long)req->target,
223 (unsigned long long) req->datalen, (unsigned long long) req->data,
225 (unsigned long long) req->offset, (unsigned long long) req->size,
226 (unsigned long long) req->serviced, req->op, req->state, req->flags,
227 (unsigned int) req->src_portno, (unsigned int) req->src_transit_portno,
228 (unsigned int) req->dst_portno, (unsigned int) req->dst_transit_portno);
233 int cmd_info(char *target)
235 uint32_t targetlen = strlen(target);
236 size_t size = sizeof(uint64_t);
239 struct xseg_request *req;
242 req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
244 fprintf(stderr, "No request!\n");
248 r = xseg_prep_request(xseg, req, targetlen, size);
250 fprintf(stderr, "Cannot prepare request! (%lu, %lu)\n",
251 (unsigned long) targetlen, (unsigned long) size);
252 xseg_put_request(xseg, req, srcport);
256 req_target = xseg_get_target(xseg, req);
257 strncpy(req_target, target, targetlen);
262 p = xseg_submit(xseg, req, srcport, X_ALLOC);
266 xseg_signal(xseg, p);
271 int cmd_read(char *target, uint64_t offset, uint64_t size)
273 uint32_t targetlen = strlen(target);
277 struct xseg_request *req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
279 fprintf(stderr, "No request\n");
283 r = xseg_prep_request(xseg, req, targetlen, size);
285 fprintf(stderr, "Cannot prepare request! (%lu, %llu)\n",
286 (unsigned long)targetlen, (unsigned long long)size);
287 xseg_put_request(xseg, req, srcport);
291 req_target = xseg_get_target(xseg, req);
292 strncpy(req_target, target, targetlen);
293 req->offset = offset;
297 p = xseg_submit(xseg, req, srcport, X_ALLOC);
301 xseg_signal(xseg, p);
305 int cmd_write(char *target, uint64_t offset)
311 char *req_target, *req_data;
312 uint32_t targetlen = strlen(target);
313 struct xseg_request *req;
315 inputbuf(stdin, &buf, &size);
317 fprintf(stderr, "No input\n");
321 req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
323 fprintf(stderr, "No request\n");
327 r = xseg_prep_request(xseg, req, targetlen, size);
329 fprintf(stderr, "Cannot prepare request! (%lu, %llu)\n",
330 (unsigned long)targetlen, (unsigned long long)size);
331 xseg_put_request(xseg, req, srcport);
335 req_target = xseg_get_target(xseg, req);
336 strncpy(req_target, target, targetlen);
338 req_data = xseg_get_data(xseg, req);
339 memcpy(req_data, buf, size);
340 req->offset = offset;
344 p = xseg_submit(xseg, req, srcport, X_ALLOC);
346 fprintf(stderr, "Cannot submit\n");
349 xseg_signal(xseg, p);
354 int cmd_truncate(char *target, uint64_t offset)
359 int cmd_delete(char *target)
361 uint32_t targetlen = strlen(target);
363 struct xseg_request *req;
365 xseg_bind_port(xseg, srcport, NULL);
367 req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
369 fprintf(stderr, "No request!\n");
373 r = xseg_prep_request(xseg, req, targetlen, 0);
375 fprintf(stderr, "Cannot prepare request! (%lu, %lu)\n",
376 (unsigned long) targetlen, (unsigned long) req->bufferlen - targetlen);
377 xseg_put_request(xseg, req, srcport);
381 char *reqtarget = xseg_get_target(xseg, req);
382 strncpy(reqtarget, target, targetlen);
385 xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
387 fprintf(stderr, "Couldn't submit request\n");
388 xseg_put_request(xseg, req, srcport);
392 xseg_signal(xseg, p);
397 int cmd_acquire(char *target)
399 uint32_t targetlen = strlen(target);
403 struct xseg_request *req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
405 fprintf(stderr, "No request\n");
409 r = xseg_prep_request(xseg, req, targetlen, 0);
411 fprintf(stderr, "Cannot prepare request! (%lu, 0)\n",
412 (unsigned long)targetlen);
413 xseg_put_request(xseg, req, srcport);
417 req_target = xseg_get_target(xseg, req);
418 strncpy(req_target, target, targetlen);
422 p = xseg_submit(xseg, req, srcport, X_ALLOC);
426 xseg_signal(xseg, p);
430 int cmd_release(char *target)
432 uint32_t targetlen = strlen(target);
436 struct xseg_request *req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
438 fprintf(stderr, "No request\n");
442 r = xseg_prep_request(xseg, req, targetlen, 0);
444 fprintf(stderr, "Cannot prepare request! (%lu, 0)\n",
445 (unsigned long)targetlen);
446 xseg_put_request(xseg, req, srcport);
450 req_target = xseg_get_target(xseg, req);
451 strncpy(req_target, target, targetlen);
455 req->flags = XF_FORCE;
456 p = xseg_submit(xseg, req, srcport, X_ALLOC);
460 xseg_signal(xseg, p);
465 int cmd_copy(char *src, char *dst)
467 uint32_t targetlen = strlen(dst);
468 uint32_t parentlen = strlen(src);
469 struct xseg_request *req;
470 struct xseg_request_copy *xcopy;
471 req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
473 fprintf(stderr, "No request\n");
477 int r = xseg_prep_request(xseg, req, targetlen,
478 sizeof(struct xseg_request_copy));
480 fprintf(stderr, "Cannot prepare request!\n");
481 xseg_put_request(xseg, req, srcport);
485 char *target = xseg_get_target(xseg, req);
486 char *data = xseg_get_data(xseg, req);
488 strncpy(target, dst, targetlen);
489 xcopy = (struct xseg_request_copy *) data;
490 strncpy(xcopy->target, src, parentlen);
491 xcopy->targetlen = parentlen;
493 req->size = sizeof(struct xseg_request_copy);
496 xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
498 fprintf(stderr, "Cannot submit request\n");
501 xseg_signal(xseg, p);
507 int cmd_clone(char *src, char *dst)
510 uint32_t targetlen = strlen(dst);
511 uint32_t parentlen = strlen(src);
512 struct xseg_request *req;
513 struct xseg_request_clone *xclone;
514 xseg_bind_port(xseg, srcport, NULL);
515 req = xseg_get_request(xseg, srcport, dstport, X_ALLOC);
517 fprintf(stderr, "No request\n");
521 int r = xseg_prep_request(xseg, req, targetlen, sizeof(struct xseg_request_clone));
523 fprintf(stderr, "Cannot prepare request!\n");
524 xseg_put_request(xseg, req, srcport);
528 char *target = xseg_get_target(xseg, req);
529 char *data = xseg_get_data(xseg, req);
531 strncpy(target, dst, targetlen);
532 xclone = (struct xseg_request_clone *) data;
533 strncpy(xclone->target, src, parentlen);
534 xclone->targetlen = parentlen;
537 req->size = sizeof(struct xseg_request_clone);
540 xport p = xseg_submit(xseg, req, srcport, X_ALLOC);
542 fprintf(stderr, "Cannot submit request\n");
545 xseg_signal(xseg, p);
550 void log_req(int logfd, uint32_t portno2, uint32_t portno1, int op, int method,
551 struct xseg_request *req)
554 char target[64], data[64];
555 char *req_target, *req_data;
556 /* null terminate name in case of req->target is less than 63 characters,
557 * and next character after name (aka first byte of next buffer) is not
560 unsigned int end = (req->targetlen > 63) ? 63 : req->targetlen;
562 req_target = xseg_get_target(xseg, req);
563 req_data = xseg_get_data(xseg, req);
565 logfp = fdopen(logfd, "a");
571 strncpy(target, req_target, end);
573 strncpy(data, req_data, 63);
577 "src port: %u, dst port: %u, op:%u offset: %llu size: %lu, reqstate: %u\n"
578 "target[%u]: '%s', data[%llu]:\n%s------------------\n\n",
579 (unsigned int)portno1,
580 (unsigned int)portno2,
581 (unsigned int)req->op,
582 (unsigned long long)req->offset,
583 (unsigned long)req->size,
584 (unsigned int)req->state,
585 (unsigned int)req->targetlen, target,
586 (unsigned long long)req->datalen, data);
590 "src port: %u, dst port: %u, op: %u\n",
591 (unsigned int)portno1,
592 (unsigned int)portno2,
593 (unsigned int)req->op);
596 fprintf(logfp, "src port: %u, dst port: %u, reqs: %llu\n",
597 (unsigned int)portno1,
598 (unsigned int)portno2,
599 (unsigned long long)++reqs);
607 #define LOG_RECEIVE 1
609 int cmd_bridge(uint32_t portno1, uint32_t portno2, char *logfile, char *how)
611 struct xseg_request *req;
613 if (!strcmp(logfile, "-"))
616 logfd = open(logfile, O_WRONLY|O_APPEND|O_CREAT, 0600);
623 if (!strcmp(how, "full"))
625 else if (!strcmp(how, "summary"))
631 int reloop = 0, active;
632 xseg_prepare_wait(xseg, portno1);
633 xseg_prepare_wait(xseg, portno2);
640 req = xseg_accept(xseg, portno1, 0);
642 xseg_submit(xseg, req, portno2, X_ALLOC);
643 log_req(logfd, portno1, portno2, LOG_ACCEPT, method, req);
647 req = xseg_accept(xseg, portno2, 0);
649 xseg_submit(xseg, req, portno1, X_ALLOC);
650 log_req(logfd, portno2, portno1, LOG_ACCEPT, method, req);
654 req = xseg_receive(xseg, portno1, 0);
656 xseg_respond(xseg, req, portno2, X_ALLOC);
657 log_req(logfd, portno1, portno2, LOG_RECEIVE, method, req);
661 req = xseg_receive(xseg, portno2, 0);
663 xseg_respond(xseg, req, portno1, X_ALLOC);
664 log_req(logfd, portno2, portno1, LOG_RECEIVE, method, req);
671 /* wait on multiple queues? */
672 xseg_wait_signal(xseg, 100000);
675 xseg_cancel_wait(xseg, portno1);
676 xseg_cancel_wait(xseg, portno2);
687 int cmd_rndwrite(long loops, int32_t seed, uint32_t targetlen, uint32_t chunksize, uint64_t size)
692 if (targetlen >= chunksize) {
693 fprintf(stderr, "targetlen >= chunksize\n");
697 char *p = realloc(namebuf, targetlen+1);
699 fprintf(stderr, "Cannot allocate memory\n");
704 p = realloc(chunk, chunksize);
706 fprintf(stderr, "Cannot allocate memory\n");
710 memset(chunk, 0, chunksize);
714 struct xseg_request *submitted = NULL, *received;
715 long nr_submitted = 0, nr_received = 0, nr_failed = 0;
719 char *req_data, *req_target;
724 xseg_prepare_wait(xseg, srcport);
725 if (nr_submitted < loops &&
726 (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
727 xseg_cancel_wait(xseg, srcport);
728 r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
730 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
731 targetlen, chunksize);
732 xseg_put_request(xseg, submitted, srcport);
736 req_target = xseg_get_target(xseg, submitted);
737 req_data = xseg_get_data(xseg, submitted);
740 mkname(namebuf, targetlen, seed);
741 namebuf[targetlen] = 0;
742 //printf("%ld: %s\n", nr_submitted, namebuf);
743 strncpy(req_target, namebuf, targetlen);
744 offset = 0;// pick(size);
745 mkchunk(req_data, chunksize, namebuf, targetlen, offset);
747 submitted->offset = offset;
748 submitted->size = chunksize;
749 submitted->op = X_WRITE;
750 submitted->flags |= XF_NOSYNC;
752 port = xseg_submit(xseg, submitted, srcport, X_ALLOC);
753 if (port == NoPort) {
754 xseg_put_request(xseg, submitted, srcport);
758 xseg_signal(xseg, port);
762 received = xseg_receive(xseg, srcport, 0);
764 xseg_cancel_wait(xseg, srcport);
766 if (!(received->state & XS_SERVED)) {
768 report_request(received);
770 if (xseg_put_request(xseg, received, srcport))
771 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
774 if (!submitted && !received)
775 xseg_wait_signal(xseg, 1000000);
777 if (nr_submitted % 1000 == 0 && !reported) {
779 fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
780 nr_submitted, nr_received, nr_failed);
783 if (nr_received >= loops)
787 fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
788 nr_submitted, nr_received, nr_failed);
792 int cmd_rnddelete(long loops, int32_t seed, uint32_t targetlen)
796 char *p = realloc(namebuf, targetlen+1);
798 fprintf(stderr, "Cannot allocate memory\n");
805 struct xseg_request *submitted = NULL, *received;
806 long nr_submitted = 0, nr_received = 0, nr_failed = 0;
814 xseg_prepare_wait(xseg, srcport);
815 if (nr_submitted < loops &&
816 (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
817 xseg_cancel_wait(xseg, srcport);
818 r = xseg_prep_request(xseg, submitted, targetlen, 0);
820 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
822 xseg_put_request(xseg, submitted, srcport);
826 req_target = xseg_get_target(xseg, submitted);
829 mkname(namebuf, targetlen, seed);
830 namebuf[targetlen] = 0;
831 //printf("%ld: %s\n", nr_submitted, namebuf);
832 strncpy(req_target, namebuf, targetlen);
833 submitted->offset = 0;
835 submitted->op = X_DELETE;
836 submitted->flags = 0;
838 port = xseg_submit(xseg, submitted, srcport, X_ALLOC);
839 if (port == NoPort) {
840 xseg_put_request(xseg, submitted, srcport);
844 xseg_signal(xseg, port);
848 received = xseg_receive(xseg, srcport, 0);
850 xseg_cancel_wait(xseg, srcport);
852 if (!(received->state & XS_SERVED)) {
854 report_request(received);
856 if (xseg_put_request(xseg, received, srcport))
857 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
860 if (!submitted && !received)
861 xseg_wait_signal(xseg, 1000000);
863 if (nr_submitted % 1000 == 0 && !reported) {
865 fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
866 nr_submitted, nr_received, nr_failed);
869 if (nr_received >= loops)
873 fprintf(stderr, "submitted %ld, received %ld, failed %ld\n",
874 nr_submitted, nr_received, nr_failed);
878 * prepare/wait rhythm,
879 * files are converted to independent chunk access patterns,
882 int cmd_rndread(long loops, int32_t seed, uint32_t targetlen, uint32_t chunksize, uint64_t size)
887 if (targetlen >= chunksize) {
888 fprintf(stderr, "targetlen >= chunksize\n");
892 char *p = realloc(namebuf, targetlen+1);
894 fprintf(stderr, "Cannot allocate memory\n");
899 p = realloc(chunk, chunksize);
901 fprintf(stderr, "Cannot allocate memory\n");
905 memset(chunk, 0, chunksize);
909 struct xseg_request *submitted = NULL, *received;
910 long nr_submitted = 0, nr_received = 0, nr_failed = 0, nr_mismatch = 0;
914 char *req_data, *req_target;
920 xseg_prepare_wait(xseg, srcport);
921 if (nr_submitted < loops &&
922 (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
923 xseg_cancel_wait(xseg, srcport);
924 r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
926 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
927 targetlen, chunksize);
928 xseg_put_request(xseg, submitted, srcport);
932 req_target = xseg_get_target(xseg, submitted);
934 mkname(namebuf, targetlen, seed);
935 namebuf[targetlen] = 0;
936 //printf("%ld: %s\n", nr_submitted, namebuf);
937 offset = 0;//pick(size);
939 strncpy(req_target, namebuf, targetlen);
940 submitted->offset = offset;
941 submitted->size = chunksize;
942 submitted->op = X_READ;
943 port = xseg_submit(xseg, submitted, srcport, X_ALLOC);
944 if (port == NoPort) {
945 xseg_put_request(xseg, submitted, srcport);
949 xseg_signal(xseg, port);
953 received = xseg_receive(xseg, srcport, 0);
955 xseg_cancel_wait(xseg, srcport);
957 req_target = xseg_get_target(xseg, received);
958 req_data = xseg_get_data(xseg, received);
959 if (!(received->state & XS_SERVED)) {
961 report_request(received);
962 } else if (!chkchunk(req_data, received->datalen,
963 req_target, received->targetlen, received->offset)) {
964 // report_request(received);
968 if (xseg_put_request(xseg, received, srcport))
969 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
972 if (!submitted && !received)
973 xseg_wait_signal(xseg, 1000000);
975 if (nr_submitted % 1000 == 0 && !reported) {
977 fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
978 nr_submitted, nr_received, nr_failed, nr_mismatch);
981 if (nr_received >= loops)
985 fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
986 nr_submitted, nr_received, nr_failed, nr_mismatch);
990 int cmd_submit_reqs(long loops, long concurrent_reqs, int op)
995 struct xseg_request *submitted = NULL, *received;
996 long nr_submitted = 0, nr_received = 0, nr_failed = 0, nr_mismatch = 0, nr_flying = 0;
999 uint32_t targetlen = 10, chunksize = 4096;
1000 struct timeval tv1, tv2;
1002 char *req_data, *req_target;
1004 xseg_bind_port(xseg, srcport, NULL);
1006 gettimeofday(&tv1, NULL);
1009 xseg_prepare_wait(xseg, srcport);
1010 if (nr_submitted < loops && nr_flying < concurrent_reqs &&
1011 (submitted = xseg_get_request(xseg, srcport, dstport, X_ALLOC))) {
1012 xseg_cancel_wait(xseg, srcport);
1013 r = xseg_prep_request(xseg, submitted, targetlen, chunksize);
1015 fprintf(stderr, "Cannot prepare request! (%u, %u)\n",
1016 targetlen, chunksize);
1017 xseg_put_request(xseg, submitted, srcport);
1025 offset = 0;//pick(size);
1027 submitted->offset = offset;
1028 submitted->size = chunksize;
1029 req_target = xseg_get_target(xseg, submitted);
1030 req_data = xseg_get_data(xseg, submitted);
1033 submitted->op = X_INFO;
1035 submitted->op = X_READ;
1037 submitted->op = X_WRITE;
1038 mkchunk(req_data, submitted->datalen, req_target, submitted->targetlen, submitted->offset);
1041 p = xseg_submit(xseg, submitted, srcport, X_ALLOC);
1043 if (xseg_signal(xseg, p) < 0)
1044 perror("Cannot signal peer");
1047 received = xseg_receive(xseg, srcport, 0);
1049 xseg_cancel_wait(xseg, srcport);
1051 if (nr_received == 0)
1052 fprintf(stderr, "latency (time for the first req to complete): %llu usecs\n",
1053 (unsigned long long)received->elapsed);
1055 if (!(received->state & XS_SERVED)) {
1057 //report_request(received);
1060 if (xseg_put_request(xseg, received, srcport))
1061 fprintf(stderr, "Cannot put request at port %u\n", received->src_portno);
1064 if (!submitted && !received)
1065 xseg_wait_signal(xseg, 10000000L);
1067 if (nr_received >= loops)
1070 gettimeofday(&tv2, NULL);
1072 fprintf(stderr, "submitted %ld, received %ld, failed %ld, mismatched %ld\n",
1073 nr_submitted, nr_received, nr_failed, nr_mismatch);
1074 long t = (tv2.tv_sec - tv1.tv_sec)*1000000 + (tv2.tv_usec - tv1.tv_usec);
1075 fprintf(stderr, "elpased time: %lf secs, throughput: %lf reqs/sec\n", (double) t / 1000000.0, (double) nr_submitted / (t / 1000000.0));
1080 static void lock_status(struct xlock *lock, char *buf, int len)
1083 if (lock->owner == Noone)
1084 r = snprintf(buf, len, "Locked: No");
1086 r = snprintf(buf, len, "Locked: Yes (Owner: %lu)", lock->owner);
1091 int cmd_report(uint32_t portno)
1093 char fls[64], rls[64], pls[64]; // buffer to store lock status
1094 struct xseg_port *port = xseg_get_port(xseg, portno);
1096 printf("port %u is not assigned\n", portno);
1099 struct xq *fq, *rq, *pq;
1100 fq = xseg_get_queue(xseg, port, free_queue);
1101 rq = xseg_get_queue(xseg, port, request_queue);
1102 pq = xseg_get_queue(xseg, port, reply_queue);
1103 lock_status(&port->fq_lock, fls, 64);
1104 lock_status(&port->rq_lock, rls, 64);
1105 lock_status(&port->pq_lock, pls, 64);
1106 fprintf(stderr, "port %u:\n"
1107 " requests: %llu/%llu src gw: %u dst gw: %u\n"
1108 " free_queue [%p] count : %4llu | %s\n"
1109 " request_queue [%p] count : %4llu | %s\n"
1110 " reply_queue [%p] count : %4llu | %s\n",
1111 portno, (unsigned long long)port->alloc_reqs,
1112 (unsigned long long)port->max_alloc_reqs,
1113 xseg->src_gw[portno],
1114 xseg->dst_gw[portno],
1115 (void *)fq, (unsigned long long)xq_count(fq), fls,
1116 (void *)rq, (unsigned long long)xq_count(rq), rls,
1117 (void *)pq, (unsigned long long)xq_count(pq), pls);
1126 xseg = xseg_join(cfg.type, cfg.name, "posix", NULL);
1128 fprintf(stderr, "cannot join segment!\n");
1133 static void print_hanlder(char *name, struct xobject_h *obj_h)
1136 lock_status(&obj_h->lock, ls, 64);
1137 fprintf(stderr, "%20s: free: %4llu, allocated: %4llu, allocated space: %7llu (object size: %llu), Lock %s\n",
1139 (unsigned long long) obj_h->nr_free,
1140 (unsigned long long) obj_h->nr_allocated,
1141 (unsigned long long) obj_h->allocated_space,
1142 (unsigned long long) obj_h->obj_size, ls);
1146 static void print_heap(struct xseg *xseg)
1156 MULT[2] = 1024*1024;
1157 MULT[3] = 1024*1024*1024;
1161 fprintf(stderr, "Heap usage: ");
1163 t = xseg->heap->cur;
1170 t = xseg->heap->cur / MULT[u];
1172 float tf = ((float)(xseg->heap->cur))/((float)MULT[u]);
1173 fprintf(stderr, "%2.1f %s/", tf, UNIT[u]);
1176 unsigned int tu = xseg->heap->cur / MULT[u];
1177 fprintf(stderr, "%3u %s/", tu, UNIT[u]);
1181 t = xseg->config.heap_size;
1188 t = xseg->config.heap_size/MULT[u];
1190 float tf = ((float)(xseg->config.heap_size))/(float)MULT[u];
1191 fprintf(stderr, "%2.1f %s ", tf, UNIT[u]);
1194 unsigned int tu = xseg->config.heap_size / MULT[u];
1195 fprintf(stderr, "%3u %s ", tu, UNIT[u]);
1198 lock_status(&xseg->heap->lock, ls, 64);
1199 fprintf(stderr, "(%llu / %llu), %s\n",
1200 (unsigned long long)xseg->heap->cur,
1201 (unsigned long long)xseg->config.heap_size,
1205 int cmd_reportall(void)
1212 fprintf(stderr, "Segment lock: %s\n",
1213 (xseg->shared->flags & XSEG_F_LOCK) ? "Locked" : "Unlocked");
1215 /* fprintf(stderr, "Heap usage: %llu / %llu\n", */
1216 /* (unsigned long long)xseg->heap->cur, */
1217 /* (unsigned long long)xseg->config.heap_size); */
1218 fprintf(stderr, "Handlers: \n");
1219 print_hanlder("Requests handler", xseg->request_h);
1220 print_hanlder("Ports handler", xseg->port_h);
1221 print_hanlder("Objects handler", xseg->object_handlers);
1222 fprintf(stderr, "\n");
1224 for (t = 0; t < xseg->config.nr_ports; t++)
1231 int finish_req(struct xseg_request *req, enum req_action action)
1233 if (action == COMPLETE){
1234 req->state &= ~XS_FAILED;
1235 req->state |= XS_SERVED;
1237 req->state |= XS_FAILED;
1238 req->state &= ~XS_SERVED;
1241 xport p = xseg_respond(xseg, req, srcport, X_ALLOC);
1243 xseg_put_request(xseg, req, srcport);
1245 xseg_signal(xseg, p);
1249 //FIXME this should be in xseg lib?
1250 static int isDangling(struct xseg_request *req)
1253 struct xseg_port *port;
1254 for (i = 0; i < xseg->config.nr_ports; i++) {
1255 if (xseg->ports[i]){
1256 port = xseg_get_port(xseg, i);
1258 fprintf(stderr, "Inconsisten port <-> portno mapping %u", i);
1261 struct xq *fq, *rq, *pq;
1262 fq = xseg_get_queue(xseg, port, free_queue);
1263 rq = xseg_get_queue(xseg, port, request_queue);
1264 pq = xseg_get_queue(xseg, port, reply_queue);
1265 xlock_acquire(&port->fq_lock, srcport);
1266 if (__xq_check(fq, XPTR_MAKE(req, xseg->segment))){
1267 xlock_release(&port->fq_lock);
1270 xlock_release(&port->fq_lock);
1271 xlock_acquire(&port->rq_lock, srcport);
1272 if (__xq_check(rq, XPTR_MAKE(req, xseg->segment))){
1273 xlock_release(&port->rq_lock);
1276 xlock_release(&port->rq_lock);
1277 xlock_acquire(&port->pq_lock, srcport);
1278 if (__xq_check(pq, XPTR_MAKE(req, xseg->segment))){
1279 xlock_release(&port->pq_lock);
1282 xlock_release(&port->pq_lock);
1288 int prompt_user(char *msg)
1291 printf("%s [y/n]: ", msg);
1294 if (c == 'y' || c == 'Y')
1296 else if (c == 'n' || c == 'N')
1298 else if (c == '\n'){
1300 printf("%s [y/n]: ", msg);
1308 //FIXME this should be in xseg lib?
1309 int cmd_verify(int fix)
1314 if (xseg->shared->flags & XSEG_F_LOCK){
1315 fprintf(stderr, "Segment lock: Locked\n");
1316 if (fix && prompt_user("Unlock it ?"))
1317 xseg->shared->flags &= ~XSEG_F_LOCK;
1320 if (xseg->heap->lock.owner != Noone){
1321 fprintf(stderr, "Heap lock: Locked (Owner: %llu)\n",
1322 (unsigned long long)xseg->heap->lock.owner);
1323 if (fix && prompt_user("Unlock it ?"))
1324 xlock_release(&xseg->heap->lock);
1327 if (xseg->request_h->lock.owner != Noone){
1328 fprintf(stderr, "Requests handler lock: Locked (Owner: %llu)\n",
1329 (unsigned long long)xseg->request_h->lock.owner);
1330 if (fix && prompt_user("Unlock it ?"))
1331 xlock_release(&xseg->request_h->lock);
1333 if (xseg->port_h->lock.owner != Noone){
1334 fprintf(stderr, "Ports handler lock: Locked (Owner: %llu)\n",
1335 (unsigned long long)xseg->port_h->lock.owner);
1336 if (fix && prompt_user("Unlock it ?"))
1337 xlock_release(&xseg->port_h->lock);
1339 if (xseg->object_handlers->lock.owner != Noone){
1340 fprintf(stderr, "Objects handler lock: Locked (Owner: %llu)\n",
1341 (unsigned long long)xseg->object_handlers->lock.owner);
1342 if (fix && prompt_user("Unlock it ?"))
1343 xlock_release(&xseg->object_handlers->lock);
1345 //take segment lock?
1347 struct xseg_port *port;
1348 for (i = 0; i < xseg->config.nr_ports; i++) {
1349 if (xseg->ports[i]){
1350 port = xseg_get_port(xseg, i);
1352 fprintf(stderr, "Inconsisten port <-> portno mapping %u", i);
1355 if (port->fq_lock.owner != Noone) {
1356 fprintf(stderr, "Free queue lock of port %u locked (Owner %llu)\n",
1357 i, (unsigned long long)port->fq_lock.owner);
1358 if (fix && prompt_user("Unlock it ?"))
1359 xlock_release(&port->fq_lock);
1361 if (port->rq_lock.owner != Noone) {
1362 fprintf(stderr, "Request queue lock of port %u locked (Owner %llu)\n",
1363 i, (unsigned long long)port->rq_lock.owner);
1364 if (fix && prompt_user("Unlock it ?"))
1365 xlock_release(&port->rq_lock);
1367 if (port->pq_lock.owner != Noone) {
1368 fprintf(stderr, "Reply queue lock of port %u locked (Owner %llu)\n",
1369 i, (unsigned long long)port->pq_lock.owner);
1370 if (fix && prompt_user("Unlock it ?"))
1371 xlock_release(&port->pq_lock);
1376 struct xobject_h *obj_h = xseg->request_h;
1377 struct xobject_iter it;
1379 struct xseg_request *req;
1380 xlock_acquire(&obj_h->lock, srcport);
1381 xobj_iter_init(obj_h, &it);
1382 while (xobj_iterate(obj_h, &it, (void **)&req)){
1383 //FIXME this will not work cause obj->magic - req->serial is not
1384 //touched when a request is get
1385 /* if (obj->magic != MAGIC_REQ && t->src_portno == portno){ */
1386 if (isDangling(req) && !__xobj_isFree(obj_h, req)){
1387 report_request(req);
1388 if (fix && prompt_user("Fail it ?")){
1389 printf("Finishing ...\n");
1390 finish_req(req, FAIL);
1394 xlock_release(&obj_h->lock);
1398 int cmd_inspectq(xport portno, enum queue qt)
1405 struct xseg_port *port = xseg_get_port(xseg, portno);
1408 if (qt == FREE_QUEUE){
1409 q = xseg_get_queue(xseg, port, free_queue);
1412 else if (qt == REQUEST_QUEUE){
1413 q = xseg_get_queue(xseg, port, request_queue);
1416 else if (qt == REPLY_QUEUE) {
1417 q = xseg_get_queue(xseg, port, reply_queue);
1422 xlock_acquire(l, srcport);
1423 xqindex i,c = xq_count(q);
1425 struct xseg_request *req;
1427 for (i = 0; i < c; i++) {
1428 xqi = __xq_pop_head(q);
1429 req = XPTR_TAKE(xqi, xseg->segment);
1430 report_request(req);
1431 __xq_append_tail(q, xqi);
1435 fprintf(stderr, "Queue is empty\n\n");
1442 int cmd_request(struct xseg_request *req, enum req_action action)
1447 struct xobject_h *obj_h = xseg->request_h;
1448 if (!xobj_check(obj_h, req))
1451 if (action == REPORT)
1452 report_request(req);
1453 else if (action == FAIL){
1454 report_request(req);
1455 if (prompt_user("fail it ?")){
1456 printf("Finishing ...\n");
1457 finish_req(req, FAIL);
1460 else if (action == COMPLETE){
1461 report_request(req);
1462 if (prompt_user("Complete it ?")){
1463 printf("Finishing ...\n");
1464 finish_req(req, COMPLETE);
1470 int cmd_create(void)
1472 int r = xseg_create(&cfg);
1474 fprintf(stderr, "cannot create segment!\n");
1478 fprintf(stderr, "Segment initialized.\n");
1482 int cmd_destroy(void)
1484 if (!xseg && cmd_join())
1489 fprintf(stderr, "Segment destroyed.\n");
1493 int cmd_alloc_requests(unsigned long nr)
1495 return xseg_alloc_requests(xseg, srcport, nr);
1498 int cmd_free_requests(unsigned long nr)
1500 return xseg_free_requests(xseg, srcport, nr);
1503 int cmd_put_requests(void)
1505 struct xseg_request *req;
1508 req = xseg_accept(xseg, dstport, 0);
1511 if (xseg_put_request(xseg, req, srcport))
1512 fprintf(stderr, "Cannot put request at port %u\n", req->src_portno);
1518 int cmd_finish(unsigned long nr, int fail)
1520 struct xseg_request *req;
1521 char *buf = malloc(sizeof(char) * 8128);
1522 char *req_target, *req_data;
1523 xseg_bind_port(xseg, srcport, NULL);
1527 xseg_prepare_wait(xseg, srcport);
1528 req = xseg_accept(xseg, srcport, 0);
1530 req_target = xseg_get_target(xseg, req);
1531 req_data = xseg_get_data(xseg, req);
1532 xseg_cancel_wait(xseg, srcport);
1534 req->state &= ~XS_SERVED;
1536 if (req->op == X_READ)
1537 mkchunk(req_data, req->datalen, req_target, req->targetlen, req->offset);
1538 else if (req->op == X_WRITE)
1539 memcpy(buf, req_data, (sizeof(*buf) > req->datalen) ? req->datalen : sizeof(*buf));
1540 else if (req->op == X_INFO)
1541 *((uint64_t *) req->data) = 4294967296;
1543 req->state |= XS_SERVED;
1544 req->serviced = req->size;
1547 p = xseg_respond(xseg, req, srcport, X_ALLOC);
1548 xseg_signal(xseg, p);
1552 xseg_wait_signal(xseg, 10000000L);
1560 void handle_reply(struct xseg_request *req)
1562 char *req_data = xseg_get_data(xseg, req);
1563 char *req_target = xseg_get_target(xseg, req);
1564 if (!(req->state & XS_SERVED)) {
1565 report_request(req);
1571 fwrite(req_data, 1, req->datalen, stdout);
1575 fprintf(stdout, "wrote: ");
1576 fwrite(req_data, 1, req->datalen, stdout);
1580 fprintf(stderr, "deleted %s\n", req_target);
1585 fprintf(stderr, "cloned %s\n", ((struct xseg_request_clone *)req_data)->target);
1588 fprintf(stderr, "size: %llu\n", (unsigned long long)*((uint64_t *)req_data));
1591 fprintf(stderr, "copied %s\n", ((struct xseg_request_copy *)req_data)->target);
1594 fprintf(stderr, "Closed %s\n", req_target);
1596 fprintf(stderr, "Opened %s\n", req_target);
1603 if (xseg_put_request(xseg, req, srcport))
1604 fprintf(stderr, "Cannot put reply at port %u\n", req->src_portno);
1607 int cmd_wait(uint32_t nr)
1609 struct xseg_request *req;
1611 init_local_signal();
1614 req = xseg_receive(xseg, srcport, 0);
1623 ret = xseg_prepare_wait(xseg, srcport);
1627 ret = xseg_wait_signal(xseg, 1000000);
1628 ret = xseg_cancel_wait(xseg, srcport);
1636 int cmd_put_replies(void)
1638 struct xseg_request *req;
1641 req = xseg_receive(xseg, dstport, 0);
1644 fprintf(stderr, "request: %08llx%08llx\n"
1647 0LL, (unsigned long long)req->serial,
1650 report_request(req);
1652 //fwrite(req->buffer, 1, req->bufferlen, stdout);
1654 if (xseg_put_request(xseg, req, srcport))
1655 fprintf(stderr, "Cannot put reply\n");
1661 int cmd_bind(long portno)
1663 struct xseg_port *port = xseg_bind_port(xseg, portno, NULL);
1665 fprintf(stderr, "failed to bind port %ld\n", portno);
1669 fprintf(stderr, "bound port %u\n", xseg_portno(xseg, port));
1673 int cmd_signal(uint32_t portno)
1675 return xseg_signal(xseg, portno);
1678 int parse_ports(char *str)
1689 if ((s > str) && isdigit(str[0])) {
1690 srcport = atol(str);
1703 if ((s > str) && isdigit(str[0])) {
1704 dstport = atol(str);
1715 int main(int argc, char **argv)
1727 if (xseg_parse_spec(spec, &cfg)) {
1728 fprintf(stderr, "Cannot parse spec\n");
1732 if (xseg_initialize()) {
1733 fprintf(stderr, "cannot initialize!\n");
1737 for (i = 2; i < argc; i++) {
1739 if (!strcmp(argv[i], "create")) {
1744 if (!strcmp(argv[i], "join")) {
1747 fprintf(stderr, "Segment joined.\n");
1751 if (!strcmp(argv[i], "destroy")) {
1752 ret = cmd_destroy();
1759 if (!strcmp(argv[i], "reportall")) {
1760 ret = cmd_reportall();
1764 if (!strcmp(argv[i], "bind") && (i + 1 < argc)) {
1765 ret = cmd_bind(atol(argv[i+1]));
1770 if (!strcmp(argv[i], "signal") && (i + 1 < argc)) {
1771 ret = cmd_signal(atol(argv[i+1]));
1776 if (!strcmp(argv[i], "bridge") && (i + 4 < argc)) {
1777 ret = cmd_bridge(atol(argv[i+1]),
1785 if (srcport == -1) {
1786 if (!parse_ports(argv[i]))
1787 fprintf(stderr, "source port undefined: %s\n", argv[i]);
1791 if (dstport == -1) {
1792 if (!parse_ports(argv[i]))
1793 fprintf(stderr, "destination port undefined: %s\n", argv[i]);
1797 if (!strcmp(argv[i], "verify")) {
1798 ret = cmd_verify(0);
1802 if (!strcmp(argv[i], "verify-fix")) {
1803 ret = cmd_verify(1);
1807 if (!strcmp(argv[i], "failreq") && (i + 1 < argc)) {
1808 struct xseg_request *req;
1809 sscanf(argv[i+1], "%lx", &req);
1810 ret = cmd_request(req, FAIL);
1815 if (!strcmp(argv[i], "inspect-freeq") && (i + 1 < argc)) {
1816 ret = cmd_inspectq(atol(argv[i+1]), FREE_QUEUE);
1821 if (!strcmp(argv[i], "inspect-requestq") && (i + 1 < argc)) {
1822 ret = cmd_inspectq(atol(argv[i+1]), REQUEST_QUEUE);
1827 if (!strcmp(argv[i], "inspect-replyq") && (i + 1 < argc)) {
1828 ret = cmd_inspectq(atol(argv[i+1]), REPLY_QUEUE);
1833 if (!strcmp(argv[i], "report")) {
1834 ret = cmd_report(dstport);
1838 if (!strcmp(argv[i], "alloc_requests") && (i + 1 < argc)) {
1839 ret = cmd_alloc_requests(atol(argv[i+1]));
1844 if (!strcmp(argv[i], "free_requests") && (i + 1 < argc)) {
1845 ret = cmd_free_requests(atol(argv[i+1]));
1850 if (!strcmp(argv[i], "put_requests")) {
1851 ret = cmd_put_requests();
1855 if (!strcmp(argv[i], "put_replies")) {
1856 ret = cmd_put_replies();
1860 if (!strcmp(argv[i], "complete") && (i + 1 < argc)) {
1861 ret = cmd_finish(atol(argv[i+1]), 0);
1866 if (!strcmp(argv[i], "fail") && (i + 1 < argc)) {
1867 ret = cmd_finish(atol(argv[i+1]), 1);
1872 if (!strcmp(argv[i], "wait") && (i + 1 < argc)) {
1873 ret = cmd_wait(atol(argv[i+1]));
1878 if (!strcmp(argv[i], "rndwrite") && (i + 5 < argc)) {
1879 long nr_loops = atol(argv[i+1]);
1880 unsigned int seed = atoi(argv[i+2]);
1881 unsigned int targetlen = atoi(argv[i+3]);
1882 unsigned int chunksize = atoi(argv[i+4]);
1883 unsigned long objectsize = atol(argv[i+5]);
1884 ret = cmd_rndwrite(nr_loops, seed, targetlen, chunksize, objectsize);
1889 if (!strcmp(argv[i], "rnddelete") && (i + 3 < argc)) {
1890 long nr_loops = atol(argv[i+1]);
1891 unsigned int seed = atoi(argv[i+2]);
1892 unsigned int targetlen = atoi(argv[i+3]);
1893 ret = cmd_rnddelete(nr_loops, seed, targetlen);
1898 if (!strcmp(argv[i], "rndread") && (i + 5 < argc)) {
1899 long nr_loops = atol(argv[i+1]);
1900 unsigned int seed = atoi(argv[i+2]);
1901 unsigned int targetlen = atoi(argv[i+3]);
1902 unsigned int chunksize = atoi(argv[i+4]);
1903 unsigned long objectsize = atol(argv[i+5]);
1904 ret = cmd_rndread(nr_loops, seed, targetlen, chunksize, objectsize);
1909 if (!strcmp(argv[i], "submit_reqs") && (i + 3 < argc)) {
1910 long nr_loops = atol(argv[i+1]);
1911 long concurrent_reqs = atol(argv[i+2]);
1912 int op = atoi(argv[i+3]);
1913 ret = cmd_submit_reqs(nr_loops, concurrent_reqs, op);
1918 if (!strcmp(argv[i], "read") && (i + 3 < argc)) {
1919 char *target = argv[i+1];
1920 uint64_t offset = atol(argv[i+2]);
1921 uint64_t size = atol(argv[i+3]);
1922 ret = cmd_read(target, offset, size);
1927 if (!strcmp(argv[i], "write") && (i + 2 < argc)) {
1928 char *target = argv[i+1];
1929 uint64_t offset = atol(argv[i+2]);
1930 ret = cmd_write(target, offset);
1935 if (!strcmp(argv[i], "truncate") && (i + 2 < argc)) {
1936 char *target = argv[i+1];
1937 uint64_t offset = atol(argv[i+2]);
1938 ret = cmd_truncate(target, offset);
1943 if (!strcmp(argv[i], "delete") && (i + 1 < argc)) {
1944 char *target = argv[i+1];
1945 ret = cmd_delete(target);
1950 if (!strcmp(argv[i], "acquire") && (i + 1 < argc)) {
1951 char *target = argv[i+1];
1952 ret = cmd_acquire(target);
1957 if (!strcmp(argv[i], "release") && (i + 1 < argc)) {
1958 char *target = argv[i+1];
1959 ret = cmd_release(target);
1964 if (!strcmp(argv[i], "copy") && (i + 2) < argc) {
1965 char *src = argv[i+1];
1966 char *dst = argv[i+2];
1967 ret = cmd_copy(src, dst);
1972 if (!strcmp(argv[i], "clone") && (i + 2 < argc)) {
1973 char *src = argv[i+1];
1974 char *dst = argv[i+2];
1975 ret = cmd_clone(src, dst);
1980 if (!strcmp(argv[i], "info") && (i + 1 < argc)) {
1981 char *target = argv[i+1];
1982 ret = cmd_info(target);
1988 if (!parse_ports(argv[i]))
1989 fprintf(stderr, "invalid argument: %s\n", argv[i]);