Index: libcfa/src/Makefile.am =================================================================== --- libcfa/src/Makefile.am (revision 0f15e3b3aebf8d0ff2af18c94c36635b87820035) +++ libcfa/src/Makefile.am (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -115,5 +115,6 @@ concurrency/mutex_stmt.hfa \ concurrency/select.hfa \ - concurrency/channel.hfa + concurrency/channel.hfa \ + concurrency/actor.hfa inst_thread_headers_src = \ Index: libcfa/src/concurrency/actor.hfa =================================================================== --- libcfa/src/concurrency/actor.hfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ libcfa/src/concurrency/actor.hfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,271 @@ +#include +#include +#include + +#ifdef __CFA_DEBUG__ +#define CFA_DEBUG( stmt ) stmt +#else +#define CFA_DEBUG( stmt ) +#endif // CFA_DEBUG + +// Define the default number of processors created in the executor. Must be greater than 0. +#define __DEFAULT_EXECUTOR_PROCESSORS__ 2 + +// Define the default number of threads created in the executor. Must be greater than 0. +#define __DEFAULT_EXECUTOR_WORKERS__ 2 + +// Define the default number of executor request-queues (mailboxes) written to by actors and serviced by the +// actor-executor threads. Must be greater than 0. +#define __DEFAULT_EXECUTOR_RQUEUES__ 2 + +// Define if executor is created in a separate cluster +#define __DEFAULT_EXECUTOR_SEPCLUS__ false + +// forward decls +struct actor; +struct message; + +enum Allocation { Nodelete, Delete, Destroy, Finished }; // allocation status + +typedef Allocation (*__receive_fn)(actor &, message &); +struct request { + actor * receiver; + message * msg; + __receive_fn fn; + bool stop; + inline dlink(request); +}; +P9_EMBEDDED( request, dlink(request) ) + +void ?{}( request & this ) { this.stop = true; } // default ctor makes a sentinel +void ?{}( request & this, actor * receiver, message * msg, __receive_fn fn ) { + this.receiver = receiver; + this.msg = msg; + this.fn = fn; + this.stop = false; +} + +struct work_queue { + futex_mutex mutex_lock; + dlist( request ) input; // unbounded list of work requests +}; // work_queue +void ?{}( work_queue & this ) with(this) { input{}; mutex_lock{}; } + +void insert( work_queue & this, request & elem ) with(this) { + lock( mutex_lock ); + insert_last( input, elem ); + unlock( mutex_lock ); +} // insert + +void transfer( work_queue & this, dlist(request) & transferTo ) with(this) { + lock( mutex_lock ); + + //C_TODO CHANGE + // transferTo->transfer( input ); // transfer input to output + + // this is awfully inefficient but Ill use it until transfer is implemented + request * r; + while ( ! input`isEmpty ) { + r = &try_pop_front( input ); + if ( r ) insert_last( transferTo, *r ); + } + + // transfer( input, transferTo ); + + unlock( mutex_lock ); +} // transfer + +thread worker { + work_queue * request_queues; + dlist( request ) current_queue; + request & req; + unsigned int start, range; +}; + +static inline void ?{}( worker & this, cluster & clu, work_queue * request_queues, unsigned int start, unsigned int range ) { + ((thread &)this){ clu }; + this.request_queues = request_queues; + this.current_queue{}; + this.start = start; + this.range = range; +} + +struct executor { + cluster * cluster; // if workers execute on separate cluster + processor ** processors; // array of virtual processors adding parallelism for workers + work_queue * request_queues; // master list of work request queues + worker ** workers; // array of workers executing work requests + unsigned int nprocessors, nworkers, nrqueues; // number of processors/threads/request queues + bool seperate_clus; // use same or separate cluster for executor +}; // executor + +static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues, bool seperate_clus ) with(this) { + if ( nrqueues < nworkers ) abort( "nrqueues needs to be >= nworkers\n" ); + this.nprocessors = nprocessors; + this.nworkers = nworkers; + this.nrqueues = nrqueues; + this.seperate_clus = seperate_clus; + + if ( seperate_clus ) { + cluster = alloc(); + (*cluster){}; + } else cluster = active_cluster(); + + request_queues = aalloc( nrqueues ); + for ( i; nrqueues ) + request_queues[i]{}; + + processors = aalloc( nprocessors ); + for ( i; nprocessors ) + (*(processors[i] = alloc())){ *cluster }; + + workers = alloc( nworkers ); + unsigned int reqPerWorker = nrqueues / nworkers, extras = nrqueues % nworkers; + for ( unsigned int i = 0, start = 0, range; i < nworkers; i += 1, start += range ) { + range = reqPerWorker + ( i < extras ? 1 : 0 ); + (*(workers[i] = alloc())){ *cluster, request_queues, start, range }; + } // for +} + +static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers, unsigned int nrqueues ) { this{ nprocessors, nworkers, nrqueues, __DEFAULT_EXECUTOR_SEPCLUS__ }; } +static inline void ?{}( executor & this, unsigned int nprocessors, unsigned int nworkers ) { this{ nprocessors, nworkers, __DEFAULT_EXECUTOR_RQUEUES__ }; } +static inline void ?{}( executor & this, unsigned int nprocessors ) { this{ nprocessors, __DEFAULT_EXECUTOR_WORKERS__ }; } +static inline void ?{}( executor & this ) { this{ __DEFAULT_EXECUTOR_PROCESSORS__ }; } + +static inline void ^?{}( executor & this ) with(this) { + request sentinels[nworkers]; + unsigned int reqPerWorker = nrqueues / nworkers; + for ( unsigned int i = 0, step = 0; i < nworkers; i += 1, step += reqPerWorker ) { + insert( request_queues[step], sentinels[i] ); // force eventually termination + } // for + + for ( i; nworkers ) + delete( workers[i] ); + + for ( i; nprocessors ) { + delete( processors[i] ); + } // for + + delete( workers ); + delete( request_queues ); + delete( processors ); + if ( seperate_clus ) delete( cluster ); +} + +// this is a static field of executor but have to forward decl for get_next_ticket +static unsigned int __next_ticket = 0; + +static inline unsigned int get_next_ticket( executor & this ) with(this) { + return __atomic_fetch_add( &__next_ticket, 1, __ATOMIC_SEQ_CST) % nrqueues; +} // tickets + +// C_TODO: update globals in this file to be static fields once the project is done +static executor * __actor_executor_ = 0p; +static bool __actor_executor_passed = false; // was an executor passed to start_actor_system +static unsigned long int __num_actors_; // number of actor objects in system +static struct thread$ * __actor_executor_thd = 0p; // used to wake executor after actors finish +struct actor { + unsigned long int ticket; // executor-queue handle to provide FIFO message execution + Allocation allocation_; // allocation action +}; + +void ?{}( actor & this ) { + // Once an actor is allocated it must be sent a message or the actor system cannot stop. Hence, its receive + // member must be called to end it + verifyf( __actor_executor_, "Creating actor before calling start_actor_system()." ); + this.allocation_ = Nodelete; + this.ticket = get_next_ticket( *__actor_executor_ ); + __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_SEQ_CST ); +} +void ^?{}( actor & this ) {} + +static inline void check_actor( actor & this ) { + if ( this.allocation_ != Nodelete ) { + switch( this.allocation_ ) { + case Delete: delete( &this ); break; + case Destroy: + CFA_DEBUG( this.ticket = MAX; ); // mark as terminated + ^?{}(this); + break; + case Finished: + CFA_DEBUG( this.ticket = MAX; ); // mark as terminated + break; + default: ; // stop warning + } + + if ( unlikely( __atomic_add_fetch( &__num_actors_, -1, __ATOMIC_SEQ_CST ) == 0 ) ) { // all actors have terminated + unpark( __actor_executor_thd ); + } + } +} + +struct message { + Allocation allocation_; // allocation action +}; + +void ?{}( message & this ) { this.allocation_ = Nodelete; } +void ?{}( message & this, Allocation allocation ) { this.allocation_ = allocation; } +void ^?{}( message & this ) {} + +static inline void check_message( message & this ) { + switch ( this.allocation_ ) { // analyze message status + case Nodelete: break; + case Delete: delete( &this ); break; + case Destroy: ^?{}(this); break; + case Finished: break; + } // switch +} + +void deliver_request( request & this ) { + Allocation actor_allocation = this.fn( *this.receiver, *this.msg ); + this.receiver->allocation_ = actor_allocation; + check_actor( *this.receiver ); + check_message( *this.msg ); +} + +void main( worker & this ) with(this) { + Exit: + for ( unsigned int i = 0;; i = (i + 1) % range ) { // cycle through set of request buffers + transfer( request_queues[i + start], current_queue ); + while ( ! current_queue`isEmpty ) { + &req = &try_pop_front( current_queue ); + if ( !&req ) continue; // possibly add some work stealing/idle sleep here + if ( req.stop ) break Exit; + deliver_request( req ); + + delete( &req ); + } // while + } // for +} + +static inline void send( executor & this, request & req, unsigned long int ticket ) with(this) { + insert( request_queues[ticket], req); +} + +static inline void send( actor & this, request & req ) { + send( *__actor_executor_, req, this.ticket ); +} + +static inline void start_actor_system( size_t num_thds ) { + __actor_executor_thd = active_thread(); + __actor_executor_ = alloc(); + (*__actor_executor_){ 0, num_thds, num_thds * 16 }; +} + +static inline void start_actor_system() { start_actor_system( active_cluster()->procs.total ); } + +static inline void start_actor_system( executor & this ) { + __actor_executor_thd = active_thread(); + __actor_executor_ = &this; + __actor_executor_passed = true; +} + +static inline void stop_actor_system() { + park( ); // will receive signal when actor system is finished + + if ( !__actor_executor_passed ) delete( __actor_executor_ ); + __actor_executor_ = 0p; + __actor_executor_thd = 0p; + __next_ticket = 0; + __actor_executor_passed = false; +} Index: libcfa/src/concurrency/locks.hfa =================================================================== --- libcfa/src/concurrency/locks.hfa (revision 0f15e3b3aebf8d0ff2af18c94c36635b87820035) +++ libcfa/src/concurrency/locks.hfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -511,9 +511,4 @@ // flag showing if lock is held volatile bool held; - - #ifdef __CFA_DEBUG__ - // for deadlock detection - struct thread$ * owner; - #endif }; @@ -526,5 +521,5 @@ static inline void ?=?( spin_queue_lock & this, spin_queue_lock this2 ) = void; -// if this is called recursively IT WILL DEADLOCK!!!!! +// if this is called recursively IT WILL DEADLOCK! static inline void lock(spin_queue_lock & this) with(this) { mcs_spin_node node; Index: src/Concurrency/Actors.cpp =================================================================== --- src/Concurrency/Actors.cpp (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ src/Concurrency/Actors.cpp (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,318 @@ +// +// Cforall Version 1.0.0 Copyright (C) 2015 University of Waterloo +// +// The contents of this file are covered under the licence agreement in the +// file "LICENCE" distributed with Cforall. +// +// Actors.cpp -- generate code needed by the actor system +// +// Author : Colby Parsons +// Created On : Thurs Jan 19 15:34:00 2023 +// Last Modified By : Colby Parsons +// Last Modified On : Thurs Jan 19 15:34:00 2023 +// Update Count : 0 +// + +#include "AST/Print.hpp" +#include "AST/Decl.hpp" +#include "AST/Pass.hpp" +#include "AST/Type.hpp" +#include "AST/Stmt.hpp" +#include "AST/TranslationUnit.hpp" +#include "AST/Expr.hpp" +using namespace ast; + +namespace Concurrency { + +struct CollectactorStructDecls : public ast::WithGuards { + std::map & actorStructDecls; + std::map & messageStructDecls; + const StructDecl ** requestDecl; + const EnumDecl ** allocationDecl; + const StructDecl ** actorDecl; + const StructDecl ** msgDecl; + StructDecl * parentDecl; + bool insideStruct = false; + + void previsit( const EnumDecl * decl ) { + if( decl->name == "Allocation" ) *allocationDecl = decl; + } + + void previsit( const StructDecl * decl ) { + GuardValue(insideStruct); + insideStruct = true; + parentDecl = mutate( decl ); + if( decl->name == "actor" ) *actorDecl = decl; + if( decl->name == "message" ) *msgDecl = decl; + if( decl->name == "request" ) *requestDecl = decl; + } + + void postvisit( const StructInstType * node ) { + if ( ! *actorDecl || ! *msgDecl ) return; + if ( insideStruct ) { + if ( node->aggr() == *actorDecl ) { + actorStructDecls.insert({parentDecl, 1}); + } else if ( node->aggr() == *msgDecl ) { + messageStructDecls.insert({parentDecl, 1}); + } + } + } + + public: + CollectactorStructDecls( std::map & actorStructDecls, std::map & messageStructDecls, + const StructDecl ** requestDecl, const EnumDecl ** allocationDecl, const StructDecl ** actorDecl, const StructDecl ** msgDecl ) + : actorStructDecls( actorStructDecls ), messageStructDecls( messageStructDecls ), requestDecl( requestDecl ), + allocationDecl( allocationDecl ), actorDecl(actorDecl), msgDecl(msgDecl) {} +}; + +struct GenReceiveDecls : public ast::WithDeclsToAdd<> { + std::map & actorStructDecls; + std::map & messageStructDecls; + const StructDecl ** requestDecl; + const EnumDecl ** allocationDecl; + const StructDecl ** actorDecl; + const StructDecl ** msgDecl; + std::vector & forwardDecls; + + void postvisit( const FunctionDecl * decl ) { + // return if not of the form receive( param1, param2 ) or if it is a forward decl + if ( decl->name != "receive" || decl->params.size() != 2 || !decl->stmts ) return; + + // the params should be references + const ReferenceType * derivedActorRef = dynamic_cast(decl->params.at(0)->get_type()); + const ReferenceType * derivedMsgRef = dynamic_cast(decl->params.at(1)->get_type()); + if ( !derivedActorRef || !derivedMsgRef ) return; + + // the references should be to struct instances + const StructInstType * arg1InstType = dynamic_cast(derivedActorRef->base.get()); + const StructInstType * arg2InstType = dynamic_cast(derivedMsgRef->base.get()); + if ( !arg1InstType || !arg2InstType ) return; + + // If the struct instances are derived actor and message types then generate the message send routine + if ( actorStructDecls.count( arg1InstType->aggr() ) && messageStructDecls.count( arg2InstType->aggr() ) ) { + + // check that we have found all the decls we need from + if ( !*allocationDecl || !*requestDecl ) + SemanticError( decl->location, "using actors requires a header, add #include \n" ); + + ////////////////////////////////////////////////////////////////////// + // The following generates this send message operator routine for all receive(derived_actor &, derived_msg &) functions + /* + static inline derived_actor & ?|?( derived_actor & receiver, derived_msg & msg ) { + request * new_req = alloc(); + Allocation (*my_work_fn)( derived_actor &, derived_msg & ) = receive; + __receive_fn fn = (__receive_fn)my_work_fn; + (*new_req){ &receiver, &msg, fn }; + send( receiver, *new_req ); + return receiver; + } + */ + CompoundStmt * sendBody = new CompoundStmt( decl->location ); + + // Generates: request * new_req = alloc(); + sendBody->push_back( new DeclStmt( + decl->location, + new ObjectDecl( + decl->location, + "new_req", + new PointerType( new StructInstType( *requestDecl ) ), + new SingleInit( decl->location, new UntypedExpr( decl->location, new NameExpr( decl->location, "alloc" ), {} ) ) + ) + )); + + // Function type is: Allocation (*)( derived_actor &, derived_msg & ) + FunctionType * derivedReceive = new FunctionType(); + derivedReceive->params.push_back( ast::deepCopy( derivedActorRef ) ); + derivedReceive->params.push_back( ast::deepCopy( derivedMsgRef ) ); + derivedReceive->returns.push_back( new EnumInstType( *allocationDecl ) ); + + // Generates: Allocation (*my_work_fn)( derived_actor &, derived_msg & ) = receive; + sendBody->push_back( new DeclStmt( + decl->location, + new ObjectDecl( + decl->location, + "my_work_fn", + new PointerType( derivedReceive ), + new SingleInit( decl->location, new NameExpr( decl->location, "receive" ) ) + ) + )); + + // Function type is: Allocation (*)( actor &, messsage & ) + FunctionType * genericReceive = new FunctionType(); + genericReceive->params.push_back( new ReferenceType( new StructInstType( *actorDecl ) ) ); + genericReceive->params.push_back( new ReferenceType( new StructInstType( *msgDecl ) ) ); + genericReceive->returns.push_back( new EnumInstType( *allocationDecl ) ); + + // Generates: Allocation (*fn)( actor &, messsage & ) = (Allocation (*)( actor &, messsage & ))my_work_fn; + // More readable synonymous code: + // typedef Allocation (*__receive_fn)(actor &, message &); + // __receive_fn fn = (__receive_fn)my_work_fn; + sendBody->push_back( new DeclStmt( + decl->location, + new ObjectDecl( + decl->location, + "fn", + new PointerType( genericReceive ), + new SingleInit( decl->location, + new CastExpr( decl->location, new NameExpr( decl->location, "my_work_fn" ), new PointerType( genericReceive ), ExplicitCast ) + ) + ) + )); + + // Generates: (*new_req){ &receiver, &msg, fn }; + sendBody->push_back( new ExprStmt( + decl->location, + new UntypedExpr ( + decl->location, + new NameExpr( decl->location, "?{}" ), + { + new UntypedExpr( decl->location, new NameExpr( decl->location, "*?" ), { new NameExpr( decl->location, "new_req" ) } ), + new AddressExpr( new NameExpr( decl->location, "receiver" ) ), + new AddressExpr( new NameExpr( decl->location, "msg" ) ), + new NameExpr( decl->location, "fn" ) + } + ) + )); + + // Generates: send( receiver, *new_req ); + sendBody->push_back( new ExprStmt( + decl->location, + new UntypedExpr ( + decl->location, + new NameExpr( decl->location, "send" ), + { + { + new NameExpr( decl->location, "receiver" ), + new UntypedExpr( decl->location, new NameExpr( decl->location, "*?" ), { new NameExpr( decl->location, "new_req" ) } ) + } + } + ) + )); + + // Generates: return receiver; + sendBody->push_back( new ReturnStmt( decl->location, new NameExpr( decl->location, "receiver" ) ) ); + + // put it all together into the complete function decl from above + FunctionDecl * sendOperatorFunction = new FunctionDecl( + decl->location, + "?|?", + {}, // forall + { + new ObjectDecl( + decl->location, + "receiver", + ast::deepCopy( derivedActorRef ) + ), + new ObjectDecl( + decl->location, + "msg", + ast::deepCopy( derivedMsgRef ) + ) + }, // params + { + new ObjectDecl( + decl->location, + "receiver_ret", + ast::deepCopy( derivedActorRef ) + ) + }, + nullptr, // body + { Storage::Static }, // storage + Linkage::Cforall, // linkage + {}, // attributes + { Function::Inline } + ); + + // forward decls to resolve use before decl problem for '|' routines + forwardDecls.push_back( ast::deepCopy( sendOperatorFunction ) ); + + sendOperatorFunction->stmts = sendBody; + declsToAddAfter.push_back( sendOperatorFunction ); + } + } + + public: + GenReceiveDecls( std::map & actorStructDecls, std::map & messageStructDecls, + const StructDecl ** requestDecl, const EnumDecl ** allocationDecl, const StructDecl ** actorDecl, const StructDecl ** msgDecl, + std::vector & forwardDecls ) : actorStructDecls(actorStructDecls), messageStructDecls(messageStructDecls), + requestDecl(requestDecl), allocationDecl(allocationDecl), actorDecl(actorDecl), msgDecl(msgDecl), forwardDecls(forwardDecls) {} +}; + +struct GenFwdDecls : public ast::WithDeclsToAdd<> { + std::map & actorStructDecls; + std::map & messageStructDecls; + std::vector & forwardDecls; + bool done; + + void postvisit( const FunctionDecl * decl ) { + if ( done ) return; + // return if not of the form receive( param1, param2 ) or if it is a forward decl + if ( decl->name != "receive" || decl->params.size() != 2 || !decl->stmts ) return; + + // the params should be references + const ReferenceType * derivedActorRef = dynamic_cast(decl->params.at(0)->get_type()); + const ReferenceType * derivedMsgRef = dynamic_cast(decl->params.at(1)->get_type()); + if ( !derivedActorRef || !derivedMsgRef ) return; + + // the references should be to struct instances + const StructInstType * arg1InstType = dynamic_cast(derivedActorRef->base.get()); + const StructInstType * arg2InstType = dynamic_cast(derivedMsgRef->base.get()); + if ( !arg1InstType || !arg2InstType ) return; + + // If the struct instances are derived actor and message types then generate the message send routine + if ( actorStructDecls.count( arg1InstType->aggr() ) && messageStructDecls.count( arg2InstType->aggr() ) ) { + done = true; + for ( const auto & func : forwardDecls ) { + declsToAddBefore.push_back( func ); + } + } + } + + public: + GenFwdDecls( std::map & actorStructDecls, std::map & messageStructDecls, + std::vector & forwardDecls ) : actorStructDecls(actorStructDecls), messageStructDecls(messageStructDecls), + forwardDecls(forwardDecls), done(false) {} +}; + +void implementActors( TranslationUnit & translationUnit ) { + // maps to collect all derived actor and message types + std::map actorStructDecls; + std::map messageStructDecls; + std::vector forwardDecls; + + // for storing through the passes + // these are populated with various important struct decls + const StructDecl * requestDeclPtr = nullptr; + const EnumDecl * allocationDeclPtr = nullptr; + const StructDecl * actorDeclPtr = nullptr; + const StructDecl * msgDeclPtr = nullptr; + + // double pointer to modify local ptrs above + const StructDecl ** requestDecl = &requestDeclPtr; + const EnumDecl ** allocationDecl = &allocationDeclPtr; + const StructDecl ** actorDecl = &actorDeclPtr; + const StructDecl ** msgDecl = &msgDeclPtr; + + // first pass collects ptrs to Allocation enum, request type, and generic receive fn typedef + // also populates maps of all derived actors and messages + Pass::run( translationUnit, actorStructDecls, messageStructDecls, requestDecl, + allocationDecl, actorDecl, msgDecl ); + + // second pass locates all receive() routines that overload the generic receive fn + // it then generates the appropriate operator '|' send routines for the receive routines + Pass::run( translationUnit, actorStructDecls, messageStructDecls, requestDecl, + allocationDecl, actorDecl, msgDecl, forwardDecls ); + + // The third pass forward declares operator '|' send routines + Pass::run( translationUnit, actorStructDecls, messageStructDecls, forwardDecls ); +} + + +} // namespace Concurrency + +// Local Variables: // +// tab-width: 4 // +// mode: c++ // +// compile-command: "make install" // +// End: // + Index: src/Concurrency/Actors.hpp =================================================================== --- src/Concurrency/Actors.hpp (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ src/Concurrency/Actors.hpp (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,32 @@ +// +// 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. +// +// Keywords.h -- Implement concurrency constructs from their keywords. +// +// Author : Colby Parsons +// Created On : Thurs Jan 19 15:16:42 2023 +// Last Modified By : +// Last Modified On : +// Update Count : 1 +// + +#pragma once + + +class Declaration; +namespace ast { + class TranslationUnit; +} + +namespace Concurrency { + void implementActors( ast::TranslationUnit & translationUnit ); +}; + +// Local Variables: // +// tab-width: 4 // +// mode: c++ // +// compile-command: "make install" // +// End: // Index: src/Concurrency/module.mk =================================================================== --- src/Concurrency/module.mk (revision 0f15e3b3aebf8d0ff2af18c94c36635b87820035) +++ src/Concurrency/module.mk (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -16,4 +16,6 @@ SRC += \ + Concurrency/Actors.cpp \ + Concurrency/Actors.hpp \ Concurrency/KeywordsNew.cpp \ Concurrency/Keywords.cc \ @@ -21,3 +23,3 @@ Concurrency/WaitforNew.cpp \ Concurrency/Waitfor.cc \ - Concurrency/Waitfor.h + Concurrency/Waitfor.h Index: src/main.cc =================================================================== --- src/main.cc (revision 0f15e3b3aebf8d0ff2af18c94c36635b87820035) +++ src/main.cc (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -46,4 +46,5 @@ #include "Common/UnimplementedError.h" // for UnimplementedError #include "Common/utility.h" // for deleteAll, filter, printAll +#include "Concurrency/Actors.hpp" // for implementActors #include "Concurrency/Keywords.h" // for implementMutex, implement... #include "Concurrency/Waitfor.h" // for generateWaitfor @@ -341,4 +342,6 @@ PASS( "Generate Autogen Routines", Validate::autogenerateRoutines( transUnit ) ); + PASS( "Implement Actors", Concurrency::implementActors( transUnit ) ); + PASS( "Implement Mutex", Concurrency::implementMutex( transUnit ) ); PASS( "Implement Thread Start", Concurrency::implementThreadStarter( transUnit ) ); Index: tests/concurrent/actors/.expect/matrix.txt =================================================================== --- tests/concurrent/actors/.expect/matrix.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/.expect/matrix.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,4 @@ +starting +started +stopping +stopped Index: tests/concurrent/actors/.expect/pingpong.txt =================================================================== --- tests/concurrent/actors/.expect/pingpong.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/.expect/pingpong.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,2 @@ +start +end Index: tests/concurrent/actors/.expect/types.txt =================================================================== --- tests/concurrent/actors/.expect/types.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/.expect/types.txt (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,14 @@ +start +basic test +1 +2 +same message and different actors test +3 +3 +same message and different actor types test +4 +4 +different message types, one actor test +-1 +5 +end Index: tests/concurrent/actors/matrix.cfa =================================================================== --- tests/concurrent/actors/matrix.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/matrix.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,128 @@ +#include +#include +#include +#include +#include + +unsigned int xr = 100, xc = 100, yc = 100, Processors = 1; // default values + +struct derived_actor { + inline actor; +}; +void ?{}( derived_actor & this ) { ((actor &)this){}; } + +struct derived_msg { + inline message; + int * Z; + int * X; + int ** Y; +}; + +void ?{}( derived_msg & this ) {} +void ?{}( derived_msg & this, int * Z, int * X, int ** Y ) { + ((message &) this){ Finished }; + this.Z = Z; + this.X = X; + this.Y = Y; +} + +Allocation receive( derived_actor & receiver, derived_msg & msg ) { + for ( unsigned int i = 0; i < yc; i += 1 ) { // multiply X_row by Y_col and sum products + msg.Z[i] = 0; + for ( unsigned int j = 0; j < xc; j += 1 ) { + msg.Z[i] += msg.X[j] * msg.Y[j][i]; + } // for + } // for + return Finished; +} + +int main( int argc, char * argv[] ) { + switch ( argc ) { + case 5: + if ( strcmp( argv[4], "d" ) != 0 ) { // default ? + Processors = atoi( argv[4] ); + if ( Processors < 1 ) goto Usage; + } // if + case 4: + if ( strcmp( argv[3], "d" ) != 0 ) { // default ? + xr = atoi( argv[3] ); + if ( xr < 1 ) goto Usage; + } // if + case 3: + if ( strcmp( argv[2], "d" ) != 0 ) { // default ? + xc = atoi( argv[2] ); + if ( xc < 1 ) goto Usage; + } // if + case 2: + if ( strcmp( argv[1], "d" ) != 0 ) { // default ? + yc = atoi( argv[1] ); + if ( yc < 1 ) goto Usage; + } // if + case 1: // use defaults + break; + default: + Usage: + sout | "Usage: " | argv[0] + | " [ yc (> 0) | 'd' (default " | yc + | ") ] [ xc (> 0) | 'd' (default " | xc + | ") ] [ xr (> 0) | 'd' (default " | xr + | ") ] [ processors (> 0) | 'd' (default " | Processors + | ") ]" ; + exit( EXIT_FAILURE ); + } // switch + + unsigned int r, c; + int * Z[xr], * X[xr], * Y[xc]; + + for ( r = 0; r < xr; r += 1 ) { // create/initialize X matrix + X[r] = aalloc( xc ); + for ( c = 0; c < xc; c += 1 ) { + X[r][c] = r * c % 37; // for timing + } // for + } // for + for ( r = 0; r < xc; r += 1 ) { // create/initialize Y matrix + Y[r] = aalloc( yc ); + for ( c = 0; c < yc; c += 1 ) { + Y[r][c] = r * c % 37; // for timing + } // for + } // for + for ( r = 0; r < xr; r += 1 ) { // create Z matrix + Z[r] = aalloc( yc ); + } // for + + executor e{ Processors, Processors, Processors == 1 ? 1 : Processors * 16, true }; + + printf("starting\n"); + + start_actor_system( e ); + + printf("started\n"); + + derived_msg messages[xr]; + + derived_actor actors[xr]; + + for ( unsigned int r = 0; r < xr; r += 1 ) { + messages[r]{ Z[r], X[r], Y }; + } // for + + for ( unsigned int r = 0; r < xr; r += 1 ) { + actors[r] | messages[r]; + } // for + + printf("stopping\n"); + + stop_actor_system(); + + printf("stopped\n"); + + for ( r = 0; r < xr; r += 1 ) { // deallocate X and Z matrices + free( X[r] ); + free( Z[r] ); + } // for + for ( r = 0; r < xc; r += 1 ) { // deallocate Y matrix + free( Y[r] ); + } // for + + return 0; +} Index: tests/concurrent/actors/pingpong.cfa =================================================================== --- tests/concurrent/actors/pingpong.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/pingpong.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,67 @@ +#include +#include +#include +#include +#include +#include + +struct ping { + inline actor; +}; +static inline void ?{}( ping & this ) { ((actor &)this){}; } + +struct pong { + inline actor; +}; +static inline void ?{}( pong & this ) { ((actor &)this){}; } + +struct p_msg { + inline message; + size_t count; +}; +static inline void ?{}( p_msg & this ) { ((message &)this){}; this.count = 0; } + +ping * pi; +pong * po; +size_t times = 100000; + +Allocation receive( ping & receiver, p_msg & msg ) { + msg.count++; + if ( msg.count > times ) return Finished; + + Allocation retval = Nodelete; + if ( msg.count == times ) retval = Finished; + *po | msg; + return retval; +} + +Allocation receive( pong & receiver, p_msg & msg ) { + msg.count++; + if ( msg.count > times ) return Finished; + + Allocation retval = Nodelete; + if ( msg.count == times ) retval = Finished; + *pi | msg; + return retval; +} + +size_t Processors = 2; + +int main( int argc, char * argv[] ) { + printf("start\n"); + + processor p[Processors - 1]; + + start_actor_system( Processors ); // test passing number of processors + + ping pi_actor; + pong po_actor; + po = &po_actor; + pi = &pi_actor; + p_msg m; + pi_actor | m; + stop_actor_system(); + + printf("end\n"); + return 0; +} Index: tests/concurrent/actors/types.cfa =================================================================== --- tests/concurrent/actors/types.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) +++ tests/concurrent/actors/types.cfa (revision 87281049cbf2fc0b061d1f07bf46eda47ba4f4fb) @@ -0,0 +1,118 @@ +#include +#include +#include +#include +#include +#include + +struct derived_actor { + inline actor; + int counter; +}; +static inline void ?{}( derived_actor & this ) { ((actor &)this){}; this.counter = 0; } + +struct d_msg { + inline message; + int num; +}; +static inline void ?{}( d_msg & this ) { ((message &)this){}; } + +// this isn't a valid receive routine since int is not a message type +Allocation receive( derived_actor & receiver, int i ) with( receiver ) { + mutex(sout) sout | i; + counter++; + if ( counter == 2 ) return Finished; + return Nodelete; +} + +Allocation receive( derived_actor & receiver, d_msg & msg ) { + return receive( receiver, msg.num ); +} + +struct derived_actor2 { + inline actor; +}; +static inline void ?{}( derived_actor2 & this ) { ((actor &)this){}; } + +Allocation receive( derived_actor2 & receiver, d_msg & msg ) { + mutex(sout) sout | msg.num; + return Finished; +} + +struct derived_actor3 { + inline actor; +}; +static inline void ?{}( derived_actor3 & this ) { ((actor &)this){}; } + +struct d_msg2 { + inline message; + int num; +}; +static inline void ?{}( d_msg2 & this ) { ((message &)this){}; } + +Allocation receive( derived_actor3 & receiver, d_msg & msg ) { + mutex(sout) sout | msg.num; + if ( msg.num == -1 ) return Nodelete; + return Finished; +} + +Allocation receive( derived_actor3 & receiver, d_msg2 & msg ) { + mutex(sout) sout | msg.num; + return Finished; +} + +size_t Processors = 3; + +int main( int argc, char * argv[] ) { + printf("start\n"); + + processor p[Processors - 1]; + + printf("basic test\n"); + start_actor_system( Processors ); // test passing number of processors + derived_actor a; + d_msg b, c; + b.num = 1; + c.num = 2; + a | b | c; + stop_actor_system(); + + printf("same message and different actors test\n"); + start_actor_system(); // let system detect # of processors + derived_actor2 d_ac2_0, d_ac2_1; + d_msg d_ac2_msg; + d_ac2_msg.num = 3; + d_ac2_0 | d_ac2_msg; + d_ac2_1 | d_ac2_msg; + stop_actor_system(); + + + { + printf("same message and different actor types test\n"); + executor e{ 0, Processors, Processors == 1 ? 1 : Processors * 4, false }; + start_actor_system( e ); // pass an explicit executor + derived_actor2 d_ac2_2; + derived_actor3 d_ac3_0; + d_msg d_ac23_msg; + d_ac23_msg.num = 4; + d_ac3_0 | d_ac23_msg; + d_ac2_2 | d_ac23_msg; + stop_actor_system(); + } // RAII to clean up executor + + { + printf("different message types, one actor test\n"); + executor e{ 1, Processors, Processors == 1 ? 1 : Processors * 4, true }; + start_actor_system( Processors ); + derived_actor3 a3; + d_msg b1; + d_msg2 c2; + b1.num = -1; + c2.num = 5; + a3 | b1 | c2; + stop_actor_system(); + } // RAII to clean up executor + + printf("end\n"); + return 0; +}