Changeset 8d462e5


Ignore:
Timestamp:
May 12, 2020, 7:21:32 PM (17 months ago)
Author:
Michael Brooks <mlbrooks@…>
Branches:
arm-eh, jacob/cs343-translation, master, new-ast, new-ast-unique-expr
Children:
979df46
Parents:
f2d05e9
Message:

Exector starter ported to standard-library dlist. Basic run works.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • libcfa/src/executor.cfa

    rf2d05e9 r8d462e5  
    44// buffer.
    55
    6 #include <bits/containers.hfa>
     6#include <containers/list.hfa>
    77#include <thread.hfa>
    88#include <stdio.h>
    99
    10 forall( dtype T )
    11 monitor Buffer {                                        // unbounded buffer
    12     __queue_t( T ) queue;                               // unbounded list of work requests
    13     condition delay;
    14 }; // Buffer
    15 forall( dtype T | is_node(T) ) {
    16     void insert( Buffer( T ) & mutex buf, T * elem ) with(buf) {
    17         append( queue, elem );                          // insert element into buffer
    18         signal( delay );                                // restart
    19     } // insert
    20 
    21     T * remove( Buffer( T ) & mutex buf ) with(buf) {
    22         if ( queue.head != 0 ) wait( delay );                   // no request to process ? => wait
    23 //      return pop_head( queue );
    24     } // remove
    25 } // distribution
    26 
    2710struct WRequest {                                       // client request, no return
    2811    void (* action)( void );
    29     WRequest * next;                                    // intrusive queue field
     12    DLISTED_MGD_IMPL_IN(WRequest)
    3013}; // WRequest
     14DLISTED_MGD_IMPL_OUT(WRequest)
    3115
    32 WRequest *& get_next( WRequest & this ) { return this.next; }
    33 void ?{}( WRequest & req ) with(req) { action = 0; next = 0; }
    34 void ?{}( WRequest & req, void (* action)( void ) ) with(req) { req.action = action; next = 0; }
     16void ?{}( WRequest & req ) with(req) { action = 0; }
     17void ?{}( WRequest & req, void (* action)( void ) ) with(req) { req.action = action; }
    3518bool stop( WRequest & req ) { return req.action == 0; }
    3619void doit( WRequest & req ) { req.action(); }
     20
     21monitor WRBuffer {                                      // unbounded buffer
     22    dlist( WRequest, WRequest ) queue;                  // unbounded list of work requests
     23    condition delay;
     24}; // WRBuffer
     25
     26void insert( WRBuffer & mutex buf, WRequest * elem ) with(buf) {
     27    insert_last( queue, *elem );                        // insert element into buffer
     28    signal( delay );                                    // restart
     29} // insert
     30
     31WRequest * remove( WRBuffer & mutex buf ) with(buf) {
     32    if ( queue`is_empty ) wait( delay );                // no request to process ? => wait
     33    return & pop_first( queue );
     34} // remove
    3735
    3836// Each worker has its own work buffer to reduce contention between client and server. Hence, work requests arrive and
     
    4038
    4139thread Worker {
    42     Buffer( WRequest ) * requests;
     40    WRBuffer * requests;
    4341    unsigned int start, range;
    4442}; // Worker
     
    5452} // Worker::main
    5553
    56 void ?{}( Worker & worker, cluster * wc, Buffer( WRequest ) * requests, unsigned int start, unsigned int range ) {
     54void ?{}( Worker & worker, cluster * wc, WRBuffer * requests, unsigned int start, unsigned int range ) {
    5755    (*get_thread(worker)){ *wc };                       // create on given cluster
    5856    worker.[requests, start, range] = [requests, start, range];
     
    6260    cluster * cluster;                                  // if workers execute on separate cluster
    6361    processor ** processors;                            // array of virtual processors adding parallelism for workers
    64     Buffer( WRequest ) * requests;                      // list of work requests
     62    WRBuffer * requests;                                // list of work requests
    6563    Worker ** workers;                                  // array of workers executing work requests
    6664    unsigned int nprocessors, nworkers, nmailboxes;     // number of mailboxes/workers/processor tasks
     
    7977    cluster = sepClus ? new( "Executor" ) : active_cluster();
    8078    processors = (processor **)anew( nprocessors );
    81     requests = anew( nmailboxes );
     79    requests = (WRBuffer *)anew( nmailboxes );
    8280    workers = (Worker **)anew( nworkers );
    8381
     
    141139        for ( i; 3000 ) {
    142140            send( exector, workie );
    143             if ( i % 100 ) yield();
     141            if ( i % 100 == 0 ) {
     142//              fprintf( stderr, "%d\n", i );
     143                yield();
     144            }
    144145        } // for
    145146    }
Note: See TracChangeset for help on using the changeset viewer.