Avoid deadlock with COI

[jlpeyton]

Summary

When an asynchronous offload task is completed, COI calls the runtime to queue a "destructor tasks". When the task deques are full, a dead-lock situation arises where the OpenMP threads are inside but cannot progress because the COI thread is stuck inside the runtime trying to find a slot in a deque.

This patch implements the solution where the task deques doubled in size when a task is being queued from a COI thread.

[tlwilmar]

LGTM

openmp/trunk/runtime/src/kmp.h

 
 # ifndef KMP_STATIC_STEAL_ENABLED
 #  define KMP_STATIC_STEAL_ENABLED 1
 # endif
 
 #define TASK_CURRENT_NOT_QUEUED  0
 #define TASK_CURRENT_QUEUED      1
 
-#define TASK_DEQUE_BITS          8  // Used solely to define TASK_DEQUE_SIZE and TASK_DEQUE_MASK.
-#define TASK_DEQUE_SIZE          ( 1 << TASK_DEQUE_BITS )
-#define TASK_DEQUE_MASK          ( TASK_DEQUE_SIZE - 1 )
-
 #ifdef BUILD_TIED_TASK_STACK
 #define TASK_STACK_EMPTY         0  // entries when the stack is empty
 
 #define TASK_STACK_BLOCK_BITS    5  // Used to define TASK_STACK_SIZE and TASK_STACK_MASK
 #define TASK_STACK_BLOCK_SIZE    ( 1 << TASK_STACK_BLOCK_BITS ) // Number of entries in each task stack array
 #define TASK_STACK_INDEX_MASK    ( TASK_STACK_BLOCK_SIZE - 1 )  // Mask for determining index into stack block
 #endif // BUILD_TIED_TASK_STACK
 

 KMP_BUILD_ASSERT( sizeof(kmp_taskdata_t) % sizeof(void *) == 0 );
 
 // Data for task team but per thread
 typedef struct kmp_base_thread_data {
     kmp_info_p *            td_thr;                // Pointer back to thread info
                                                    // Used only in __kmp_execute_tasks_template, maybe not avail until task is queued?
     kmp_bootstrap_lock_t    td_deque_lock;         // Lock for accessing deque
     kmp_taskdata_t **       td_deque;              // Deque of tasks encountered by td_thr, dynamically allocated
+    kmp_int32               td_deque_size;         // Size of deck
     kmp_uint32              td_deque_head;         // Head of deque (will wrap)
     kmp_uint32              td_deque_tail;         // Tail of deque (will wrap)
     kmp_int32               td_deque_ntasks;       // Number of tasks in deque
                                                    // GEH: shouldn't this be volatile since used in while-spin?
     kmp_int32               td_deque_last_stolen;  // Thread number of last successful steal
 #ifdef BUILD_TIED_TASK_STACK
     kmp_task_stack_t        td_susp_tied_tasks;    // Stack of suspended tied tasks for task scheduling constraint
 #endif // BUILD_TIED_TASK_STACK
 } kmp_base_thread_data_t;
 
+#define TASK_DEQUE_BITS          8  // Used solely to define INITIAL_TASK_DEQUE_SIZE
+#define INITIAL_TASK_DEQUE_SIZE  ( 1 << TASK_DEQUE_BITS )
+
+#define TASK_DEQUE_SIZE(td)     ((td).td_deque_size)
+#define TASK_DEQUE_MASK(td)     ((td).td_deque_size - 1)
+
 typedef union KMP_ALIGN_CACHE kmp_thread_data {
     kmp_base_thread_data_t  td;
     double                  td_align;       /* use worst case alignment */
     char                    td_pad[ KMP_PAD(kmp_base_thread_data_t, CACHE_LINE) ];
 } kmp_thread_data_t;
 
 
 // Data for task teams which are used when tasking is enabled for the team

openmp/trunk/runtime/src/kmp_debugger.c

#endif
     addr_and_size_of( func_fork ),
     addr_and_size_of( func_fork_teams ),
     addr_and_size_of( __kmp_team_counter ),
     addr_and_size_of( __kmp_task_counter ),
     addr_and_size_of( kmp_omp_nthr_info_buffer ),
     sizeof( void * ),
     OMP_LOCK_T_SIZE < sizeof(void *),
     bs_last_barrier,
-    TASK_DEQUE_SIZE,
+    INITIAL_TASK_DEQUE_SIZE,
 
     // thread structure information
     sizeof( kmp_base_info_t ),
     offset_and_size_of( kmp_base_info_t, th_info ),
     offset_and_size_of( kmp_base_info_t, th_team ),
     offset_and_size_of( kmp_base_info_t, th_root ),
     offset_and_size_of( kmp_base_info_t, th_serial_team ),
     offset_and_size_of( kmp_base_info_t, th_ident ),
#if OMP_40_ENABLED
     offset_and_size_of( kmp_base_depnode_t, npredecessors   ),
     offset_and_size_of( kmp_base_depnode_t, nrefs           ),
 #endif
     offset_and_size_of( kmp_task_t, routine                 ),
 
     // thread_data_t.
     sizeof( kmp_thread_data_t ),
     offset_and_size_of( kmp_base_thread_data_t, td_deque             ),
+    offset_and_size_of( kmp_base_thread_data_t, td_deque_size        ),
     offset_and_size_of( kmp_base_thread_data_t, td_deque_head        ),
     offset_and_size_of( kmp_base_thread_data_t, td_deque_tail        ),
     offset_and_size_of( kmp_base_thread_data_t, td_deque_ntasks      ),
     offset_and_size_of( kmp_base_thread_data_t, td_deque_last_stolen ),
 
     // The last field.
     KMP_OMP_VERSION,
 

openmp/trunk/runtime/src/kmp_omp.h

#if OMP_40_ENABLED
     offset_and_size_t  dn_npredecessors;
     offset_and_size_t  dn_nrefs;
 #endif
     offset_and_size_t  dn_routine;
 
     /* kmp_thread_data_t */
     kmp_int32          hd_sizeof_struct;
     offset_and_size_t  hd_deque;
+    offset_and_size_t  hd_deque_size;
     offset_and_size_t  hd_deque_head;
     offset_and_size_t  hd_deque_tail;
     offset_and_size_t  hd_deque_ntasks;
     offset_and_size_t  hd_deque_last_stolen;
 
     // The last field of stable version.
     kmp_uint64         last_field;
 
 } kmp_omp_struct_info_t;
 
 #endif /* USE_DEBUGGER */
 
 /* end of file */

openmp/trunk/runtime/src/kmp_tasking.c

__kmp_push_task(kmp_int32 gtid, kmp_task_t * task )
     thread_data = & task_team -> tt.tt_threads_data[ tid ];
 
     // No lock needed since only owner can allocate
     if (thread_data -> td.td_deque == NULL ) {
         __kmp_alloc_task_deque( thread, thread_data );
     }
 
     // Check if deque is full
-    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE )
+    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE(thread_data->td) )
     {
         KA_TRACE(20, ( "__kmp_push_task: T#%d deque is full; returning TASK_NOT_PUSHED for task %p\n",
                        gtid, taskdata ) );
         return TASK_NOT_PUSHED;
     }
 
     // Lock the deque for the task push operation
     __kmp_acquire_bootstrap_lock( & thread_data -> td.td_deque_lock );
 
 #if OMP_41_ENABLED
     // Need to recheck as we can get a proxy task from a thread outside of OpenMP
-    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE )
+    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE(thread_data->td) )
     {
         __kmp_release_bootstrap_lock( & thread_data -> td.td_deque_lock );
         KA_TRACE(20, ( "__kmp_push_task: T#%d deque is full on 2nd check; returning TASK_NOT_PUSHED for task %p\n",
                        gtid, taskdata ) );
         return TASK_NOT_PUSHED;
     }
 #else
     // Must have room since no thread can add tasks but calling thread
-    KMP_DEBUG_ASSERT( TCR_4(thread_data -> td.td_deque_ntasks) < TASK_DEQUE_SIZE );
+    KMP_DEBUG_ASSERT( TCR_4(thread_data -> td.td_deque_ntasks) < TASK_DEQUE_SIZE(thread_data->td) );
 #endif
 
     thread_data -> td.td_deque[ thread_data -> td.td_deque_tail ] = taskdata;  // Push taskdata
     // Wrap index.
-    thread_data -> td.td_deque_tail = ( thread_data -> td.td_deque_tail + 1 ) & TASK_DEQUE_MASK;
+    thread_data -> td.td_deque_tail = ( thread_data -> td.td_deque_tail + 1 ) & TASK_DEQUE_MASK(thread_data->td);
     TCW_4(thread_data -> td.td_deque_ntasks, TCR_4(thread_data -> td.td_deque_ntasks) + 1);             // Adjust task count
 
     __kmp_release_bootstrap_lock( & thread_data -> td.td_deque_lock );
 
     KA_TRACE(20, ("__kmp_push_task: T#%d returning TASK_SUCCESSFULLY_PUSHED: "
                   "task=%p ntasks=%d head=%u tail=%u\n",
                   gtid, taskdata, thread_data->td.td_deque_ntasks,
                   thread_data->td.td_deque_tail, thread_data->td.td_deque_head) );
__kmp_remove_my_task( kmp_info_t * thread, kmp_int32 gtid, kmp_task_team_t *task_team,
     if (TCR_4(thread_data -> td.td_deque_ntasks) == 0) {
         __kmp_release_bootstrap_lock( & thread_data -> td.td_deque_lock );
         KA_TRACE(10, ("__kmp_remove_my_task(exit #2): T#%d No tasks to remove: ntasks=%d head=%u tail=%u\n",
                       gtid, thread_data->td.td_deque_ntasks, thread_data->td.td_deque_head,
                       thread_data->td.td_deque_tail) );
         return NULL;
     }
 
-    tail = ( thread_data -> td.td_deque_tail - 1 ) & TASK_DEQUE_MASK;  // Wrap index.
+    tail = ( thread_data -> td.td_deque_tail - 1 ) & TASK_DEQUE_MASK(thread_data->td);  // Wrap index.
     taskdata = thread_data -> td.td_deque[ tail ];
 
     if (is_constrained) {
         // we need to check if the candidate obeys task scheduling constraint:
         // only child of current task can be scheduled
         kmp_taskdata_t * current = thread->th.th_current_task;
         kmp_int32        level = current->td_level;
         kmp_taskdata_t * parent = taskdata->td_parent;
if ( (TCR_4(victim_td -> td.td_deque_ntasks) == 0) ||
         return NULL;
     }
 
     KMP_DEBUG_ASSERT( victim_td -> td.td_deque != NULL );
 
     if ( !is_constrained ) {
         taskdata = victim_td -> td.td_deque[ victim_td -> td.td_deque_head ];
         // Bump head pointer and Wrap.
-        victim_td -> td.td_deque_head = ( victim_td -> td.td_deque_head + 1 ) & TASK_DEQUE_MASK;
+        victim_td -> td.td_deque_head = ( victim_td -> td.td_deque_head + 1 ) & TASK_DEQUE_MASK(victim_td->td);
     } else {
         // While we have postponed tasks let's steal from tail of the deque (smaller tasks)
-        kmp_int32 tail = ( victim_td -> td.td_deque_tail - 1 ) & TASK_DEQUE_MASK;  // Wrap index.
+        kmp_int32 tail = ( victim_td -> td.td_deque_tail - 1 ) & TASK_DEQUE_MASK(victim_td->td);  // Wrap index.
         taskdata = victim_td -> td.td_deque[ tail ];
         // we need to check if the candidate obeys task scheduling constraint:
         // only child of current task can be scheduled
         kmp_taskdata_t * current = __kmp_threads[ gtid ]->th.th_current_task;
         kmp_int32        level = current->td_level;
         kmp_taskdata_t * parent = taskdata->td_parent;
         while ( parent != current && parent->td_level > level ) {
             parent = parent->td_parent;  // check generation up to the level of the current task
__kmp_alloc_task_deque( kmp_info_t *thread, kmp_thread_data_t *thread_data )
     // Initialize last stolen task field to "none"
     thread_data -> td.td_deque_last_stolen = -1;
 
     KMP_DEBUG_ASSERT( TCR_4(thread_data -> td.td_deque_ntasks) == 0 );
     KMP_DEBUG_ASSERT( thread_data -> td.td_deque_head == 0 );
     KMP_DEBUG_ASSERT( thread_data -> td.td_deque_tail == 0 );
 
     KE_TRACE( 10, ( "__kmp_alloc_task_deque: T#%d allocating deque[%d] for thread_data %p\n",
-                   __kmp_gtid_from_thread( thread ), TASK_DEQUE_SIZE, thread_data ) );
+                   __kmp_gtid_from_thread( thread ), INITIAL_TASK_DEQUE_SIZE, thread_data ) );
     // Allocate space for task deque, and zero the deque
     // Cannot use __kmp_thread_calloc() because threads not around for
     // kmp_reap_task_team( ).
     thread_data -> td.td_deque = (kmp_taskdata_t **)
-            __kmp_allocate( TASK_DEQUE_SIZE * sizeof(kmp_taskdata_t *));
+            __kmp_allocate( INITIAL_TASK_DEQUE_SIZE * sizeof(kmp_taskdata_t *));
+  thread_data -> td.td_deque_size = INITIAL_TASK_DEQUE_SIZE;
 }
 
+//------------------------------------------------------------------------------
+// __kmp_realloc_task_deque:
+// Re-allocates a task deque for a particular thread, copies the content from the old deque
+// and adjusts the necessary data structures relating to the deque.
+// This operation must be done with a the deque_lock being held
+
+static void __kmp_realloc_task_deque ( kmp_info_t *thread, kmp_thread_data_t *thread_data )
+{
+    kmp_int32 size = TASK_DEQUE_SIZE(thread_data->td);
+    kmp_int32 new_size = 2 * size;
+
+    KE_TRACE( 10, ( "__kmp_realloc_task_deque: T#%d reallocating deque[from %d to %d] for thread_data %p\n",
+                  __kmp_gtid_from_thread( thread ), size, new_size, thread_data ) );
+
+    kmp_taskdata_t ** new_deque = (kmp_taskdata_t **) __kmp_allocate( new_size * sizeof(kmp_taskdata_t *));
+
+    int i,j;
+    for ( i = thread_data->td.td_deque_head, j = 0; j < size; i = (i+1) & TASK_DEQUE_MASK(thread_data->td), j++ )
+       new_deque[j] = thread_data->td.td_deque[i];
+
+    __kmp_free(thread_data->td.td_deque);
+
+    thread_data -> td.td_deque_head = 0;
+    thread_data -> td.td_deque_tail = size;
+    thread_data -> td.td_deque = new_deque;
+    thread_data -> td.td_deque_size = new_size;
+}
 
 //------------------------------------------------------------------------------
 // __kmp_free_task_deque:
 // Deallocates a task deque for a particular thread.
 // Happens at library deallocation so don't need to reset all thread data fields.
 
 static void
 __kmp_free_task_deque( kmp_thread_data_t *thread_data )

 #if OMP_41_ENABLED
 
 /* __kmp_give_task puts a task into a given thread queue if:
     - the queue for that thread was created
     - there's space in that queue
 
     Because of this, __kmp_push_task needs to check if there's space after getting the lock
  */
-static bool __kmp_give_task ( kmp_info_t *thread, kmp_int32 tid, kmp_task_t * task )
+static bool __kmp_give_task ( kmp_info_t *thread, kmp_int32 tid, kmp_task_t * task, kmp_int32 pass )
 {
     kmp_taskdata_t *    taskdata = KMP_TASK_TO_TASKDATA(task);
     kmp_task_team_t * task_team = taskdata->td_task_team;
 
     KA_TRACE(20, ("__kmp_give_task: trying to give task %p to thread %d.\n", taskdata, tid ) );
 
     // If task_team is NULL something went really bad...
     KMP_DEBUG_ASSERT( task_team != NULL );
 
     bool result = false;
     kmp_thread_data_t * thread_data = & task_team -> tt.tt_threads_data[ tid ];
 
     if (thread_data -> td.td_deque == NULL ) {
         // There's no queue in this thread, go find another one
         // We're guaranteed that at least one thread has a queue
         KA_TRACE(30, ("__kmp_give_task: thread %d has no queue while giving task %p.\n", tid, taskdata ) );
         return result;
     }
 
-    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE )
+    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE(thread_data->td) )
     {
         KA_TRACE(30, ("__kmp_give_task: queue is full while giving task %p to thread %d.\n", taskdata, tid ) );
-        return result;
-    }
+
+        // if this deque is bigger than the pass ratio give a chance to another thread
+        if ( TASK_DEQUE_SIZE(thread_data->td)/INITIAL_TASK_DEQUE_SIZE >= pass ) return result;
 
         __kmp_acquire_bootstrap_lock( & thread_data-> td.td_deque_lock );
+        __kmp_realloc_task_deque(thread,thread_data);
 
-    if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE )
+    } else {
+
+       __kmp_acquire_bootstrap_lock( & thread_data-> td.td_deque_lock );
+
+       if ( TCR_4(thread_data -> td.td_deque_ntasks) >= TASK_DEQUE_SIZE(thread_data->td) )
        {
            KA_TRACE(30, ("__kmp_give_task: queue is full while giving task %p to thread %d.\n", taskdata, tid ) );
+
+           // if this deque is bigger than the pass ratio give a chance to another thread
+           if ( TASK_DEQUE_SIZE(thread_data->td)/INITIAL_TASK_DEQUE_SIZE >= pass )
               goto release_and_exit;
+
+           __kmp_realloc_task_deque(thread,thread_data);
+       }
     }
 
+    // lock is held here, and there is space in the deque
+
     thread_data -> td.td_deque[ thread_data -> td.td_deque_tail ] = taskdata;
     // Wrap index.
-    thread_data -> td.td_deque_tail = ( thread_data -> td.td_deque_tail + 1 ) & TASK_DEQUE_MASK;
+    thread_data -> td.td_deque_tail = ( thread_data -> td.td_deque_tail + 1 ) & TASK_DEQUE_MASK(thread_data->td);
     TCW_4(thread_data -> td.td_deque_ntasks, TCR_4(thread_data -> td.td_deque_ntasks) + 1);
 
     result = true;
     KA_TRACE(30, ("__kmp_give_task: successfully gave task %p to thread %d.\n", taskdata, tid ) );
 
 release_and_exit:
     __kmp_release_bootstrap_lock( & thread_data-> td.td_deque_lock );
 
void __kmpc_proxy_task_completed_ooo ( kmp_task_t *ptask )
     KMP_DEBUG_ASSERT( taskdata->td_flags.proxy == TASK_PROXY );
 
     __kmp_first_top_half_finish_proxy(taskdata);
 
     // Enqueue task to complete bottom half completion from a thread within the corresponding team
     kmp_team_t * team = taskdata->td_team;
     kmp_int32 nthreads = team->t.t_nproc;
     kmp_info_t *thread;
-    kmp_int32 k = 0;
+
+    //This should be similar to start_k = __kmp_get_random( thread ) % nthreads but we cannot use __kmp_get_random here
+    kmp_int32 start_k = 0;
+    kmp_int32 pass = 1;
+    kmp_int32 k = start_k;
 
     do {
-        //This should be similar to k = __kmp_get_random( thread ) % nthreads but we cannot use __kmp_get_random here
         //For now we're just linearly trying to find a thread
-        k = (k+1) % nthreads;
         thread = team->t.t_threads[k];
-    } while ( !__kmp_give_task( thread, k,  ptask ) );
+        k = (k+1) % nthreads;
+
+        // we did a full pass through all the threads
+        if ( k == start_k ) pass = pass << 1;
+
+    } while ( !__kmp_give_task( thread, k,  ptask, pass ) );
 
     __kmp_second_top_half_finish_proxy(taskdata);
 
     KA_TRACE(10, ("__kmp_proxy_task_completed_ooo(exit): proxy task completing ooo %p\n", taskdata ) );
 }
 
 //---------------------------------------------------------------------------------
 // __kmp_task_dup_alloc: Allocate the taskdata and make a copy of source task for taskloop