[BBT#332] Abstraction of threading library, implementation with the Argobots

CMakeLists.txt

@@ -361,6 +361,17 @@ foreach( ITEM ${TMP_LIST})
   set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${ITEM}")
 endforeach()
 
+option(PaRSEC_USE_ARGOBOTS "PaRSEC will use the argobots threading library" OFF)
+if (PaRSEC_USE_ARGOBOTS)
+  find_package(argobots)
+  if( LIBARGOBOTS_FOUND )
+    list(APPEND CMAKE_REQUIRED_LIBRARIES "${LIBARGOBOTS_LIBRARIES}" )
+    list(APPEND EXTRA_LIBS ${LIBARGOBOTS_LIBRARIES})
+    include_directories( ${LIBARGOBOTS_INCLUDE_DIRS} )
+    add_definitions(-DARGOBOTS)
+  endif (LIBARGOBOTS_FOUND)
+endif (PaRSEC_USE_ARGOBOTS)
+
 find_package(Threads)
 if(Threads_FOUND)
   set(PARSEC_HAVE_PTHREAD true)

cmake_modules/Findargobots.cmake

@@ -0,0 +1,33 @@
+# - This module determines if the Argobots Threading, Tasks,
+#   and Synchronization routines library is available, and where
+#   to find it.
+# The following variables are set
+#  LIBARGOBOTS_FOUND - System has libabt
+#  LIBARGOBOTS_INCLUDE_DIRS - The libabt include directories
+#  LIBARGOBOTS_LIBRARIES - The libraries needed to use libabt
+#  LIBARGOBOTS_DEFINITIONS - Compiler switches required for using libabt
+
+set(LIBARGOBOTS_DIR "" CACHE PATH "Root directory containing argobots install")
+
+find_package(PkgConfig)
+pkg_check_modules(PC_LIBARGOBOTS QUIET argobots)
+set(LIBARGOBOTS_DEFINITIONS ${PC_LIBARGOBOTS_CFLAGS_OTHER})
+
+find_path(LIBARGOBOTS_INCLUDE_DIR abt.h
+  HINTS
+  ${LIBARGOBOTS_DIR}/include ${PC_LIBARGOBOTS_INCLUDEDIR} ${PC_LIBARGOBOTS_INCLUDE_DIRS} )
+
+find_library(LIBARGOBOTS_LIBRARY NAME abt
+  HINTS
+  ${LIBARGOBOTS_DIR}/lib ${PC_LIBARGOBOTS_LIBDIR} ${PC_LIBARGOBOTS_LIBRARY_DIRS} )
+
+set(LIBARGOBOTS_LIBRARIES ${LIBARGOBOTS_LIBRARY} )
+set(LIBARGOBOTS_INCLUDE_DIRS ${LIBARGOBOTS_INCLUDE_DIR} )
+
+include(FindPackageHandleStandardArgs)
+# handle the QUIETLY and REQUIRED arguments and set LIBARGOBOTS_FOUND to TRUE
+# if all listed variables are TRUE
+find_package_handle_standard_args(LIBARGOBOTS DEFAULT_MSG
+                                  LIBARGOBOTS_LIBRARY LIBARGOBOTS_INCLUDE_DIR)
+
+mark_as_advanced(LIBARGOBOTS_INCLUDE_DIR LIBARGOBOTS_LIBRARY )

contrib/platforms/config.saturn

@@ -22,8 +22,12 @@ elif [ "x$USER" = "xdgenet" ]; then
   CC=gcc
   CXX=g++
   FC=gfortran
-  MPI_DIR="/sw/openmpi/2.0.2/"
-  USER_OPTIONS+=" -DDPLASMA_PRECISIONS=d"
+  MPI_DIR="/sw/openmpi/2.1.1/"
+  USER_OPTIONS+=" -DDPLASMA_PRECISIONS=d -DPARSEC_PROF_TRACE=ON"
+  if [ "x$1" = "xARGOBOTS" ]; then
+      USER_OPTIONS+=" -DPaRSEC_USE_ARGOBOTS=ON"
+      MPI_DIR="/home/dgenet/Argobots/install/mpich/argobots/gcc"
+  fi
 elif [ "x$USER" = "xYOURSELF" ]; then
   PLASMA_DIR="SOMEWHERE"
   USER_OPTIONS+="-DPARSEC_OPTION=ON"

dplasma/testing/testing_zgemm.c

@@ -41,7 +41,11 @@ int main(int argc, char ** argv)
 #if defined(PARSEC_HAVE_CUDA) && 1
     iparam[IPARAM_NGPUS] = 0;
 #endif
-    /* Initialize PaRSEC */
+    int i=1;
+    fprintf(stdout, "pid: %d\n", getpid());
+    while(i) { sleep(1); }
+
+/* Initialize PaRSEC */
     parsec = setup_parsec(argc, argv, iparam);
     PASTE_CODE_IPARAM_LOCALS(iparam);
 

parsec/CMakeLists.txt

@@ -72,6 +72,24 @@ set(SOURCES
   ${MCA_EXTRA_SOURCES}
 )
 
+if (LIBARGOBOTS_FOUND)
+  set (SOURCES ${SOURCES} thread/argobots/comm_scheduler.c)
+
+  # Argobots demonstrations codes are not necessarily exclusive
+  option(BUILD_ARGO_DEMO1 "Build the demonstration using the backbone" OFF)
+  if (BUILD_ARGO_DEMO1)
+    set (SOURCES ${SOURCES} thread/argobots/publisher_ult.c)
+    add_definitions (-DARGO_DEMO1)
+  endif (BUILD_ARGO_DEMO1)
+
+  option(BUILD_ARGO_DEMO2 "Build the demonstration handling the communicating threads" OFF)
+  if (BUILD_ARGO_DEMO2)
+    # both features should be available in pthread implementation
+    set (SOURCES ${SOURCES} thread/argobots/listener_ult.c thread/argobots/monitoring_ult.c)
+    add_definitions (-DARGO_DEMO2)
+  endif (BUILD_ARGO_DEMO2)
+endif (LIBARGOBOTS_FOUND)
+
 if( NOT MPI_C_FOUND )
   list(APPEND SOURCES datatype/datatype.c)
 else( NOT MPI_C_FOUND )
@@ -92,6 +110,7 @@ endif( PARSEC_PROF_GRAPHER )
 #
 if( BUILD_PARSEC )
   add_library(parsec-base ${BASE_SOURCES})
+  target_link_libraries(parsec-base ${EXTRA_LIBS})
   set_target_properties(parsec-base PROPERTIES COMPILE_FLAGS "-DYYERROR_VERBOSE")
   set_target_properties(parsec-base PROPERTIES POSITION_INDEPENDENT_CODE ON)
   #  if (MPI_C_FOUND)

parsec/barrier.c

@@ -6,22 +6,24 @@
 
 #include "parsec/parsec_config.h"
 #include "parsec/class/barrier.h"
+#include "parsec/sys/atomic.h"
+#include "thread/thread.h"
 
 #if PARSEC_IMPLEMENT_BARRIERS
 
 int parsec_barrier_init(parsec_barrier_t* barrier, const void* attr, unsigned int count)
 {
     int rc;
 
-    if( 0 != (rc = pthread_mutex_init(&(barrier->mutex), attr)) ) {
+    if( 0 != (rc = PARSEC_THREAD_MUTEX_CREATE(&(barrier->mutex), attr)) ) {
         return rc;
     }
 
     barrier->count      = count;
     barrier->curcount   = 0;
     barrier->generation = 0;
-    if( 0 != (rc = pthread_cond_init(&(barrier->cond), NULL)) ) {
-        pthread_mutex_destroy( &(barrier->mutex) );
+    if( 0 != (rc = PARSEC_THREAD_COND_CREATE(&(barrier->cond), NULL)) ) {
+        PARSEC_THREAD_MUTEX_DESTROY( &(barrier->mutex) );
         return rc;
     }
     return 0;
@@ -30,31 +32,32 @@ int parsec_barrier_init(parsec_barrier_t* barrier, const void* attr, unsigned in
 int parsec_barrier_wait(parsec_barrier_t* barrier)
 {
     int generation;
+    parsec_mfence();
 
-    pthread_mutex_lock( &(barrier->mutex) );
+    PARSEC_THREAD_MUTEX_LOCK( &(barrier->mutex) );
     if( (barrier->curcount + 1) == barrier->count) {
         barrier->generation++;
         barrier->curcount = 0;
-        pthread_cond_broadcast( &(barrier->cond) );
-        pthread_mutex_unlock( &(barrier->mutex) );
+        PARSEC_THREAD_COND_BROADCAST( &(barrier->cond) );
+        PARSEC_THREAD_MUTEX_UNLOCK( &(barrier->mutex) );
         return 1;
     }
     barrier->curcount++;
     generation = barrier->generation;
     for(;;) {
-        pthread_cond_wait( &(barrier->cond), &(barrier->mutex) );
+        PARSEC_THREAD_COND_WAIT( &(barrier->cond), &(barrier->mutex) );
         if( generation != barrier->generation ) {
             break;
         }
     }
-    pthread_mutex_unlock( &(barrier->mutex) );
+    PARSEC_THREAD_MUTEX_UNLOCK( &(barrier->mutex) );
     return 0;
 }
 
 int parsec_barrier_destroy(parsec_barrier_t* barrier)
 {
-    pthread_mutex_destroy( &(barrier->mutex) );
-    pthread_cond_destroy( &(barrier->cond) );
+    PARSEC_THREAD_MUTEX_DESTROY( &(barrier->mutex) );
+    PARSEC_THREAD_COND_DESTROY( &(barrier->cond) );
     barrier->count    = 0;
     barrier->curcount = 0;
     return 0;

parsec/class/barrier.h

@@ -10,7 +10,7 @@
 #include "parsec/parsec_config.h"
 
 #include <unistd.h>
-#include <pthread.h>
+#include "parsec/thread/thread.h"
 
 /**
  * @defgroup parsec_internal_classes_barrier Barrier
@@ -52,13 +52,13 @@ typedef pthread_barrier_t parsec_barrier_t;
  *          providing a simple barrier mechanism.
  */
 typedef struct parsec_barrier_t {
-    int                 count;       /**< Number of threads expected to enter the barrier */
-    volatile int        curcount;    /**< Number of threads currently inside the barrier */
-    volatile int        generation;  /**< Unique number used to count how many times this
-                                      *   barrier was used, and enable debugging unmatching barriers */
-    pthread_mutex_t     mutex;       /**< Lock on the barrier, to make threads wait passively */
-    pthread_cond_t      cond;        /**< Condition on the barrier, to allow waking up threads that wait
-                                      *   passively once all threads have joined the barrier */
+    int                   count;       /**< Number of threads expected to enter the barrier */
+    volatile int          curcount;    /**< Number of threads currently inside the barrier */
+    volatile int          generation;  /**< Unique number used to count how many times this
+                                        *   barrier was used, and enable debugging unmatching barriers */
+    parsec_thread_mutex_t mutex;       /**< Lock on the barrier, to make threads wait passively */
+    parsec_thread_cond_t  cond;        /**< Condition on the barrier, to allow waking up threads that wait
+                                        *   passively once all threads have joined the barrier */
 } parsec_barrier_t;
 
 /**
@@ -104,6 +104,7 @@ int parsec_barrier_wait(parsec_barrier_t* barrier);
  * @return 0 if success another code otherwise.
  */
 int parsec_barrier_destroy(parsec_barrier_t* barrier);
+
 #define PARSEC_IMPLEMENT_BARRIERS 1
 
 #endif

parsec/include/parsec/execution_stream.h

@@ -12,8 +12,8 @@
 #include <hwloc.h>
 #endif
 
-#include <pthread.h>
 #include <stdint.h>
+#include "parsec/thread/thread.h"
 #include "parsec/hbbuffer.h"
 #include "parsec/mempool.h"
 #include "parsec/profiling.h"
@@ -38,7 +38,12 @@ struct parsec_execution_stream_s {
     int core_id;            /**< Core on which the thread is bound (hwloc in order numbering) */
     int socket_id;          /**< Socket on which the thread is bound (hwloc in order numerotation) */
 
-    pthread_t pthread_id;     /**< POSIX thread identifier. */
+	//TODO: Assess a best way and the real need to do it this way.
+#ifdef ARGOBOTS
+    int ABT_thread_id;      /**< ABT UL thread identifier. */
+#else
+    pthread_t pthread_id;
+#endif /*ARGOBOTS*/
 
 #if defined(PARSEC_PROF_TRACE)
     parsec_thread_profiling_t *es_profile;
@@ -121,12 +126,14 @@ struct parsec_context_s {
 
     size_t remote_dep_fw_mask_sizeof; /* Size of the remote dep fw mask */
 
-    pthread_t *pthreads; /**< all POSIX threads used for computation are stored here in order
-                          *   threads[0] is uninitialized, this is the user's thread
-                          *   threads[1] = thread for vp=0, th=1, if vp[0]->nbcores > 1
-                          *   threads[n] = thread(vp=1, th=0) if vp[0]->nb_cores = n
-                          *   etc...
-                          */
+    parsec_thread_t *monitoring_steering_threads;
+
+    parsec_thread_t *parsec_threads; /**< all the threads used for computation are stored here in order
+                                    *   threads[0] is user's calling thread
+                                    *   threads[1] = thread for vp=0, th=1 if vp[0]->nbcores > 1
+                                    *   threads[n] = thread(vp=1, th=0) if vp[0]->nbcores = n
+                                    *   etc...
+                                    */
 
     int32_t nb_vp; /**< number of virtual processes in this physical process */
 

parsec/mca/pins/alperf/ValidateModule.CMake

@@ -0,0 +1,12 @@
+if (PINS_ENABLE)
+  if (NOT PARSEC_PROF_TRACE)
+    MESSAGE(STATUS "Module ${MODULE} not selectable: PARSEC_PROF_TRACE disabled.")
+  else (NOT PARSEC_PROF_TRACE)
+#    SET(MCA_${COMPONENT}_${MODULE} ON)
+#    FILE(GLOB MCA_${COMPONENT}_${MODULE}_SOURCES ${MCA_BASE_DIR}/${COMPONENT}/${MODULE}/[^\\.]*.c)
+#    SET(MCA_${COMPONENT}_${MODULE}_CONSTRUCTOR "${COMPONENT}_${MODULE}_static_component")
+  endif (NOT PARSEC_PROF_TRACE)
+else (PINS_ENABLE)
+  MESSAGE(STATUS "Module ${MODULE} not enabled: PINS disabled.")
+  SET(MCA_${COMPONENT}_${MODULE} OFF)
+endif (PINS_ENABLE)

parsec/mca/pins/alperf/pins_alperf.h

@@ -0,0 +1,63 @@
+#ifndef PINS_ALPERF_H
+#define PINS_ALPERF_H
+
+#include "parsec_config.h"
+#include "parsec/mca/mca.h"
+#include "parsec/mca/pins/pins.h"
+#include "parsec.h"
+
+BEGIN_C_DECLS
+
+/**
+ * This module gathers user-selected / programmer-defined statistics into
+ * a single structure. That structure is of variable size (decided at runtime),
+ * so it is pointed to with a void * (it’s the field pins_alperf_counter_store.counters,
+ * and its size is what is returned by pins_alperf_counter_store_size()).
+ * The format of that buffer is the following:
+     double <date of last read, in seconds, must be updated by the monitoring thread>
+     uint16_t <Number of times the following sequence repeats itself>
+       char name[8] <name of the selected stat. e.g.: “task\0\0\0\0”, or “flops\0\0\0”>
+       uint64_t value0 <value for eu 0>
+       uint64_t value1 <value for eu 1>
+       …
+       uint64_t value<n-1> <value for eu n-1, where n is the number of execution units for the sending process>
+
+ * The structure is not padded at all: it is packed, and all alignment is done by converting
+ * to a uintptr_t and adding the right number of bytes with the macro PINS_ALPERF_COUNTER().
+ */
+
+#define PINS_ALPERF_COUNTER_NAME_MAX 8
+
+typedef struct {
+    char name[PINS_ALPERF_COUNTER_NAME_MAX];
+    uint64_t value_per_eu[1];
+} pins_alperf_counter_t;
+
+typedef struct {
+    size_t    ct_size;
+    uint16_t  nb_counters;
+    void     *counters;
+} pins_alperf_counter_store_t;
+
+PARSEC_DECLSPEC extern pins_alperf_counter_store_t pins_alperf_counter_store;
+
+static inline size_t pins_alperf_counter_store_size(void)
+{
+    return sizeof(double) + sizeof(uint16_t) + pins_alperf_counter_store.ct_size * pins_alperf_counter_store.nb_counters;
+}
+
+#define PINS_ALPERF_DATE       ((double*)pins_alperf_counter_store.counters)
+#define PINS_ALPERF_NBCOUNTERS ((uint16_t*)((uintptr_t)pins_alperf_counter_store.counters + sizeof(double)))
+#define PINS_ALPERF_COUNTER(i) ((pins_alperf_counter_t*)((uintptr_t)pins_alperf_counter_store.counters + (sizeof(double) + sizeof(uint16_t) + ((i) * pins_alperf_counter_store.ct_size))))
+
+/**
+ * Globally exported variable
+ */
+PARSEC_DECLSPEC extern const parsec_pins_base_component_t parsec_pins_alperf_component;
+PARSEC_DECLSPEC extern const parsec_pins_module_t parsec_pins_alperf_module;
+/* static accessor */
+mca_base_component_t * pins_alperf_static_component(void);
+
+END_C_DECLS
+
+#endif // PINS_ALPERF_H