re-enable passing in providers/domains to try an use

Author: rdfriese

src/api.c

@@ -21,21 +21,24 @@
  * state and to start the underlying OFI and runtime layer. Using any other APIs
  * without prior call to rofi_init() may result in failure and unexpected behavior.
  *
+ * @param provs A priority list of providers to try and be used.
+ * @param domains A priority list of domains to try and be used.
+ *
  *
  * @return 0 on success, -1 in case of errors
  *
  * \b blocking: yes
  * \b thread-safe: no
  *
  */
-int rofi_init(char *prov) {
+int rofi_init(char *provs, char *domains) {
     int ret = 0;
 
     rofi.desc.status = ROFI_STATUS_NONE;
 
     DEBUG_MSG("Initilizing ROFI runtime...");
 
-    ret = rofi_init_internal(prov);
+    ret = rofi_init_internal(provs, domains);
 
     if (ret) {
         ERR_MSG("Error initializing ROFI library");

src/core.c

@@ -25,7 +25,7 @@ void *rofi_get_remote_addr_internal(void *addr, unsigned int id) {
         return NULL;
     }
 
-    DEBUG_MSG("\t Found MR [0x%lx - 0x%lx] Addr: %p Key: 0x%lx ", el->start, el->start + el->size, (void *)(addr - (uintptr_t)el->start + el->iov[id].addr), el->mr_key);
+    DEBUG_MSG("\t Found MR [0x%lx - 0x%lx] Addr: %p Key: 0x%lx ", el->start, el->start + el->size, (void *)(addr - (uintptr_t)el->start + el->iov[id].addr), el->iov[id].key);
     return (void *)(addr - (uintptr_t)el->start + el->iov[id].addr);
 }
 
@@ -38,7 +38,7 @@ void *rofi_get_local_addr_from_remote_addr_internal(void *addr, unsigned int id)
         return NULL;
     }
 
-    DEBUG_MSG("\t Found MR [0x%lx - 0x%lx] Addr: %p Key: 0x%lx", el->start, el->start + el->size, (void *)(addr - el->iov[id].addr + (uintptr_t)el->start), el->mr_key);
+    DEBUG_MSG("\t Found MR [0x%lx - 0x%lx] Addr: %p Key: 0x%lx", el->start, el->start + el->size, (void *)(addr - el->iov[id].addr + (uintptr_t)el->start), el->iov[id].key);
     return (void *)(addr - el->iov[id].addr + (uintptr_t)el->start);
 }
 
@@ -206,10 +206,34 @@ int rofi_send_internal(unsigned long id, void *buf, size_t size, unsigned long f
 int rofi_recv_internal(unsigned long id, void *buf, size_t size, unsigned long flags) {
 }
 
-int rofi_init_internal(char *prov) {
-    if (!prov) {
-        ERR_MSG("ROFI provider not specified. Currently ROFI only supports \"verbs\".");
+rofi_names_t *rofi_parse_names_internal(char *names_list) {
+    char token = ';';
+    int name_cnt = 0;
+    for (int i = 0; i < strlen(names_list); i++) {
+        if (names_list[i] == token) {
+            name_cnt++;
+        }
+    }
+    name_cnt += 1;
+    char **name_strs = (char **)calloc(name_cnt, sizeof(char *));
+
+    int p = 0;
+    int i = 0;
+    for (int k = 0; k < strlen(names_list); k++) {
+        if (names_list[k] == token) {
+            name_strs[p] = strndup(&names_list[i], k - i);
+            p++;
+            i = k + 1;
+        }
     }
+    name_strs[p] = strndup(&names_list[i], strlen(names_list) - i);
+    rofi_names_t *names = (rofi_names_t *)calloc(1, sizeof(rofi_names_t));
+    names->num = name_cnt;
+    names->names = name_strs;
+    return names;
+}
+
+int rofi_init_internal(char *provs, char *domains) {
     pthread_rwlock_init(&rofi.mr_lock, NULL);
     pthread_mutex_init(&rofi.lock, NULL);
     int ret = 0;
@@ -224,12 +248,7 @@ int rofi_init_internal(char *prov) {
     rofi.desc.nodes = rt_get_size();
     rofi.desc.nid = rt_get_rank();
 
-    rofi.info = (struct fi_info *)calloc(1, sizeof(struct fi_info));
-    if (!rofi.info) {
-        ERR_MSG("Error allocating memory for rofi. Aborting.");
-        ret = EXIT_FAILURE;
-        goto err;
-    }
+    rofi.info = NULL;
 
     DEBUG_MSG("Initializing process %d/%d...", rofi.desc.nid, rofi.desc.nodes);
 
@@ -238,7 +257,7 @@ int rofi_init_internal(char *prov) {
         return EXIT_FAILURE;
     }
 
-    hints->caps = FI_RMA | FI_ATOMIC | FI_COLLECTIVE; // eventually want FI_ATOMIC
+    hints->caps = FI_RMA | FI_ATOMIC | FI_COLLECTIVE;
     hints->addr_format = FI_FORMAT_UNSPEC;
     hints->domain_attr->resource_mgmt = FI_RM_ENABLED;
     hints->domain_attr->threading = FI_THREAD_DOMAIN;
@@ -249,9 +268,21 @@ int rofi_init_internal(char *prov) {
     hints->ep_attr->type = FI_EP_RDM;
     hints->tx_attr->op_flags = FI_DELIVERY_COMPLETE; // maybe need to change this to FI_INJECT_COMPLETE or FI_TRANSMIT_COMPLETE
 
-    if (prov) {
-        hints->fabric_attr->prov_name = strdup(prov);
+    rofi_names_t *prov_names = NULL;
+    if (provs) {
+        prov_names = rofi_parse_names_internal(provs);
     }
+    // else {
+    //     names = rofi_parse_names_internal("verbs");
+    // }
+
+    rofi_names_t *domain_names = NULL;
+    if (domains) {
+        domain_names = rofi_parse_names_internal(domains);
+    }
+    // else {
+    //     rofi.domains = rofi_parse_names_internal("ib");
+    // }
 
     // this isn't really needed for verbs since it is a connected endpoint
     rofi.remote_addrs = (fi_addr_t *)malloc(rofi.desc.nodes * sizeof(fi_addr_t));
@@ -264,11 +295,27 @@ int rofi_init_internal(char *prov) {
         rofi.remote_addrs[i] = i;
     }
 
-    rofi_transport_init(hints, &rofi);
+    rofi_transport_init(hints, &rofi, prov_names, domain_names);
+
+    if (prov_names) {
+        for (int i = 0; i < prov_names->num; i++) {
+            free(prov_names->names[i]);
+        }
+        free(prov_names);
+    }
+
+    if (domain_names) {
+        for (int i = 0; i < domain_names->num; i++) {
+            free(domain_names->names[i]);
+        }
+        free(domain_names);
+    }
 
     mr_init();
     uint64_t global_barrier_size = rofi.desc.nodes * sizeof(uint64_t);
-    int rofi_mr_size = global_barrier_size;
+    uint64_t sub_alloc_barrier_size = rofi.desc.nodes * sizeof(uint64_t);
+    uint64_t sub_alloc_size = rofi.desc.nodes * sizeof(struct fi_rma_iov);
+    int rofi_mr_size = global_barrier_size + sub_alloc_barrier_size + sub_alloc_size;
 
     rofi.mr = mr_add(&rofi, rofi_mr_size, 0);
     if (!rofi.mr) {
@@ -284,9 +331,14 @@ int rofi_init_internal(char *prov) {
 
     rofi.global_barrier_id = 0;
     rofi.global_barrier_buf = (uint64_t *)rofi.mr->start;
+    rofi.sub_alloc_barrier_buf = (uint64_t *)(rofi.mr->start + global_barrier_size);
+    rofi.sub_alloc_buf = (struct fi_rma_iov *)(rofi.mr->start + global_barrier_size + sub_alloc_barrier_size);
 
     for (int i = 0; i < rofi.desc.nid; i++) {
         rofi.global_barrier_buf[i] = 0;
+        rofi.sub_alloc_barrier_buf[i] = 0;
+        rofi.sub_alloc_buf[i].key = 0;
+        rofi.sub_alloc_buf[i].addr = 0;
     }
     fi_freeinfo(hints);
     return 0;

src/include/rofi.h

@@ -5,21 +5,21 @@
 
 #define ROFI_ERR_ALLOC 0x01
 
-int          rofi_init(char*);
-int          rofi_finit(void);
+int rofi_init(char *, char *);
+int rofi_finit(void);
 unsigned int rofi_get_size(void);
 unsigned int rofi_get_id(void);
-int          rofi_flush(void);
-int          rofi_put(void*, void*, size_t, unsigned int, unsigned long);
-int          rofi_iput(void*, void*, size_t, unsigned int, unsigned long);
-int          rofi_get(void*, void*, size_t, unsigned int, unsigned long);
-int          rofi_iget(void*, void*, size_t, unsigned int, unsigned long);
-int          rofi_alloc(size_t, unsigned long, void**);
-int          rofi_sub_alloc(size_t, unsigned long, void**,uint64_t*, uint64_t);
-int          rofi_release(void*);
-int          rofi_sub_release(void*,uint64_t*, uint64_t);
-void         rofi_barrier(void);
-int          rofi_wait(void);
-void*        rofi_get_remote_addr(void*, unsigned int);
-void*        rofi_get_local_addr_from_remote_addr(void*, unsigned int);
+int rofi_flush(void);
+int rofi_put(void *, void *, size_t, unsigned int, unsigned long);
+int rofi_iput(void *, void *, size_t, unsigned int, unsigned long);
+int rofi_get(void *, void *, size_t, unsigned int, unsigned long);
+int rofi_iget(void *, void *, size_t, unsigned int, unsigned long);
+int rofi_alloc(size_t, unsigned long, void **);
+int rofi_sub_alloc(size_t, unsigned long, void **, uint64_t *, uint64_t);
+int rofi_release(void *);
+int rofi_sub_release(void *, uint64_t *, uint64_t);
+void rofi_barrier(void);
+int rofi_wait(void);
+void *rofi_get_remote_addr(void *, unsigned int);
+void *rofi_get_local_addr_from_remote_addr(void *, unsigned int);
 #endif

src/include/rofi_debug.h

@@ -1,18 +1,22 @@
 #ifndef ROFI_DEBUG_H
 #define ROFI_DEBUG_H
 
+#include <sys/syscall.h>
 #include <sys/types.h>
 #include <unistd.h>
-#include <sys/syscall.h>
 
 #ifdef _DEBUG
-#define DEBUG_MSG(fmt, ...) \
-	do { fprintf(stderr, "[%d][%d][DEBUG][%s][%s:%d] " fmt "\n", rt_get_rank(),syscall(__NR_gettid),__func__, __FILE__, \
-		     __LINE__, ##__VA_ARGS__); } while (0)
+#define DEBUG_MSG(fmt, ...)                                                                                                            \
+    do {                                                                                                                               \
+        fprintf(stderr, "[ROFI DEBUG][PE: %d][TID: %d][%s][%s:%d] " fmt "\n", rt_get_rank(), syscall(__NR_gettid), __func__, __FILE__, \
+                __LINE__, ##__VA_ARGS__);                                                                                              \
+    } while (0)
 #else
 #define DEBUG_MSG(fmt, args...)
 #endif
 
-#define ERR_MSG(fmt, args...) fprintf(stderr,"[PE %d, TID: %d][ROFI ERR][%s][%s:%d] " fmt "\n", rt_get_rank(),syscall(__NR_gettid),__func__, __FILE__, __LINE__, ##args)
+#define WARN_MSG(fmt, args...) fprintf(stderr, "[ROFI WARNING][PE: %d][TID: %d][%s][%s:%d] " fmt "\n", rt_get_rank(), syscall(__NR_gettid), __func__, __FILE__, __LINE__, ##args)
+
+#define ERR_MSG(fmt, args...) fprintf(stderr, "[ROFI ERROR][PE: %d][TID: %d][%s][%s:%d] " fmt "\n", rt_get_rank(), syscall(__NR_gettid), __func__, __FILE__, __LINE__, ##args)
 
 #endif

src/include/rofi_internal.h

@@ -15,7 +15,7 @@
 #define ROFI_FI_VERSION FI_VERSION(1, 20)
 #endif
 
-typedef struct rofi_transport_s rofi_transport_t;
+typedef struct rofi_transport_t rofi_transport_t;
 
 #include "context.h"
 #include "mr.h"
@@ -54,7 +54,12 @@ typedef struct {
     uint64_t inject_size;
 } rofi_desc_t;
 
-struct rofi_transport_s {
+typedef struct rofi_prov_names_t {
+    char **names;
+    int num;
+} rofi_names_t;
+
+struct rofi_transport_t {
     struct fi_info *info;
     struct fid_fabric *fabric;
     struct fid_domain *domain;
@@ -71,13 +76,16 @@ struct rofi_transport_s {
     rofi_mr_desc *mr;
     uint64_t global_barrier_id;
     uint64_t *global_barrier_buf;
+    uint64_t *sub_alloc_barrier_buf;
+    struct fi_rma_iov *sub_alloc_buf;
     pthread_mutex_t lock;
     pthread_rwlock_t mr_lock;
+    uint64_t fi_collective;
 };
 
 extern rofi_transport_t rofi;
 
-int rofi_init_internal(char *);
+int rofi_init_internal(char *, char *);
 int rofi_finit_internal(void);
 unsigned int rofi_get_size_internal(void);
 unsigned int rofi_get_id_internal(void);

src/include/transport.h

@@ -31,39 +31,40 @@
 #define _TRANSPORT_H_
 
 #if HAVE_CONFIG_H
-#  include <config.h>
+#include <config.h>
 #endif /* HAVE_CONFIG_H */
 
-#include <stdlib.h>
 #include <inttypes.h>
 #include <netinet/tcp.h>
-#include <sys/uio.h>
 #include <stdbool.h>
+#include <stdlib.h>
+#include <sys/uio.h>
 
 #include <rdma/fabric.h>
-#include <rdma/fi_rma.h>
 #include <rdma/fi_domain.h>
+#include <rdma/fi_rma.h>
 
-#define ROFI_TRANSPORT_ERR_MSG(call,retv) \
-    do { fprintf(stderr, "[PE %d][ROFI TRANSPORT ERR][%s:%d] " call " failed: %s (%d)\n", \
-           rt_get_rank(),__FILE__,__LINE__, fi_strerror(retv),(int)(retv)); } while(0)
+#define ROFI_TRANSPORT_ERR_MSG(call, retv)                                               \
+    do {                                                                                 \
+        fprintf(stderr, "[PE %d][ROFI TRANSPORT ERR][%s:%d] " call " failed: %s (%d)\n", \
+                rt_get_rank(), __FILE__, __LINE__, fi_strerror(retv), (int)(retv));      \
+    } while (0)
 
- #define MIN(a,b) \
-   ({ __typeof__ (a) _a = (a); \
+#define MIN(a, b) \
+    ({ __typeof__ (a) _a = (a); \
        __typeof__ (b) _b = (b); \
      _a < _b ? _a : _b; })
 
 int rofi_transport_fini(rofi_transport_t *rofi);
-int rofi_transport_init(struct fi_info *hints,  rofi_transport_t *rofi);
-int rofi_transport_init_fabric_resources( rofi_transport_t *rofi);
-int rofi_transport_init_endpoint_resources( rofi_transport_t *rofi);
+int rofi_transport_init(struct fi_info *hints, rofi_transport_t *rofi, rofi_names_t *prov_names, rofi_names_t *domain_names);
+int rofi_transport_init_fabric_resources(rofi_transport_t *rofi);
+int rofi_transport_init_endpoint_resources(rofi_transport_t *rofi);
 int rofi_transport_init_av(rofi_transport_t *rofi);
 
-
 int rofi_transport_progress(rofi_transport_t *rofi);
 int rofi_transport_ctx_check_err(rofi_transport_t *rofi, int err);
 int rofi_transport_check_rma_err(rofi_transport_t *rofi, int ret);
-int rofi_transport_wait_on_cntr(rofi_transport_t *rofi, uint64_t *pending_cntr, struct fid_cntr *cntr); 
+int rofi_transport_wait_on_cntr(rofi_transport_t *rofi, uint64_t *pending_cntr, struct fid_cntr *cntr);
 
 int rofi_transport_put_inject(rofi_transport_t *rofi, struct fi_rma_iov *rma_iov, uint64_t pe, const void *src_addr, size_t len);
 int rofi_transport_put_large(rofi_transport_t *rofi, struct fi_rma_iov *rma_iov, uint64_t pe, const void *src_addr, size_t len, void *desc, void *context);
@@ -75,10 +76,9 @@ int rofi_transport_get_large(rofi_transport_t *rofi, struct fi_rma_iov *rma_iov,
 int rofi_transport_get(rofi_transport_t *rofi, struct fi_rma_iov *rma_iov, uint64_t pe, void *dst_addr, size_t len, void *desc, void *context);
 int rofi_transport_get_wait_all(rofi_transport_t *rofi);
 
-
 int rofi_transport_exchange_mr_info(rofi_transport_t *rofi, rofi_mr_desc *mr);
 int rofi_transport_sub_exchange_mr_info(rofi_transport_t *rofi, rofi_mr_desc *mr, uint64_t *pes, uint64_t num_pes);
+int rofi_transport_inner_barrier(rofi_transport_t *rofi, uint64_t *barrier_id, uint64_t *barrier_buf, uint64_t *pes, uint64_t me, uint64_t num_pes);
 int rofi_transport_barrier(rofi_transport_t *rofi);
 
-
 #endif /* _TRANSPORT_H_ */