sync

.clang-format

@@ -2,7 +2,8 @@
 BasedOnStyle: LLVM
 AlignEscapedNewlines: DontAlign
 AlignConsecutiveAssignments: true
-AlignConsecutiveDeclarations: false
+AlignConsecutiveDeclarations: true
+AlignConsecutiveStructMembers: true
 AlignConsecutiveMacros: true
 AlignAfterOpenBracket: true
 AlignOperands: true
@@ -20,13 +21,13 @@ AlwaysBreakAfterReturnType: None
 PenaltyReturnTypeOnItsOwnLine: 100
 PenaltyBreakAssignment: 100
 PenaltyExcessCharacter: 10
-ColumnLimit: 85
+ColumnLimit: 80
 AlwaysBreakBeforeMultilineStrings: false
 BinPackArguments: true
 BinPackParameters: true
 BreakBeforeBraces: Custom
 BraceWrapping:
-  AfterClass: true
+  AfterClass: false
   AfterControlStatement: false
   AfterEnum: false
   AfterFunction: true
@@ -58,7 +59,36 @@ SpaceInEmptyParentheses: false
 SpaceBeforeParens: ControlStatementsExceptForEachMacros
 SpaceBeforeAssignmentOperators: true
 SpaceAfterCStyleCast: false
-ForEachMacros: ["ucs_for_each_submask"]
+SortIncludes: false
+ForEachMacros: ['FOR_EACH_ENTITY',
+                'kh_foreach',
+                'kh_foreach_key',
+                'kh_foreach_value',
+                'ucp_unpacked_address_for_each',
+                'ucs_array_for_each',
+                'UCS_BITMAP_FOR_EACH_BIT',
+                'ucs_for_each_bit',
+                'ucs_for_each_submask',
+                'ucs_hlist_for_each',
+                'ucs_hlist_for_each_extract',
+                'ucs_hlist_for_each_extract_if',
+                'ucs_list_for_each',
+                'ucs_list_for_each_safe',
+                'ucs_memory_type_for_each',
+                'UCS_PP_FOREACH',
+                'UCS_PP_FOREACH_SEP',
+                'ucs_profile_for_each_location',
+                'ucs_ptr_array_for_each',
+                'ucs_ptr_array_locked_for_each',
+                'ucs_queue_for_each',
+                'ucs_queue_for_each_extract',
+                'ucs_queue_for_each_safe',
+                'ucs_timerq_for_each_expired',
+                'UCT_IB_IFACE_VERBS_FOREACH_RXWQE',
+                'UCT_RC_VERBS_IFACE_FOREACH_TXWQE',
+                'UCS_INIT_ONCE',
+                'UCS_TEST_F',
+                'UCX_PERF_TEST_FOREACH']
 
 # CPP
 Standard: Cpp11
@@ -73,7 +103,6 @@ FixNamespaceComments: true
 NamespaceIndentation: None
 UseTab: Never
 ReflowComments: true
-SortIncludes: true
 IncludeCategories:
  - Regex: '^"'
    Priority: 1
@@ -85,13 +114,7 @@ SpacesInAngles: false
 SpacesBeforeTrailingComments: 1
 SpaceAfterTemplateKeyword: false
 SpacesInContainerLiterals: false
-
-# Issues:
-# 1.
-# Pointer alignment + declaration alignment:
-# long_type_name var;
-# void *         ptr;
-# Instead of:
-#void          *ptr;
-
-...
+---
+# Java
+Language: Java
+DisableFormat: true

.gitignore

@@ -88,3 +88,6 @@ GTAGS
 *.swp
 compile_commands.json
 .idea/
+.externalToolBuilders
+.classpath
+.vscode

Makefile.am

@@ -37,6 +37,7 @@ SUBDIRS += $(UCG_SUBDIR)
 endif
 
 SUBDIRS += \
+	src/tools/vfs \
 	src/tools/info \
 	src/tools/perf \
 	src/tools/profile \

NEWS

@@ -9,7 +9,8 @@
 
 ## Current
 ### Features: TBD
-#### UCX Core TBD
+#### UCX Core
+- Added ucp_tag_msg_recv_nbx routine.
 #### UCX Java (API Preview) TBD
 ### Bugfixes: TBD
 

bindings/java/src/main/java/org/openucx/jucx/ucp/UcpConstants.java

@@ -38,6 +38,7 @@ public class UcpConstants {
     static long UCP_FEATURE_AMO64;
     static long UCP_FEATURE_WAKEUP;
     static long UCP_FEATURE_STREAM;
+    static long UCP_FEATURE_AM;
 
     /**
      * UCP worker parameters field mask.

bindings/java/src/main/java/org/openucx/jucx/ucp/UcpEndpoint.java

@@ -9,11 +9,13 @@
 import java.io.Closeable;
 import java.nio.ByteBuffer;
 
+import static org.openucx.jucx.ucs.UcsConstants.MEMORY_TYPE.UCS_MEMORY_TYPE_UNKNOWN;
+
 public class UcpEndpoint extends UcxNativeStruct implements Closeable {
-    private final String paramsString;
+    private String paramsString;
     // Keep a reference to errorHandler to prevent it from GC and have valid ref
     // from JNI error handler.
-    private final UcpEndpointErrorHandler errorHandler;
+    private UcpEndpointErrorHandler errorHandler;
 
     @Override
     public String toString() {
@@ -81,9 +83,15 @@ public UcpRequest putNonBlocking(ByteBuffer src, long remoteAddress, UcpRemoteKe
     public UcpRequest putNonBlocking(long localAddress, long size,
                                      long remoteAddress, UcpRemoteKey remoteKey,
                                      UcxCallback callback) {
+        return putNonBlocking(localAddress, size, remoteAddress, remoteKey, callback,
+            UCS_MEMORY_TYPE_UNKNOWN);
+    }
 
+    public UcpRequest putNonBlocking(long localAddress, long size,
+                                     long remoteAddress, UcpRemoteKey remoteKey,
+                                     UcxCallback callback, int memoryType) {
         return putNonBlockingNative(getNativeId(), localAddress,
-            size, remoteAddress, remoteKey.getNativeId(), callback);
+            size, remoteAddress, remoteKey.getNativeId(), callback, memoryType);
     }
 
     /**
@@ -136,8 +144,16 @@ public UcpRequest getNonBlocking(long remoteAddress, UcpRemoteKey remoteKey,
     public UcpRequest getNonBlocking(long remoteAddress, UcpRemoteKey remoteKey,
                                      long localAddress, long size, UcxCallback callback) {
 
+        return getNonBlocking(remoteAddress, remoteKey, localAddress, size, callback,
+            UCS_MEMORY_TYPE_UNKNOWN);
+    }
+
+    public UcpRequest getNonBlocking(long remoteAddress, UcpRemoteKey remoteKey,
+                                     long localAddress, long size, UcxCallback callback,
+                                     int memoryType) {
+
         return getNonBlockingNative(getNativeId(), remoteAddress, remoteKey.getNativeId(),
-            localAddress, size, callback);
+            localAddress, size, callback, memoryType);
     }
 
     /**
@@ -192,7 +208,13 @@ public UcpRequest sendTaggedNonBlocking(ByteBuffer sendBuffer, long tag, UcxCall
 
     public UcpRequest sendTaggedNonBlocking(long localAddress, long size,
                                             long tag, UcxCallback callback) {
-        return sendTaggedNonBlockingNative(getNativeId(), localAddress, size, tag, callback);
+        return sendTaggedNonBlocking(localAddress, size, tag, callback, UCS_MEMORY_TYPE_UNKNOWN);
+    }
+
+    public UcpRequest sendTaggedNonBlocking(long localAddress, long size,
+                                            long tag, UcxCallback callback, int memoryType) {
+        return sendTaggedNonBlockingNative(getNativeId(), localAddress, size, tag, callback,
+            memoryType);
     }
 
     /**
@@ -207,11 +229,17 @@ public UcpRequest sendTaggedNonBlocking(ByteBuffer sendBuffer, UcxCallback callb
      * Iov version of non blocking send operaation
      */
     public UcpRequest sendTaggedNonBlocking(long[] localAddresses, long[] sizes,
-                                            long tag, UcxCallback callback) {
+                                            long tag, UcxCallback callback, int memoryType) {
         UcxParams.checkArraySizes(localAddresses, sizes);
 
         return sendTaggedIovNonBlockingNative(getNativeId(), localAddresses, sizes,
-            tag, callback);
+            tag, callback, memoryType);
+    }
+
+    public UcpRequest sendTaggedNonBlocking(long[] localAddresses, long[] sizes,
+                                            long tag, UcxCallback callback) {
+
+        return sendTaggedNonBlocking(localAddresses, sizes, tag, callback, UCS_MEMORY_TYPE_UNKNOWN);
     }
 
     /**
@@ -222,14 +250,27 @@ public UcpRequest sendTaggedNonBlocking(long[] localAddresses, long[] sizes,
      * completion of the send operation.
      */
     public UcpRequest sendStreamNonBlocking(long localAddress, long size, UcxCallback callback) {
-        return sendStreamNonBlockingNative(getNativeId(), localAddress, size, callback);
+        return sendStreamNonBlocking(localAddress, size, callback, UCS_MEMORY_TYPE_UNKNOWN);
+    }
+
+    public UcpRequest sendStreamNonBlocking(long localAddress, long size, UcxCallback callback,
+                                            int memoryType) {
+        return sendStreamNonBlockingNative(getNativeId(), localAddress, size, callback, memoryType);
     }
 
     public UcpRequest sendStreamNonBlocking(long[] localAddresses, long[] sizes,
                                             UcxCallback callback) {
         UcxParams.checkArraySizes(localAddresses, sizes);
 
-        return sendStreamIovNonBlockingNative(getNativeId(), localAddresses, sizes, callback);
+        return sendStreamNonBlocking(localAddresses, sizes, callback, UCS_MEMORY_TYPE_UNKNOWN);
+    }
+
+    public UcpRequest sendStreamNonBlocking(long[] localAddresses, long[] sizes,
+                                            UcxCallback callback, int memoryType) {
+        UcxParams.checkArraySizes(localAddresses, sizes);
+
+        return sendStreamIovNonBlockingNative(getNativeId(), localAddresses, sizes, callback,
+            memoryType);
     }
 
     public UcpRequest sendStreamNonBlocking(ByteBuffer buffer, UcxCallback callback) {
@@ -244,17 +285,30 @@ public UcpRequest sendStreamNonBlocking(ByteBuffer buffer, UcxCallback callback)
      * the UCP library will invoke the call-back when data is in the receive buffer
      * and ready for application access.
      */
+    public UcpRequest recvStreamNonBlocking(long localAddress, long size, long flags,
+                                            UcxCallback callback, int memoryType) {
+        return recvStreamNonBlockingNative(getNativeId(), localAddress, size, flags, callback,
+            memoryType);
+    }
+
     public UcpRequest recvStreamNonBlocking(long localAddress, long size, long flags,
                                             UcxCallback callback) {
-        return recvStreamNonBlockingNative(getNativeId(), localAddress, size, flags, callback);
+        return recvStreamNonBlocking(localAddress, size, flags, callback, UCS_MEMORY_TYPE_UNKNOWN);
     }
 
     public UcpRequest recvStreamNonBlocking(long[] localAddresses, long[] sizes, long flags,
-                                            UcxCallback callback) {
+                                            UcxCallback callback, int memoryType) {
         UcxParams.checkArraySizes(localAddresses, sizes);
 
         return recvStreamIovNonBlockingNative(getNativeId(), localAddresses, sizes, flags,
-            callback);
+            callback, memoryType);
+    }
+
+    public UcpRequest recvStreamNonBlocking(long[] localAddresses, long[] sizes, long flags,
+                                            UcxCallback callback) {
+
+        return recvStreamNonBlocking(localAddresses, sizes, flags, callback,
+            UCS_MEMORY_TYPE_UNKNOWN);
     }
 
     public UcpRequest recvStreamNonBlocking(ByteBuffer buffer, long flags, UcxCallback callback) {
@@ -297,45 +351,53 @@ public UcpRequest closeNonBlockingFlush() {
 
     private static native UcpRequest putNonBlockingNative(long enpointId, long localAddress,
                                                           long size, long remoteAddr,
-                                                          long ucpRkeyId, UcxCallback callback);
+                                                          long ucpRkeyId, UcxCallback callback,
+                                                          int memoryType);
 
     private static native void putNonBlockingImplicitNative(long enpointId, long localAddress,
                                                             long size, long remoteAddr,
                                                             long ucpRkeyId);
 
     private static native UcpRequest getNonBlockingNative(long enpointId, long remoteAddress,
                                                           long ucpRkeyId, long localAddress,
-                                                          long size, UcxCallback callback);
+                                                          long size, UcxCallback callback,
+                                                          int memoryType);
 
     private static native void getNonBlockingImplicitNative(long enpointId, long remoteAddress,
                                                             long ucpRkeyId, long localAddress,
                                                             long size);
 
     private static native UcpRequest sendTaggedNonBlockingNative(long enpointId, long localAddress,
                                                                  long size, long tag,
-                                                                 UcxCallback callback);
+                                                                 UcxCallback callback,
+                                                                 int memoryType);
 
     private static native UcpRequest sendTaggedIovNonBlockingNative(long enpointId,
                                                                     long[] localAddresses,
                                                                     long[] sizes, long tag,
-                                                                    UcxCallback callback);
+                                                                    UcxCallback callback,
+                                                                    int memoryType);
 
     private static native UcpRequest sendStreamNonBlockingNative(long enpointId, long localAddress,
-                                                                 long size, UcxCallback callback);
+                                                                 long size, UcxCallback callback,
+                                                                 int memoryType);
 
     private static native UcpRequest sendStreamIovNonBlockingNative(long enpointId,
                                                                     long[] localAddresses,
                                                                     long[] sizes,
-                                                                    UcxCallback callback);
+                                                                    UcxCallback callback,
+                                                                    int memoryType);
 
     private static native UcpRequest recvStreamNonBlockingNative(long enpointId, long localAddress,
                                                                  long size, long flags,
-                                                                 UcxCallback callback);
+                                                                 UcxCallback callback,
+                                                                 int memoryType);
 
     private static native UcpRequest recvStreamIovNonBlockingNative(long enpointId,
                                                                     long[] localAddresses,
                                                                     long[] sizes, long flags,
-                                                                    UcxCallback callback);
+                                                                    UcxCallback callback,
+                                                                    int memoryType);
 
     private static native UcpRequest flushNonBlockingNative(long enpointId, UcxCallback callback);
 

bindings/java/src/main/java/org/openucx/jucx/ucp/UcpParams.java

@@ -102,6 +102,15 @@ public UcpParams requestTagFeature() {
         return this;
     }
 
+    /**
+     * Request Active Message support feature.
+     */
+    public UcpParams requestAmFeature() {
+        this.fieldMask |= UcpConstants.UCP_PARAM_FIELD_FEATURES;
+        this.features |= UcpConstants.UCP_FEATURE_AM;
+        return this;
+    }
+
     /**
      * Request remote memory access support.
      */

bindings/java/src/main/java/org/openucx/jucx/ucp/UcpRequest.java

@@ -7,6 +7,7 @@
 
 import org.openucx.jucx.UcxCallback;
 import org.openucx.jucx.UcxNativeStruct;
+import org.openucx.jucx.ucs.UcsConstants;
 
 import java.io.Closeable;
 import java.nio.ByteBuffer;
@@ -21,6 +22,8 @@ public class UcpRequest extends UcxNativeStruct implements Closeable {
 
     private long senderTag;
 
+    private int status = UcsConstants.STATUS.UCS_INPROGRESS;
+
     private UcpRequest(long nativeId) {
         setNativeId(nativeId);
     }
@@ -52,6 +55,13 @@ public boolean isCompleted() {
         return (getNativeId() == null) || isCompletedNative(getNativeId());
     }
 
+    /**
+     * @return status of the current request
+     */
+    public int getStatus() {
+        return status;
+    }
+
     /**
      * This routine releases the non-blocking request back to the library, regardless
      * of its current state. Communications operations associated with this request