P2p collective (#274)

* add dual directional test for nixl

* add send recv over ipc

* add simple memcpy_peer py code: one GPU only has one DMA, on mi300x, this can reach 45GB/s

* handling local and remote connections in collectives

* add working ipc transfer but hit data consistency problems

* fixing gpuSetDevice bugs for gpuIpcCloseMemHandle. Now p2p has not memory consistency bugs

* fixing value errors

* slightly improve perf

* adding multi-stream

* find the performance anomaly is the cudaIpcCloudMemHande

* expose remote_gpu_idx via uccl_accept()

* nits

Author: YangZhou1997

include/util/gpu_rt.h

@@ -42,8 +42,24 @@
 #define gpuMemcpyPeerAsync cudaMemcpyPeerAsync
 #define gpuMemcpyDeviceToDevice cudaMemcpyDeviceToDevice
 #define gpuMemcpyFromSymbol cudaMemcpyFromSymbol
+#define gpuMemsetAsync cudaMemsetAsync
 #define gpuGetLastError cudaGetLastError
 #define gpuErrorPeerAccessAlreadyEnabled cudaErrorPeerAccessAlreadyEnabled
+#define gpuEvent_t cudaEvent_t
+#define gpuEventCreate cudaEventCreate
+#define gpuEventDestroy cudaEventDestroy
+#define gpuEventRecord cudaEventRecord
+#define gpuEventQuery cudaEventQuery
+#define gpuEventSynchronize cudaEventSynchronize
+#define gpuStreamWaitEvent cudaStreamWaitEvent
+#define gpuEventCreateWithFlags cudaEventCreateWithFlags
+#define gpuEventDefault cudaEventDefault
+#define gpuEventDisableTiming cudaEventDisableTiming
+#define gpuEventInterprocess cudaEventInterprocess
+#define gpuIpcEventHandle_t cudaIpcEventHandle_t
+#define gpuIpcGetEventHandle cudaIpcGetEventHandle
+#define gpuIpcOpenEventHandle cudaIpcOpenEventHandle
+#define gpuIpcCloseEventHandle cudaIpcCloseEventHandle
 #else
 #include <hip/hip_runtime.h>
 #include <hip/hip_runtime_api.h>
@@ -87,8 +103,24 @@
 #define gpuMemcpyPeerAsync hipMemcpyPeerAsync
 #define gpuMemcpyDeviceToDevice hipMemcpyDeviceToDevice
 #define gpuMemcpyFromSymbol hipMemcpyFromSymbol
+#define gpuMemsetAsync hipMemsetAsync
 #define gpuGetLastError hipGetLastError
 #define gpuErrorPeerAccessAlreadyEnabled hipErrorPeerAccessAlreadyEnabled
+#define gpuEvent_t hipEvent_t
+#define gpuEventCreate hipEventCreate
+#define gpuEventDestroy hipEventDestroy
+#define gpuEventRecord hipEventRecord
+#define gpuEventSynchronize hipEventSynchronize
+#define gpuEventQuery hipEventQuery
+#define gpuStreamWaitEvent hipStreamWaitEvent
+#define gpuEventCreateWithFlags hipEventCreateWithFlags
+#define gpuEventDefault hipEventDefault
+#define gpuEventDisableTiming hipEventDisableTiming
+#define gpuEventInterprocess hipEventInterprocess
+#define gpuIpcEventHandle_t hipIpcEventHandle_t
+#define gpuIpcGetEventHandle hipIpcGetEventHandle
+#define gpuIpcOpenEventHandle hipIpcOpenEventHandle
+#define gpuIpcCloseEventHandle(handle) (gpuSuccess)
 #endif
 
 #define GPU_RT_CHECK(call)                                         \

include/util/util.h

@@ -98,6 +98,40 @@ inline int send_message(int sockfd, void const* buffer, size_t n_bytes) {
   return bytes_sent;
 }
 
+inline int receive_message_nonblock(int sockfd, void* buffer, size_t n_bytes) {
+  int bytes_read = 0;
+  int r;
+  while (bytes_read < static_cast<int>(n_bytes)) {
+    r = read(sockfd, static_cast<char*>(buffer) + bytes_read,
+             static_cast<size_t>(n_bytes - bytes_read));
+    if (r < 0 && !(errno == EAGAIN || errno == EWOULDBLOCK)) {
+      CHECK(false) << "ERROR reading from socket";
+    }
+    if (r > 0) {
+      bytes_read += r;
+    }
+  }
+  return bytes_read;
+}
+
+inline int send_message_nonblock(int sockfd, void const* buffer,
+                                 size_t n_bytes) {
+  int bytes_sent = 0;
+  int r;
+  while (bytes_sent < static_cast<int>(n_bytes)) {
+    // Make sure we write exactly n_bytes
+    r = write(sockfd, static_cast<char const*>(buffer) + bytes_sent,
+              n_bytes - bytes_sent);
+    if (r < 0 && !(errno == EAGAIN || errno == EWOULDBLOCK)) {
+      CHECK(false) << "ERROR writing to socket";
+    }
+    if (r > 0) {
+      bytes_sent += r;
+    }
+  }
+  return bytes_sent;
+}
+
 inline void send_ready(int bootstrap_fd) {
   bool ready = true;
   int ret = send_message(bootstrap_fd, &ready, sizeof(bool));