Changeset 22f94a4 for benchmark/io/readv.cfa

Timestamp:

Aug 11, 2020, 4:40:15 PM (5 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

0d070ca

Parents:

07d867b (diff), 129674b (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' into new-ast

File:

• benchmark/io/readv.cfa (modified) (7 diffs)

benchmark/io/readv.cfa

@@ +1 @@
+#define _GNU_SOURCE
+
 #include <stdlib.h>
 #include <stdio.h>
@@ -10 +12 @@
 }
 
+#include <errno.h>
 #include <unistd.h>
 
 #include <clock.hfa>
+#include <iofwd.hfa>
 #include <kernel.hfa>
 #include <thread.hfa>
 #include <time.hfa>
+#include <stats.hfa>
+
+#include "../benchcltr.hfa"
 
 extern bool traceHeapOn();
-extern ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
 
 int fd;
@@ -24 +30 @@
 volatile size_t count = 0;
 
-unsigned long int buflen = 50;
+unsigned long int buflen = 512;
+bool fixed_file = false;
 
-cluster * the_cluster;
-
-thread Reader {};
+thread __attribute__((aligned(128))) Reader {};
 void ?{}( Reader & this ) {
-        ((thread&)this){ "Reader Thread", *the_cluster };
+        ((thread&)this){ "Reader Thread", *the_benchmark_cluster };
 }
 
-struct my_processor {
-        processor p;
-};
-
-void ?{}( my_processor & this ) {
-        (this.p){ "I/O Processor", *the_cluster };
+int do_read(int fd, struct iovec * iov) {
+        // extern ssize_t cfa_preadv2(int, const struct iovec *, int, off_t, int, int = 0, Duration = -1`s, io_cancellation * = 0p, io_context * = 0p);
+        int sflags = 0;
+        if(fixed_file) {
+                sflags |= CFA_IO_FIXED_FD1;
+        }
+        return cfa_preadv2(fd, iov, 1, 0, 0, sflags, -1`s, 0p, 0p);
 }
 
 void main( Reader & ) {
-        while(!__atomic_load_n(&run, __ATOMIC_RELAXED)) yield();
+        park( __cfaabi_dbg_ctx );
+        /* paranoid */ assert( true == __atomic_load_n(&run, __ATOMIC_RELAXED) );
 
-        char data[buflen];
+        __attribute__((aligned(512)))  char data[buflen];
         struct iovec iov = { data, buflen };
 
         while(__atomic_load_n(&run, __ATOMIC_RELAXED)) {
-                int r = cfa_preadv2(fd, &iov, 1, 0, 0);
-                if(r < 0) abort("%s\n", strerror(-r));
+                int r = do_read(fd, &iov);
+                if(r < 0) abort("%s\n", strerror(errno));
 
                 __atomic_fetch_add( &count, 1, __ATOMIC_SEQ_CST );
@@ -56 +63 @@
 
 int main(int argc, char * argv[]) {
-        double duration   = 5.0;
-        unsigned long int nthreads = 2;
-        unsigned long int nprocs   = 1;
-        int flags = 0;
+        BENCH_DECL
+        unsigned num_io = 1;
+        io_context_params params;
+        int file_flags = 0;
+        unsigned sublen = 16;
 
         arg_loop:
         for(;;) {
                 static struct option options[] = {
-                        {"duration",   required_argument, 0, 'd'},
-                        {"nthreads",   required_argument, 0, 't'},
-                        {"nprocs",     required_argument, 0, 'p'},
-                        {"bufsize",    required_argument, 0, 'b'},
-                        {"userthread", no_argument      , 0, 'u'},
+                        BENCH_OPT_LONG
+                        {"bufsize",       required_argument, 0, 'b'},
+                        {"submitthread",  no_argument      , 0, 's'},
+                        {"eagersubmit",   no_argument      , 0, 'e'},
+                        {"kpollsubmit",   no_argument      , 0, 'k'},
+                        {"kpollcomplete", no_argument      , 0, 'i'},
+                        {"fixed-files",   no_argument      , 0, 'f'},
+                        {"open-direct",   no_argument      , 0, 'o'},
+                        {"submitlength",  required_argument, 0, 'l'},
                         {0, 0, 0, 0}
                 };
 
                 int idx = 0;
-                int opt = getopt_long(argc, argv, "d:t:p:b:u", options, &idx);
+                int opt = getopt_long(argc, argv, BENCH_OPT_SHORT "b:sekil:", options, &idx);
 
                 const char * arg = optarg ? optarg : "";
@@ -81 +93 @@
                         case -1:
                                 break arg_loop;
-                        // Numeric Arguments
-                        case 'd':
-                                duration = strtod(arg, &end);
-                                if(*end != '\0') {
-                                        fprintf(stderr, "Duration must be a valid double, was %s\n", arg);
-                                        goto usage;
-                                }
-                                break;
-                        case 't':
-                                nthreads = strtoul(arg, &end, 10);
-                                if(*end != '\0' || nthreads < 1) {
-                                        fprintf(stderr, "Number of threads must be a positive integer, was %s\n", arg);
-                                        goto usage;
-                                }
-                                break;
-                        case 'p':
-                                nprocs = strtoul(arg, &end, 10);
-                                if(*end != '\0' || nprocs < 1) {
-                                        fprintf(stderr, "Number of processors must be a positive integer, was %s\n", arg);
-                                        goto usage;
-                                }
-                                break;
+                        BENCH_OPT_CASE
                         case 'b':
                                 buflen = strtoul(arg, &end, 10);
@@ -110 +101 @@
                                 }
                                 break;
-                        case 'u':
-                                flags |= CFA_CLUSTER_IO_POLLER_USER_THREAD;
+                        case 's':
+                                params.poller_submits = true;
                                 break;
-                        // Other cases
+                        case 'e':
+                                params.eager_submits = true;
+                                break;
+                        case 'k':
+                                params.poll_submit = true;
+                        case 'f':
+                                fixed_file = true;
+                                break;
+                        case 'i':
+                                params.poll_complete = true;
+                        case 'o':
+                                file_flags |= O_DIRECT;
+                                break;
+                        case 'l':
+                                sublen = strtoul(arg, &end, 10);
+                                if(*end != '\0' && sublen < 16) {
+                                        fprintf(stderr, "Submit length must be at least 16, was %s\n", arg);
+                                        goto usage;
+                                }
+                                // flags |= (sublen << CFA_CLUSTER_IO_BUFFLEN_OFFSET);
+                                break;
                         default: /* ? */
                                 fprintf(stderr, "%d\n", opt);
                         usage:
-                                fprintf(stderr, "Usage: %s : [options]\n", argv[0]);
-                                fprintf(stderr, "\n");
-                                fprintf(stderr, "  -d, --duration=DURATION  Duration of the experiment, in seconds\n");
-                                fprintf(stderr, "  -t, --nthreads=NTHREADS  Number of user threads\n");
-                                fprintf(stderr, "  -p, --nprocs=NPROCS      Number of kernel threads\n");
-                                fprintf(stderr, "  -b, --buflen=SIZE        Number of bytes to read per request\n");
+                                bench_usage( argv );
+                                fprintf( stderr, "  -b, --buflen=SIZE        Number of bytes to read per request\n" );
+                                fprintf( stderr, "  -u, --userthread         If set, cluster uses user-thread to poll I/O\n" );
+                                fprintf( stderr, "  -s, --submitthread       If set, cluster uses polling thread to submit I/O\n" );
+                                fprintf( stderr, "  -e, --eagersubmit        If set, cluster submits I/O eagerly but still aggregates submits\n" );
+                                fprintf( stderr, "  -k, --kpollsubmit        If set, cluster uses IORING_SETUP_SQPOLL\n" );
+                                fprintf( stderr, "  -i, --kpollcomplete      If set, cluster uses IORING_SETUP_IOPOLL\n" );
+                                fprintf( stderr, "  -l, --submitlength=LEN   Max number of submitions that can be submitted together\n" );
                                 exit(EXIT_FAILURE);
                 }
         }
 
-        fd = open(__FILE__, 0);
-        if(fd < 0) {
+        int lfd = open(__FILE__, file_flags);
+        if(lfd < 0) {
                 fprintf(stderr, "Could not open source file\n");
                 exit(EXIT_FAILURE);
         }
 
-        printf("Running %lu threads, reading %lu bytes each, over %lu processors for %lf seconds\n", nthreads, buflen, nprocs, duration);
+        printf("Running %d threads, reading %lu bytes each, over %d processors for %f seconds\n", nthreads, buflen, nprocs, duration);
 
         {
                 Time start, end;
-                cluster cl = { "IO Cluster", flags };
-                the_cluster = &cl;
-                #if !defined(__CFA_NO_STATISTICS__)
-                        print_stats_at_exit( cl );
-                #endif
+                BenchCluster cl = { num_io, params, CFA_STATS_READY_Q | CFA_STATS_IO };
+
+                if(fixed_file) {
+                        fd = 0;
+                        register_fixed_files( cl.self, &lfd, 1 );
+                }
+                else {
+                        fd = lfd;
+                }
+
                 {
-                        my_processor procs[nprocs];
+                        BenchProc procs[nprocs];
                         {
                                 Reader threads[nthreads];
 
                                 printf("Starting\n");
-                                start = getTime();
+                                bool is_tty = isatty(STDOUT_FILENO);
+                                start = getTimeNsec();
                                 run = true;
-                                do {
-                                        sleep(500`ms);
-                                        end = getTime();
-                                } while( (end - start) < duration`s );
+
+                                for(i; nthreads) {
+                                        unpark( threads[i] __cfaabi_dbg_ctx2 );
+                                }
+                                wait(duration, start, end, is_tty);
+
                                 run = false;
-                                end = getTime();
-                                printf("Done\n");
+                                end = getTimeNsec();
+                                printf("\nDone\n");
                         }
+                        printf("Readers closed\n");
                 }
                 printf("Took %'ld ms\n", (end - start)`ms);
@@ -166 +188 @@
         }
 
-        close(fd);
+        close(lfd);
 }