Changes in / [ff443e5:9082d7e8]

Files:

• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/barrier.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/churn.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/contend.cfa (modified) (9 diffs)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/daisy_chain.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/hot_potato.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/ping_pong.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/cfa/pub_sub.cfa (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/barrier/barrier.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/barrier/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/churn/churn.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/churn/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend.go (deleted)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend/contend.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/contend/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/daisy_chain/daisy_chain.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/daisy_chain/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/hot_potato/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/hot_potato/hot_potato.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/ping_pong/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/ping_pong/ping_pong.go (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/pub_sub/go.mod (added)
• doc/theses/colby_parsons_MMAth/benchmarks/channels/go/pub_sub/pub_sub.go (added)
• 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 (added)
• tests/concurrent/channels/.expect/contend.txt (added)
• tests/concurrent/channels/.expect/daisy_chain.txt (added)
• tests/concurrent/channels/.expect/hot_potato.txt (added)
• tests/concurrent/channels/.expect/ping_pong.txt (added)
• tests/concurrent/channels/.expect/pub_sub.txt (added)
• tests/concurrent/channels/barrier.cfa (added)
• tests/concurrent/channels/churn.cfa (added)
• tests/concurrent/channels/contend.cfa (added)
• tests/concurrent/channels/daisy_chain.cfa (added)
• tests/concurrent/channels/hot_potato.cfa (added)
• tests/concurrent/channels/ping_pong.cfa (added)
• tests/concurrent/channels/pub_sub.cfa (added)

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

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

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

@@ -36 +36 @@
     procs.append(int(val))
 
-# 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
+# 3rd line has number of variants
 line = readfile.readline()
 names = line.split()
@@ -50 +44 @@
 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
-currLocks = -1 # default val
+currBench = Bench.Unset # default val
 count = 0
 procCount = 0
 currVariant = 0
-name = "Aggregate Lock"
+name = ""
 var_name = ""
 sendData = [0.0 for j in range(numVariants)]
@@ -64 +68 @@
     # print(line)
     
-    if currLocks == -1:
-        lineArr = line.split()
-        currLocks = lineArr[-1]
+    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()
         continue
 
@@ -95 +122 @@
             if currVariant == numVariants:
                 fig, ax = plt.subplots()
-                plt.title(name + " Benchmark: " + str(currLocks) + " Locks")
-                plt.ylabel("Throughput (entries)")
+                plt.title(name + " Benchmark")
+                plt.ylabel("Throughput (channel operations)")
                 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 + "Aggregate_Lock_" + str(currLocks) + ".png")
-                plt.savefig("plots/" + machineName + "Aggregate_Lock_" + str(currLocks) + ".pgf")
+                # fig.savefig("plots/" + machineName + name + ".png")
+                plt.savefig("plots/" + machineName + name + ".pgf")
                 fig.clf()
 
                 # reset
-                currLocks = -1
+                currBench = Bench.Unset
                 currVariant = 0

doc/theses/colby_parsons_MMAth/benchmarks/channels/run

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

libcfa/src/concurrency/actor.hfa

@@ -35 +35 @@
 
 // show stats
-// #define STATS
+// #define ACTOR_STATS
 
 // forward decls
@@ -122 +122 @@
     copy_queue * c_queue;           // current queue
     volatile bool being_processed;  // flag to prevent concurrent processing
-    #ifdef STATS
+    #ifdef ACTOR_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 STATS
+    #ifdef ACTOR_STATS
     id = i;
     missed = 0;
@@ -153 +153 @@
     // check if queue is being processed elsewhere
     if ( unlikely( being_processed ) ) {
-        #ifdef STATS
+        #ifdef ACTOR_STATS
         missed++;
         #endif
@@ -175 +175 @@
 struct worker_info {
     volatile unsigned long long stamp;
-    #ifdef STATS
+    #ifdef ACTOR_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 STATS
+    #ifdef ACTOR_STATS
     this.stolen_from = 0;
     this.try_steal = 0;                             // attempts to steal
@@ -194 +194 @@
 }
 
-// #ifdef STATS
+// #ifdef ACTOR_STATS
 // unsigned int * stolen_arr;
 // unsigned int * replaced_queue;
@@ -206 +206 @@
 };
 
-#ifdef STATS
+#ifdef ACTOR_STATS
 // aggregate counters for statistics
 size_t __total_tries = 0, __total_stolen = 0, __total_workers, __all_gulps = 0,
@@ -235 +235 @@
 }; // executor
 
-// #ifdef STATS
+// #ifdef ACTOR_STATS
 // __spinlock_t out_lock;
 // #endif
 static inline void ^?{}( worker & mutex this ) with(this) { 
-    #ifdef STATS
+    #ifdef ACTOR_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 STATS
+    #ifdef ACTOR_STATS
     // stolen_arr = aalloc( nrqueues );
     // replaced_queue = aalloc( nrqueues );
@@ -341 +341 @@
     } // for
 
-    #ifdef STATS
+    #ifdef ACTOR_STATS
     size_t misses = 0;
     for ( i; nrqueues ) {
@@ -358 +358 @@
     if ( seperate_clus ) delete( cluster );
 
-    #ifdef STATS // print formatted stats
+    #ifdef ACTOR_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 STATS
+    #ifdef ACTOR_STATS
     __atomic_fetch_add( &__num_actors_stats, 1, __ATOMIC_SEQ_CST );
     #endif
@@ -513 +513 @@
             continue;
 
-        #ifdef STATS
+        #ifdef ACTOR_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 // STATS
+        #endif // ACTOR_STATS
 
         return;
@@ -558 +558 @@
 
 void main( worker & this ) with(this) {
-    // #ifdef STATS
+    // #ifdef ACTOR_STATS
     // for ( i; executor_->nrqueues ) {
     //     replaced_queue[i] = 0;
@@ -587 +587 @@
         }
         transfer( *curr_work_queue, &current_queue );
-        #ifdef STATS
+        #ifdef ACTOR_STATS
         executor_->w_infos[id].gulps++;
-        #endif // STATS
+        #endif // ACTOR_STATS
         #ifdef __STEAL
         if ( isEmpty( *current_queue ) ) {
@@ -599 +599 @@
             __atomic_store_n( &executor_->w_infos[id].stamp, rdtscl(), __ATOMIC_RELAXED );
             
-            #ifdef STATS
+            #ifdef ACTOR_STATS
             executor_->w_infos[id].try_steal++;
-            #endif // STATS
+            #endif // ACTOR_STATS
             
             steal_work( this, start + prng( range ) );
@@ -608 +608 @@
         #endif // __STEAL
         while ( ! isEmpty( *current_queue ) ) {
-            #ifdef STATS
+            #ifdef ACTOR_STATS
             executor_->w_infos[id].processed++;
             #endif
@@ -636 +636 @@
 
 static inline void __reset_stats() {
-    #ifdef STATS
+    #ifdef ACTOR_STATS
     __total_tries = 0;
     __total_stolen = 0;

libcfa/src/concurrency/channel.hfa

@@ -3 +3 @@
 #include <locks.hfa>
 #include <list.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
+#include <mutex_stmt.hfa>
 
 // link field used for threads waiting on channel
@@ -148 +23 @@
 }
 
+// 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 channel {
-    size_t size;
-    size_t front, back, count;
+struct __attribute__((aligned(128))) channel {
+    size_t size, front, back, count;
     T * buffer;
-    dlist( wait_link ) prods, cons;
-    exp_backoff_then_block_lock mutex_lock;
+    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
 };
 
@@ -165 +74 @@
     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) { delete( buffer ); }
+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 size_t get_count( channel(T) & chan ) with(chan) { return count; }
 static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
@@ -175 +95 @@
 static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
 
-static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
+// 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) {
     memcpy((void *)&buffer[back], (void *)&elem, sizeof(T));
     count += 1;
@@ -182 +131 @@
 }
 
-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();
+// 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) {
+    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) {
-    lock( mutex_lock );
+    // 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
@@ -202 +187 @@
         wake_one( cons );
         unlock( mutex_lock );
-        return;
+        return true;
     }
 
     // wait if buffer is full, work will be completed by someone else
     if ( count == size ) {
-        block( prods, &elem, mutex_lock );
+        #ifdef CHAN_STATS
+        blocks++;
+        #endif
+
+        // check for if woken due to close
+        if ( unlikely( block( prods, &elem, mutex_lock ) ) )
+            __closed_insert( chan, elem );
         return;
     } // if
@@ -214 +205 @@
         memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
         wake_one( cons );
-    } else insert_( chan, elem );
+    } else __buf_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) {
+    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) {
-    lock( mutex_lock );
     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
@@ -233 +288 @@
     // wait if buffer is empty, work will be completed by someone else
     if (count == 0) {
-        block( cons, &retval, mutex_lock );
+        #ifdef CHAN_STATS
+        blocks++;
+        #endif
+        // check for if woken due to close
+        if ( unlikely( block( cons, &retval, mutex_lock ) ) )
+            __closed_remove( chan, retval );
         return retval;
     }
 
     // Remove from buffer
-    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 );
-    }
+    __do_remove( chan, retval );
 
     unlock( mutex_lock );
@@ -251 +304 @@
 }
 } // 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 */
@@ -155 +154 @@
 // futex_mutex
 
-// - No cond var support
 // - Kernel thd blocking alternative to the spinlock
 // - No ownership (will deadlock on reacq)
@@ -185 +183 @@
         int state;
 
-        
-        // // 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;
+        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
@@ -209 +205 @@
 
 static inline void unlock(futex_mutex & this) with(this) {
-        // if uncontended do atomice unlock and then return
-        if (__atomic_fetch_sub(&val, 1, __ATOMIC_RELEASE) == 1) return; // TODO: try acq/rel
+        // 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
-        __atomic_store_n((int *)&val, 0, __ATOMIC_RELEASE);
         futex((int *)&val, FUTEX_WAKE, 1);
 }
@@ -222 +217 @@
 // 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 ) {}
 
 //-----------------------------------------------------------------------------
@@ -271 +336 @@
         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) {