Changeset ece0e80

Timestamp:

Jan 9, 2021, 4:27:57 PM (3 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:

561dd26

Parents:

35fd2c4

Message:

Added prints.
Naive implementation of cancel.
Server now shutdown cleanly.

Files:

• benchmark/io/http/main.cfa (modified) (4 diffs)
• benchmark/io/http/protocol.cfa (modified) (4 diffs)
• benchmark/io/http/protocol.hfa (modified) (2 diffs)
• benchmark/io/http/worker.cfa (modified) (5 diffs)
• benchmark/io/http/worker.hfa (modified) (1 diff)
• libcfa/src/concurrency/io.cfa (modified) (8 diffs)
• libcfa/src/concurrency/io/call.cfa.in (modified) (1 diff)
• libcfa/src/concurrency/io/setup.cfa (modified) (7 diffs)

benchmark/io/http/main.cfa

@@ -114 +114 @@
                 options.clopts.instance = &cl;
 
-                init_protocol();
 
                 int pipe_cnt = options.clopts.nworkers * 2;
@@ -131 +130 @@
                 {
                         ServerProc procs[options.clopts.nprocs];
+
+                        init_protocol();
                         {
                                 Worker workers[options.clopts.nworkers];
@@ -158 +159 @@
                                         printf("Shutting Down\n");
                                 }
+
+                                for(i; options.clopts.nworkers) {
+                                        printf("Cancelling %p\n", (void*)workers[i].cancel.target);
+                                        cancel(workers[i].cancel);
+                                }
+
+                                printf("Shutting down socket\n");
+                                int ret = shutdown( server_fd, SHUT_RD );
+                                if( ret < 0 ) { abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); }
+
+                                //===================
+                                // Close Socket
+                                printf("Closing Socket\n");
+                                ret = close( server_fd );
+                                if(ret < 0) {
+                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
+                                }
                         }
                         printf("Workers Closed\n");
+
+                        deinit_protocol();
                 }
 
@@ -170 +190 @@
                 free(fds);
 
-                deinit_protocol();
-        }
-
-        //===================
-        // Close Socket
-        printf("Closing Socket\n");
-        ret = close( server_fd );
-        if(ret < 0) {
-                abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
         }
 

benchmark/io/http/protocol.cfa

@@ -71 +71 @@
 }
 
-[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len, io_cancellation * cancel) {
         char * it = buffer;
         size_t count = len - 1;
@@ -77 +77 @@
         READ:
         for() {
-                int ret = cfa_read(fd, (void*)it, count, 0, -1`s, 0p, 0p);
+                int ret = cfa_read(fd, (void*)it, count, 0, -1`s, cancel, 0p);
                 // int ret = read(fd, (void*)it, count);
                 if(ret == 0 ) return [OK200, true, 0, 0];
@@ -148 +148 @@
 
 void ?{}( DateFormater & this ) {
-        ((thread&)this){ *options.clopts.instance };
+        ((thread&)this){ "Server Date Thread", *options.clopts.instance };
         this.idx = 0;
         memset( this.buffers[0].buff, 0, sizeof(this.buffers[0]) );
@@ -162 +162 @@
 
                 Time now = getTimeNsec();
-                // Date: Wed, 17 Apr 2013 12:00:00 GMT
+
                 strftime( this.buffers[this.idx].buff, 100, "%a, %d %b %Y %H:%M:%S %Z", now );
-                printf("Changing date to %s\n", this.buffers[this.idx].buff);
 
                 char * next = this.buffers[this.idx].buff;

benchmark/io/http/protocol.hfa

@@ -1 +1 @@
 #pragma once
+
+struct io_cancellation;
 
 enum HttpCode {
@@ -15 +17 @@
 int answer_header( int fd, size_t size );
 
-[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len);
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len, io_cancellation *);
 
 void sendfile( int pipe[2], int fd, int ans_fd, size_t count );

benchmark/io/http/worker.cfa

@@ -28 +28 @@
         CONNECTION:
         for() {
-                int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], 0, -1`s, 0p, 0p );
+                printf("=== Accepting connection ===\n");
+                int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], 0, -1`s, &this.cancel, 0p );
                 if(fd < 0) {
                         if( errno == ECONNABORTED ) break;
+                        if( errno == EINVAL ) break;
                         abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                 }
 
-                printf("New connection %d, waiting for requests\n", fd);
+                printf("=== New connection %d, waiting for requests ===\n", fd);
                 REQUEST:
                 for() {
@@ -45 +47 @@
                         size_t len = options.socket.buflen;
                         char buffer[len];
-                        printf("Reading request\n");
-                        [code, closed, file, name_size] = http_read(fd, buffer, len);
+                        printf("=== Reading request ===\n");
+                        [code, closed, file, name_size] = http_read(fd, buffer, len, &this.cancel);
 
                         // if we are done, break out of the loop
                         if( closed ) {
-                                printf("Connection closed\n");
+                                printf("=== Connection closed ===\n");
                                 continue CONNECTION;
                         }
@@ -56 +58 @@
                         // If this wasn't a request retrun 400
                         if( code != OK200 ) {
-                                printf("Invalid Request : %d\n", code_val(code));
+                                printf("=== Invalid Request : %d ===\n", code_val(code));
                                 answer_error(fd, code);
                                 continue REQUEST;
                         }
 
-                        printf("Request for file %.*s\n", (int)name_size, file);
+                        printf("=== Request for file %.*s ===\n", (int)name_size, file);
 
                         // Get the fd from the file cache
@@ -70 +72 @@
                         // If we can't find the file, return 404
                         if( ans_fd < 0 ) {
-                                printf("File Not Found\n");
+                                printf("=== File Not Found ===\n");
                                 answer_error(fd, E404);
                                 continue REQUEST;
@@ -81 +83 @@
                         sendfile( this.pipe, fd, ans_fd, count);
 
-                        printf("File sent\n");
+                        printf("=== File sent ===\n");
                 }
         }

benchmark/io/http/worker.hfa

@@ -17 +17 @@
         socklen_t * addrlen;
         int flags;
+        io_cancellation cancel;
 };
 void ?{}( Worker & this);

libcfa/src/concurrency/io.cfa

@@ -134 +134 @@
                 int ret = 0;
                 if( need_sys_to_submit || need_sys_to_complete ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O : IO_URING enter %d %u %u\n", ring.fd, to_submit, flags);
                         ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, flags, (sigset_t *)0p, _NSIG / 8);
                         if( ret < 0 ) {
@@ -230 +231 @@
                 __cfadbg_print_safe(io_core, "Kernel I/O : IO poller %p for ring %p ready\n", &this, &this.ring);
 
-                int reset = 0;
+                const int reset_cnt = 5;
+                int reset = reset_cnt;
                 // Then loop until we need to start
+                LOOP:
                 while(!__atomic_load_n(&this.done, __ATOMIC_SEQ_CST)) {
                         // Drain the io
@@ -239 +242 @@
                                 [count, again] = __drain_io( *this.ring );
 
-                                if(!again) reset++;
+                                if(!again) reset--;
 
                                 // Update statistics
@@ -249 +252 @@
 
                         // If we got something, just yield and check again
-                        if(reset < 5) {
+                        if(reset > 1) {
                                 yield();
-                        }
-                        // We didn't get anything baton pass to the slow poller
-                        else {
+                                continue LOOP;
+                        }
+
+                        // We alread failed to find events a few time.
+                        if(reset == 1) {
+                                // Rearm the context so it can block
+                                // but don't block right away
+                                // we need to retry one last time in case
+                                // something completed *just now*
+                                __ioctx_prepare_block( this, ev );
+                                continue LOOP;
+                        }
+
                                 __STATS__( false,
                                         io.complete_q.blocks += 1;
                                 )
                                 __cfadbg_print_safe(io_core, "Kernel I/O : Parking io poller %p\n", &this.self);
-                                reset = 0;
 
                                 // block this thread
-                                __ioctx_prepare_block( this, ev );
                                 wait( this.sem );
-                        }
+
+                        // restore counter
+                        reset = reset_cnt;
                 }
 
@@ -319 +332 @@
                                         )
 
+                                        __cfadbg_print_safe( io, "Kernel I/O : allocated [%p, %u] for %p (%p)\n", sqe, idx, active_thread(), (void*)data );
 
                                         // Success return the data
@@ -376 +390 @@
 
         void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
+                __cfadbg_print_safe( io, "Kernel I/O : submitting %u for %p\n", idx, active_thread() );
+
                 __io_data & ring = *ctx->thrd.ring;
                 // Get now the data we definetely need
@@ -443 +459 @@
                                 unlock(ring.submit_q.submit_lock);
                         #endif
-                        if( ret < 0 ) return;
+                        if( ret < 0 ) {
+                                return;
+                        }
 
                         // Release the consumed SQEs
@@ -454 +472 @@
                                 io.submit_q.submit_avg.cnt += 1;
                         )
+
+                        __cfadbg_print_safe( io, "Kernel I/O : submitted %u (among %u) for %p\n", idx, ret, active_thread() );
                 }
                 else {

libcfa/src/concurrency/io/call.cfa.in

@@ -464 +464 @@
 
 print("""
+//-----------------------------------------------------------------------------
+bool cancel(io_cancellation & this) {
+        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_ASYNC_CANCEL)
+                return false;
+        #else
+                io_future_t future;
+
+                io_context * context = __get_io_context();
+
+                __u8 sflags = 0;
+                struct __io_data & ring = *context->thrd.ring;
+
+                __u32 idx;
+                struct io_uring_sqe * sqe;
+                [sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&future );
+
+                sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0;
+                sqe->opcode = IORING_OP_ASYNC_CANCEL;
+                sqe->flags = sflags;
+                sqe->addr = this.target;
+
+                verify( sqe->user_data == (__u64)(uintptr_t)&future );
+                __submit( context, idx );
+
+                wait(future);
+
+                if( future.result == 0 ) return true; // Entry found
+                if( future.result == -EALREADY) return true; // Entry found but in progress
+                if( future.result == -ENOENT ) return false; // Entry not found
+                return false;
+        #endif
+}
+
 //-----------------------------------------------------------------------------
 // Check if a function is has asynchronous

libcfa/src/concurrency/io/setup.cfa

@@ -169 +169 @@
                 // Main loop
                 while( iopoll.run ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : waiting on io_uring contexts\n");
+
                         // Wait for events
                         int nfds = epoll_pwait( iopoll.epollfd, events, 10, -1, &mask );
+
+                        __cfadbg_print_safe(io_core, "Kernel I/O - epoll : %d io contexts events, waking up\n", nfds);
 
                         // Check if an error occured
@@ -181 +185 @@
                                 $io_ctx_thread * io_ctx = ($io_ctx_thread *)(uintptr_t)events[i].data.u64;
                                 /* paranoid */ verify( io_ctx );
-                                __cfadbg_print_safe(io_core, "Kernel I/O : Unparking io poller %p\n", io_ctx);
+                                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Unparking io poller %p\n", io_ctx);
                                 #if !defined( __CFA_NO_STATISTICS__ )
                                         __cfaabi_tls.this_stats = io_ctx->self.curr_cluster->stats;
@@ -233 +237 @@
                 $thread & thrd = this.thrd.self;
                 if( cluster_context ) {
+                        // We are about to do weird things with the threads
+                        // we don't need interrupts to complicate everything
+                        disable_interrupts();
+
+                        // Get cluster info
                         cluster & cltr = *thrd.curr_cluster;
                         /* paranoid */ verify( cltr.idles.total == 0 || &cltr == mainCluster );
@@ -239 +248 @@
                         // We need to adjust the clean-up based on where the thread is
                         if( thrd.state == Ready || thrd.preempted != __NO_PREEMPTION ) {
+                                // This is the tricky case
+                                // The thread was preempted or ready to run and now it is on the ready queue
+                                // but the cluster is shutting down, so there aren't any processors to run the ready queue
+                                // the solution is to steal the thread from the ready-queue and pretend it was blocked all along
 
                                 ready_schedule_lock();
-
-                                        // This is the tricky case
-                                        // The thread was preempted and now it is on the ready queue
+                                        // The thread should on the list
+                                        /* paranoid */ verify( thrd.link.next != 0p );
+
+                                        // Remove the thread from the ready queue of this cluster
                                         // The thread should be the last on the list
-                                        /* paranoid */ verify( thrd.link.next != 0p );
-
-                                        // Remove the thread from the ready queue of this cluster
                                         __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
                                         /* paranoid */ verify( removed );
@@ -263 +274 @@
                         }
                         // !!! This is not an else if !!!
+                        // Ok, now the thread is blocked (whether we cheated to get here or not)
                         if( thrd.state == Blocked ) {
-
                                 // This is the "easy case"
                                 // The thread is parked and can easily be moved to active cluster
@@ -274 +285 @@
                         }
                         else {
-
                                 // The thread is in a weird state
                                 // I don't know what to do here
                                 abort("io_context poller thread is in unexpected state, cannot clean-up correctly\n");
                         }
+
+                        // The weird thread kidnapping stuff is over, restore interrupts.
+                        enable_interrupts( __cfaabi_dbg_ctx );
                 } else {
                         post( this.thrd.sem );
@@ -463 +476 @@
 
         void __ioctx_prepare_block($io_ctx_thread & ctx, struct epoll_event & ev) {
+                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %p\n", &ctx);
                 int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_MOD, ctx.ring->fd, &ev);
                 if (ret < 0) {