Changeset b18830e for src

Timestamp:

Sep 20, 2017, 2:07:57 PM (7 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, aaron-thesis, arm-eh, ast-experimental, cleanup-dtors, deferred_resn, demangler, enum, forall-pointer-decay, jacob/cs343-translation, jenkins-sandbox, master, new-ast, new-ast-unique-expr, new-env, no_list, persistent-indexer, pthread-emulation, qualifiedEnum, resolv-new, with_gc

Children:

0895cba

Parents:

f980549

Message:

Refactoring monitor code in prevision for proper waitfor support

• added monitor group struct
• else and timeout now return negative results

Location:

src

Files:

• Concurrency/Waitfor.cc (modified) (2 diffs)
• libcfa/concurrency/invoke.h (modified) (1 diff)
• libcfa/concurrency/kernel.c (modified) (6 diffs)
• libcfa/concurrency/monitor (modified) (1 diff)
• libcfa/concurrency/monitor.c (modified) (16 diffs)
• libcfa/concurrency/thread (modified) (1 diff)
• libcfa/concurrency/thread.c (modified) (1 diff)

src/Concurrency/Waitfor.cc

@@ -484 +484 @@
                                 new CaseStmt(
                                         noLabels,
-                                        new ConstantExpr( Constant::from_ulong( i++ ) ),
+                                        new ConstantExpr( Constant::from_int( -2 ) ),
                                         {
                                                 waitfor->timeout.statement,
@@ -501 +501 @@
                                 new CaseStmt(
                                         noLabels,
-                                        new ConstantExpr( Constant::from_ulong( i++ ) ),
+                                        new ConstantExpr( Constant::from_int( -1 ) ),
                                         {
                                                 waitfor->orelse.statement,

src/libcfa/concurrency/invoke.h

@@ -94 +94 @@
             unsigned short acceptable_count;          // number of acceptable functions
             short accepted_index;                     // the index of the accepted function, -1 if none
-       };
+      };
+
+      struct __monitor_group {
+            struct monitor_desc ** list;              // currently held monitors
+            short                  size;              // number of currently held monitors
+            fptr_t                 func;              // last function that acquired monitors
+      };
 
       struct thread_desc {
             // Core threading fields
-            struct coroutine_desc cor;                // coroutine body used to store context
-            struct monitor_desc mon;                  // monitor body used for mutual exclusion
+            struct coroutine_desc  self_cor;          // coroutine body used to store context
+            struct monitor_desc    self_mon;          // monitor body used for mutual exclusion
+            struct monitor_desc *  self_mon_p;        // pointer to monitor with sufficient lifetime for current monitors
+            struct __monitor_group monitors;          // monitors currently held by this thread
 
             // Link lists fields
             struct thread_desc * next;                // instrusive link field for threads
 
-            // Current status related to monitors
-            struct monitor_desc ** current_monitors;  // currently held monitors
-            unsigned short current_monitor_count;     // number of currently held monitors
-            fptr_t current_monitor_func;              // last function that acquired monitors
+
      };
+
+     #ifdef __CFORALL__
+     extern "Cforall" {
+            static inline monitor_desc * ?[?]( const __monitor_group & this, ptrdiff_t index ) {
+                  return this.list[index];
+            }
+
+            static inline bool ?==?( const __monitor_group & lhs, const __monitor_group & rhs ) {
+                  if( lhs.size != rhs.size ) return false;
+                  if( lhs.func != rhs.func ) return false;
+
+                  // Check that all the monitors match
+                  for( int i = 0; i < lhs.size; i++ ) {
+                        // If not a match, check next function
+                        if( lhs[i] != rhs[i] ) return false;
+                  }
+
+                  return true;
+            }
+      }
+      #endif
 
 #endif //_INVOKE_H_

src/libcfa/concurrency/kernel.c

@@ -106 +106 @@
 
 void ?{}( thread_desc & this, current_stack_info_t * info) {
-        (this.cor){ info };
+        (this.self_cor){ info };
 }
 
@@ -115 +115 @@
 void ?{}(processorCtx_t & this, processor * proc) {
         (this.__cor){ "Processor" };
-        this.__cor.starter = &mainThread->cor;
+        this.__cor.starter = &mainThread->self_cor;
         this.proc = proc;
         proc->runner = &this;
@@ -328 +328 @@
         // if( !thrd ) return;
         verify( thrd );
-        verify( thrd->cor.state != Halted );
+        verify( thrd->self_cor.state != Halted );
 
         verify( disable_preempt_count > 0 );
@@ -373 +373 @@
         assert(thrd);
         disable_interrupts();
-        assert( thrd->cor.state != Halted );
+        assert( thrd->self_cor.state != Halted );
         this_processor->finish.action_code = Schedule;
         this_processor->finish.thrd = thrd;
@@ -466 +466 @@
         this_processor = mainProcessor;
         this_thread = mainThread;
-        this_coroutine = &mainThread->cor;
+        this_coroutine = &mainThread->self_cor;
 
         // Enable preemption
@@ -547 +547 @@
         thread_desc * thrd = kernel_data;
 
-        int len = snprintf( abort_text, abort_text_size, "Error occurred while executing task %.256s (%p)", thrd->cor.name, thrd );
+        int len = snprintf( abort_text, abort_text_size, "Error occurred while executing task %.256s (%p)", thrd->self_cor.name, thrd );
         __lib_debug_write( STDERR_FILENO, abort_text, len );
 

src/libcfa/concurrency/monitor

@@ -105 +105 @@
 
 struct __acceptable_t {
-        fptr_t func;
-        unsigned short count;
-        monitor_desc ** monitors;
+        __monitor_group monitors;
         bool is_dtor;
 };

src/libcfa/concurrency/monitor.c

@@ -25 +25 @@
 static inline void set_owner( monitor_desc * this, thread_desc * owner );
 static inline thread_desc * next_thread( monitor_desc * this );
-static inline int is_accepted( thread_desc * owner, monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() );
+static inline int is_accepted( thread_desc * owner, monitor_desc * this, const __monitor_group & monitors );
 
 static inline void lock_all( spinlock ** locks, unsigned short count );
@@ -42 +42 @@
 static inline unsigned short insert_unique( thread_desc ** thrds, unsigned short end, thread_desc * val );
 
-static inline thread_desc * search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count );
+static inline [thread_desc *, int] search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count );
+
+static inline short count_max( short acc_count, __acceptable_t * acceptables );
+static inline short aggregate( monitor_desc ** storage, short count, __acceptable_t * acceptables );
+static inline void  set_mask ( monitor_desc ** storage, short count, __acceptable_t * acceptables, short acc_count );
 
 //-----------------------------------------------------------------------------
@@ -68 +72 @@
 extern "C" {
         // Enter single monitor
-        static void __enter_monitor_desc( monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() ) {
+        static void __enter_monitor_desc( const __monitor_group & group ) {
+                monitor_desc * this = group.list[0];
+
                 // Lock the monitor spinlock, lock_yield to reduce contention
                 lock_yield( &this->lock DEBUG_CTX2 );
@@ -89 +95 @@
                         LIB_DEBUG_PRINT_SAFE("Kernel :  mon already owned \n");
                 }
-                else if( (this->accepted_index = is_accepted( thrd, this, group, group_cnt, func)) >= 0 ) {
+                else if( (this->accepted_index = is_accepted( thrd, this, group)) >= 0 ) {
                         // Some one was waiting for us, enter
                         set_owner( this, thrd );
@@ -146 +152 @@
         // Should never return
         void __leave_thread_monitor( thread_desc * thrd ) {
-                monitor_desc * this = &thrd->mon;
+                monitor_desc * this = &thrd->self_mon;
 
                 // Lock the monitor now
@@ -153 +159 @@
                 disable_interrupts();
 
-                thrd->cor.state = Halted;
+                thrd->self_cor.state = Halted;
 
                 verifyf( thrd == this->owner, "Expected owner to be %p, got %p (r: %i)", thrd, this->owner, this->recursion );
@@ -178 +184 @@
 // Enter multiple monitor
 // relies on the monitor array being sorted
-static inline void enter(monitor_desc ** monitors, int count, void (*func)() ) {
-        for(int i = 0; i < count; i++) {
-                __enter_monitor_desc( monitors[i], monitors, count, func );
+static inline void enter( __monitor_group monitors ) {
+        for(int i = 0; i < monitors.size; i++) {
+                __enter_monitor_desc( monitors );
         }
 }
@@ -203 +209 @@
 
         // Save previous thread context
-        this.prev_mntrs = this_thread->current_monitors;
-        this.prev_count = this_thread->current_monitor_count;
-        this.prev_func  = this_thread->current_monitor_func;
+        this.prev_mntrs = this_thread->monitors.list;
+        this.prev_count = this_thread->monitors.size;
+        this.prev_func  = this_thread->monitors.func;
 
         // Update thread context (needed for conditions)
-        this_thread->current_monitors      = m;
-        this_thread->current_monitor_count = count;
-        this_thread->current_monitor_func  = func;
+        this_thread->monitors.list = m;
+        this_thread->monitors.size = count;
+        this_thread->monitors.func = func;
 
         // Enter the monitors in order
-        enter( this.m, this.count, func );
+        __monitor_group group = {this.m, this.count, func};
+        enter( group );
 }
 
@@ -223 +230 @@
 
         // Restore thread context
-        this_thread->current_monitors      = this.prev_mntrs;
-        this_thread->current_monitor_count = this.prev_count;
-        this_thread->current_monitor_func  = this.prev_func;
+        this_thread->monitors.list = this.prev_mntrs;
+        this_thread->monitors.size = this.prev_count;
+        this_thread->monitors.func = this.prev_func;
 }
 
@@ -315 +322 @@
         LIB_DEBUG_DO(
                 thread_desc * this_thrd = this_thread;
-                if ( this->monitor_count != this_thrd->current_monitor_count ) {
-                        abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", this, this->monitor_count, this_thrd->current_monitor_count );
+                if ( this->monitor_count != this_thrd->monitors.size ) {
+                        abortf( "Signal on condition %p made with different number of monitor(s), expected %i got %i", this, this->monitor_count, this_thrd->monitors.size );
                 }
 
                 for(int i = 0; i < this->monitor_count; i++) {
-                        if ( this->monitors[i] != this_thrd->current_monitors[i] ) {
-                                abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->current_monitors[i] );
+                        if ( this->monitors[i] != this_thrd->monitors.list[i] ) {
+                                abortf( "Signal on condition %p made with different monitor, expected %p got %i", this, this->monitors[i], this_thrd->monitors.list[i] );
                         }
                 }
@@ -397 +404 @@
 
 //-----------------------------------------------------------------------------
-// Internal scheduling
-int __waitfor_internal( unsigned short acc_count, __acceptable_t * acceptables ) {
-        thread_desc * thrd = this_thread;
+// External scheduling
+// cases to handle :
+//      - target already there :
+//              block and wake
+//      - dtor already there
+//              put thread on signaller stack
+//      - non-blocking
+//              return else
+//      - timeout
+//              return timeout
+//      - block
+//              setup mask
+//              block
+int __waitfor_internal( unsigned short acc_count, __acceptable_t * acceptables, int duration ) {
+        // This statment doesn't have a contiguous list of monitors...
+        // Create one!
+        short max = count_max( acc_count, acceptables );
+        monitor_desc * mon_storage[max];
+        short actual_count = aggregate( mon_storage, acc_count, acceptables );
 
         // Create storage for monitor context
-        monitor_ctx( acceptables->monitors, acceptables->count );
+        monitor_ctx( mon_storage, actual_count );
 
         // Lock all monitors (aggregates the lock them as well)
         lock_all( monitors, locks, count );
 
-        // Create the node specific to this wait operation
-        wait_ctx_primed( thrd, 0 );
-
-        // Check if the entry queue
-        thread_desc * next = search_entry_queue( acceptables, acc_count, monitors, count );
-
-        LIB_DEBUG_PRINT_SAFE("Owner(s) :");
-        for(int i = 0; i < count; i++) {
-                LIB_DEBUG_PRINT_SAFE(" %p", monitors[i]->owner );
-        }
-        LIB_DEBUG_PRINT_SAFE("\n");
-
-        LIB_DEBUG_PRINT_SAFE("Passing mon to %p\n", next);
-
-        if( !next ) {
-                // Update acceptables on the current monitors
-                for(int i = 0; i < count; i++) {
-                        monitors[i]->acceptables = acceptables;
-                        monitors[i]->acceptable_count = acc_count;
-                }
-        }
-        else {
-                for(int i = 0; i < count; i++) {
-                        set_owner( monitors[i], next );
-                }
-        }
+        {
+                // Check if the entry queue
+                thread_desc * next;
+                int index;
+                [next, index] = search_entry_queue( acceptables, acc_count, monitors, count );
+
+                if( next ) {
+                        if( acceptables[index].is_dtor ) {
+                                #warning case not implemented
+                        }
+                        else {
+                                save_recursion( monitors, recursions, count );
+
+                                // Everything is ready to go to sleep
+                                BlockInternal( locks, count, &next, 1 );
+
+
+                                //WE WOKE UP
+
+
+                                //We are back, restore the owners and recursions
+                                lock_all( locks, count );
+                                restore_recursion( monitors, recursions, count );
+                                unlock_all( locks, count );
+                        }
+
+                        return index;
+                }
+        }
+
+
+        if( duration == 0 ) return -1;
+
+        set_mask( monitors, count, acceptables, acc_count );
+
+        verifyf( duration < 0, "Timeout on waitfor statments not supported yet.");
 
 
@@ -439 +472 @@
 
         // Everything is ready to go to sleep
-        BlockInternal( locks, count, &next, next ? 1 : 0 );
+        BlockInternal( locks, count );
 
 
@@ -485 +518 @@
 }
 
-static inline int is_accepted( thread_desc * owner, monitor_desc * this, monitor_desc ** group, int group_cnt, void (*func)() ) {
-        __acceptable_t* accs = this->acceptables; // Optim
-        int acc_cnt = this->acceptable_count;
-
-        // Check if there are any acceptable functions
-        if( !accs ) return -1;
-
-        // If this isn't the first monitor to test this, there is no reason to repeat the test.
-        if( this != group[0] ) return group[0]->accepted_index;
-
-        // For all acceptable functions check if this is the current function.
-        OUT_LOOP:
-        for( int i = 0; i < acc_cnt; i++ ) {
-                __acceptable_t * acc = &accs[i];
-
-                // if function matches, check the monitors
-                if( acc->func == func ) {
-
-                        // If the group count is different then it can't be a match
-                        if( acc->count != group_cnt ) return -1;
-
-                        // Check that all the monitors match
-                        for( int j = 0; j < group_cnt; j++ ) {
-                                // If not a match, check next function
-                                if( acc->monitors[j] != group[j] ) continue OUT_LOOP;
-                        }
-
-                        // It's a complete match, accept the call
-                        return i;
-                }
-        }
-
-        // No function matched
-        return -1;
-}
-
 static inline void init( int count, monitor_desc ** monitors, __condition_node_t * waiter, __condition_criterion_t * criteria ) {
         for(int i = 0; i < count; i++) {
@@ -607 +604 @@
         if( !this->monitors ) {
                 // LIB_DEBUG_PRINT_SAFE("Branding\n");
-                assertf( thrd->current_monitors != NULL, "No current monitor to brand condition %p", thrd->current_monitors );
-                this->monitor_count = thrd->current_monitor_count;
+                assertf( thrd->monitors.list != NULL, "No current monitor to brand condition %p", thrd->monitors.list );
+                this->monitor_count = thrd->monitors.size;
 
                 this->monitors = malloc( this->monitor_count * sizeof( *this->monitors ) );
                 for( int i = 0; i < this->monitor_count; i++ ) {
-                        this->monitors[i] = thrd->current_monitors[i];
+                        this->monitors[i] = thrd->monitors.list[i];
                 }
         }
@@ -628 +625 @@
 }
 
-
-static inline bool match( __acceptable_t * acc, thread_desc * thrd ) {
-        verify( thrd );
-        verify( acc );
-        if( acc->func != thrd->current_monitor_func ) return false;
-
-        return true;
-}
-
-static inline thread_desc * search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count ) {
+static inline int is_accepted( thread_desc * owner, monitor_desc * this, const __monitor_group & group ) {
+        __acceptable_t* accs = this->acceptables; // Optim
+        int acc_cnt = this->acceptable_count;
+
+        // Check if there are any acceptable functions
+        if( !accs ) return -1;
+
+        // If this isn't the first monitor to test this, there is no reason to repeat the test.
+        if( this != group[0] ) return group[0]->accepted_index;
+
+        // For all acceptable functions check if this is the current function.
+        for( int i = 0; i < acc_cnt; i++ ) {
+                __acceptable_t * acc = &accs[i];
+
+                if( acc->monitors == group ) return i;
+        }
+
+        // No function matched
+        return -1;
+}
+
+static inline [thread_desc *, int] search_entry_queue( __acceptable_t * acceptables, int acc_count, monitor_desc ** monitors, int count ) {
 
         __thread_queue_t * entry_queue = &monitors[0]->entry_queue;
@@ -647 +656 @@
         {
                 // For each acceptable check if it matches
+                int i;
                 __acceptable_t * acc_end = acceptables + acc_count;
-                for( __acceptable_t * acc_it = acceptables; acc_it != acc_end; acc_it++ ) {
+                for( __acceptable_t * acc_it = acceptables; acc_it != acc_end; acc_it++, i++ ) {
                         // Check if we have a match
-                        if( match( acc_it, *thrd_it ) ) {
+                        if( acc_it->monitors == (*thrd_it)->monitors ) {
 
                                 // If we have a match return it
                                 // after removeing it from the entry queue
-                                return remove( entry_queue, thrd_it );
+                                return [remove( entry_queue, thrd_it ), i];
                         }
                 }
         }
 
-        return NULL;
-}
+        return [0, -1];
+}
+
+static inline short count_max( short acc_count, __acceptable_t * acceptables ) {
+        short max = 0;
+        for( int i = 0; i < acc_count; i++ ) {
+                max += acceptables[i].monitors.size;
+        }
+        return max;
+}
+
+static inline short aggregate( monitor_desc ** storage, short count, __acceptable_t * acceptables ) {
+        #warning function not implemented
+        return 0;
+}
+
+static inline void set_mask( monitor_desc ** storage, short count, __acceptable_t * acceptables, short acc_count ) {
+        for(int i = 0; i < count; i++) {
+                storage[i]->acceptables      = acceptables;
+                storage[i]->acceptable_count = acc_count;
+                storage[i]->accepted_index   = -1;
+        }
+}
+
+
 void ?{}( __condition_blocked_queue_t & this ) {
         this.head = NULL;

src/libcfa/concurrency/thread

@@ -36 +36 @@
 forall( dtype T | is_thread(T) )
 static inline coroutine_desc* get_coroutine(T & this) {
-        return &get_thread(this)->cor;
+        return &get_thread(this)->self_cor;
 }
 
 forall( dtype T | is_thread(T) )
 static inline monitor_desc* get_monitor(T & this) {
-        return &get_thread(this)->mon;
+        return &get_thread(this)->self_mon;
 }
 
 static inline coroutine_desc* get_coroutine(thread_desc * this) {
-        return &this->cor;
+        return &this->self_cor;
 }
 
 static inline monitor_desc* get_monitor(thread_desc * this) {
-        return &this->mon;
+        return &this->self_mon;
 }
 

src/libcfa/concurrency/thread.c

@@ -33 +33 @@
 
 void ?{}(thread_desc& this) {
-        (this.cor){};
-        this.cor.name = "Anonymous Coroutine";
-        this.mon.owner = &this;
-        this.mon.recursion = 1;
+        (this.self_cor){};
+        this.self_cor.name = "Anonymous Coroutine";
+        this.self_mon.owner = &this;
+        this.self_mon.recursion = 1;
+        this.self_mon_p = &this.self_mon;
         this.next = NULL;
 
-        this.current_monitors      = &this.mon;
-        this.current_monitor_count = 1;
+        (this.monitors){ &this.self_mon_p, 1, (fptr_t)0 };
 }
 
 void ^?{}(thread_desc& this) {
-        ^(this.cor){};
+        ^(this.self_cor){};
 }