Changes in libcfa/src/executor.cfa [8d462e5:cca568e]

File:

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

libcfa/src/executor.cfa

@@ -4 +4 @@
 // buffer.
 
-#include <containers/list.hfa>
+#include <bits/containers.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 );
-    DLISTED_MGD_IMPL_IN(WRequest)
+    WRequest * next;                                    // intrusive queue field
 }; // WRequest
-DLISTED_MGD_IMPL_OUT(WRequest)
 
-void ?{}( WRequest & req ) with(req) { action = 0; }
-void ?{}( WRequest & req, void (* action)( void ) ) with(req) { req.action = action; }
+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; }
 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
@@ -38 +40 @@
 
 thread Worker {
-    WRBuffer * requests;
+    Buffer( WRequest ) * requests;
     unsigned int start, range;
 }; // Worker
@@ -52 +54 @@
 } // Worker::main
 
-void ?{}( Worker & worker, cluster * wc, WRBuffer * requests, unsigned int start, unsigned int range ) {
+void ?{}( Worker & worker, cluster * wc, Buffer( WRequest ) * requests, unsigned int start, unsigned int range ) {
     (*get_thread(worker)){ *wc };                       // create on given cluster
     worker.[requests, start, range] = [requests, start, range];
@@ -60 +62 @@
     cluster * cluster;                                  // if workers execute on separate cluster
     processor ** processors;                            // array of virtual processors adding parallelism for workers
-    WRBuffer * requests;                                // list of work requests
+    Buffer( WRequest ) * requests;                      // list of work requests
     Worker ** workers;                                  // array of workers executing work requests
     unsigned int nprocessors, nworkers, nmailboxes;     // number of mailboxes/workers/processor tasks
@@ -77 +79 @@
     cluster = sepClus ? new( "Executor" ) : active_cluster();
     processors = (processor **)anew( nprocessors );
-    requests = (WRBuffer *)anew( nmailboxes );
+    requests = anew( nmailboxes );
     workers = (Worker **)anew( nworkers );
 
@@ -139 +141 @@
         for ( i; 3000 ) {
             send( exector, workie );
-            if ( i % 100 == 0 ) {
-//              fprintf( stderr, "%d\n", i );
-                yield();
-            }
+            if ( i % 100 ) yield();
         } // for
     }