merge

Author: ritvikrao

CMakeLists.txt

@@ -37,8 +37,8 @@ option(RECONVERSE_TRY_ENABLE_COMM_LCI2 "whether to enable the LCIv2 backend" ON)
 option(RECONVERSE_AUTOFETCH_LCI2
         "whether to autofetch LCIv2 if LCI2 cannot be found" OFF)
 set(RECONVERSE_AUTOFETCH_LCI2_TAG
-    "163cb084f9c1892792c36e21e31e07998bd7a088"
-    CACHE STRING "The tag to fetch for LCIv2") # master branch as of 2025-10-13
+    "5c526672b7a9548013d8b45d8ebf06d5cac36e66"
+    CACHE STRING "The tag to fetch for LCIv2") # master branch as of 2025-11-7
 
 option(RECONVERSE_TRY_ENABLE_COMM_LCW "whether to enable the LCW backend" ON)
 option(RECONVERSE_AUTOFETCH_LCW
@@ -52,6 +52,7 @@ option(SPANTREE "whether to enable spanning tree collectives" OFF) #should turn
 # Reconverse uses SMP always, but turn on this option for backwards compatibility (see pingpong_multipairs for example)
 option(CMK_SMP "whether to enable SMP support" ON)
 option(CMK_CPV_IS_SMP "whether to enable SMP support for cpvs" ON)
+option(CMK_USE_SHMEM "whether to use POSIX shared memory for IPC" OFF)
 
 option(RECONVERSE_ATOMIC_QUEUE "whether to use atomic queue" ON)
 

include/converse.h

@@ -6,6 +6,7 @@
 #ifdef __cplusplus
 #include <atomic>
 #include <queue>
+#include <utility>
 #else
 #include <stdatomic.h>
 #endif
@@ -54,6 +55,8 @@ extern CmiNodeLock CmiMemLock_lock;
 
 #define CmiMemUnlock() do{if (CmiMemLock_lock) CmiUnlock(CmiMemLock_lock);} while (0)
 
+#define CmiInCommThread()  (0)
+
 #define CMK_TAG(x, y) x##y##_
 #define CMK_CONCAT(x, y) x##y
 
@@ -279,6 +282,8 @@ void CthFree(CthThread t);
 
 void CthResume(CthThread t);
 
+int CthIsSuspendable(CthThread t);
+
 void CthSuspend(void);
 
 void CthAwaken(CthThread th);
@@ -536,7 +541,11 @@ int CmiGetArgc(char **argv);
 int CmiScanf(const char *format, ...);
 int CmiError(const char *format, ...);
 
-#define ConverseExit(...) CmiExit(__VA_ARGS__ + 0)
+#ifdef __cplusplus
+void ConverseExit(int status=0);
+#else
+void ConverseExit(int status);
+#endif
 #define CmiMemcpy(dest, src, size) memcpy((dest), (src), (size))
 
 #define setMemoryTypeChare(p) /* empty memory debugging method */
@@ -546,15 +555,25 @@ void CmiInitCPUTopology(char **argv);
 void CmiInitCPUAffinity(char **argv);
 int CmiOnCore(void);
 
-void __CmiEnforceMsgHelper(const char *expr, const char *fileName, int lineNum,
-                           const char *msg, ...);
+#define __CMK_STRING(x) #x
+#define __CMK_XSTRING(x) __CMK_STRING(x)
 
-#define CmiEnforce(condition)                                                  \
-  do {                                                                         \
-    if (!(condition)) {                                                        \
-      __CmiEnforceMsgHelper(#condition, __FILE__, __LINE__, "");               \
-    }                                                                          \
-  } while (0)
+void __CmiEnforceHelper(const char* expr, const char* fileName, const char* lineNum);
+void __CmiEnforceMsgHelper(const char* expr, const char* fileName,
+			   const char* lineNum, const char* msg, ...);
+
+#define CmiEnforce(expr)                                             \
+  ((void)(CMI_LIKELY(expr) ? 0                                       \
+                 : (__CmiEnforceHelper(__CMK_STRING(expr), __FILE__, \
+                                       __CMK_XSTRING(__LINE__)),     \
+                    0)))
+
+#define CmiEnforceMsg(expr, ...)                                              \
+  ((void)(CMI_LIKELY(expr)                                                    \
+              ? 0                                                             \
+              : (__CmiEnforceMsgHelper(__CMK_STRING(expr), __FILE__,          \
+                                       __CMK_XSTRING(__LINE__), __VA_ARGS__), \
+                 0)))
 
 double getCurrentTime(void);
 double CmiWallTimer(void);
@@ -1196,4 +1215,109 @@ void CmiInterSyncNodeSendAndFreeFn(int destNode, int partition, int messageSize,
 
 /* end of variables and functions for partition */
 
+#define CMI_IPC_CUTOFF_ARG "ipccutoff"
+#define CMI_IPC_CUTOFF_DESC "max message size for cmi-shmem (in bytes)"
+#define CMI_IPC_POOL_SIZE_ARG "ipcpoolsize"
+#define CMI_IPC_POOL_SIZE_DESC "size of cmi-shmem pool (in bytes)"
+
+struct CmiIpcManager;
+
+#ifndef __cplusplus
+typedef struct CmiIpcManager CmiIpcManager;
+#endif
+
+#ifdef __cplusplus
+namespace cmi {
+namespace ipc {
+// recommended cutoff for block sizes
+CpvExtern(size_t, kRecommendedCutoff);
+// used to represent an empty linked list
+constexpr auto nil = uintptr_t(0);
+// used to represent the tail of a linked list
+constexpr auto max = UINTPTR_MAX;
+// used to indicate a message bound for a node
+constexpr auto nodeDatagram = UINT16_MAX;
+// default number of attempts to alloc before timing out
+constexpr auto defaultTimeout = 4;
+}  // namespace ipc
+}  // namespace cmi
+
+// alignas is used for padding here, rather than for alignment of the
+// CmiIpcBlock itself.
+struct alignas(ALIGN_BYTES) CmiIpcBlock {
+public:
+  // "home" rank of the block
+  int src;
+  int dst;
+  uintptr_t orig;
+  uintptr_t next;
+  size_t size;
+
+  CmiIpcBlock(size_t size_, uintptr_t orig_)
+      : orig(orig_), next(cmi::ipc::nil), size(size_) {}
+};
+
+enum CmiIpcAllocStatus {
+  CMI_IPC_OUT_OF_MEMORY,
+  CMI_IPC_REMOTE_DESTINATION,
+  CMI_IPC_SUCCESS,
+  CMI_IPC_TIMEOUT
+};
+
+// sets up ipc environment
+void CmiIpcInit(char** argv);
+
+// creates an ipc manager, waking the thread when it's done
+// ( this must be called in the same order on all pes! )
+CmiIpcManager* CmiMakeIpcManager(CthThread th);
+
+// push/pop blocks from the manager's send/recv queue
+bool CmiPushIpcBlock(CmiIpcManager*, CmiIpcBlock*);
+CmiIpcBlock* CmiPopIpcBlock(CmiIpcManager*);
+
+// tries to allocate a block, returning null if unsucessful
+// (fails when other PEs are contending resources)
+// second value of pair indicates failure cause
+std::pair<CmiIpcBlock*, CmiIpcAllocStatus> CmiAllocIpcBlock(CmiIpcManager*, int node, std::size_t size);
+
+// frees a block -- enabling it to be used again
+void CmiFreeIpcBlock(CmiIpcManager*, CmiIpcBlock*);
+
+// currently a no-op but may be eventually usable
+// intended to "capture" blocks from remote pes
+inline void CmiCacheIpcBlock(CmiIpcBlock*) { return; }
+
+// identifies whether a void* is the payload of a block
+// belonging to the given node
+CmiIpcBlock* CmiIsIpcBlock(CmiIpcManager*, void*, int node);
+
+// if (init) is true -- initializes the
+// memory segment for use as a message
+void* CmiIpcBlockToMsg(CmiIpcBlock*, bool init);
+
+// equivalent to calling above with (init = false)
+inline void* CmiIpcBlockToMsg(CmiIpcBlock* block) {
+  auto res = (char*)block + sizeof(CmiIpcBlock) + sizeof(CmiChunkHeader);
+  return (void*)res;
+}
+
+inline CmiIpcBlock* CmiMsgToIpcBlock(CmiIpcManager* manager, void* msg) {
+  return CmiIsIpcBlock(manager, (char*)msg - sizeof(CmiChunkHeader), CmiMyNode());
+}
+
+CmiIpcBlock* CmiMsgToIpcBlock(CmiIpcManager*, char* msg, std::size_t len, int node,
+                           int rank = cmi::ipc::nodeDatagram,
+                           int timeout = cmi::ipc::defaultTimeout);
+
+// deliver a block as a message
+void CmiDeliverIpcBlockMsg(CmiIpcBlock*);
+
+inline const std::size_t& CmiRecommendedIpcBlockCutoff(void) {
+  using namespace cmi::ipc;
+  return CpvAccess(kRecommendedCutoff);
+}
+#endif /* __cplusplus */
+
+CsvExtern(CmiIpcManager*, coreIpcManager_);
+
 #endif // CONVERSE_H

src/CMakeLists.txt

@@ -2,7 +2,7 @@ target_include_directories(reconverse PRIVATE .)
 target_sources(reconverse PRIVATE conv-conds.cpp convcore.cpp random.cpp 
                 scheduler.cpp scheduler_helpers.cpp cpuaffinity.cpp collectives.cpp
                 comm_backend/comm_backend_internal.cpp threads.cpp cldb.rand.cpp cldb.cpp cmirdmautils.cpp 
-                conv-rdma.cpp conv-topology.cpp msgmgr.cpp queueing.cpp)
+                conv-rdma.cpp conv-topology.cpp msgmgr.cpp queueing.cpp cmishmem.cpp)
 target_include_directories(
   reconverse PRIVATE $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>
                      $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>)

src/cldb.cpp

@@ -6,8 +6,8 @@
 
 typedef char *BitVector;
 
-/*
 CpvDeclare(int, CldHandlerIndex);
+/*
 CpvDeclare(int, CldNodeHandlerIndex);
 CpvDeclare(BitVector, CldPEBitVector);
 CpvDeclare(int, CldBalanceHandlerIndex);

src/cldb.h

@@ -5,7 +5,7 @@
 
 #define MAXMSGBFRSIZE 100000
 
-extern thread_local int CldHandlerIndex;
+CpvExtern(int, CldHandlerIndex);
 extern thread_local int CldNodeHandlerIndex;
 extern thread_local int CldBalanceHandlerIndex;
 extern thread_local int CldRelocatedMessages;

src/cldb.rand.cpp

@@ -40,7 +40,7 @@ void CldEnqueueGroup(CmiGroup grp, void *msg, int infofn) {
     pfn(&msg);
     ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
   }
-  CldSwitchHandler((char *)msg, CldHandlerIndex);
+  CldSwitchHandler((char *)msg, CpvAccess(CldHandlerIndex));
   CmiSetInfo(msg, infofn);
 
   CmiSyncMulticastAndFree(grp, len, msg);
@@ -59,7 +59,7 @@ void CldEnqueueWithinNode(void *msg, int infofn) {
     pfn(&msg);
     ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
   }
-  CldSwitchHandler((char *)msg, CldHandlerIndex);
+  CldSwitchHandler((char *)msg, CpvAccess(CldHandlerIndex));
   CmiSetInfo(msg, infofn);
 
   CmiWithinNodeBroadcast(len, (char *)msg);
@@ -75,7 +75,7 @@ void CldEnqueueMulti(int npes, const int *pes, void *msg, int infofn) {
     pfn(&msg);
     ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
   }
-  CldSwitchHandler((char *)msg, CldHandlerIndex);
+  CldSwitchHandler((char *)msg, CpvAccess(CldHandlerIndex));
   CmiSetInfo(msg, infofn);
 
   CmiSyncListSendAndFree(npes, pes, len, msg);
@@ -108,7 +108,7 @@ void CldEnqueue(int pe, void *msg, int infofn) {
       pfn(&msg);
       ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
     }
-    CldSwitchHandler((char *)msg, CldHandlerIndex);
+    CldSwitchHandler((char *)msg, CpvAccess(CldHandlerIndex));
     CmiSetInfo(msg, infofn);
     if (pe == CLD_BROADCAST) {
       CmiSyncBroadcastAndFree(len, msg);
@@ -153,7 +153,8 @@ void CldNodeEnqueue(int node, void *msg, int infofn) {
 }
 
 void CldModuleInit(char **argv) {
-  CldHandlerIndex = CmiRegisterHandler((CmiHandler)CldHandler);
+  CpvInitialize(int, CldHandlerIndex);
+  CpvAccess(CldHandlerIndex) = CmiRegisterHandler((CmiHandler)CldHandler);
   CldNodeHandlerIndex = CmiRegisterHandler((CmiHandler)CldNodeHandler);
   CldRelocatedMessages = 0;
   CldLoadBalanceMessages = 0;