UCP Backend

Makefile

@@ -17,6 +17,12 @@ SRCS += $(wildcard $(SDIR)src/backends/tcp/*.c)
 IFLAGS += $(TCP_INC)
 LDLIBS += $(TCP_LIBS)
 endif
+ifdef USE_UCP
+SRCS += $(wildcard $(SDIR)src/backends/ucp/*.c)
+IFLAGS += $(UCP_INC)
+LDLIBS += $(UCP_LIBS)
+endif
+
 HEADERS = $(wildcard $(SDIR)include/*.h $(SDIR)include/laik/*.h)
 OBJS = $(SRCS:$(SDIR)%.c=%.o)
 

configure

@@ -31,14 +31,45 @@ if "CC" in os.environ:
 # Command line parsing
 
 parser = argparse.ArgumentParser()
+parser.add_argument("--no-ucp", help="disable UCP backend", action="store_true")
+parser.add_argument("--no-fabric", help="disable Libfabric backend",
+        action="store_true")
 parser.add_argument("--no-tcp", help="disable TCP backend", action="store_true")
 parser.add_argument("--no-mpi", help="disable MPI backend", action="store_true")
 parser.add_argument("--no-mqtt", help="disable MQTT support", action="store_true")
 args = parser.parse_args()
+use_fabric = not args.no_fabric
 use_mpi = not args.no_mpi
+use_ucp = not args.no_ucp
 use_tcp = not args.no_tcp
 use_mqtt = not args.no_mqtt
 
+pkgc_found = bool(shutil.which("pkg-config"))
+# Enable a backend if pkg-config finds its dependency
+# This behaviour is exactly the same for TCP and Libfabric backends,
+# thus a common function makes sense
+def enable_if_found(name, pkgname, ubuntu, pkgopts=[]):
+    global pkgc_found, defs, miscvars, test_subdirs
+    if not pkgc_found:
+        print("%s backend disabled: pkg-config not found." % (name))
+        return
+
+    found = not subprocess.call(['pkg-config'] + pkgopts + [pkgname])
+    if not found:
+        print("%s backend disabled: %s not found." % (name, pkgname))
+        print("  On Ubuntu, install '%s'" % (ubuntu))
+        return
+
+    print("%s backend enabled (%s found)." % (name, pkgname))
+    pkg_inc = os.popen("pkg-config --cflags %s" % (pkgname)).read().strip()
+    pkg_libs = os.popen("pkg-config --libs %s" % (pkgname)).read().strip()
+    up = name.upper()
+    defs += " -DUSE_%s" % (up)
+    miscvars += "USE_%s=1\n" % (up)
+    miscvars += "%s_INC=%s\n" % (up, pkg_inc)
+    miscvars += "%s_LIBS=%s\n" % (up, pkg_libs)
+    test_subdirs += " %s" % (name.lower())
+
 #------------------------------------
 # Mosquitto/MQTT
 
@@ -116,34 +147,31 @@ else:
 # TCP backend support
 
 if use_tcp:
-    pkgc_found = bool(shutil.which("pkg-config"))
-    if pkgc_found:
-        # check gio version
-        glib_versionok = not subprocess.call(['pkg-config', '--atleast-version=2.44', 'gio-2.0'])
-if use_tcp and pkgc_found and glib_versionok:
-    print("TCP backend enabled (glib-2.0/gio-2.0 found).")
-    glib_inc = os.popen('pkg-config --cflags gio-2.0').read().strip()
-    glib_libs = os.popen('pkg-config --libs gio-2.0').read().strip()
-    defs += " -DUSE_TCP"
-    miscvars += "USE_TCP=1\n"
-    miscvars += "TCP_INC=" + glib_inc + "\n"
-    miscvars += "TCP_LIBS=" + glib_libs + "\n"
-    test_subdirs += " tcp"
+    enable_if_found("TCP", "gio-2.0", "libglib2.0-dev", ["--atleast-version=2.44"])
+else:
+    print("TCP backend disabled.")
+
+#------------------------------------
+# Libfabric backend support
+
+if use_fabric:
+    enable_if_found("Fabric", "libfabric", "libfabric-dev")
 else:
-    if not use_tcp:
-        print("TCP backend disabled.")
-    elif not pkgc_found:
-        print("TCP backend disabled: pkg-config required to detect glib-2.0 dependency.")
-    else:
-        print("TCP backend disabled: glib-2.0/gio-2.0 not found.")
-        print("  On Ubuntu, install 'libglib2.0-dev'")
+    print("Fabric backend disabled.")
 
 #------------------------------------
 # TCP2 backend support: always enable
 print("TCP2 backend enabled.")
 defs += " -DUSE_TCP2"
 test_subdirs += " tcp2"
 
+#------------------------------------
+# UCP backend support
+if use_ucp:
+    enable_if_found("UCP", "ucx", "libucx-dev")
+else:
+    print("UCP backend disabled.")
+
 #------------------------------------
 # C++ support
 # LAIK does not use C++ itself, but there is a C++ example
@@ -236,11 +264,11 @@ sdir = os.path.dirname(os.path.realpath(__file__))
 if bdir != sdir:
     print("Detected src != build directory")
     # generate a mirror directory hierarchy for generated files
-    for dir in ["src", "src/backends", "src/backends/tcp",
+    for dir in ["src", "src/backends", "src/backends/tcp", "src/backends/ucp"
                 "examples","examples/c++","external",
                 "external/MQTT","external/simple",
                 "tests","tests/src","tests/mpi",
-                "tests/tcp","tests/tcp2"]:
+                "tests/tcp","tests/tcp2","tests/ucp"]:
         if not os.path.exists(dir):
             os.makedirs(dir)
             print("    created directory '" + dir + "'")
@@ -249,7 +277,7 @@ if bdir != sdir:
     for dir in ["","examples/","examples/c++/",
                 "external/MQTT/", "external/simple/",
                 "tests/", "tests/src/", "tests/mpi/",
-                "tests/tcp/", "tests/tcp2/"]:
+                "tests/tcp/", "tests/tcp2/", "tests/ucp"]:
         mfile = open(dir + "Makefile", 'w')
         mfile.write("# Generated by 'configure'.\n")
         mfile.write("SDIR=" + sdir + "/" + dir + "\n")

include/laik-backend-ucp.h

@@ -0,0 +1,8 @@
+#ifndef LAIK_BACKEND_UCP_H
+#define LAIK_BACKEND_UCP_H
+
+#include "laik.h"
+
+Laik_Instance* laik_init_ucp(int *argc, char ***argv);
+
+#endif /* LAIK_BACKEND_UCP_H */

include/laik/backend.h

@@ -85,6 +85,8 @@ struct _Laik_Backend {
   // for elasticity: removal of processes which got started in a previous
   // resize is finished. They can be marked as dead and resources freed
   void (*finish_resize)();
+
+  Laik_Allocator* (*allocator)();
 };
 
 

src/backends/ucp/backend-ucp-types.h

@@ -0,0 +1,43 @@
+//*********************************************************************************
+#pragma once
+
+//*********************************************************************************
+#include <ucp/api/ucp.h>
+
+//*********************************************************************************
+// Initialized as NEW
+typedef enum _State
+{
+    NEW = 1,   // process is only in new group
+    INHERITED, // process is in old and new group
+    INREMOVE1, // process is marked to be removed
+    INREMOVE2, // process will be removed from laik group (is only in old group)
+    DEAD       // process is no longer used
+} State;
+
+//*********************************************************************************
+typedef struct _Peer
+{
+    State state;
+    size_t addrlen;
+    ucp_address_t *address;
+} Peer;
+
+//*********************************************************************************
+// Global struct used to describe each processes' current state
+typedef struct _InstData
+{
+    State state;
+    int number_dead;
+    char host[64];          // my hostname
+    char location[128];     // my location
+    int mylid;              // location id
+    int world_size;         // total number of location ids/ peers (can only grow)
+    int phase;              // current pohase
+    int epoch;              // current epoch
+    size_t addrlen;         // local ucx address length
+    ucp_address_t *address; // local ucx address, memory is handled by the ucp worker
+    Peer *peer;
+} InstData;
+
+//*********************************************************************************