osc/pt2pt: various threading fixes

This commit fixes several bugs identified by a new multi-threaded RMA
benchmarking suite. The following bugs have been identified and fixed:

 - The code that signaled the actual start of an access epoch changed
   the eager_send_active flag on a synchronization object without
   holding the object's lock. This could cause another thread waiting
   on eager sends to block indefinitely because the entirety of
   ompi_osc_pt2pt_sync_expected could exectute between the check of
   eager_send_active and the conditon wait of
   ompi_osc_pt2pt_sync_wait.

 - The bookkeeping of fragments could get screwed up when performing
   long put/accumulate operations from different threads. This was
   caused by the fragment flush code at the end of both put and
   accumulate. This code was put in place to avoid sending a large
   number of unexpected messages to a peer. To fix the bookkeeping
   issue we now 1) wait for eager sends to be active before stating
   any large isend's, and 2) keep track of the number of large isends
   associated with a fragment. If the number of large isends reaches
   32 the active fragment is flushed.

 - Use atomics to update the large receive/send tag counters. This
   prevents duplicate tags from being used. The tag space has also
   been updated to use the entire 16-bits of the tag space.

These changes should also fix #1299.

Signed-off-by: Nathan Hjelm <hjelmn@lanl.gov>

Author: hjelmn

ompi/mca/osc/pt2pt/osc_pt2pt.h

@@ -8,7 +8,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2007-2015 Los Alamos National Security, LLC.  All rights
+ * Copyright (c) 2007-2016 Los Alamos National Security, LLC.  All rights
  *                         reserved.
  * Copyright (c) 2010      Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
@@ -149,20 +149,20 @@ struct ompi_osc_pt2pt_module_t {
     uint32_t *epoch_outgoing_frag_count;
 
     /** cyclic counter for a unique tage for long messages. */
-    unsigned int tag_counter;
-    unsigned int rtag_counter;
+    uint32_t tag_counter;
+    uint32_t rtag_counter;
 
     /* Number of outgoing fragments that have completed since the
        begining of time */
-    uint32_t outgoing_frag_count;
+    volatile uint32_t outgoing_frag_count;
     /* Next outgoing fragment count at which we want a signal on cond */
-    uint32_t outgoing_frag_signal_count;
+    volatile uint32_t outgoing_frag_signal_count;
 
     /* Number of incoming fragments that have completed since the
        begining of time */
-    uint32_t active_incoming_frag_count;
+    volatile uint32_t active_incoming_frag_count;
     /* Next incoming buffer count at which we want a signal on cond */
-    uint32_t active_incoming_frag_signal_count;
+    volatile uint32_t active_incoming_frag_signal_count;
 
     /** Number of targets locked/being locked */
     unsigned int passive_target_access_epoch;
@@ -409,14 +409,6 @@ int ompi_osc_pt2pt_component_irecv(ompi_osc_pt2pt_module_t *module,
                                   int tag,
                                   struct ompi_communicator_t *comm);
 
-int ompi_osc_pt2pt_component_isend(ompi_osc_pt2pt_module_t *module,
-                                  const void *buf,
-                                  size_t count,
-                                  struct ompi_datatype_t *datatype,
-                                  int dest,
-                                  int tag,
-                                  struct ompi_communicator_t *comm);
-
 /**
  * ompi_osc_pt2pt_progress_pending_acc:
  *
@@ -639,8 +631,8 @@ static inline void osc_pt2pt_add_pending (ompi_osc_pt2pt_pending_t *pending)
                             opal_list_append (&mca_osc_pt2pt_component.pending_operations, &pending->super));
 }
 
-#define OSC_PT2PT_FRAG_TAG   0x10000
-#define OSC_PT2PT_FRAG_MASK  0x0ffff
+#define OSC_PT2PT_FRAG_TAG   0x80000
+#define OSC_PT2PT_FRAG_MASK  0x7ffff
 
 /**
  * get_tag:
@@ -658,23 +650,17 @@ static inline int get_tag(ompi_osc_pt2pt_module_t *module)
     /* the LSB of the tag is used be the receiver to determine if the
        message is a passive or active target (ie, where to mark
        completion). */
-    int tmp = module->tag_counter + !!(module->passive_target_access_epoch);
-
-    module->tag_counter = (module->tag_counter + 4) & OSC_PT2PT_FRAG_MASK;
-
-    return tmp;
+    int32_t tmp = OPAL_THREAD_ADD32((volatile int32_t *) &module->tag_counter, 4);
+    return (tmp & OSC_PT2PT_FRAG_MASK) | !!(module->passive_target_access_epoch);
 }
 
 static inline int get_rtag(ompi_osc_pt2pt_module_t *module)
 {
     /* the LSB of the tag is used be the receiver to determine if the
        message is a passive or active target (ie, where to mark
        completion). */
-    int tmp = module->rtag_counter + !!(module->passive_target_access_epoch);
-
-    module->rtag_counter = (module->rtag_counter + 4) & OSC_PT2PT_FRAG_MASK;
-
-    return tmp;
+    int32_t tmp = OPAL_THREAD_ADD32((volatile int32_t *) &module->rtag_counter, 4);
+    return (tmp & OSC_PT2PT_FRAG_MASK) | !!(module->passive_target_access_epoch);
 }
 /**
  * ompi_osc_pt2pt_accumulate_lock:

ompi/mca/osc/pt2pt/osc_pt2pt_active_target.c

@@ -8,7 +8,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2007-2015 Los Alamos National Security, LLC.  All rights
+ * Copyright (c) 2007-2016 Los Alamos National Security, LLC.  All rights
  *                         reserved.
  * Copyright (c) 2010      IBM Corporation.  All rights reserved.
  * Copyright (c) 2012-2013 Sandia National Laboratories.  All rights reserved.
@@ -211,7 +211,7 @@ int ompi_osc_pt2pt_start (ompi_group_t *group, int assert, ompi_win_t *win)
     ompi_osc_pt2pt_module_t *module = GET_MODULE(win);
     ompi_osc_pt2pt_sync_t *sync = &module->all_sync;
 
-    OPAL_THREAD_LOCK(&sync->lock);
+    OPAL_THREAD_LOCK(&module->lock);
 
     /* check if we are already in an access epoch */
     if (ompi_osc_pt2pt_access_epoch_active (module)) {