Index: benchmark/io/http/Makefile.am =================================================================== --- benchmark/io/http/Makefile.am (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ benchmark/io/http/Makefile.am (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -21,5 +21,5 @@ include $(top_srcdir)/tools/build/cfa.make -AM_CFLAGS = -O3 -Wall -Wextra -I$(srcdir) -lrt -pthread # -Werror +AM_CFLAGS = -O3 -Wall -Wextra -I$(srcdir) -lrt -pthread -g # -Werror AM_CFAFLAGS = -quiet -nodebug AM_LDFLAGS = -quiet -nodebug @@ -39,4 +39,6 @@ protocol.cfa \ protocol.hfa \ + socket.cfa \ + socket.hfa \ worker.cfa \ worker.hfa Index: benchmark/io/http/main.cfa =================================================================== --- benchmark/io/http/main.cfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ benchmark/io/http/main.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -2,4 +2,5 @@ #include +#include #include #include @@ -8,4 +9,5 @@ #include #include + #include #include #include @@ -14,4 +16,5 @@ #include #include +#include #include #include @@ -21,4 +24,5 @@ #include "filecache.hfa" #include "options.hfa" +#include "socket.hfa" #include "worker.hfa" @@ -36,4 +40,5 @@ Worker * workers; int worker_cnt; + condition_variable(fast_block_lock) var; }; @@ -54,5 +59,5 @@ or else {} - sleep(10`s); + wait(this.var, 10`s); print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO ); @@ -177,7 +182,21 @@ //============================================================================================= +// Termination +//============================================================================================= + +int closefd; +void cleanstop(int) { + eventfd_t buffer = 1; + char * buffer_s = (char*)&buffer; + int ret = write(closefd, buffer_s, sizeof(buffer)); + if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) ); + return; +} + +//============================================================================================= // Main //=============================================================================================' int main( int argc, char * argv[] ) { + int ret; __sighandler_t s = 1p; signal(SIGPIPE, s); @@ -186,4 +205,18 @@ // Parse args parse_options(argc, argv); + + //=================== + // Setup non-interactive termination + if(!options.interactive) { + closefd = eventfd(0, 0); + if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) ); + + sighandler_t prev = signal(SIGTERM, cleanstop); + intptr_t prev_workaround = (intptr_t) prev; + // can't use SIG_ERR it crashes the compiler + if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) ); + + sout | "Signal termination ready"; + } //=================== @@ -197,45 +230,11 @@ // Open Socket sout | getpid() | ": Listening on port" | options.socket.port; - int server_fd = socket(AF_INET, SOCK_STREAM, 0); - if(server_fd < 0) { - abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) ); - } - - int ret = 0; + struct sockaddr_in address; - int addrlen = sizeof(address); - memset( (char *)&address, '\0' ); - address.sin_family = AF_INET; - address.sin_addr.s_addr = htonl(INADDR_ANY); - address.sin_port = htons( options.socket.port ); - - int waited = 0; - for() { - int sockfd = server_fd; - __CONST_SOCKADDR_ARG addr; - addr.__sockaddr__ = (struct sockaddr *)&address; - socklen_t addrlen = sizeof(address); - ret = bind( sockfd, addr, addrlen ); - if(ret < 0) { - if(errno == EADDRINUSE) { - if(waited == 0) { - if(!options.interactive) abort | "Port already in use in non-interactive mode. Aborting"; - sout | "Waiting for port"; - } else { - sout | "\r" | waited | nonl; - flush( sout ); - } - waited ++; - sleep( 1`s ); - continue; - } - abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) ); - } - break; - } - - ret = listen( server_fd, options.socket.backlog ); - if(ret < 0) { - abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) ); + int addrlen = prepaddr(address); + + int server_fd; + if(!options.socket.manyreuse) { + server_fd = listener(address, addrlen); } @@ -257,9 +256,11 @@ { + // Stats printer makes a copy so this needs to persist longer than normal + Worker * workers; ServerCluster cl[options.clopts.nclusters]; init_protocol(); { - Worker * workers = anew(options.clopts.nworkers); + workers = anew(options.clopts.nworkers); cl[0].prnt->workers = workers; cl[0].prnt->worker_cnt = options.clopts.nworkers; @@ -273,5 +274,5 @@ workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0]; workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1]; - workers[i].sockfd = server_fd; + workers[i].sockfd = options.socket.manyreuse ? listener(address, addrlen) : server_fd; workers[i].addr = (struct sockaddr *)&address; workers[i].addrlen = (socklen_t*)&addrlen; @@ -285,14 +286,20 @@ } sout | nl; - if(!options.interactive) park(); { - char buffer[128]; - for() { - int ret = cfa_read(0, buffer, 128, 0); - if(ret == 0) break; + if(options.interactive) { + char buffer[128]; + for() { + int ret = cfa_read(0, buffer, 128, 0); + if(ret == 0) break; + if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) ); + sout | "User wrote '" | "" | nonl; + write(sout, buffer, ret - 1); + sout | "'"; + } + } + else { + char buffer[sizeof(eventfd_t)]; + int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0); if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) ); - sout | "User wrote '" | "" | nonl; - write(sout, buffer, ret - 1); - sout | "'"; } @@ -307,7 +314,15 @@ sout | "Shutting down socket..." | nonl; flush( sout ); - int ret = shutdown( server_fd, SHUT_RD ); - if( ret < 0 ) { - abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); + if(options.socket.manyreuse) { + for(i; options.clopts.nworkers) { + ret = shutdown( workers[i].sockfd, SHUT_RD ); + if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); + } + } + else { + ret = shutdown( server_fd, SHUT_RD ); + if( ret < 0 ) { + abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); + } } sout | "done"; @@ -316,12 +331,26 @@ // Close Socket sout | "Closing Socket..." | nonl; flush( sout ); - ret = close( server_fd ); - if(ret < 0) { - abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); + if(options.socket.manyreuse) { + for(i; options.clopts.nworkers) { + ret = close(workers[i].sockfd); + if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); + } + } + else { + ret = close( server_fd ); + if(ret < 0) { + abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); + } } sout | "done"; sout | "Stopping connection threads..." | nonl; flush( sout ); - adelete(workers); + for(i; options.clopts.nworkers) { + for(j; 2) { + ret = close(workers[i].pipe[j]); + if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); + } + join(workers[i]); + } } sout | "done"; @@ -331,15 +360,28 @@ sout | "done"; + sout | "Stopping printer threads..." | nonl; flush( sout ); + for(i; options.clopts.nclusters) { + StatsPrinter * p = cl[i].prnt; + if(p) { + notify_one(p->var); + join(*p); + } + } + sout | "done"; + + // Now that the stats printer is stopped, we can reclaim this + adelete(workers); + sout | "Stopping processors/clusters..." | nonl; flush( sout ); } sout | "done"; - sout | "Closing splice fds..." | nonl; flush( sout ); - for(i; pipe_cnt) { - ret = close( fds[pipe_off + i] ); - if(ret < 0) { - abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) ); - } - } + // sout | "Closing splice fds..." | nonl; flush( sout ); + // for(i; pipe_cnt) { + // ret = close( fds[pipe_off + i] ); + // if(ret < 0) { + // abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) ); + // } + // } free(fds); sout | "done"; Index: benchmark/io/http/options.cfa =================================================================== --- benchmark/io/http/options.cfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ benchmark/io/http/options.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -35,7 +35,9 @@ { // socket - 8080, // port - 10, // backlog - 1024 // buflen + 8080, // port + 10, // backlog + 1024, // buflen + false, // onereuse + false // manyreuse }, @@ -70,4 +72,6 @@ {'\0', "shell", "Disable interactive mode", options.interactive, parse_setfalse}, {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog}, + {'\0', "reuseport-one", "Create a single listen socket with SO_REUSEPORT", options.socket.onereuse, parse_settrue}, + {'\0', "reuseport", "Use many listen sockets with SO_REUSEPORT", options.socket.manyreuse, parse_settrue}, {'\0', "request_len", "Maximum number of bytes in the http request, requests with more data will be answered with Http Code 414", options.socket.buflen}, {'\0', "seed", "seed to use for hashing", options.file_cache.hash_seed }, Index: benchmark/io/http/options.hfa =================================================================== --- benchmark/io/http/options.hfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ benchmark/io/http/options.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -27,4 +27,6 @@ int backlog; int buflen; + bool onereuse; + bool manyreuse; } socket; Index: benchmark/io/http/socket.cfa =================================================================== --- benchmark/io/http/socket.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/socket.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -0,0 +1,72 @@ +#define _GNU_SOURCE + +#include "socket.hfa" + +#include +#include +extern "C" { + #include + #include +} + +#include +#include +#include + +#include "options.hfa" + +int prepaddr(struct sockaddr_in & address) { + int addrlen = sizeof(address); + memset( (char *)&address, '\0', addrlen ); + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_ANY); + address.sin_port = htons( options.socket.port ); + return addrlen; +} + +int listener(struct sockaddr_in & address, int addrlen) { + int sockfd = socket(AF_INET, SOCK_STREAM, 0); + if(sockfd < 0) { + abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) ); + } + + if(options.socket.onereuse || options.socket.manyreuse) { + int value = 1; + // if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const void*)&on, sizeof(on))) + // abort( "setsockopt SO_REUSEADDR error: (%d) %s\n", (int)errno, strerror(errno) ); + if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &value, sizeof(int)) < 0) + abort( "setsockopt SO_REUSEPORT error: (%d) %s\n", (int)errno, strerror(errno) ); + } + + int ret = 0; + int waited = 0; + for() { + __CONST_SOCKADDR_ARG addr; + addr.__sockaddr__ = (struct sockaddr *)&address; + socklen_t addrlen = sizeof(address); + ret = bind( sockfd, addr, addrlen ); + if(ret < 0) { + if(errno == EADDRINUSE) { + if(waited == 0) { + if(!options.interactive) abort | "Port already in use in non-interactive mode. Aborting"; + sout | "Waiting for port"; + } else { + sout | "\r" | waited | nonl; + flush( sout ); + } + waited ++; + sleep( 1`s ); + continue; + } + abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) ); + } + break; + } + + ret = listen( sockfd, options.socket.backlog ); + if(ret < 0) { + abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) ); + } + + return sockfd; +} Index: benchmark/io/http/socket.hfa =================================================================== --- benchmark/io/http/socket.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ benchmark/io/http/socket.hfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -0,0 +1,5 @@ +#pragma once + +struct sockaddr_in; +int prepaddr(struct sockaddr_in & addr); +int listener(struct sockaddr_in & address, int addrlen); Index: benchmark/io/http/worker.cfa =================================================================== --- benchmark/io/http/worker.cfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ benchmark/io/http/worker.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -43,4 +43,6 @@ /* paranoid */ assert( this.pipe[0] != -1 ); /* paranoid */ assert( this.pipe[1] != -1 ); + + const bool reuse = options.socket.manyreuse; CONNECTION: Index: libcfa/src/concurrency/io.cfa =================================================================== --- libcfa/src/concurrency/io.cfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ libcfa/src/concurrency/io.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -159,30 +159,39 @@ const __u32 mask = *ctx->cq.mask; + const __u32 num = *ctx->cq.num; unsigned long long ts_prev = ctx->cq.ts; - - // re-read the head and tail in case it already changed. - const __u32 head = *ctx->cq.head; - const __u32 tail = *ctx->cq.tail; - const __u32 count = tail - head; - __STATS__( false, io.calls.drain++; io.calls.completed += count; ) - - for(i; count) { - unsigned idx = (head + i) & mask; - volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx]; - - /* paranoid */ verify(&cqe); - - struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data; - // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future ); - - __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL ); - } - - unsigned long long ts_next = ctx->cq.ts = rdtscl(); - - // Mark to the kernel that the cqe has been seen - // Ensure that the kernel only sees the new value of the head index after the CQEs have been read. - __atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST ); - ctx->proc->idle_wctx.drain_time = ts_next; + unsigned long long ts_next; + + // We might need to do this multiple times if more events completed than can fit in the queue. + for() { + // re-read the head and tail in case it already changed. + const __u32 head = *ctx->cq.head; + const __u32 tail = *ctx->cq.tail; + const __u32 count = tail - head; + __STATS__( false, io.calls.drain++; io.calls.completed += count; ) + + for(i; count) { + unsigned idx = (head + i) & mask; + volatile struct io_uring_cqe & cqe = ctx->cq.cqes[idx]; + + /* paranoid */ verify(&cqe); + + struct io_future_t * future = (struct io_future_t *)(uintptr_t)cqe.user_data; + // __cfadbg_print_safe( io, "Kernel I/O : Syscall completed : cqe %p, result %d for %p\n", &cqe, cqe.res, future ); + + __kernel_unpark( fulfil( *future, cqe.res, false ), UNPARK_LOCAL ); + } + + ts_next = ctx->cq.ts = rdtscl(); + + // Mark to the kernel that the cqe has been seen + // Ensure that the kernel only sees the new value of the head index after the CQEs have been read. + __atomic_store_n( ctx->cq.head, head + count, __ATOMIC_SEQ_CST ); + ctx->proc->idle_wctx.drain_time = ts_next; + + if(likely(count < num)) break; + + ioring_syscsll( *ctx, 0, IORING_ENTER_GETEVENTS); + } __cfadbg_print_safe(io, "Kernel I/O : %u completed age %llu\n", count, ts_next); Index: libcfa/src/concurrency/io/setup.cfa =================================================================== --- libcfa/src/concurrency/io/setup.cfa (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ libcfa/src/concurrency/io/setup.cfa (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -138,5 +138,5 @@ __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256; if( !is_pow2(nentries) ) { - abort("ERROR: I/O setup 'num_entries' must be a power of 2\n"); + abort("ERROR: I/O setup 'num_entries' must be a power of 2, was %u\n", nentries); } Index: src/Concurrency/Waitfor.cc =================================================================== --- src/Concurrency/Waitfor.cc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/Concurrency/Waitfor.cc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -56,5 +56,5 @@ | | | | - | | + | | | | | | Index: src/Concurrency/Waitfor.h =================================================================== --- src/Concurrency/Waitfor.h (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/Concurrency/Waitfor.h (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -19,7 +19,12 @@ class Declaration; +namespace ast { + class TranslationUnit; +} namespace Concurrency { void generateWaitFor( std::list< Declaration * > & translationUnit ); + +void generateWaitFor( ast::TranslationUnit & translationUnit ); }; Index: src/Concurrency/WaitforNew.cpp =================================================================== --- src/Concurrency/WaitforNew.cpp (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) +++ src/Concurrency/WaitforNew.cpp (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -0,0 +1,580 @@ +// +// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo +// +// The contents of this file are covered under the licence agreement in the +// file "LICENCE" distributed with Cforall. +// +// WaitforNew.cpp -- Expand waitfor clauses into code. +// +// Author : Andrew Beach +// Created On : Fri May 27 10:31:00 2022 +// Last Modified By : Andrew Beach +// Last Modified On : Tue Jun 13 13:30:00 2022 +// Update Count : 0 +// + +#include "Waitfor.h" + +#include + +#include "AST/Pass.hpp" +#include "Common/UniqueName.h" +#include "InitTweak/InitTweak.h" +#include "ResolvExpr/Resolver.h" + +#include "AST/Print.hpp" + +using namespace std::string_literals; +using ResolvExpr::ResolveContext; + +/* So this is what this file dones: + +void f(int i, float f, A & mutex b, struct foo * ); +void f(int ); + +...{ + when ( a < 1 ) waitfor( f : a ) { fee(); } + or timeout( getWaitTime() ) { fy(); } + or waitfor( g : a ) { foe(); } + or waitfor( ^?{} : a ) { break; } + or waitfor( ^?{} ) { break; } + or when ( a < 1 ) else { fum(); } +}... + + || + || + \||/ + \/ + +...{ + { + __acceptable_t __acceptables_#[4 ]; + bool __do_run_# = false; + + monitor$ * __monitors_#[1 ] = { a }; + if ( a < 1) { + void (*__function_#)() = f; + __acceptables_#[0].is_dtor = false; + __acceptables_#[0].func = __function_#; + __acceptables_#[0].data = __monitors_#; + __acceptables_#[0].size = 1; + __do_run_# = true; + } + + // Remaining waitfor clauses go here. + + long long unsigned int __timeout_# = -1; + if ( true ) { + __timeout_# = getWaitTime(); + __do_run_# = true; + } + + if ( a < 1 ) { + __timeout_# = 0 + __do_run_# = true; + } + + short int __index_# = -1; + __waitfor_mask_t __mask_# = {&__index_#, {__acceptables_#, ?}}; + __waitfor_internal((__waitfor_mask_t&)__mask_#, __timeout_#); + + switch (__index_#) { + case 0: + { { fee(); } break; } + case 1: + { { foe(); } break; } + case 2: + { break; } + case 3: + { break; } + case -2: + { { fy(); } break; } + case -1: + { { foe(); } break; } + } + } +}... +*/ + +namespace Concurrency { + +namespace { + +class GenerateWaitForCore : + public ast::WithSymbolTable, public ast::WithConstTranslationUnit { + const ast::FunctionDecl * decl_waitfor = nullptr; + const ast::StructDecl * decl_mask = nullptr; + const ast::StructDecl * decl_acceptable = nullptr; + const ast::StructDecl * decl_monitor = nullptr; + + UniqueName namer_acc = "__acceptables_"s; + UniqueName namer_idx = "__index_"s; + UniqueName namer_flg = "__do_run_"s; + UniqueName namer_msk = "__mask_"s; + UniqueName namer_mon = "__monitors_"s; + UniqueName namer_tim = "__timeout_"s; + UniqueName namer_fun = "__function_"s; + + ast::ObjectDecl * declareAcceptables( ast::CompoundStmt * out, + const CodeLocation & location, unsigned long numClauses ); + ast::ObjectDecl * declareFlag( + ast::CompoundStmt * out, const CodeLocation & location ); + ast::ExprStmt * makeSetter( + const CodeLocation & location, ast::ObjectDecl * flag ); + ast::ObjectDecl * declMonitors( + ast::CompoundStmt * out, const ast::WaitForClause * clause ); + void init_clause( ast::CompoundStmt * out, ast::ObjectDecl * acceptables, + int index, const ast::WaitForClause * clause, ast::Stmt * setter ); + ast::Expr * init_timeout( + ast::CompoundStmt * out, const CodeLocation & topLocation, + const ast::Expr * timeout_time, const ast::Expr * timeout_cond, + const ast::Stmt * else_stmt, const ast::Expr * else_cond, + const ast::Stmt * setter ); + ast::Expr * call( + ast::CompoundStmt * out, const CodeLocation & location, + size_t numClauses, ast::ObjectDecl * acceptables, + ast::Expr * timeout ); +public: + void previsit( const ast::FunctionDecl * decl ); + void previsit( const ast::StructDecl * decl ); + ast::Stmt * postvisit( const ast::WaitForStmt * stmt ); +}; + +ast::Expr * makeOpIndex( const CodeLocation & location, + const ast::DeclWithType * array, unsigned long index ) { + return new ast::UntypedExpr( location, + new ast::NameExpr( location, "?[?]" ), + { + new ast::VariableExpr( location, array ), + ast::ConstantExpr::from_ulong( location, index ), + } + ); +} + +ast::Expr * makeOpAssign( const CodeLocation & location, + const ast::Expr * lhs, const ast::Expr * rhs ) { + return new ast::UntypedExpr( location, + new ast::NameExpr( location, "?=?" ), + { lhs, rhs } + ); +} + +ast::Expr * makeOpMember( const CodeLocation & location, + const std::string & mem, const ast::Expr * sue ) { + return new ast::UntypedMemberExpr( location, + new ast::NameExpr( location, mem ), + sue + ); +} + +ast::Stmt * makeAccStmt( + const CodeLocation & location, ast::DeclWithType * object, + unsigned long index, const std::string & member, + const ast::Expr * value, const ResolveContext & context +) { + ast::Expr * expr = makeOpAssign( location, + makeOpMember( location, + member, + makeOpIndex( location, + object, + index + ) + ), + value + ); + + auto result = ResolvExpr::findVoidExpression( expr, context ); + return new ast::ExprStmt( location, result.get() ); +} + +const ast::Stmt * maybeCond( const CodeLocation & location, + const ast::Expr * cond, std::list> && stmts ) { + ast::Stmt * block = new ast::CompoundStmt( location, std::move( stmts ) ); + return (cond) ? new ast::IfStmt( location, cond, block ) : block; +} + +const ast::VariableExpr * extractVariable( const ast::Expr * func ) { + if ( auto var = dynamic_cast( func ) ) { + return var; + } + auto cast = strict_dynamic_cast( func ); + return cast->arg.strict_as(); +} + +const ast::Expr * detectIsDtor( + const CodeLocation & location, const ast::Expr * func ) { + const ast::VariableExpr * typed_func = extractVariable( func ); + bool is_dtor = InitTweak::isDestructor( + typed_func->var.strict_as() ); + return ast::ConstantExpr::from_bool( location, is_dtor ); +} + +ast::ObjectDecl * GenerateWaitForCore::declareAcceptables( + ast::CompoundStmt * out, + const CodeLocation & location, unsigned long numClauses ) { + ast::ObjectDecl * acceptables = new ast::ObjectDecl( location, + namer_acc.newName(), + new ast::ArrayType( + new ast::StructInstType( decl_acceptable ), + ast::ConstantExpr::from_ulong( location, numClauses ), + ast::FixedLen, + ast::DynamicDim + ) + ); + out->push_back( new ast::DeclStmt( location, acceptables ) ); + + ast::Expr * set = new ast::UntypedExpr( location, + new ast::NameExpr( location, "__builtin_memset" ), + { + new ast::VariableExpr( location, acceptables ), + ast::ConstantExpr::from_int( location, 0 ), + new ast::SizeofExpr( location, + new ast::VariableExpr( location, acceptables ) ), + } + ); + ResolveContext context{ symtab, transUnit().global }; + auto result = ResolvExpr::findVoidExpression( set, context ); + out->push_back( new ast::ExprStmt( location, result.get() ) ); + + return acceptables; +} + +ast::ObjectDecl * GenerateWaitForCore::declareFlag( + ast::CompoundStmt * out, const CodeLocation & location ) { + ast::ObjectDecl * flag = new ast::ObjectDecl( location, + namer_flg.newName(), + new ast::BasicType( ast::BasicType::Bool ), + new ast::SingleInit( location, + ast::ConstantExpr::from_ulong( location, 0 ) + ) + ); + out->push_back( new ast::DeclStmt( location, flag ) ); + return flag; +} + +ast::ExprStmt * GenerateWaitForCore::makeSetter( + const CodeLocation & location, ast::ObjectDecl * flag ) { + ast::Expr * expr = new ast::UntypedExpr( location, + new ast::NameExpr( location, "?=?" ), + { + new ast::VariableExpr( location, flag ), + ast::ConstantExpr::from_ulong( location, 1 ), + } + ); + ResolveContext context{ symtab, transUnit().global }; + auto result = ResolvExpr::findVoidExpression( expr, context ); + return new ast::ExprStmt( location, result.get() ); +} + +ast::ObjectDecl * GenerateWaitForCore::declMonitors( + ast::CompoundStmt * out, + const ast::WaitForClause * clause ) { + const CodeLocation & location = clause->location; + ast::ObjectDecl * monitor = new ast::ObjectDecl( location, + namer_mon.newName(), + new ast::ArrayType( + new ast::PointerType( + new ast::StructInstType( decl_monitor ) + ), + ast::ConstantExpr::from_ulong( location, clause->target_args.size() ), + ast::FixedLen, + ast::DynamicDim + ), + new ast::ListInit( location, + map_range>>( + clause->target_args, + []( const ast::Expr * expr ){ + return new ast::SingleInit( expr->location, expr ); } + ) + ) + ); + out->push_back( new ast::DeclStmt( location, monitor ) ); + return monitor; +} + +void GenerateWaitForCore::init_clause( + ast::CompoundStmt * out, + ast::ObjectDecl * acceptables, + int index, + const ast::WaitForClause * clause, + ast::Stmt * setter ) { + const CodeLocation & location = clause->location; + const ast::ObjectDecl * monitors = declMonitors( out, clause ); + ast::Type * fptr_t = new ast::PointerType( + new ast::FunctionType( ast::VariableArgs ) ); + + const ast::VariableExpr * variableExpr = + clause->target_func.as(); + ast::Expr * castExpr = new ast::CastExpr( + location, + new ast::CastExpr( + location, + clause->target_func, + ast::deepCopy( variableExpr->result.get() ), + ast::GeneratedCast ), + fptr_t, + ast::GeneratedCast ); + + ast::ObjectDecl * funcDecl = new ast::ObjectDecl( location, + namer_fun.newName(), + ast::deepCopy( fptr_t ), + new ast::SingleInit( location, castExpr ) + ); + ast::Expr * funcExpr = new ast::VariableExpr( location, funcDecl ); + out->push_back( new ast::DeclStmt( location, funcDecl ) ); + + ResolveContext context{ symtab, transUnit().global }; + out->push_back( maybeCond( location, clause->cond.get(), { + makeAccStmt( location, acceptables, index, "is_dtor", + detectIsDtor( location, clause->target_func ), context ), + makeAccStmt( location, acceptables, index, "func", + funcExpr, context ), + makeAccStmt( location, acceptables, index, "data", + new ast::VariableExpr( location, monitors ), context ), + makeAccStmt( location, acceptables, index, "size", + ast::ConstantExpr::from_ulong( location, + clause->target_args.size() ), context ), + ast::deepCopy( setter ), + } ) ); +} + +ast::Expr * GenerateWaitForCore::init_timeout( + ast::CompoundStmt * out, + const CodeLocation & topLocation, + const ast::Expr * timeout_time, + const ast::Expr * timeout_cond, + const ast::Stmt * else_stmt, + const ast::Expr * else_cond, + const ast::Stmt * setter ) { + ast::ObjectDecl * timeout = new ast::ObjectDecl( topLocation, + namer_tim.newName(), + new ast::BasicType( ast::BasicType::LongLongUnsignedInt ), + new ast::SingleInit( topLocation, + ast::ConstantExpr::from_int( topLocation, -1 ) + ) + ); + out->push_back( new ast::DeclStmt( topLocation, timeout ) ); + + if ( timeout_time ) { + const CodeLocation & location = timeout_time->location; + out->push_back( maybeCond( location, timeout_cond, { + new ast::ExprStmt( location, + makeOpAssign( + location, + new ast::VariableExpr( location, timeout ), + timeout_time + ) + ), + ast::deepCopy( setter ), + } ) ); + } + + // We only care about the else_stmt's presence and location. + if ( else_stmt ) { + const CodeLocation & location = else_stmt->location; + out->push_back( maybeCond( location, else_cond, { + new ast::ExprStmt( location, + makeOpAssign( + location, + new ast::VariableExpr( location, timeout ), + ast::ConstantExpr::from_ulong( location, 0 ) + ) + ), + ast::deepCopy( setter ), + } ) ); + } + + return new ast::VariableExpr( topLocation, timeout ); +} + +ast::Expr * GenerateWaitForCore::call( + ast::CompoundStmt * out, + const CodeLocation & location, + size_t numClauses, + ast::ObjectDecl * acceptables, + ast::Expr * timeout +) { + ast::ObjectDecl * index = new ast::ObjectDecl( location, + namer_idx.newName(), + new ast::BasicType( ast::BasicType::ShortSignedInt ), + new ast::SingleInit( location, + ast::ConstantExpr::from_int( location, -1 ) + ) + ); + out->push_back( new ast::DeclStmt( location, index ) ); + + ast::ObjectDecl * mask = new ast::ObjectDecl( location, + namer_msk.newName(), + new ast::StructInstType( decl_mask ), + new ast::ListInit( location, { + new ast::SingleInit( location, + new ast::AddressExpr( location, + new ast::VariableExpr( location, index ) + ) + ), + new ast::ListInit( location, { + new ast::SingleInit( location, + new ast::VariableExpr( location, acceptables ) + ), + new ast::SingleInit( location, + ast::ConstantExpr::from_ulong( location, numClauses ) + ), + }), + }) + ); + out->push_back( new ast::DeclStmt( location, mask ) ); + + ast::ApplicationExpr * waitforMask = new ast::ApplicationExpr( location, + ast::VariableExpr::functionPointer( location, decl_waitfor ), + { + new ast::CastExpr( + new ast::VariableExpr( location, mask ), + new ast::ReferenceType( + new ast::StructInstType( decl_mask ) + ) + ), + timeout + } + ); + out->push_back( new ast::ExprStmt( location, waitforMask ) ); + + return new ast::VariableExpr( location, index ); +} + +ast::Stmt * choose( const ast::WaitForStmt * waitfor, ast::Expr * result ) { + const CodeLocation & location = waitfor->location; + + ast::SwitchStmt * theSwitch = new ast::SwitchStmt( location, + result, + std::vector>() + ); + + // For some reason, enumerate doesn't work here because of references. + for ( size_t i = 0 ; i < waitfor->clauses.size() ; ++i ) { + theSwitch->cases.push_back( + new ast::CaseClause( location, + ast::ConstantExpr::from_ulong( location, i ), + { + new ast::CompoundStmt( location, { + waitfor->clauses[i]->stmt, + new ast::BranchStmt( location, + ast::BranchStmt::Break, + ast::Label( location ) + ) + }) + } + ) + ); + } + + if ( waitfor->timeout_stmt ) { + theSwitch->cases.push_back( + new ast::CaseClause( location, + ast::ConstantExpr::from_int( location, -2 ), + { + new ast::CompoundStmt( location, { + waitfor->timeout_stmt, + new ast::BranchStmt( location, + ast::BranchStmt::Break, + ast::Label( location ) + ) + }) + } + ) + ); + } + + if ( waitfor->else_stmt ) { + theSwitch->cases.push_back( + new ast::CaseClause( location, + ast::ConstantExpr::from_int( location, -1 ), + { + new ast::CompoundStmt( location, { + waitfor->else_stmt, + new ast::BranchStmt( location, + ast::BranchStmt::Break, + ast::Label( location ) + ) + }) + } + ) + ); + } + + return theSwitch; +} + +void GenerateWaitForCore::previsit( const ast::FunctionDecl * decl ) { + if ( "__waitfor_internal" == decl->name ) { + decl_waitfor = decl; + } +} + +void GenerateWaitForCore::previsit( const ast::StructDecl * decl ) { + if ( !decl->body ) { + return; + } else if ( "__acceptable_t" == decl->name ) { + assert( !decl_acceptable ); + decl_acceptable = decl; + } else if ( "__waitfor_mask_t" == decl->name ) { + assert( !decl_mask ); + decl_mask = decl; + } else if ( "monitor$" == decl->name ) { + assert( !decl_monitor ); + decl_monitor = decl; + } +} + +ast::Stmt * GenerateWaitForCore::postvisit( const ast::WaitForStmt * stmt ) { + if ( !decl_monitor || !decl_acceptable || !decl_mask ) { + SemanticError( stmt, "waitfor keyword requires monitors to be in scope, add #include " ); + } + + const CodeLocation & location = stmt->location; + ast::CompoundStmt * comp = new ast::CompoundStmt( location ); + + ast::ObjectDecl * acceptables = declareAcceptables( comp, location, stmt->clauses.size() ); + ast::ObjectDecl * flag = declareFlag( comp, location ); + ast::Stmt * setter = makeSetter( location, flag ); + + // For some reason, enumerate doesn't work here because of references. + for ( size_t i = 0 ; i < stmt->clauses.size() ; ++i ) { + init_clause( comp, acceptables, i, stmt->clauses[i], setter ); + } + + ast::Expr * timeout = init_timeout( + comp, + location, + stmt->timeout_time, + stmt->timeout_cond, + stmt->else_stmt, + stmt->else_cond, + setter + ); + + ast::CompoundStmt * compound = new ast::CompoundStmt( location ); + comp->push_back( new ast::IfStmt( location, + new ast::VariableExpr( location, flag ), + compound, + nullptr + )); + + ast::Expr * result = call( + compound, location, stmt->clauses.size(), acceptables, timeout ); + compound->push_back( choose( stmt, result ) ); + return comp; +} + +} // namespace + +void generateWaitFor( ast::TranslationUnit & translationUnit ) { + ast::Pass::run( translationUnit ); +} + +} // namespace Concurrency + +// Local Variables: // +// tab-width: 4 // +// mode: c++ // +// compile-command: "make install" // +// End: // Index: src/Concurrency/module.mk =================================================================== --- src/Concurrency/module.mk (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/Concurrency/module.mk (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -19,4 +19,5 @@ Concurrency/Keywords.cc \ Concurrency/Keywords.h \ + Concurrency/WaitforNew.cpp \ Concurrency/Waitfor.cc \ Concurrency/Waitfor.h Index: src/InitTweak/FixGlobalInit.cc =================================================================== --- src/InitTweak/FixGlobalInit.cc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/InitTweak/FixGlobalInit.cc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -162,5 +162,5 @@ } // if if ( Statement * ctor = ctorInit->ctor ) { - addDataSectonAttribute( objDecl ); + addDataSectionAttribute( objDecl ); initStatements.push_back( ctor ); objDecl->init = nullptr; Index: src/InitTweak/FixInit.cc =================================================================== --- src/InitTweak/FixInit.cc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/InitTweak/FixInit.cc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -806,5 +806,5 @@ // The attribute works, and is meant to apply, both for leaving the static local alone, // and for hoisting it out as a static global. - addDataSectonAttribute( objDecl ); + addDataSectionAttribute( objDecl ); // originally wanted to take advantage of gcc nested functions, but Index: src/InitTweak/InitTweak.cc =================================================================== --- src/InitTweak/InitTweak.cc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/InitTweak/InitTweak.cc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -587,5 +587,5 @@ bool isConstructable( const ast::Type * type ) { - return ! dynamic_cast< const ast::VarArgsType * >( type ) && ! dynamic_cast< const ast::ReferenceType * >( type ) + return ! dynamic_cast< const ast::VarArgsType * >( type ) && ! dynamic_cast< const ast::ReferenceType * >( type ) && ! dynamic_cast< const ast::FunctionType * >( type ) && ! Tuples::isTtype( type ); } @@ -1025,5 +1025,5 @@ if (!assign) { auto td = new ast::TypeDecl({}, "T", {}, nullptr, ast::TypeDecl::Dtype, true); - assign = new ast::FunctionDecl({}, "?=?", {}, + assign = new ast::FunctionDecl({}, "?=?", {}, { new ast::ObjectDecl({}, "_dst", new ast::ReferenceType(new ast::TypeInstType("T", td))), new ast::ObjectDecl({}, "_src", new ast::TypeInstType("T", td))}, @@ -1095,7 +1095,7 @@ // address of a variable or member expression is constexpr - if ( ! dynamic_cast< const ast::NameExpr * >( arg ) - && ! dynamic_cast< const ast::VariableExpr * >( arg ) - && ! dynamic_cast< const ast::MemberExpr * >( arg ) + if ( ! dynamic_cast< const ast::NameExpr * >( arg ) + && ! dynamic_cast< const ast::VariableExpr * >( arg ) + && ! dynamic_cast< const ast::MemberExpr * >( arg ) && ! dynamic_cast< const ast::UntypedMemberExpr * >( arg ) ) result = false; } @@ -1241,24 +1241,25 @@ } - void addDataSectonAttribute( ObjectDecl * objDecl ) { + #if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message + #define ASM_COMMENT "#" + #else // defined( __ARM_ARCH ) + #define ASM_COMMENT "//" + #endif + static const char * const data_section = ".data" ASM_COMMENT; + static const char * const tlsd_section = ".tdata" ASM_COMMENT; + void addDataSectionAttribute( ObjectDecl * objDecl ) { + const bool is_tls = objDecl->get_storageClasses().is_threadlocal; + const char * section = is_tls ? tlsd_section : data_section; objDecl->attributes.push_back(new Attribute("section", { - new ConstantExpr( Constant::from_string(".data" -#if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message - "#" -#else // defined( __ARM_ARCH ) - "//" -#endif - ))})); + new ConstantExpr( Constant::from_string( section ) ) + })); } void addDataSectionAttribute( ast::ObjectDecl * objDecl ) { + const bool is_tls = objDecl->storage.is_threadlocal; + const char * section = is_tls ? tlsd_section : data_section; objDecl->attributes.push_back(new ast::Attribute("section", { - ast::ConstantExpr::from_string(objDecl->location, ".data" -#if defined( __x86_64 ) || defined( __i386 ) // assembler comment to prevent assembler warning message - "#" -#else // defined( __ARM_ARCH ) - "//" -#endif - )})); + ast::ConstantExpr::from_string(objDecl->location, section) + })); } Index: src/InitTweak/InitTweak.h =================================================================== --- src/InitTweak/InitTweak.h (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/InitTweak/InitTweak.h (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -127,5 +127,5 @@ /// .section .data#,"a" /// to avoid assembler warning "ignoring changed section attributes for .data" - void addDataSectonAttribute( ObjectDecl * objDecl ); + void addDataSectionAttribute( ObjectDecl * objDecl ); void addDataSectionAttribute( ast::ObjectDecl * objDecl ); Index: src/main.cc =================================================================== --- src/main.cc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ src/main.cc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -10,6 +10,6 @@ // Created On : Fri May 15 23:12:02 2015 // Last Modified By : Andrew Beach -// Last Modified On : Fri Apr 29 9:52:00 2022 -// Update Count : 673 +// Last Modified On : Tue Jun 7 13:29:00 2022 +// Update Count : 674 // @@ -447,5 +447,6 @@ PASS( "Expand Unique Expr", Tuples::expandUniqueExpr( transUnit ) ); // xxx - is this the right place for this? want to expand ASAP so tha, sequent passes don't need to worry about double-visiting a unique expr - needs to go after InitTweak::fix so that copy constructed return declarations are reused - PASS( "Translate Tries" , ControlStruct::translateTries( transUnit ) ); + PASS( "Translate Tries", ControlStruct::translateTries( transUnit ) ); + PASS( "Gen Waitfor", Concurrency::generateWaitFor( transUnit ) ); translationUnit = convert( move( transUnit ) ); @@ -517,9 +518,7 @@ PASS( "Expand Unique Expr", Tuples::expandUniqueExpr( translationUnit ) ); // xxx - is this the right place for this? want to expand ASAP so tha, sequent passes don't need to worry about double-visiting a unique expr - needs to go after InitTweak::fix so that copy constructed return declarations are reused - - PASS( "Translate Tries" , ControlStruct::translateTries( translationUnit ) ); + PASS( "Translate Tries", ControlStruct::translateTries( translationUnit ) ); + PASS( "Gen Waitfor", Concurrency::generateWaitFor( translationUnit ) ); } - - PASS( "Gen Waitfor" , Concurrency::generateWaitFor( translationUnit ) ); PASS( "Convert Specializations", GenPoly::convertSpecializations( translationUnit ) ); // needs to happen before tuple types are expanded Index: tools/cfa.nanorc =================================================================== --- tools/cfa.nanorc (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ tools/cfa.nanorc (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -10,10 +10,10 @@ color green "\<(forall|trait|(o|d|f|t)type|mutex|_Bool|volatile|virtual)\>" color green "\<(float|double|bool|char|int|short|long|enum|void|auto)\>" -color green "\<(static|const|extern|(un)?signed|inline)\>" "\<(sizeof)\>" +color green "\<(static|const|extern|(un)?signed|inline|sizeof|vtable)\>" color green "\<((s?size)|one|zero|((u_?)?int(8|16|32|64|ptr)))_t\>" # Declarations color brightgreen "\<(struct|union|typedef|trait|coroutine|generator)\>" -color brightgreen "\<(monitor|thread|with)\>" +color brightgreen "\<(monitor|thread|with|exception)\>" # Control Flow Structures Index: tools/jenkins/setup.sh.in =================================================================== --- tools/jenkins/setup.sh.in (revision 55422cfcdbdcf5a152f33907bf39eaa0a6b0095a) +++ tools/jenkins/setup.sh.in (revision 6e2b04e05bb3d9f5059628fbcd9d7304e2577458) @@ -29,5 +29,5 @@ function getrunpath() { - local elfout=$(readelf -d $1 | grep "RUNPATH") + local elfout=$(readelf -d $1 | grep -E "RPATH|RUNPATH") regex='\[/([[:alpha:][:digit:]@/_.-]+)\]' if [[ $elfout =~ $regex ]]; then @@ -43,14 +43,14 @@ { local deps=$(ldd $1) + retlcldeps=() retsysdeps=() - retlcldeps=() while IFS= read -r line; do - regex1='/([[:alpha:][:digit:]@/_.-]+)' - regex2='(libcfa[[:alpha:][:digit:].]+) => not found' + regex1='(libcfa[[:alpha:][:digit:].]+)' + regex2='/([[:alpha:][:digit:]@/_.-]+)' regex3='linux-vdso.so.1|linux-gate.so.1' if [[ $line =~ $regex1 ]]; then + retlcldeps+=(${BASH_REMATCH[1]}) + elif [[ $line =~ $regex2 ]]; then retsysdeps+=(${BASH_REMATCH[1]}) - elif [[ $line =~ $regex2 ]]; then - retlcldeps+=(${BASH_REMATCH[1]}) elif [[ $line =~ $regex3 ]]; then # echo "ignoring '$line}': intrinsic"