Remove addition of LPF mutexes (split off into GitHub MT #55)

Author: anyzelman

include/lpf/core.h

@@ -2336,68 +2336,6 @@ 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)

src/MPI/core.cpp

@@ -268,42 +268,6 @@ lpf_err_t lpf_sync( lpf_t ctx, lpf_sync_attr_t attr )
 }
 
 
-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
-)
-{
-    (void) attr; // ignore parameter 'msg' since this implementation only 
-                 // implements core functionality
-    lpf::Interface * i = realContext(ctx);
-    if (!i->isAborted())
-        i->lockSlot( src_slot, src_offset, dst_pid, dst_slot, dst_offset, size );
-    return LPF_SUCCESS;
-}
-
-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
-)
-{
-    (void) attr; // ignore parameter 'msg' since this implementation only 
-                 // implements core functionality
-    lpf::Interface * i = realContext(ctx);
-    if (!i->isAborted())
-        i->unlockSlot( src_slot, src_offset, dst_pid, dst_slot, dst_offset, size );
-    return LPF_SUCCESS;
-}
-
 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)
 {
     (void) attr; // ignore attr parameter since this implementation only

src/MPI/ibverbsZero.cpp

@@ -679,73 +679,6 @@ void IBVerbs :: dereg( SlotID id )
 }
 
 
-void IBVerbs :: blockingCompareAndSwap(SlotID srcSlot, size_t srcOffset, int dstPid, SlotID dstSlot, size_t dstOffset, size_t size, uint64_t compare_add, uint64_t swap)
-{
-	const MemorySlot & src = m_memreg.lookup( srcSlot );
-	const MemorySlot & dst = m_memreg.lookup( dstSlot);
-
-    char * localAddr
-        = static_cast<char *>(src.glob[m_pid]._addr) + srcOffset;
-        const char * remoteAddr
-            = static_cast<const char *>(dst.glob[dstPid]._addr) + dstOffset;
-
-	struct ibv_sge sge;
-	memset(&sge, 0, sizeof(sge));
-        sge.addr = reinterpret_cast<uintptr_t>( localAddr );
-	sge.length =  std::min<size_t>(size, m_maxMsgSize );
-        sge.lkey = src.mr->lkey;
-
-	struct ibv_send_wr wr;
-	memset(&wr, 0, sizeof(wr));
-	wr.wr_id = srcSlot;
-	wr.sg_list = &sge;
-	wr.next = NULL; // this needs to be set, otherwise EINVAL return error in ibv_post_send
-	wr.num_sge = 1;
-	wr.opcode     = IBV_WR_ATOMIC_CMP_AND_SWP;
-	wr.send_flags = IBV_SEND_SIGNALED;
-	wr.wr.atomic.remote_addr = reinterpret_cast<uintptr_t>(remoteAddr);
-	wr.wr.atomic.compare_add = compare_add;
-	wr.wr.atomic.swap = swap;
-	wr.wr.atomic.rkey = dst.glob[dstPid]._rkey;
-	struct ibv_send_wr *bad_wr;
-	int error;
-    std::vector<ibv_wc_opcode> opcodes;
-
-blockingCompareAndSwap:
-	if (int err = ibv_post_send(m_connectedQps[dstPid].get(), &wr, &bad_wr ))
-	{
-		LOG(1, "Error while posting RDMA requests: " << std::strerror(err) );
-		throw Exception("Error while posting RDMA requests");
-	}
-
-    /**
-     * Keep waiting on a completion of events until you
-     * register a completed atomic compare-and-swap
-     */
-    do {
-        opcodes = wait_completion(error);
-         if (error) {
-            LOG(1, "Error in wait_completion");
-            std::abort();
-        }
-    } while (std::find(opcodes.begin(), opcodes.end(), IBV_WC_COMP_SWAP) == opcodes.end());
-
-	uint64_t * remoteValueFound = reinterpret_cast<uint64_t *>(localAddr);
-	/*
-     * if we fetched the value we expected, then
-     * we are holding the lock now (that is, we swapped successfully!)
-     * else, re-post your request for the lock
-     */
-	if (remoteValueFound[0] != compare_add)  {
-        LOG(4, "Process " << m_pid <<  " couldn't get the lock. remoteValue = " << remoteValueFound[0] << " compare_add = " << compare_add  << " go on, iterate\n");
-		goto blockingCompareAndSwap;
-    }
-    else {
-        LOG(4, "Process " << m_pid << " reads value " << remoteValueFound[0] << " and expected = " << compare_add  <<" gets the lock, done\n");
-    }
-	// else we hold the lock and swap value into the remote slot ...
-}
-
 void IBVerbs :: put( SlotID srcSlot, size_t srcOffset,
               int dstPid, SlotID dstSlot, size_t dstOffset, size_t size)
 {

src/MPI/interface.cpp

@@ -100,27 +100,6 @@ void Interface :: put( memslot_t srcSlot, size_t srcOffset,
             size );
 }
 
-// only for HiCR
-//#ifdef 
-
-void Interface :: lockSlot( memslot_t srcSlot, size_t srcOffset, 
-        pid_t dstPid, memslot_t dstSlot, size_t dstOffset,
-        size_t size ) 
-{
-    m_mesgQueue.lockSlot( srcSlot, srcOffset,
-            dstPid, dstSlot, dstOffset, 
-            size );
-}
-
-void Interface :: unlockSlot( memslot_t srcSlot, size_t srcOffset, 
-        pid_t dstPid, memslot_t dstSlot, size_t dstOffset,
-        size_t size ) 
-{
-    m_mesgQueue.unlockSlot( srcSlot, srcOffset,
-            dstPid, dstSlot, dstOffset, 
-            size );
-}
-
 void Interface :: getRcvdMsgCountPerSlot(size_t * msgs, SlotID slot) {
     m_mesgQueue.getRcvdMsgCountPerSlot(msgs, slot);
 }
@@ -172,9 +151,6 @@ err_t Interface :: syncPerSlot(memslot_t slot)
     }
 }
 
-// only for HiCR
-//#endif
-
 void Interface :: get( pid_t srcPid, memslot_t srcSlot, size_t srcOffset, 
         memslot_t dstSlot, size_t dstOffset,
         size_t size )

src/MPI/interface.hpp

@@ -70,8 +70,6 @@ class _LPFLIB_LOCAL Interface
 
     static err_t hook( const mpi::Comm & comm , spmd_t spmd, args_t args );
 
-    // only for HiCR
-    // #if
     err_t countingSyncPerSlot(memslot_t slot, size_t expected_sent, size_t expected_rcvd);
 
     err_t syncPerSlot(memslot_t slot);
@@ -90,16 +88,6 @@ class _LPFLIB_LOCAL Interface
 
     void flushReceived();
 
-    void lockSlot( memslot_t srcSlot, size_t srcOffset, 
-		    pid_t dstPid, memslot_t dstSlot, size_t dstOffset,
-		    size_t size );
-
-    void unlockSlot( memslot_t srcSlot, size_t srcOffset, 
-		    pid_t dstPid, memslot_t dstSlot, size_t dstOffset,
-		    size_t size );
-
-    // only for HiCR
-//#endif
     err_t rehook( spmd_t spmd, args_t args);
 
     void probe( machine_t & machine ) ;

src/MPI/mesgqueue.cpp

@@ -322,34 +322,6 @@ void MessageQueue :: get( pid_t srcPid, memslot_t srcSlot, size_t srcOffset,
 #endif
 }
 
-void MessageQueue :: lockSlot( memslot_t srcSlot, size_t srcOffset,
-        pid_t dstPid, memslot_t dstSlot, size_t dstOffset, size_t size )
-{
-    ASSERT(srcSlot != LPF_INVALID_MEMSLOT);
-    ASSERT(dstSlot != LPF_INVALID_MEMSLOT);
-    (void) srcOffset;
-    (void) dstOffset;
-    (void) dstPid;
-    (void) size;
-#ifdef LPF_CORE_MPI_USES_zero
-m_ibverbs.blockingCompareAndSwap(m_memreg.getVerbID(srcSlot), srcOffset, dstPid, m_memreg.getVerbID(dstSlot), dstOffset, size, 0ULL, 1ULL);
-#endif
-}
-
-void MessageQueue :: unlockSlot( memslot_t srcSlot, size_t srcOffset,
-        pid_t dstPid, memslot_t dstSlot, size_t dstOffset, size_t size )
-{
-    ASSERT(srcSlot != LPF_INVALID_MEMSLOT);
-    ASSERT(dstSlot != LPF_INVALID_MEMSLOT);
-    (void) srcOffset;
-    (void) dstOffset;
-    (void) dstPid;
-    (void) size;
-#ifdef LPF_CORE_MPI_USES_zero
-m_ibverbs.blockingCompareAndSwap(m_memreg.getVerbID(srcSlot), srcOffset, dstPid, m_memreg.getVerbID(dstSlot), dstOffset, size, 1ULL, 0ULL);
-#endif
-}
-
 void MessageQueue :: put( memslot_t srcSlot, size_t srcOffset,
         pid_t dstPid, memslot_t dstSlot, size_t dstOffset, size_t size )
 {

src/MPI/mesgqueue.hpp

@@ -66,13 +66,6 @@ class _LPFLIB_LOCAL MessageQueue
     // returns how many processes have entered in an aborted state
     int sync( bool abort );
 
-//only for HiCR
-    void lockSlot( memslot_t srcSlot, size_t srcOffset,
-            pid_t dstPid, memslot_t dstSlot, size_t dstOffset, size_t size );
-
-    void unlockSlot( memslot_t srcSlot, size_t srcOffset,
-		    pid_t dstPid, memslot_t dstSlot, size_t dstOffset, size_t size );
-
     void getRcvdMsgCountPerSlot(size_t * msgs, memslot_t slot);
 
     void getRcvdMsgCount(size_t * msgs);
@@ -88,7 +81,6 @@ class _LPFLIB_LOCAL MessageQueue
     int countingSyncPerSlot(memslot_t slot, size_t expected_sent, size_t expected_rcvd);
 
     int syncPerSlot(memslot_t slot);
-// end only for HiCR
 
 private:
     enum Msgs { BufPut ,

src/imp/core.c

@@ -146,32 +146,6 @@ lpf_err_t lpf_counting_sync_per_slot( lpf_t lpf, lpf_sync_attr_t attr, lpf_memsl
     return LPF_SUCCESS;
 }
 
-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
-) {
-	return LPF_SUCCESS;
-}
-
-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
-) {
-	return LPF_SUCCESS;
-}
-
 static double messageGap( lpf_pid_t p, size_t min_msg_size, lpf_sync_attr_t attr)
 { 
     (void) p;