Changeset 8d462e5 for libcfa/src

Timestamp:

May 12, 2020, 7:21:32 PM (4 years ago)

Author:

Michael Brooks <mlbrooks@…>

Branches:

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

Children:

979df46

Parents:

f2d05e9

Message:

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

File:

• libcfa/src/executor.cfa (modified) (6 diffs)

libcfa/src/executor.cfa

@@ -4 +4 @@
 // buffer.
 
-#include <bits/containers.hfa>
+#include <containers/list.hfa>
 #include <thread.hfa>
 #include <stdio.h>
 
-forall( dtype T )
-monitor Buffer {                                        // unbounded buffer
-    __queue_t( T ) queue;                               // unbounded list of work requests
-    condition delay;
-}; // Buffer
-forall( dtype T | is_node(T) ) {
-    void insert( Buffer( T ) & mutex buf, T * elem ) with(buf) {
-        append( queue, elem );                          // insert element into buffer
-        signal( delay );                                // restart
-    } // insert
-
-    T * remove( Buffer( T ) & mutex buf ) with(buf) {
-        if ( queue.head != 0 ) wait( delay );                   // no request to process ? => wait
-//      return pop_head( queue );
-    } // remove
-} // distribution
-
 struct WRequest {                                       // client request, no return
     void (* action)( void );
-    WRequest * next;                                    // intrusive queue field
+    DLISTED_MGD_IMPL_IN(WRequest)
 }; // WRequest
+DLISTED_MGD_IMPL_OUT(WRequest)
 
-WRequest *& get_next( WRequest & this ) { return this.next; }
-void ?{}( WRequest & req ) with(req) { action = 0; next = 0; }
-void ?{}( WRequest & req, void (* action)( void ) ) with(req) { req.action = action; next = 0; }
+void ?{}( WRequest & req ) with(req) { action = 0; }
+void ?{}( WRequest & req, void (* action)( void ) ) with(req) { req.action = action; }
 bool stop( WRequest & req ) { return req.action == 0; }
 void doit( WRequest & req ) { req.action(); }
+
+monitor WRBuffer {                                      // unbounded buffer
+    dlist( WRequest, WRequest ) queue;                  // unbounded list of work requests
+    condition delay;
+}; // WRBuffer
+
+void insert( WRBuffer & mutex buf, WRequest * elem ) with(buf) {
+    insert_last( queue, *elem );                        // insert element into buffer
+    signal( delay );                                    // restart
+} // insert
+
+WRequest * remove( WRBuffer & mutex buf ) with(buf) {
+    if ( queue`is_empty ) wait( delay );                // no request to process ? => wait
+    return & pop_first( queue );
+} // remove
 
 // Each worker has its own work buffer to reduce contention between client and server. Hence, work requests arrive and
@@ -40 +38 @@
 
 thread Worker {
-    Buffer( WRequest ) * requests;
+    WRBuffer * requests;
     unsigned int start, range;
 }; // Worker
@@ -54 +52 @@
 } // Worker::main
 
-void ?{}( Worker & worker, cluster * wc, Buffer( WRequest ) * requests, unsigned int start, unsigned int range ) {
+void ?{}( Worker & worker, cluster * wc, WRBuffer * requests, unsigned int start, unsigned int range ) {
     (*get_thread(worker)){ *wc };                       // create on given cluster
     worker.[requests, start, range] = [requests, start, range];
@@ -62 +60 @@
     cluster * cluster;                                  // if workers execute on separate cluster
     processor ** processors;                            // array of virtual processors adding parallelism for workers
-    Buffer( WRequest ) * requests;                      // list of work requests
+    WRBuffer * requests;                                // list of work requests
     Worker ** workers;                                  // array of workers executing work requests
     unsigned int nprocessors, nworkers, nmailboxes;     // number of mailboxes/workers/processor tasks
@@ -79 +77 @@
     cluster = sepClus ? new( "Executor" ) : active_cluster();
     processors = (processor **)anew( nprocessors );
-    requests = anew( nmailboxes );
+    requests = (WRBuffer *)anew( nmailboxes );
     workers = (Worker **)anew( nworkers );
 
@@ -141 +139 @@
         for ( i; 3000 ) {
             send( exector, workie );
-            if ( i % 100 ) yield();
+            if ( i % 100 == 0 ) {
+//              fprintf( stderr, "%d\n", i );
+                yield();
+            }
         } // for
     }