Changes in / [9082d7e8:ff443e5]

Files:

• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/barrier.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/churn.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/contend.cfa (modified) (9 diffs)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/daisy_chain.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/hot_potato.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/ping_pong.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/pub_sub.cfa (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/barrier/barrier.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/barrier/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/churn/churn.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/churn/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend/contend.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/daisy_chain/daisy_chain.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/daisy_chain/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/hot_potato/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/hot_potato/hot_potato.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/ping_pong/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/ping_pong/ping_pong.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/pub_sub/go.mod (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/pub_sub/pub_sub.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/plotData.py (modified) (4 diffs)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/run (modified) (4 diffs)
• libcfa/src/concurrency/actor.hfa (modified) (20 diffs)
• libcfa/src/concurrency/channel.hfa (modified) (9 diffs)
• libcfa/src/concurrency/locks.hfa (modified) (6 diffs)
• tests/concurrent/channels/.expect/churn.txt (deleted)
• tests/concurrent/channels/.expect/contend.txt (deleted)
• tests/concurrent/channels/.expect/daisy_chain.txt (deleted)
• tests/concurrent/channels/.expect/hot_potato.txt (deleted)
• tests/concurrent/channels/.expect/ping_pong.txt (deleted)
• tests/concurrent/channels/.expect/pub_sub.txt (deleted)
• tests/concurrent/channels/barrier.cfa (deleted)
• tests/concurrent/channels/churn.cfa (deleted)
• tests/concurrent/channels/contend.cfa (deleted)
• tests/concurrent/channels/daisy_chain.cfa (deleted)
• tests/concurrent/channels/hot_potato.cfa (deleted)
• tests/concurrent/channels/ping_pong.cfa (deleted)
• tests/concurrent/channels/pub_sub.cfa (deleted)

doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/contend.cfa

@@ -8 +8 @@
 
 // user defines this
-// #define BIG 1
+#define BIG 1
 
 owner_lock o;
 
-size_t total_operations = 0;
+unsigned long long total_operations = 0;
 
 struct bigObject {
@@ -36 +36 @@
 Channel * channels;
 
-bool cons_done = false, prod_done = false;
+volatile bool cons_done = false, prod_done = false;
 volatile int cons_done_count = 0;
 size_t cons_check = 0, prod_check = 0;
@@ -48 +48 @@
 }
 void main(Consumer & this) {
-    size_t runs = 0;
+    unsigned long long runs = 0;
     size_t my_check = 0;
     for ( ;; ) {
@@ -78 +78 @@
 }
 void main(Producer & this) {
-    size_t runs = 0;
+    unsigned long long runs = 0;
     size_t my_check = 0;
-    size_t my_id = this.i;
     for ( ;; ) {
         if ( prod_done ) break;
         #ifdef BIG
         bigObject j{(size_t)runs};
-        insert( channels[ my_id ], j );
+        insert( channels[ this.i ], j );
         my_check = my_check ^ (j.a + j.b + j.c + j.d + j.d + j.e + j.f + j.g + j.h);
         #else
-        insert( channels[ my_id ], (size_t)runs );
+        insert( channels[ this.i ], (size_t)runs );
         my_check = my_check ^ ((size_t)runs);
         #endif
@@ -100 +99 @@
 }
 
-static inline int test( size_t Processors, size_t Channels, size_t Producers, size_t Consumers, size_t ChannelSize ) {
-    size_t Clusters = 1;
+int test( size_t Processors, size_t Channels, size_t Producers, size_t Consumers, size_t ChannelSize ) {
+    size_t Clusters = Processors;
     // create a cluster
     cluster clus[Clusters];
     processor * proc[Processors];
     for ( i; Processors ) {
-        (*(proc[i] = malloc())){clus[i % Clusters]};
+        (*(proc[i] = alloc())){clus[i % Clusters]};
     }
 
@@ -121 +120 @@
     Producer * p[Producers * Channels];
 
-    for ( j; Channels ) {
-        for ( i; Producers ) {
-            (*(p[i] = malloc())){ j, clus[j % Clusters] };
-        }
-
-        for ( i; Consumers ) {
-            (*(c[i] = malloc())){ j, clus[j % Clusters] };
-        }
+    for ( i; Consumers * Channels ) {
+        (*(c[i] = alloc())){ i % Channels, clus[i % Clusters] };
+    }
+
+    for ( i; Producers * Channels ) {
+        (*(p[i] = alloc())){ i % Channels, clus[i % Clusters] };
     }
 
@@ -151 +148 @@
             }
         }
+        
     }
 
@@ -187 +185 @@
 
 int main( int argc, char * argv[] ) {
-    size_t Processors = 1, Channels = 1, Producers = 1, Consumers = 1, ChannelSize = 128;
+    size_t Processors = 10, Channels = 1, Producers = 10, Consumers = 10, ChannelSize = 128;
     switch ( argc ) {
           case 3:
@@ -208 +206 @@
                 exit( EXIT_FAILURE );
         } // switch
-    Producers = Processors / 2;
-    Consumers = Processors / 2;
+    Producers = Processors;
+    Consumers = Processors;
     test(Processors, Channels, Producers, Consumers, ChannelSize);
 }

doc/theses/colby_parsons_MMAth/benchmarks/channels/plotData.py

@@ -36 +36 @@
     procs.append(int(val))
 
-# 3rd line has number of variants
+# 3rd line has num locks args
+line = readfile.readline()
+locks = []
+for val in line.split():
+    locks.append(int(val))
+
+# 4th line has number of variants
 line = readfile.readline()
 names = line.split()
@@ -44 +50 @@
 lines = (line for line in lines if line) # Non-blank lines
 
-class Bench(Enum):
-    Unset = 0
-    Contend = 1
-    Zero = 2
-    Barrier = 3
-    Churn = 4
-    Daisy_Chain = 5
-    Hot_Potato = 6
-    Pub_Sub = 7
-
 nameSet = False
-currBench = Bench.Unset # default val
+currLocks = -1 # default val
 count = 0
 procCount = 0
 currVariant = 0
-name = ""
+name = "Aggregate Lock"
 var_name = ""
 sendData = [0.0 for j in range(numVariants)]
@@ -68 +64 @@
     # print(line)
     
-    if currBench == Bench.Unset:
-        if line == "contend:":
-            name = "Contend"
-            currBench = Bench.Contend
-        elif line == "zero:":
-            name = "Zero"
-            currBench = Bench.Zero
-        elif line == "barrier:":
-            name = "Barrier"
-            currBench = Bench.Barrier
-        elif line == "churn:":
-            name = "Churn"
-            currBench = Bench.Churn
-        elif line == "daisy_chain:":
-            name = "Daisy_Chain"
-            currBench = Bench.Daisy_Chain
-        elif line == "hot_potato:":
-            name = "Hot_Potato"
-            currBench = Bench.Hot_Potato
-        elif line == "pub_sub:":
-            name = "Pub_Sub"
-            currBench = Bench.Pub_Sub
-        else:
-            print("Expected benchmark name")
-            print("Line: " + line)
-            sys.exit()
+    if currLocks == -1:
+        lineArr = line.split()
+        currLocks = lineArr[-1]
         continue
 
@@ -122 +95 @@
             if currVariant == numVariants:
                 fig, ax = plt.subplots()
-                plt.title(name + " Benchmark")
-                plt.ylabel("Throughput (channel operations)")
+                plt.title(name + " Benchmark: " + str(currLocks) + " Locks")
+                plt.ylabel("Throughput (entries)")
                 plt.xlabel("Cores")
                 for idx, arr in enumerate(data):
                     plt.errorbar( procs, arr, [bars[idx][0], bars[idx][1]], capsize=2, marker='o' )
-                
                 plt.yscale("log")
-                # plt.ylim(1, None)
-                # ax.get_yaxis().set_major_formatter(ticks.ScalarFormatter())
-                # else:
-                #     plt.ylim(0, None)
                 plt.xticks(procs)
                 ax.legend(names)
-                # fig.savefig("plots/" + machineName + name + ".png")
-                plt.savefig("plots/" + machineName + name + ".pgf")
+                # fig.savefig("plots/" + machineName + "Aggregate_Lock_" + str(currLocks) + ".png")
+                plt.savefig("plots/" + machineName + "Aggregate_Lock_" + str(currLocks) + ".pgf")
                 fig.clf()
 
                 # reset
-                currBench = Bench.Unset
+                currLocks = -1
                 currVariant = 0

doc/theses/colby_parsons_MMAth/benchmarks/channels/run

@@ -85 +85 @@
 }
 
-numtimes=5
+numtimes=11
+
+# locks=('-DLOCKS=L1' '-DLOCKS=L2' '-DLOCKS=L3' '-DLOCKS=L4' '-DLOCKS=L5' '-DLOCKS=L6' '-DLOCKS=L7' '-DLOCKS=L8')
+# locks='1 2 3 4 5 6 7 8'
+lock_flags=('-DLOCKS=L2' '-DLOCKS=L4' '-DLOCKS=L8')
+locks=('2' '4' '8')
 
 num_threads='2 4 8 16 24 32'
-# num_threads='2'
+# num_threads='2 4 8'
 
 # toggle benchmarks
-zero=${false}
-contend=${true}
-barrier=${false}
-churn=${false}
-daisy_chain=${false}
-hot_potato=${false}
-pub_sub=${false}
+order=${true}
+rand=${true}
+baseline=${true}
 
 runCFA=${true}
-runGO=${true}
+runCPP=${true}
 # runCFA=${false}
-# runGO=${false}
+# runCPP=${false}
 
 cfa=~/cfa-cc/driver/cfa
+cpp=g++
 
 # Helpers to minimize code duplication
@@ -150 +152 @@
 echo $num_threads
 
-if [ ${runCFA} -eq ${true} ]; then
-    echo -n 'CFA '
-fi
-if [ ${runGO} -eq ${true} ]; then
-    echo -n 'Go '
+for i in ${!locks[@]}; do
+        echo -n ${locks[$i]}' '
+done
+echo ""
+
+if [ ${runCFA} -eq ${true} ] && [ ${order} -eq ${true} ]; then
+    echo -n 'CFA-order '
+fi
+if [ ${runCPP} -eq ${true} ] && [ ${order} -eq ${true} ]; then
+    echo -n 'CPP-order '
+fi
+if [ ${runCFA} -eq ${true} ] && [ ${baseline} -eq ${true} ]; then
+    echo -n 'CFA-baseline '
+fi
+if [ ${runCPP} -eq ${true} ] && [ ${baseline} -eq ${true} ]; then
+    echo -n 'CPP-baseline '
+fi
+if [ ${runCFA} -eq ${true} ] && [ ${rand} -eq ${true} ]; then
+    echo -n 'CFA-rand '
+fi
+if [ ${runCPP} -eq ${true} ] && [ ${rand} -eq ${true} ]; then
+    echo -n 'CPP-rand '
 fi
 echo ""
@@ -163 +182 @@
 cfa_flags='-quiet -O3 -nodebug -DNDEBUG'
 
+# cpp flagse
+cpp_flags='-O3 -std=c++17 -lpthread -pthread -DNDEBUG'
+
 # run the benchmarks
 
-run_contend() {
+run_order() {
     post_args=${1}
 
     if [ ${runCFA} -eq ${true} ] ; then
         cd cfa # CFA RUN
-        print_header 'CFA'
-        ${cfa} ${cfa_flags} ${2}.cfa -o a.${hostname} > /dev/null 2>&1
+        print_header 'CFA-'${3}
+        ${cfa} ${cfa_flags} ${2} ${3}.cfa -o a.${hostname} > /dev/null 2>&1
         run_bench
         rm a.${hostname}
@@ -177 +199 @@
     fi # done CFA
 
-    if [ ${runGO} -eq ${true} ] ; then
-        cd go/${2} # Go RUN
-        print_header 'Go'
-        go build -o a.${hostname} > /dev/null 2>&1
+    if [ ${runCPP} -eq ${true} ] ; then
+        cd cpp # CPP RUN
+        print_header 'CPP-'${3}
+        ${cpp} ${cpp_flags} ${2} ${3}.cc -o a.${hostname} > /dev/null 2>&1
         run_bench
         rm a.${hostname}
         cd - > /dev/null
-    fi # done Go
+    fi # done CPP
 }
 
 # /usr/bin/time -f "%Uu %Ss %Er %Mkb"
-if [ ${contend} -eq ${true} ] ; then
-    echo "contend: "
-    run_contend '128' 'contend'
-fi
-
-if [ ${zero} -eq ${true} ] ; then
-    echo "zero: "
-    run_contend '0' 'contend'
-fi
-
-if [ ${barrier} -eq ${true} ] ; then
-    echo "barrier: "
-    run_contend '' 'barrier'
-fi
-
-if [ ${churn} -eq ${true} ] ; then
-    echo "churn: "
-    run_contend '' 'churn'
-fi
-
-if [ ${daisy_chain} -eq ${true} ] ; then
-    echo "daisy_chain: "
-    run_contend '' 'daisy_chain'
-fi
-
-if [ ${hot_potato} -eq ${true} ] ; then
-    echo "hot_potato: "
-    run_contend '' 'hot_potato'
-fi
-
-if [ ${pub_sub} -eq ${true} ] ; then
-    echo "pub_sub: "
-    run_contend '' 'pub_sub'
-fi
+
+for i in ${!locks[@]}; do
+    echo "locks: "${locks[$i]}
+    if [ ${order} -eq ${true} ] ; then
+        run_order ${locks[$i]} ${lock_flags[$i]} 'order'
+    fi
+    if [ ${baseline} -eq ${true} ] ; then
+        run_order ${locks[$i]} ${lock_flags[$i]} 'baseline'
+    fi
+    if [ ${rand} -eq ${true} ] ; then
+        run_order ${locks[$i]} '-DLOCKS=L8' 'rand'
+    fi
+done
+
+

libcfa/src/concurrency/actor.hfa

@@ -35 +35 @@
 
 // show stats
-// #define ACTOR_STATS
+// #define STATS
 
 // forward decls
@@ -122 +122 @@
     copy_queue * c_queue;           // current queue
     volatile bool being_processed;  // flag to prevent concurrent processing
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     unsigned int id;
     size_t missed;                  // transfers skipped due to being_processed flag being up
@@ -132 +132 @@
     c_queue = owned_queue;
     being_processed = false;
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     id = i;
     missed = 0;
@@ -153 +153 @@
     // check if queue is being processed elsewhere
     if ( unlikely( being_processed ) ) {
-        #ifdef ACTOR_STATS
+        #ifdef STATS
         missed++;
         #endif
@@ -175 +175 @@
 struct worker_info {
     volatile unsigned long long stamp;
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     size_t stolen_from, try_steal, stolen, failed_swaps, msgs_stolen;
     unsigned long long processed;
@@ -182 +182 @@
 };
 static inline void ?{}( worker_info & this ) {
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     this.stolen_from = 0;
     this.try_steal = 0;                             // attempts to steal
@@ -194 +194 @@
 }
 
-// #ifdef ACTOR_STATS
+// #ifdef STATS
 // unsigned int * stolen_arr;
 // unsigned int * replaced_queue;
@@ -206 +206 @@
 };
 
-#ifdef ACTOR_STATS
+#ifdef STATS
 // aggregate counters for statistics
 size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0,
@@ -235 +235 @@
 }; // executor
 
-// #ifdef ACTOR_STATS
+// #ifdef STATS
 // __spinlock_t out_lock;
 // #endif
 static inline void ^?{}( worker & mutex this ) with(this) { 
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     __atomic_add_fetch(&__all_gulps, executor_->w_infos[id].gulps,__ATOMIC_SEQ_CST);
     __atomic_add_fetch(&__all_processed, executor_->w_infos[id].processed,__ATOMIC_SEQ_CST);
@@ -276 +276 @@
         no_steal = true;
     
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     // stolen_arr = aalloc( nrqueues );
     // replaced_queue = aalloc( nrqueues );
@@ -341 +341 @@
     } // for
 
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     size_t misses = 0;
     for ( i; nrqueues ) {
@@ -358 +358 @@
     if ( seperate_clus ) delete( cluster );
 
-    #ifdef ACTOR_STATS // print formatted stats
+    #ifdef STATS // print formatted stats
     printf("    Actor System Stats:\n");
     printf("\tActors Created:\t\t\t\t%lu\n\tMessages Sent:\t\t\t\t%lu\n", __num_actors_stats, __all_processed);
@@ -404 +404 @@
     ticket = __get_next_ticket( *__actor_executor_ );
     __atomic_fetch_add( &__num_actors_, 1, __ATOMIC_RELAXED );
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     __atomic_fetch_add( &__num_actors_stats, 1, __ATOMIC_SEQ_CST );
     #endif
@@ -513 +513 @@
             continue;
 
-        #ifdef ACTOR_STATS
+        #ifdef STATS
         curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
         if ( curr_steal_queue ) {
@@ -526 +526 @@
         #else
         curr_steal_queue = try_swap_queues( this, i + vic_start, swap_idx );
-        #endif // ACTOR_STATS
+        #endif // STATS
 
         return;
@@ -558 +558 @@
 
 void main( worker & this ) with(this) {
-    // #ifdef ACTOR_STATS
+    // #ifdef STATS
     // for ( i; executor_->nrqueues ) {
     //     replaced_queue[i] = 0;
@@ -587 +587 @@
         }
         transfer( *curr_work_queue, &current_queue );
-        #ifdef ACTOR_STATS
+        #ifdef STATS
         executor_->w_infos[id].gulps++;
-        #endif // ACTOR_STATS
+        #endif // STATS
         #ifdef __STEAL
         if ( isEmpty( *current_queue ) ) {
@@ -599 +599 @@
             __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED );
             
-            #ifdef ACTOR_STATS
+            #ifdef STATS
             executor_->w_infos[id].try_steal++;
-            #endif // ACTOR_STATS
+            #endif // STATS
             
             steal_work( this, start + prng( range ) );
@@ -608 +608 @@
         #endif // __STEAL
         while ( ! isEmpty( *current_queue ) ) {
-            #ifdef ACTOR_STATS
+            #ifdef STATS
             executor_->w_infos[id].processed++;
             #endif
@@ -636 +636 @@
 
 static inline void __reset_stats() {
-    #ifdef ACTOR_STATS
+    #ifdef STATS
     __total_tries = 0;
     __total_stolen = 0;

libcfa/src/concurrency/channel.hfa

@@ -3 +3 @@
 #include <locks.hfa>
 #include <list.hfa>
-#include <mutex_stmt.hfa>
+
+#define __COOP_CHANNEL
+#ifdef __PREVENTION_CHANNEL
+forall( T ) {
+struct channel {
+    size_t size, count, front, back;
+    T * buffer;
+    thread$ * chair;
+    T * chair_elem;
+    exp_backoff_then_block_lock c_lock, p_lock;
+    __spinlock_t mutex_lock;
+    char __padding[64]; // avoid false sharing in arrays of channels
+};
+
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    buffer = aalloc( size );
+    chair = 0p;
+    mutex_lock{};
+    c_lock{};
+    p_lock{};
+}
+
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
+static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
+static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
+static inline bool has_waiters( channel(T) & chan ) with(chan) { return chair != 0p; }
+
+static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
+    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+
+static inline void insert( channel(T) & chan, T elem ) with( chan ) {
+    lock( p_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+
+    // have to check for the zero size channel case
+    if ( size == 0 && chair != 0p ) {
+        memcpy((void *)chair_elem, (void *)&elem, sizeof(T));
+        unpark( chair );
+        chair = 0p;
+        unlock( mutex_lock );
+        unlock( p_lock );
+        unlock( c_lock );
+        return;
+    }
+
+    // wait if buffer is full, work will be completed by someone else
+    if ( count == size ) {
+        chair = active_thread();
+        chair_elem = &elem;
+        unlock( mutex_lock );
+        park( );
+        return;
+    } // if
+
+    if ( chair != 0p ) {
+        memcpy((void *)chair_elem, (void *)&elem, sizeof(T));
+        unpark( chair );
+        chair = 0p;
+        unlock( mutex_lock );
+        unlock( p_lock );
+        unlock( c_lock );
+        return;
+    }
+    insert_( chan, elem );
+
+    unlock( mutex_lock );
+    unlock( p_lock );
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
+    lock( c_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+    T retval;
+
+    // have to check for the zero size channel case
+    if ( size == 0 && chair != 0p ) {
+        memcpy((void *)&retval, (void *)chair_elem, sizeof(T));
+        unpark( chair );
+        chair = 0p;
+        unlock( mutex_lock );
+        unlock( p_lock );
+        unlock( c_lock );
+        return retval;
+    }
+
+    // wait if buffer is empty, work will be completed by someone else
+    if ( count == 0 ) {
+        chair = active_thread();
+        chair_elem = &retval;
+        unlock( mutex_lock );
+        park( );
+        return retval;
+    }
+
+    // Remove from buffer
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front++;
+    if ( front == size ) front = 0;
+
+    if ( chair != 0p ) {
+        insert_( chan, *chair_elem );  // do waiting producer work
+        unpark( chair );
+        chair = 0p;
+        unlock( mutex_lock );
+        unlock( p_lock );
+        unlock( c_lock );
+        return retval;
+    }
+
+    unlock( mutex_lock );
+    unlock( c_lock );
+    return retval;
+}
+
+} // forall( T )
+#endif
+
+#ifdef __COOP_CHANNEL
 
 // link field used for threads waiting on channel
@@ -23 +148 @@
 }
 
-// wake one thread from the list
-static inline void wake_one( dlist( wait_link ) & queue ) {
-    wait_link & popped = try_pop_front( queue );
-    unpark( popped.t );
-}
-
-// returns true if woken due to shutdown
-// blocks thread on list and releases passed lock
-static inline bool block( dlist( wait_link ) & queue, void * elem_ptr, go_mutex & lock ) {
-    wait_link w{ active_thread(), elem_ptr };
-    insert_last( queue, w );
-    unlock( lock );
-    park();
-    return w.elem == 0p;
-}
-
-// void * used for some fields since exceptions don't work with parametric polymorphism currently
-exception channel_closed {
-    // on failed insert elem is a ptr to the element attempting to be inserted
-    // on failed remove elem ptr is 0p
-    // on resumption of a failed insert this elem will be inserted
-    // so a user may modify it in the resumption handler
-    void * elem;
-
-    // pointer to chan that is closed
-    void * closed_chan;
-};
-vtable(channel_closed) channel_closed_vt;
-
-// #define CHAN_STATS // define this to get channel stats printed in dtor
-
 forall( T ) {
 
-struct __attribute__((aligned(128))) channel {
-    size_t size, front, back, count;
+struct channel {
+    size_t size;
+    size_t front, back, count;
     T * buffer;
-    dlist( wait_link ) prods, cons; // lists of blocked threads
-    go_mutex mutex_lock;            // MX lock
-    bool closed;                    // indicates channel close/open
-    #ifdef CHAN_STATS
-    size_t blocks, operations;      // counts total ops and ops resulting in a blocked thd
-    #endif
+    dlist( wait_link ) prods, cons;
+    exp_backoff_then_block_lock mutex_lock;
 };
 
@@ -74 +165 @@
     cons{};
     mutex_lock{};
-    closed = false;
-    #ifdef CHAN_STATS
-    blocks = 0;
-    operations = 0;
-    #endif
 }
 
 static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
-static inline void ^?{}( channel(T) &c ) with(c) {
-    #ifdef CHAN_STATS
-    printf("Channel %p Blocks: %lu, Operations: %lu, %.2f%% of ops blocked\n", &c, blocks, operations, ((double)blocks)/operations * 100);
-    #endif
-    verifyf( cons`isEmpty && prods`isEmpty, "Attempted to delete channel with waiting threads (Deadlock).\n" );
-    delete( buffer );
-}
+static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
 static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
 static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
@@ -95 +175 @@
 static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
 
-// closes the channel and notifies all blocked threads
-static inline void close( channel(T) & chan ) with(chan) {
-    lock( mutex_lock );
-    closed = true;
-
-    // flush waiting consumers and producers
-    while ( has_waiting_consumers( chan ) ) {
-        cons`first.elem = 0p;
-        wake_one( cons );
-    }
-    while ( has_waiting_producers( chan ) ) {
-        prods`first.elem = 0p;
-        wake_one( prods );
-    }
-    unlock(mutex_lock);
-}
-
-static inline void is_closed( channel(T) & chan ) with(chan) { return closed; }
-
-static inline void flush( channel(T) & chan, T elem ) with(chan) {
-    lock( mutex_lock );
-    while ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    }
-    unlock( mutex_lock );
-}
-
-// handles buffer insert
-static inline void __buf_insert( channel(T) & chan, T & elem ) with(chan) {
+static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
     memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
     count += 1;
@@ -131 +182 @@
 }
 
-// does the buffer insert or hands elem directly to consumer if one is waiting
-static inline void __do_insert( channel(T) & chan, T & elem ) with(chan) {
-    if ( count == 0 && !cons`isEmpty ) {
-        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
-        wake_one( cons );
-    } else __buf_insert( chan, elem );
-}
-
-// needed to avoid an extra copy in closed case
-static inline bool __internal_try_insert( channel(T) & chan, T & elem ) with(chan) {
+static inline void wake_one( dlist( wait_link ) & queue ) {
+    wait_link & popped = try_pop_front( queue );
+    unpark( popped.t );
+}
+
+static inline void block( dlist( wait_link ) & queue, void * elem_ptr, exp_backoff_then_block_lock & lock ) {
+    wait_link w{ active_thread(), elem_ptr };
+    insert_last( queue, w );
+    unlock( lock );
+    park();
+}
+
+static inline void insert( channel(T) & chan, T elem ) with(chan) {
     lock( mutex_lock );
-    #ifdef CHAN_STATS
-    operations++;
-    #endif
-    if ( count == size ) { unlock( mutex_lock ); return false; }
-    __do_insert( chan, elem );
-    unlock( mutex_lock );
-    return true;
-}
-
-// attempts a nonblocking insert
-// returns true if insert was successful, false otherwise
-static inline bool try_insert( channel(T) & chan, T elem ) { return __internal_try_insert( chan, elem ); }
-
-// handles closed case of insert routine
-static inline void __closed_insert( channel(T) & chan, T & elem ) with(chan) {
-    channel_closed except{&channel_closed_vt, &elem, &chan };
-    throwResume except; // throw closed resumption
-    if ( !__internal_try_insert( chan, elem ) ) throw except; // if try to insert fails (would block), throw termination
-}
-
-static inline void insert( channel(T) & chan, T elem ) with(chan) {
-    // check for close before acquire mx
-    if ( unlikely(closed) ) {
-        __closed_insert( chan, elem );
-        return;
-    } 
-
-    lock( mutex_lock );
-
-    #ifdef CHAN_STATS
-    if ( !closed ) operations++;
-    #endif
-
-    // if closed handle
-    if ( unlikely(closed) ) {
-        unlock( mutex_lock );
-        __closed_insert( chan, elem );
-        return;
-    }
 
     // have to check for the zero size channel case
@@ -187 +202 @@
         wake_one( cons );
         unlock( mutex_lock );
-        return true;
+        return;
     }
 
     // wait if buffer is full, work will be completed by someone else
     if ( count == size ) {
-        #ifdef CHAN_STATS
-        blocks++;
-        #endif
-
-        // check for if woken due to close
-        if ( unlikely( block( prods, &elem, mutex_lock ) ) )
-            __closed_insert( chan, elem );
+        block( prods, &elem, mutex_lock );
         return;
     } // if
@@ -205 +214 @@
         memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
         wake_one( cons );
-    } else __buf_insert( chan, elem );
+    } else insert_( chan, elem );
     
     unlock( mutex_lock );
-    return;
-}
-
-// handles buffer remove
-static inline void __buf_remove( channel(T) & chan, T & retval ) with(chan) {
-    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
-    count -= 1;
-    front = (front + 1) % size;
-}
-
-// does the buffer remove and potentially does waiting producer work
-static inline void __do_remove( channel(T) & chan, T & retval ) with(chan) {
-    __buf_remove( chan, retval );
-    if (count == size - 1 && !prods`isEmpty ) {
-        __buf_insert( chan, *(T *)prods`first.elem );  // do waiting producer work
-        wake_one( prods );
-    }
-}
-
-// needed to avoid an extra copy in closed case and single return val case
-static inline bool __internal_try_remove( channel(T) & chan, T & retval ) with(chan) {
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
     lock( mutex_lock );
-    #ifdef CHAN_STATS
-    operations++;
-    #endif
-    if ( count == 0 ) { unlock( mutex_lock ); return false; }
-    __do_remove( chan, retval );
-    unlock( mutex_lock );
-    return true;
-}
-
-// attempts a nonblocking remove
-// returns [T, true] if insert was successful
-// returns [T, false] if insert was successful (T uninit)
-static inline [T, bool] try_remove( channel(T) & chan ) {
     T retval;
-    return [ retval, __internal_try_remove( chan, retval ) ];
-}
-
-static inline T try_remove( channel(T) & chan, T elem ) {
-    T retval;
-    __internal_try_remove( chan, retval );
-    return retval;
-}
-
-// handles closed case of insert routine
-static inline void __closed_remove( channel(T) & chan, T & retval ) with(chan) {
-    channel_closed except{&channel_closed_vt, 0p, &chan };
-    throwResume except; // throw resumption
-    if ( !__internal_try_remove( chan, retval ) ) throw except; // if try to remove fails (would block), throw termination
-}
-
-static inline T remove( channel(T) & chan ) with(chan) {
-    T retval;
-    if ( unlikely(closed) ) {
-        __closed_remove( chan, retval );
-        return retval;
-    } 
-    lock( mutex_lock );
-
-    #ifdef CHAN_STATS
-    if ( !closed ) operations++;
-    #endif
-
-    if ( unlikely(closed) ) {
-        unlock( mutex_lock );
-        __closed_remove( chan, retval );
-        return retval;
-    } 
 
     // have to check for the zero size channel case
@@ -288 +233 @@
     // wait if buffer is empty, work will be completed by someone else
     if (count == 0) {
-        #ifdef CHAN_STATS
-        blocks++;
-        #endif
-        // check for if woken due to close
-        if ( unlikely( block( cons, &retval, mutex_lock ) ) )
-            __closed_remove( chan, retval );
+        block( cons, &retval, mutex_lock );
         return retval;
     }
 
     // Remove from buffer
-    __do_remove( chan, retval );
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front = (front + 1) % size;
+
+    if (count == size - 1 && !prods`isEmpty ) {
+        insert_( chan, *(T *)prods`first.elem );  // do waiting producer work
+        wake_one( prods );
+    }
 
     unlock( mutex_lock );
@@ -304 +251 @@
 }
 } // forall( T )
+#endif
+
+#ifdef __BARGE_CHANNEL
+forall( T ) {
+struct channel {
+    size_t size;
+    size_t front, back, count;
+    T * buffer;
+    fast_cond_var( exp_backoff_then_block_lock ) prods, cons;
+    exp_backoff_then_block_lock mutex_lock;
+};
+
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    buffer = aalloc( size );
+    prods{};
+    cons{};
+    mutex_lock{};
+}
+
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
+static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
+static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
+static inline bool has_waiters( channel(T) & chan ) with(chan) { return !empty( cons ) || !empty( prods ); }
+static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !empty( cons ); }
+static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !empty( prods ); }
+
+static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
+    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+
+
+static inline void insert( channel(T) & chan, T elem ) with(chan) {
+    lock( mutex_lock );
+
+    while ( count == size ) { 
+        wait( prods, mutex_lock );
+    } // if
+
+    insert_( chan, elem );
+    
+    if ( !notify_one( cons ) && count < size )
+        notify_one( prods );
+
+    unlock( mutex_lock );
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
+    lock( mutex_lock );
+    T retval;
+
+    while (count == 0) { 
+        wait( cons, mutex_lock );
+    }
+
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front = (front + 1) % size;
+
+    if ( !notify_one( prods ) && count > 0 )
+        notify_one( cons );
+
+    unlock( mutex_lock );
+    return retval;
+}
+
+} // forall( T )
+#endif
+
+#ifdef __NO_WAIT_CHANNEL
+forall( T ) {
+struct channel {
+    size_t size;
+    size_t front, back, count;
+    T * buffer;
+    thread$ * chair;
+    T * chair_elem;
+    exp_backoff_then_block_lock c_lock, p_lock;
+    __spinlock_t mutex_lock;
+};
+
+static inline void ?{}( channel(T) &c, size_t _size ) with(c) {
+    size = _size;
+    front = back = count = 0;
+    buffer = aalloc( size );
+    chair = 0p;
+    mutex_lock{};
+    c_lock{};
+    p_lock{};
+    lock( c_lock );
+}
+
+static inline void ?{}( channel(T) &c ){ ((channel(T) &)c){ 0 }; }
+static inline void ^?{}( channel(T) &c ) with(c) { delete( buffer ); }
+static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
+static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
+static inline bool has_waiters( channel(T) & chan ) with(chan) { return c_lock.lock_value != 0; }
+
+static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
+    memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
+    count += 1;
+    back++;
+    if ( back == size ) back = 0;
+}
+
+static inline void insert( channel(T) & chan, T elem ) with( chan ) {
+    lock( p_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+
+    insert_( chan, elem );
+
+    if ( count != size )
+        unlock( p_lock );
+
+    if ( count == 1 )
+        unlock( c_lock );
+        
+    unlock( mutex_lock );
+}
+
+static inline T remove( channel(T) & chan ) with(chan) {
+    lock( c_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
+    T retval;
+
+    // Remove from buffer
+    memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
+    count -= 1;
+    front = (front + 1) % size;
+
+    if ( count != 0 )
+        unlock( c_lock );
+
+    if ( count == size - 1 )
+        unlock( p_lock );
+        
+    unlock( mutex_lock );
+    return retval;
+}
+
+} // forall( T )
+#endif

libcfa/src/concurrency/locks.hfa

@@ -32 +32 @@
 #include <fstream.hfa>
 
+
 // futex headers
 #include <linux/futex.h>      /* Definition of FUTEX_* constants */
@@ -154 +155 @@
 // futex_mutex
 
+// - No cond var support
 // - Kernel thd blocking alternative to the spinlock
 // - No ownership (will deadlock on reacq)
@@ -183 +185 @@
         int state;
 
-        for( int spin = 4; spin < 1024; spin += spin) {
-                state = 0;
-                // if unlocked, lock and return
-                if (internal_try_lock(this, state)) return;
-                if (2 == state) break;
-                for (int i = 0; i < spin; i++) Pause();
-        }
-
-        // // no contention try to acquire
-        // if (internal_try_lock(this, state)) return;
+        
+        // // linear backoff omitted for now
+        // for( int spin = 4; spin < 1024; spin += spin) {
+        //      state = 0;
+        //      // if unlocked, lock and return
+        //      if (internal_try_lock(this, state)) return;
+        //      if (2 == state) break;
+        //      for (int i = 0; i < spin; i++) Pause();
+        // }
+
+        // no contention try to acquire
+        if (internal_try_lock(this, state)) return;
         
         // if not in contended state, set to be in contended state
@@ -205 +209 @@
 
 static inline void unlock(futex_mutex & this) with(this) {
-        // if uncontended do atomic unlock and then return
-    if (__atomic_exchange_n(&val, 0, __ATOMIC_RELEASE) == 1) return;
+        // if uncontended do atomice unlock and then return
+        if (__atomic_fetch_sub(&val, 1, __ATOMIC_RELEASE) == 1) return; // TODO: try acq/rel
         
         // otherwise threads are blocked so we must wake one
+        __atomic_store_n((int *)&val, 0, __ATOMIC_RELEASE);
         futex((int *)&val, FUTEX_WAKE, 1);
 }
@@ -217 +222 @@
 // to set recursion count after getting signalled;
 static inline void on_wakeup( futex_mutex & f, size_t recursion ) {}
-
-//-----------------------------------------------------------------------------
-// go_mutex
-
-// - Kernel thd blocking alternative to the spinlock
-// - No ownership (will deadlock on reacq)
-// - Golang's flavour of mutex
-// - Impl taken from Golang: src/runtime/lock_futex.go
-struct go_mutex {
-        // lock state any state other than UNLOCKED is locked
-        // enum LockState { UNLOCKED = 0, LOCKED = 1, SLEEPING = 2 };
-        
-        // stores a lock state
-        int val; 
-};
-
-static inline void  ?{}( go_mutex & this ) with(this) { val = 0; }
-
-static inline bool internal_try_lock(go_mutex & this, int & compare_val, int new_val ) with(this) {
-        return __atomic_compare_exchange_n((int*)&val, (int*)&compare_val, new_val, false, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE);
-}
-
-static inline int internal_exchange(go_mutex & this, int swap ) with(this) {
-        return __atomic_exchange_n((int*)&val, swap, __ATOMIC_ACQUIRE);
-}
-
-const int __go_mtx_spins = 4;
-const int __go_mtx_pauses = 30;
-// if this is called recursively IT WILL DEADLOCK!!!!!
-static inline void lock(go_mutex & this) with(this) {
-        int state, init_state;
-
-    // speculative grab
-    state = internal_exchange(this, 1);
-    if ( !state ) return; // state == 0
-    init_state = state;
-    for (;;) {
-        for( int i = 0; i < __go_mtx_spins; i++ ) {
-            while( !val ) { // lock unlocked
-                state = 0;
-                if (internal_try_lock(this, state, init_state)) return;
-            }
-            for (int i = 0; i < __go_mtx_pauses; i++) Pause();
-        }
-
-        while( !val ) { // lock unlocked
-            state = 0;
-            if (internal_try_lock(this, state, init_state)) return;
-        }
-        sched_yield();
-        
-        // if not in contended state, set to be in contended state
-        state = internal_exchange(this, 2);
-        if ( !state ) return; // state == 0
-        init_state = 2;
-        futex((int*)&val, FUTEX_WAIT, 2); // if val is not 2 this returns with EWOULDBLOCK
-    }
-}
-
-static inline void unlock( go_mutex & this ) with(this) {
-        // if uncontended do atomic unlock and then return
-    if (__atomic_exchange_n(&val, 0, __ATOMIC_RELEASE) == 1) return;
-        
-        // otherwise threads are blocked so we must wake one
-        futex((int *)&val, FUTEX_WAKE, 1);
-}
-
-static inline void on_notify( go_mutex & f, thread$ * t){ unpark(t); }
-static inline size_t on_wait( go_mutex & f ) {unlock(f); return 0;}
-static inline void on_wakeup( go_mutex & f, size_t recursion ) {}
 
 //-----------------------------------------------------------------------------
@@ -336 +271 @@
         this.lock_value = 0;
 }
-
-static inline void  ^?{}( exp_backoff_then_block_lock & this ){}
 
 static inline bool internal_try_lock(exp_backoff_then_block_lock & this, size_t & compare_val) with(this) {