Provide local asynchronous access consistency

CMakeExt/Threading.cmake

@@ -14,9 +14,9 @@ if (ENABLE_THREADSUPPORT)
     if (DART_SYNC_BUILTINS)
       MESSAGE(STATUS "Found builtin __sync_add_and_fetch")
       set(CMAKE_C_FLAGS
-        "${CMAKE_C_FLAGS_RELEASE} -DDART_HAVE_SYNC_BUILTINS")
+        "${CMAKE_C_FLAGS} -DDART_HAVE_SYNC_BUILTINS")
       set(CMAKE_CXX_FLAGS
-        "${CMAKE_CXX_FLAGS_RELEASE} -DDART_HAVE_SYNC_BUILTINS")
+        "${CMAKE_CXX_FLAGS} -DDART_HAVE_SYNC_BUILTINS")
     else()
       # error out for now
       MESSAGE(STATUS "Compiling builtin __sync_add_and_fetch failed with error " 

CMakeLists.txt

@@ -99,6 +99,8 @@ option(ENABLE_HDF5
        "Specify whether HDF5 features are enabled" on)
 option(ENABLE_NASTYMPI
        "Specify whether the NastyMPI proxy should be enabled" off)
+option(ENABLE_DART_PROGRESS
+       "Specify whether DART explicitely triggers progress" off)
 
 if (BUILD_COVERAGE_TESTS)
   set(BUILD_TESTS TRUE CACHE BOOLEAN
@@ -264,6 +266,8 @@ message(INFO "Unified RMA memory model: (ENABLE_UNIFIED_MEMORY_MODEL)    "
         ${ENABLE_UNIFIED_MEMORY_MODEL})
 message(INFO "MPI shared windows:       (ENABLE_SHARED_WINDOWS)          "
         ${ENABLE_SHARED_WINDOWS})
+message(INFO "Explicit DART progress:   (ENABLE_DART_PROGRESS)           "
+        ${ENABLE_DART_PROGRESS})
 message(INFO "Enable multithreading:    (ENABLE_THREADSUPPORT)           "
         ${ENABLE_THREADSUPPORT})
 message(INFO "Default index type long:  (ENABLE_DEFAULT_INDEX_TYPE_LONG) "

build.debug.sh

@@ -67,6 +67,7 @@ rm -Rf $BUILD_DIR/*
                         -DENABLE_ASSERTIONS=ON \
                         \
                         -DENABLE_SHARED_WINDOWS=ON \
+                        -DENABLE_DART_PROGRESS=ON \
                         -DENABLE_UNIFIED_MEMORY_MODEL=ON \
                         -DENABLE_DEFAULT_INDEX_TYPE_LONG=ON \
                         \

build.dev.sh

@@ -67,6 +67,7 @@ rm -Rf $BUILD_DIR/*
                         -DENABLE_ASSERTIONS=ON \
                         \
                         -DENABLE_SHARED_WINDOWS=ON \
+                        -DENABLE_DART_PROGRESS=ON \
                         -DENABLE_UNIFIED_MEMORY_MODEL=ON \
                         -DENABLE_DEFAULT_INDEX_TYPE_LONG=ON \
                         \

build.minimal.sh

@@ -59,6 +59,7 @@ rm -Rf $BUILD_DIR/*
                         -DENABLE_ASSERTIONS=ON \
                         \
                         -DENABLE_SHARED_WINDOWS=OFF \
+                        -DENABLE_DART_PROGRESS=OFF \
                         -DENABLE_UNIFIED_MEMORY_MODEL=ON \
                         -DENABLE_DEFAULT_INDEX_TYPE_LONG=OFF \
                         \

dart-if/include/dash/dart/if/dart_communication.h

@@ -26,6 +26,18 @@ extern "C" {
 #define DART_INTERFACE_ON
 /** \endcond */
 
+/*
+ * Set of flags to be passed to \ref dart_get, \ref dart_get_blocking, and
+ * \ref dart_get_handle to control their behavior.
+ */
+typedef enum {
+  /// Empty flag
+  DART_FLAG_NONE    = 0,
+  /// Enforce odering of communication operations
+  DART_FLAG_ORDERED = 1
+} dart_communication_flags_t;
+
+
 /**
  * \name Collective operations
  * Collective operations involving all units of a given team.
@@ -170,8 +182,10 @@ dart_ret_t dart_allgatherv(
  *
  * \param sendbuf The buffer containing the data to be sent by each unit.
  * \param recvbuf The buffer to hold the received data.
- * \param nelem   Number of elements sent by each process and received from each unit.
- * \param dtype   The data type of values in \c sendbuf and \c recvbuf to use in \c op.
+ * \param nelem   Number of elements sent by each process and received from
+ *                each unit.
+ * \param dtype   The data type of values in \c sendbuf and \c recvbuf to use
+ *                in \c op.
  * \param op      The reduction operation to perform.
  * \param team The team to participate in the allreduce.
  *
@@ -192,8 +206,10 @@ dart_ret_t dart_allreduce(
  * DART Equivalent to MPI_Reduce.
  *
  * \param sendbuf Buffer containing \c nelem elements to reduce using \c op.
- * \param recvbuf Buffer of size \c nelem to store the result of the element-wise operation \c op in.
- * \param nelem   The number of elements of type \c dtype in \c sendbuf and \c recvbuf.
+ * \param recvbuf Buffer of size \c nelem to store the result of the
+ *                element-wise operation \c op in.
+ * \param nelem   The number of elements of type \c dtype in \c sendbuf
+ *                and \c recvbuf.
  * \param dtype   The data type of values stored in \c sendbuf and \c recvbuf.
  * \param op      The reduce operation to perform.
  * \param root    The unit receiving the reduced values.
@@ -333,6 +349,8 @@ dart_ret_t dart_compare_and_swap(
  * \param gptr   A global pointer determining the source of the get operation.
  * \param nelem  The number of elements of type \c dtype to transfer.
  * \param dtype  The data type of the values in buffer \c dest.
+ * \param flags  Flags controling the behavior of \c dart_get,
+ *               see \ref dart_communication_flags_t.
  *
  * \return \c DART_OK on success, any other of \ref dart_ret_t otherwise.
  *
@@ -343,7 +361,8 @@ dart_ret_t dart_get(
   void            * dest,
   dart_gptr_t       gptr,
   size_t            nelem,
-  dart_datatype_t   dtype) DART_NOTHROW;
+  dart_datatype_t   dtype,
+  int               flags) DART_NOTHROW;
 
 /**
  * 'REGULAR' variant of dart_put.
@@ -352,6 +371,9 @@ dart_ret_t dart_get(
  * is guaranteed. A later flush operation is needed to guarantee
  * local and remote completion.
  *
+ * However, a \ref dart_get or \ref dart_get_blocking following this put will
+ * be guaranteed to read the updated value.
+ *
  * \param gptr   A global pointer determining the target of the put operation.
  * \param src    The local source buffer to load the data from.
  * \param nelem  The number of elements of type \c dtype to transfer.
@@ -370,14 +392,16 @@ dart_ret_t dart_put(
 
 
 /**
- * Guarantee completion of all outstanding operations involving a segment on a certain unit
+ * Guarantee completion of all outstanding operations involving a segment on
+ * a certain unit.
  *
  * Guarantees local and remote completion of all pending puts and
  * gets on a certain memory allocation / window / segment for the
  * target unit specified in gptr.
  * Similar to \c MPI_Win_flush().
  *
- * \param gptr Global pointer identifying the segment and unit to complete outstanding operations for.
+ * \param gptr Global pointer identifying the segment and unit to complete
+ *             outstanding operations for.
  * \return \c DART_OK on success, any other of \ref dart_ret_t otherwise.
  *
  * \threadsafe
@@ -463,7 +487,10 @@ typedef struct dart_handle_struct * dart_handle_t;
  * \param gptr   Global pointer being the source of the data transfer.
  * \param nelem  The number of elements of \c dtype in buffer \c dest.
  * \param dtype  The data type of the values in buffer \c dest.
- * \param[out] handle Pointer to DART handle to instantiate for later use with \c dart_wait, \c dart_wait_all etc.
+ * \param flags  Flags controling the behavior of \c dart_get_handle,
+ *               see \ref dart_communication_flags_t.
+ * \param[out] handle Pointer to DART handle to instantiate for later use
+ *             with \c dart_wait, \c dart_wait_all etc.
  *
  * \return \c DART_OK on success, any other of \ref dart_ret_t otherwise.
  *
@@ -475,19 +502,24 @@ dart_ret_t dart_get_handle(
   dart_gptr_t       gptr,
   size_t            nelem,
   dart_datatype_t   dtype,
-  dart_handle_t   * handle) DART_NOTHROW;
+  dart_handle_t   * handle,
+  int               flags) DART_NOTHROW;
 
 /**
  * 'HANDLE' variant of dart_put.
  * Neither local nor remote completion is guaranteed. A later
  * dart_wait*() call or a fence/flush operation is needed to guarantee
  * completion.
  *
+ * The update will not be visible in a following call to \ref dart_get variant
+ * unless the operation has been completed.
+ *
  * \param gptr   Global pointer being the target of the data transfer.
  * \param src    Local source memory to transfer data from.
  * \param nelem  The number of elements of type \c dtype to transfer.
  * \param dtype  The data type of the values in buffer \c dest.
- * \param[out] handle Pointer to DART handle to instantiate for later use with \c dart_wait, \c dart_wait_all etc.
+ * \param[out] handle Pointer to DART handle to instantiate for later use with
+ *                    \c dart_wait, \c dart_wait_all etc.
  *
  * \return \c DART_OK on success, any other of \ref dart_ret_t otherwise.
  *
@@ -606,6 +638,8 @@ dart_ret_t dart_testall_local(
  * \param gptr   Global pointer being the source of the data transfer.
  * \param nelem  The number of elements of type \c dtype to transfer.
  * \param dtype  The data type of the values in buffer \c dest.
+ * \param flags  Flags controling the behavior of \c dart_get_blocking,
+ *               see \ref dart_communication_flags_t.
  *
  * \return \c DART_OK on success, any other of \ref dart_ret_t otherwise.
  *
@@ -616,7 +650,8 @@ dart_ret_t dart_get_blocking(
   void         *  dest,
   dart_gptr_t     gptr,
   size_t          nelem,
-  dart_datatype_t dtype) DART_NOTHROW;
+  dart_datatype_t dtype,
+  int             flags) DART_NOTHROW;
 
 /**
  * 'BLOCKING' variant of dart_put.

dart-if/include/dash/dart/if/dart_globmem.h

@@ -322,6 +322,13 @@ dart_ret_t dart_team_memalloc_aligned(
   dart_datatype_t   dtype,
   dart_gptr_t     * gptr) DART_NOTHROW;
 
+
+dart_ret_t dart_team_memalloc_aligned_full(
+  dart_team_t       teamid,
+  size_t            nelem,
+  dart_datatype_t   dtype,
+  dart_gptr_t     * gptr) DART_NOTHROW;
+
 /**
  * Collective function to free global memory previously allocated
  * using \ref dart_team_memalloc_aligned.

dart-impl/mpi/CMakeLists.txt

@@ -39,6 +39,8 @@ set(ENABLE_PAPI ${ENABLE_PAPI}
     PARENT_SCOPE)
 set(ENABLE_HDF5 ${ENABLE_HDF5}
     PARENT_SCOPE)
+set(ENABLE_DART_PROGRESS ${ENABLE_DART_PROGRESS}
+    PARENT_SCOPE)
 
 message(STATUS "MPI include path:         " ${MPI_INCLUDE_PATH})
 message(STATUS "MPI libraries:")
@@ -143,6 +145,11 @@ if (NOT ENABLE_SHARED_WINDOWS)
        ${ADDITIONAL_COMPILE_FLAGS} -DDART_MPI_DISABLE_SHARED_WINDOWS)
 endif()
 
+if (ENABLE_DART_PROGRESS)
+  set(ADDITIONAL_COMPILE_FLAGS 
+      ${ADDITIONAL_COMPILE_FLAGS} -DDART_ENABLE_PROGRESS)
+endif()
+
 if(MPI_COMPILE_FLAGS)
   set (ADDITIONAL_COMPILE_FLAGS
        ${ADDITIONAL_COMPILE_FLAGS} ${MPI_COMPILE_FLAGS})

dart-impl/mpi/include/dash/dart/mpi/dart_globmem_priv.h

@@ -6,8 +6,5 @@
 
 /* Global object for one-sided communication on memory region allocated with 'local allocation'. */
 extern MPI_Win dart_win_local_alloc DART_INTERNAL;
-#if !defined(DART_MPI_DISABLE_SHARED_WINDOWS)
-extern MPI_Win dart_sharedmem_win_local_alloc DART_INTERNAL;
-#endif
 
 #endif /* DART__MPI__DART_GLOBMEM_PRIV_H__ */

dart-impl/mpi/include/dash/dart/mpi/dart_segment.h

@@ -8,6 +8,8 @@
 #ifndef DART_SEGMENT_H_
 #define DART_SEGMENT_H_
 #include <mpi.h>
+#include <stdbool.h>
+#include <stdlib.h>
 
 #include <dash/dart/if/dart_types.h>
 #include <dash/dart/base/macro.h>
@@ -18,13 +20,16 @@ typedef int16_t dart_segid_t;
 
 typedef struct
 {
-  dart_segid_t segid; /* seg_id determines a global pointer uniquely */
   size_t       size;
-  MPI_Aint   * disp;   /* address set of memory location of all units in certain team. */
-  char      ** baseptr;
-  char       * selfbaseptr;
-  MPI_Win      win;
-  uint16_t     flags;
+  MPI_Aint   * disp;        /* offsets at all units in the team */
+  char      ** baseptr;     /* baseptr of all units in the sharedmem group */
+  char       * selfbaseptr; /* baseptr of the current unit */
+  MPI_Win      shmwin;      /* sharedmem window */
+  MPI_Win      win;         /* window used to access this segment */
+  uint16_t     flags;       /* 16 bit flags */
+  dart_segid_t segid;       /* ID of the segment, globally unique in a team */
+  bool         dirty;       /* whether the segment has pending writes */
+  bool         isshm;       /* whether this is a shared memory segment */
 } dart_segment_info_t;
 
 // forward declaration to make the compiler happy
@@ -50,10 +55,60 @@ typedef struct {
 } dart_segmentdata_t;
 
 typedef enum {
+  DART_SEGMENT_LOCAL_ALLOC,
   DART_SEGMENT_ALLOC,
   DART_SEGMENT_REGISTER
 } dart_segment_type;
 
+struct dart_seghash_elem {
+  dart_seghash_elem_t *next;
+  dart_segment_info_t  data;
+};
+
+static inline int hash_segid(dart_segid_t segid)
+{
+  /* Simply use the lower bits of the segment ID.
+   * Since segment IDs are allocated continuously, this is likely to cause
+   * collisions starting at (segment number == DART_SEGMENT_HASH_SIZE)
+   * TODO: come up with a random distribution to account for random free'd
+   * segments?
+   * */
+  return (abs(segid) % DART_SEGMENT_HASH_SIZE);
+}
+
+static inline
+MPI_Aint
+dart_segment_disp(dart_segment_info_t *seginfo, dart_team_unit_t team_unit_id)
+{
+  return (seginfo->disp != NULL) ? seginfo->disp[team_unit_id.id] : 0;
+}
+
+/**
+ * Returns the segment info for the segment with ID \c segid.
+ */
+static inline dart_segment_info_t * dart_segment_get_info(
+    dart_segmentdata_t *segdata,
+    dart_segid_t        segid)
+{
+  int slot = hash_segid(segid);
+  dart_seghash_elem_t *elem = segdata->hashtab[slot];
+
+  while (elem != NULL) {
+    if (elem->data.segid == segid) {
+      break;
+    }
+    elem = elem->next;
+  }
+
+//  if (elem == NULL) {
+//    DART_LOG_ERROR("dart_segment__get_segment : "
+//                   "Invalid segment ID %i on team %i",
+//                   segid, segdata->team_id);
+//    return NULL;
+//  }
+
+  return &(elem->data);
+}
 
 /**
  * Initialize the segment data hash table.
@@ -80,6 +135,13 @@ dart_segment_register(
   dart_segmentdata_t  *segdata,
   dart_segment_info_t *seg) DART_INTERNAL;
 
+/**
+ * Returns the segment info for the segment with ID \c segid.
+ */
+//dart_segment_info_t * dart_segment_get_info(
+//  dart_segmentdata_t *segdata,
+//  dart_segid_t        segid) DART_INTERNAL;
+
 
 #if !defined(DART_MPI_DISABLE_SHARED_WINDOWS)
 
@@ -92,7 +154,7 @@ dart_segment_register(
  * \retval non-negative integer Search successfully.
  * \retval negative integer Failure.
  */
-dart_ret_t dart_segment_get_win(
+dart_ret_t dart_segment_get_shmwin(
   dart_segmentdata_t * segdata,
   int16_t              seg_id,
   MPI_Win            * win) DART_INTERNAL;
@@ -104,6 +166,16 @@ dart_ret_t dart_segment_get_baseptr(
   char               **  baseptr_s) DART_INTERNAL;
 #endif
 
+dart_ret_t dart_segment_get_dirty(
+  dart_segmentdata_t  * segdata,
+  int16_t               segid,
+  bool                * dirty) DART_INTERNAL;
+
+dart_ret_t dart_segment_set_dirty(
+  dart_segmentdata_t  * segdata,
+  int16_t               segid,
+  bool                  dirty) DART_INTERNAL;
+
 dart_ret_t dart_segment_get_selfbaseptr(
   dart_segmentdata_t * segdata,
   int16_t              seg_id,

dart-impl/mpi/include/dash/dart/mpi/dart_team_private.h

@@ -194,11 +194,6 @@ typedef struct dart_team_data {
 
 } dart_team_data_t;
 
-
-#if !defined(DART_MPI_DISABLE_SHARED_WINDOWS)
-
-extern char* *dart_sharedmem_local_baseptr_set DART_INTERNAL;
-#endif
 /* @brief Initiate the free-team-list and allocated-team-list.
  *
  * This call will be invoked within dart_init(), and the free teamlist consist of