Lpf mutex extension

CMakeLists.txt

@@ -175,7 +175,7 @@ if ( LIB_MATH AND LIB_DL AND MPI_FOUND )
     endif()
 
     if (ENABLE_IBVERBS)
-        list(APPEND ENGINES "ibverbs")
+        list(APPEND ENGINES "ibverbs" "zero")
     endif()
 
 endif()

NOTICE

@@ -33,6 +33,8 @@ Implementation
         1) BSMP
         2) Collectives
         3) Pthread implementation
+    - 2022 - 2024, Kiril Dichev
+        1) Develop zero engine for LPF
 
     - 2018, Pierre Leca
         1) Usability improvements of compiler frontends and CMake integration
@@ -50,6 +52,8 @@ Quality Assurance
 
     - 2015 - 2017, Albert-Jan Yzelman
         1) Performance test suite
+    - 2022 - 2024, Kiril Dichev
+        1) Rewrite all functional tests to use CTest/Gtest
 
 
 Miscellaneous / Acknowledgments

bootstrap.sh

@@ -278,13 +278,13 @@ echo "--------------------------------------------------"
 echo
 ${CMAKE_EXE} -Wno-dev \
       -DCMAKE_INSTALL_PREFIX="$installdir" \
-      -DCMAKE_BUILD_TYPE=$config           \
-      -DLPFLIB_MAKE_DOC=$doc         \
-      -DLPFLIB_MAKE_TEST_DOC=$doc    \
+      -DCMAKE_BUILD_TYPE=$config \
+      -DLPFLIB_MAKE_DOC=$doc \
+      -DLPFLIB_MAKE_TEST_DOC=$doc \
       -DLPF_ENABLE_TESTS=$functests \
       -DGTEST_AGREE_TO_LICENSE=$googletest_license_agreement \
-      -DLPFLIB_PERFTESTS=$perftests  \
-      -DLPFLIB_CONFIG_NAME=${config_name:-${config}}\
+      -DLPFLIB_PERFTESTS=$perftests \
+      -DLPFLIB_CONFIG_NAME=${config_name:-${config}} \
       -DLPF_HWLOC="${hwloc}" \
       $hwloc_found_flag \
       $mpi_cmake_flags \

include/debug/lpf/core.h

@@ -64,6 +64,12 @@ extern "C" {
 #define lpf_sync( ctx, attrs ) \
     lpf_debug_sync( __FILE__, __LINE__, (ctx), (attrs) )
 
+#define lpf_counting_sync_per_tag( ctx, attrs, slot, expected_sends, expected_rcvs ) \
+    lpf_debug_counting_sync_per_tag( __FILE__, __LINE__, (ctx), (attrs), (slot), (expected_sends), (expected_rcvs) )
+
+#define lpf_sync_per_tag( ctx, attrs, slot) \
+    lpf_debug_sync_per_tag( __FILE__, __LINE__, (ctx), (attrs), (slot))
+
 #define lpf_resize_memory_register( ctx, size ) \
     lpf_debug_resize_memory_register( __FILE__, __LINE__, (ctx), (size) )
 
@@ -128,6 +134,9 @@ extern _LPFLIB_API
 lpf_err_t lpf_debug_sync( const char * file, int line, 
         lpf_t ctx, lpf_sync_attr_t attr );
 
+lpf_err_t lpf_debug_counting_sync_per_tag( const char * file, int line, 
+        lpf_t ctx, lpf_sync_attr_t attr, lpf_memslot_t slot, size_t expected_sends, size_t expected_rcvs);
+
 extern _LPFLIB_API 
 lpf_err_t lpf_debug_resize_memory_register( const char * file, int line,
         lpf_t ctx, size_t max_regs );

include/lpf/collectives.h

@@ -116,6 +116,16 @@ typedef void (*lpf_combiner_t) (size_t n, const void * combine, void * into );
  */
 extern _LPFLIB_API const lpf_coll_t LPF_INVALID_COLL;
 
+/**
+ * ToDo: document allgatherv
+ */
+lpf_err_t lpf_allgatherv(
+        lpf_coll_t coll,
+        lpf_memslot_t src,
+        lpf_memslot_t dst,
+        size_t *sizes, 
+        bool exclude_myself
+        );
 /**
  * Initialises a collectives struct, which allows the scheduling of collective
  * calls. The initialised struct is only valid after a next call to lpf_sync().

include/lpf/core.h

@@ -705,7 +705,7 @@ extern "C" {
  * released, and NN the number of the specifications released before this one in
  * the same year.
  */
-#define _LPF_VERSION 202000L
+#define _LPF_VERSION 202400L
 
 /**
  * An implementation that has defined this macro may never define the
@@ -984,7 +984,7 @@ typedef struct lpf_machine {
      * byte. This value may depend on the actual number of processes \a p used,
      * the minimum message size \a min_msg_size the user aims to send and
      * receive, and the type of synchronisation requested via \a attr. The
-	 * value is bitwise equivalent across all processes.
+     * value is bitwise equivalent across all processes.
      *
      * \param[in] p            A value between 1 and #lpf_machine_t.p, where
      *                         both bounds are inclusive.
@@ -2060,6 +2060,25 @@ lpf_err_t lpf_get(
 extern _LPFLIB_API
 lpf_err_t lpf_sync( lpf_t ctx, lpf_sync_attr_t attr );
 
+/**
+ * This synchronisation waits on memory slot #slot to complete sending
+ * and receiving @expected_sent and @expected_rcvd messages. The counts are
+ * checked in the ibv_poll_cq calls and associated to certain LPF slots.
+ * This call is only implemented for IB verbs at the moment.
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_counting_sync_per_slot( lpf_t ctx, lpf_sync_attr_t attr, lpf_memslot_t slot, size_t expected_sent, size_t expected_rcvd);
+
+/**
+ * This synchronisation waits on memory slot #slot to complete sending
+ * or receiving all outstanding messages. For the current implementation 
+ * in IB verbs, this means all scheduled sends via ibv_post_send are 
+ * checked for completion via ibv_poll_cq. Currently, there is no logic
+ * scheduling receives, but only sends -- for either get or put.
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_sync_per_slot( lpf_t ctx, lpf_sync_attr_t attr, lpf_memslot_t slot);
+
 /**
  * This primitive allows a user to inspect the machine that this LPF program
  * has been assigned. All resources reported in the #lpf_machine_t struct are
@@ -2317,6 +2336,122 @@ lpf_err_t lpf_resize_memory_register( lpf_t ctx, size_t max_regs );
 extern _LPFLIB_API
 lpf_err_t lpf_resize_message_queue( lpf_t ctx, size_t max_msgs );
 
+/**
+ * This call blockingly locks a destination slot #dst_slot, relying
+ * on IBVerbs Compare-and-Swap atomics.
+ * For an example, check tests/functional/func_lpf_compare_and_swap.ibverbs.c
+ * It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ * \param[in] src_slot Local slot used as source for the 
+ * operation to lock the destination slot, registered via lpf_register_local()
+ * \param[in] src_offset Source offset to use (0 in most cases)
+ * \param[in] dst_pid The process ID of the destination process
+ * \param[in] dst_slot The memory slot of the remote destination memory area
+ * registered via lpf_register_global().
+ * \param[in] dst_offset Destinaton offset (0 in most cases)
+ * \param[in] size The number of bytes to copy from the source memory area to
+ *                 the destination memory area (#lpf_memslot_t in most cases)
+ * \param[in] attr A #lpf_sync_attr_t value (use #LPF_MSG_DEFAULT)
+ * \returns #LPF_SUCCESS
+ *            When this process successfully locks the slot
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_lock_slot(
+    lpf_t ctx,
+    lpf_memslot_t src_slot,
+    size_t src_offset,
+    lpf_pid_t dst_pid,
+    lpf_memslot_t dst_slot,
+    size_t dst_offset,
+    size_t size,
+    lpf_msg_attr_t attr
+);
+
+/**
+ * This call blockingly unlocks a destination slot #dst_slot, relying
+ * on IBVerbs Compare-and-Swap atomics.
+ * For an example, check tests/functional/func_lpf_compare_and_swap.ibverbs.c
+ * It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ * \param[in] src_slot Local slot used as source for the 
+ * operation to lock the destination slot, registered via lpf_register_local()
+ * \param[in] src_offset Source offset to use (0 in most cases)
+ * \param[in] dst_pid The process ID of the destination process
+ * \param[in] dst_slot The memory slot of the remote destination memory area
+ * registered via lpf_register_global().
+ * \param[in] dst_offset Destinaton offset (0 in most cases)
+ * \param[in] size The number of bytes to copy from the source memory area to
+ *                 the destination memory area (#lpf_memslot_t in most cases)
+ * \param[in] attr A #lpf_sync_attr_t value (use #LPF_MSG_DEFAULT)
+ * \returns #LPF_SUCCESS
+ *            When this process successfully locks the slot
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_unlock_slot(
+    lpf_t ctx,
+    lpf_memslot_t src_slot,
+    size_t src_offset,
+    lpf_pid_t dst_pid,
+    lpf_memslot_t dst_slot,
+    size_t dst_offset,
+    size_t size,
+    lpf_msg_attr_t attr
+);
+
+/**
+ * This function returns in @rcvd_msgs the received message count on 
+ * LPF slot #slot. It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ * \param[out] rcvd_msgs Received message count
+ * \param[in] slot LPF slot to check received messages for
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_get_rcvd_msg_count_per_slot( lpf_t ctx, size_t *rcvd_msgs, lpf_memslot_t slot);
+
+/**
+ * This function returns in @rcvd_msgs the total received message count.
+ * It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ * \param[out] rcvd_msgs Received message count
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_get_rcvd_msg_count( lpf_t ctx, size_t *rcvd_msgs);
+
+/**
+ * This function returns in @sent_msgs the sent message count on LPF
+ * slot #slot. It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ * \param[out] sent_msgs Total messages sent on #slot
+ * \param[in] slot
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_get_sent_msg_count_per_slot( lpf_t ctx, size_t *sent_msgs, lpf_memslot_t slot);
+
+/**
+ * This function blocks until all the scheduled messages via
+ * ibv_post_send are completed (via ibv_poll_cq). This includes
+ * both put and get calls on the local process.
+ * No concept of slots is used here.
+ * This allows to reuse the send buffers e.g. in higher-level channel
+ * libraries.
+ * It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_flush_sent( lpf_t ctx);
+
+/**
+ * This function blocks until all the incoming received messages
+ * waiting on the receive completion queue are handled (via ibv_poll_cq).
+ * No concept of slots is used here.
+ * This allows to reuse the send buffers e.g. in higher-level channel
+ * libraries.
+ * It is only implemented for the zero backend (on Infiniband)
+ * \param[in] ctx The LPF context
+ */
+extern _LPFLIB_API
+lpf_err_t lpf_flush_received( lpf_t ctx);
+
 #ifdef __cplusplus
 }
 #endif

include/lpf/static_dispatch.h

@@ -40,8 +40,15 @@
 #undef lpf_get
 #undef lpf_put
 #undef lpf_sync
+#undef lpf_counting_sync_per_slot
+#undef lpf_sync_per_slot
 #undef lpf_register_local
+#undef lpf_get_rcvd_msg_count
+#undef lpf_get_rcvd_msg_count_per_slot
+#undef lpf_get_sent_msg_count_per_slot
 #undef lpf_register_global
+#undef lpf_flush_sent
+#undef lpf_flush_received
 #undef lpf_deregister
 #undef lpf_probe
 #undef lpf_resize_memory_register
@@ -85,7 +92,14 @@
 #define lpf_get             LPF_FUNC(get)
 #define lpf_put             LPF_FUNC(put)
 #define lpf_sync            LPF_FUNC(sync)
+#define lpf_counting_sync_per_slot            LPF_FUNC(counting_sync_per_slot)
+#define lpf_sync_per_slot            LPF_FUNC(sync_per_slot)
 #define lpf_register_local  LPF_FUNC(register_local)
+#define lpf_get_rcvd_msg_count LPF_FUNC(get_rcvd_msg_count)
+#define lpf_get_rcvd_msg_count_per_slot LPF_FUNC(get_rcvd_msg_count_per_slot)
+#define lpf_get_sent_msg_count_per_slot LPF_FUNC(get_sent_msg_count_per_slot)
+#define lpf_flush_sent LPF_FUNC(flush_sent)
+#define lpf_flush_received LPF_FUNC(flush_received)
 #define lpf_register_global LPF_FUNC(register_global)
 #define lpf_deregister      LPF_FUNC(deregister)
 #define lpf_probe           LPF_FUNC(probe)

lpfrun.in

@@ -57,7 +57,7 @@ function printhelp()
     echo
     echo "   -engine <engine>"
     echo "               Allow you to choose the engine. Currently supported"
-    echo "               are: pthread, mpirma, mpimsg, ibverbs, hybrid"
+    echo "               are: pthread, mpirma, mpimsg, ibverbs, zero, hybrid"
     echo 
     echo "   -probe <seconds>"
     echo "               Set the number of seconds to probe the system for BSP"
@@ -846,7 +846,7 @@ case $engine in
         exit_status=$?
         ;;
 
-    mpirma|mpimsg|ibverbs)
+    mpirma|mpimsg|ibverbs|zero)
 
         mpi_impl=$(mpi_detect)
         proc_args=
@@ -1128,8 +1128,8 @@ case $engine in
         ;;
 
     *)
-        echo "Engine '$engine' is not supported. Please choose 'pthread',"
-        echo "'mpirma', or 'hybrid'"
+        echo "Engine '$engine' is not supported. Please choose "
+        echo "'pthread', 'mpirma', 'mpimsg', 'ibverbs, 'zero', 'hybrid'"
         exit_status=1
         ;;
 esac

post-install/post-install-test.cmake.in

@@ -353,6 +353,9 @@ endif()
 ######   CMake integration using generated CMake module file ############
 
 foreach(engine @ENGINES@)
+    if ("${engine}" STREQUAL "zero") 
+        continue()
+    endif()
     message("Testing generated CMake module files for engine ${engine}")
 
     set(test_dir @builddir@/cmake-module-test-${engine})

post-install/test-lpf-nprocs.c

@@ -53,6 +53,8 @@ void spmd( lpf_t lpf, lpf_pid_t pid, lpf_pid_t nprocs, lpf_args_t args )
     lpf_memslot_t mem_slot = LPF_INVALID_MEMSLOT;
     lpf_register_global( lpf, mem, nprocs, &mem_slot );
 
+    lpf_sync(lpf, LPF_SYNC_DEFAULT);
+
     if (pid != 0) 
         lpf_get( lpf, 0, params_slot, 0, params_slot, 0, sizeof(params), LPF_MSG_DEFAULT );
 

src/MPI/CMakeLists.txt

@@ -23,7 +23,7 @@ if (MPI_FOUND)
     endif()
 
     if (ENABLE_IBVERBS)
-        list(APPEND MPI_ENGINES ibverbs)
+        list(APPEND MPI_ENGINES ibverbs zero)
     endif()
 
     if (MPI_IBARRIER)
@@ -52,6 +52,10 @@ if (MPI_FOUND)
             set(ibverbs_sources ibverbs.cpp)
         endif()
 
+        if (LPF_IMPL_ID STREQUAL zero)
+            set(ibverbs_sources ibverbsZero.cpp)
+        endif()
+
         add_library(raw_${libname} OBJECT
                 memorytable.cpp
                 mesgqueue.cpp
@@ -127,9 +131,9 @@ if (MPI_FOUND)
                 ${LIB_POSIX_THREADS}
         )
 
-        if (engine STREQUAL ibverbs)
-           target_link_libraries(${target} ${LIB_IBVERBS})
-        endif()
+    if (engine STREQUAL ibverbs OR engine STREQUAL zero)
+        target_link_libraries(${target} ${LIB_IBVERBS})
+    endif()
     endfunction()
 
 
@@ -176,6 +180,10 @@ if (MPI_FOUND)
         add_gtest( ibverbs_test "ibverbs" ON ${CMAKE_CURRENT_SOURCE_DIR}/ibverbs.t.cpp 
             ${CMAKE_CURRENT_SOURCE_DIR}/ibverbs.cpp 
             ${CMAKE_CURRENT_SOURCE_DIR}/mpilib.cpp)
+
+        add_gtest( zero_test "zero" ON ${CMAKE_CURRENT_SOURCE_DIR}/ibverbs.t.cpp 
+            ${CMAKE_CURRENT_SOURCE_DIR}/ibverbsZero.cpp 
+            ${CMAKE_CURRENT_SOURCE_DIR}/mpilib.cpp)
     endif()
 
     foreach (engine ${MPI_ENGINES})