Changeset d11d6eb

Timestamp:

Aug 28, 2020, 1:26:28 PM (4 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

8e3034d

Parents:

8e2cb4a

Message:

Fixed some compilation errors.
Fixed file descriptor support in progress.

Location:

benchmark/io/http

Files:

• filecache.cfa (modified) (7 diffs)
• main.cfa (modified) (5 diffs)
• options.cfa (modified) (4 diffs)
• options.hfa (modified) (2 diffs)
• protocol.cfa (modified) (6 diffs)
• worker.cfa (modified) (1 diff)

benchmark/io/http/filecache.cfa

@@ -73 +73 @@
         cache_line * entries;
         size_t size;
+        int * rawfds;
+        int nfds;
 } file_cache;
 
@@ -98 +100 @@
 }
 
-int put_file( cache_line & entry ) {
+int put_file( cache_line & entry, int fd ) {
         uint32_t idx = murmur3_32( (const uint8_t *)entry.file, strlen(entry.file), options.file_cache.hash_seed ) % file_cache.size;
 
@@ -108 +110 @@
 
         file_cache.entries[idx] = entry;
+        file_cache.entries[idx].fd = fd;
         return i > 0 ? 1 : 0;
 }
@@ -121 +124 @@
         size_t fcount = 0;
         size_t fsize = 16;
-        cache_line * raw = 0p;
-        raw = alloc(raw, fsize, true);
+        cache_line * raw = alloc(fsize);
         // Step 1 get a dense array of all files
         int walk(const char *fpath, const struct stat *sb, int typeflag) {
@@ -131 +133 @@
                 if(fcount > fsize) {
                         fsize *= 2;
-                        raw = alloc(raw, fsize, true);
+                        raw = alloc(fsize, raw`realloc);
                 }
 
@@ -162 +164 @@
         file_cache.entries = anew(file_cache.size);
 
+        if(options.file_cache.fixed_fds) {
+                file_cache.nfds   = fcount;
+                file_cache.rawfds = alloc(fcount);
+        }
+
         // Step 3 fill the cache
         int conflicts = 0;
         for(i; fcount) {
-                conflicts += put_file( raw[i] );
+                int fd;
+                if(options.file_cache.fixed_fds) {
+                        file_cache.rawfds[i] = raw[i].fd;
+                        fd = i;
+                }
+                else {
+                        fd = raw[i].fd;
+                }
+                conflicts += put_file( raw[i], fd );
         }
         printf("Filled cache from path \"%s\" with %zu files\n", path, fcount);
@@ -197 +212 @@
         }
 
-        return [aalloc(extra), 0];
+        size_t s = file_cache.nfds + extra;
+        int * data = alloc(s, file_cache.rawfds`realloc);
+        return [data, file_cache.nfds];
 }
 

benchmark/io/http/main.cfa

@@ -12 +12 @@
 #include <kernel.hfa>
 #include <stats.hfa>
+#include <time.hfa>
 #include <thread.hfa>
 
@@ -18 +19 @@
 #include "options.hfa"
 #include "worker.hfa"
+
+extern void register_fixed_files( cluster &, int *, unsigned count );
+
+Duration default_preemption() {
+        return 0;
+}
 
 //=============================================================================================
@@ -84 +91 @@
         // Run Server Cluster
         {
-                cluster cl = { "Server Cluster", options.clopts.flags };
+                cluster cl = { "Server Cluster", options.clopts.params };
                 #if !defined(__CFA_NO_STATISTICS__)
                         print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
@@ -102 +109 @@
                 }
 
+                if(options.file_cache.fixed_fds) {
+                        register_fixed_files(cl, fds, pipe_off);
+                }
+
                 {
                         ServerProc procs[options.clopts.nprocs];
@@ -107 +118 @@
                                 Worker workers[options.clopts.nworkers];
                                 for(i; options.clopts.nworkers) {
-                                        if( options.file_cache.fixed_fds ) {
-                                                workers[i].pipe[0] = pipe_off + (i * 2) + 0;
-                                                workers[i].pipe[1] = pipe_off + (i * 2) + 1;
-                                        }
-                                        else {
+                                        // if( options.file_cache.fixed_fds ) {
+                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
+                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
+                                        // }
+                                        // else
+                                        {
                                                 workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
                                                 workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];

benchmark/io/http/options.cfa

@@ -39 +39 @@
 
 const char * parse_options( int argc, char * argv[] ) {
-        bool uthrdpo = false;
         bool subthrd = false;
         bool eagrsub = false;
@@ -57 +56 @@
                 {'C', "cache-size",     "Size of the cache to use, if set to small, will uses closes power of 2", options.file_cache.size },
                 {'l', "list-files",     "List the files in the specified path and exit", options.file_cache.list, parse_settrue },
-                {'u', "userthread",     "If set, cluster uses user-thread to poll I/O", uthrdpo, parse_settrue },
                 {'s', "submitthread",   "If set, cluster uses polling thread to submit I/O", subthrd, parse_settrue },
                 {'e', "eagersubmit",    "If set, cluster submits I/O eagerly but still aggregates submits", eagrsub, parse_settrue},
@@ -71 +69 @@
         parse_args( argc, argv, opt, opt_cnt, "[OPTIONS]... [PATH]\ncforall http server", left );
 
-        if( uthrdpo ) {
-                options.clopts.flags |= CFA_CLUSTER_IO_POLLER_USER_THREAD;
-        }
-
-        if( subthrd ) {
-                options.clopts.flags |= CFA_CLUSTER_IO_POLLER_THREAD_SUBMITS;
-        }
-
-        if( eagrsub ) {
-                options.clopts.flags |= CFA_CLUSTER_IO_EAGER_SUBMITS;
-        }
+        options.clopts.params.poller_submits = subthrd;
+        options.clopts.params.eager_submits  = eagrsub;
 
         if( fixedfd ) {
@@ -88 +77 @@
 
         if( sqkpoll ) {
-                options.clopts.flags |= CFA_CLUSTER_IO_KERNEL_POLL_SUBMITS;
+                options.clopts.params.poll_submit = true;
                 options.file_cache.fixed_fds = true;
         }
 
         if( iokpoll ) {
-                options.clopts.flags |= CFA_CLUSTER_IO_KERNEL_POLL_COMPLETES;
+                options.clopts.params.poll_complete = true;
                 options.file_cache.open_flags |= O_DIRECT;
         }
 
-        options.clopts.flags |= (sublen << CFA_CLUSTER_IO_BUFFLEN_OFFSET);
+        options.clopts.params.num_ready = sublen;
 
         if( left[0] == 0p ) { return "."; }

benchmark/io/http/options.hfa

@@ -2 +2 @@
 
 #include <stdint.h>
+
+#include <kernel.hfa>
 
 struct cluster;
@@ -23 +25 @@
                 int nprocs;
                 int nworkers;
-                int flags;
+                io_context_params params;
                 int chan_size;
                 bool procstats;

benchmark/io/http/protocol.cfa

@@ -11 +11 @@
 extern "C" {
       int snprintf ( char * s, size_t n, const char * format, ... );
+        #include <linux/io_uring.h>
 }
 #include <string.h>
-
 #include <errno.h>
 
+#include "options.hfa"
 
 const char * http_msgs[] = {
@@ -74 +75 @@
         READ:
         for() {
-                int ret = cfa_read(fd, it, count);
-                if(ret == 0 ) return [OK200, true, 0p, 0];
+                int ret = cfa_read(fd, (void*)it, count, 0, -1`s, 0p, 0p);
+                if(ret == 0 ) return [OK200, true, 0, 0];
                 if(ret < 0 ) {
                         if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
@@ -88 +89 @@
                 count -= ret;
 
-                if( count < 1 ) return [E414, false, 0p, 0];
+                if( count < 1 ) return [E414, false, 0, 0];
         }
 
@@ -95 +96 @@
         it = buffer;
         int ret = memcmp(it, "GET /", 5);
-        if( ret != 0 ) return [E400, false, 0p, 0];
+        if( ret != 0 ) return [E400, false, 0, 0];
         it += 5;
 
@@ -106 +107 @@
         ssize_t ret;
         SPLICE1: while(count > 0) {
-                ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, SPLICE_F_MOVE | SPLICE_F_MORE);
+                ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, SPLICE_F_MOVE | SPLICE_F_MORE, 0, -1`s, 0p, 0p);
                 if( ret < 0 ) {
                         if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1;
@@ -116 +117 @@
                 size_t in_pipe = ret;
                 SPLICE2: while(in_pipe > 0) {
-                        ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, SPLICE_F_MOVE | SPLICE_F_MORE);
+                        ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, SPLICE_F_MOVE | SPLICE_F_MORE, 0, -1`s, 0p, 0p);
                         if( ret < 0 ) {
                                 if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2;

benchmark/io/http/worker.cfa

@@ -96 +96 @@
 void main( Acceptor & this ) {
         for() {
-                int ret = cfa_accept4( this.[sockfd, addr, addrlen, flags] );
+                int ret = cfa_accept4( this.[sockfd, addr, addrlen, flags], 0, -1`s, 0p, 0p );
                 if(ret < 0) {
                         if( errno == ECONNABORTED ) break;