Merge pull request #3426 from hiroyuki-sato/event_set_functions

UCS/API event_set

src/ucs/Makefile.am

@@ -36,6 +36,7 @@ nobase_dist_libucs_la_HEADERS = \
 	profile/profile_on.h \
 	stats/stats_fwd.h \
 	stats/libstats.h \
+	sys/event_set.h \
 	sys/compiler_def.h\
 	sys/math.h \
 	sys/preprocessor.h \
@@ -122,6 +123,7 @@ libucs_la_SOURCES = \
 	memory/memtype_cache.c \
 	profile/profile.c \
 	stats/stats.c \
+	sys/event_set.c \
 	sys/init.c \
 	sys/math.c \
 	sys/module.c \

src/ucs/sys/event_set.c

@@ -0,0 +1,168 @@
+/**
+ * Copyright (C) Hiroyuki Sato. 2019. ALL RIGHTS RESERVED.
+ *
+ * See file LICENSE for terms.
+ */
+
+#include "event_set.h"
+
+#include <ucs/debug/memtrack.h>
+#include <ucs/debug/log.h>
+#include <ucs/debug/assert.h>
+
+#include <sys/epoll.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+
+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+#define UCS_EVENT_EPOLL_MAX_EVENTS 16
+
+struct ucs_sys_event_set {
+    int epfd;
+};
+
+
+static inline int ucs_event_set_map_to_raw_events(int events)
+{
+    int raw_events = 0;
+
+    if (events & UCS_EVENT_SET_EVREAD) {
+         raw_events |= EPOLLIN;
+    }
+    if (events & UCS_EVENT_SET_EVWRITE) {
+         raw_events |= EPOLLOUT;
+    }
+    return raw_events;
+}
+
+static inline int ucs_event_set_map_to_events(int raw_events)
+{
+    int events = 0;
+
+    if (raw_events & EPOLLIN) {
+         events |= UCS_EVENT_SET_EVREAD;
+    }
+    if (raw_events & EPOLLOUT) {
+         events |= UCS_EVENT_SET_EVWRITE;
+    }
+    return events;
+}
+
+ucs_status_t ucs_event_set_create(ucs_sys_event_set_t **event_set_p)
+{
+    ucs_sys_event_set_t *event_set;
+    ucs_status_t status;
+
+    event_set = ucs_malloc(sizeof(ucs_sys_event_set_t), "ucs_sys_event_set");
+    if (event_set == NULL) {
+        status = UCS_ERR_NO_MEMORY;
+        goto out_create;
+    }
+
+    /* Create epoll set the thread will wait on */
+    event_set->epfd = epoll_create(1);
+    if (event_set->epfd < 0) {
+        ucs_error("epoll_create() failed: %m");
+        status = UCS_ERR_IO_ERROR;
+        goto err_free;
+    }
+
+    *event_set_p = event_set;
+    return UCS_OK;
+
+err_free:
+    ucs_free(event_set);
+out_create:
+    return status;
+}
+
+ucs_status_t ucs_event_set_add(ucs_sys_event_set_t *event_set, int event_fd,
+                               ucs_event_set_type_t events)
+{
+    struct epoll_event raw_event;
+    int ret;
+
+    memset(&raw_event, 0, sizeof(raw_event));
+    raw_event.events   = ucs_event_set_map_to_raw_events(events);
+    raw_event.data.fd  = event_fd;
+
+    ret = epoll_ctl(event_set->epfd, EPOLL_CTL_ADD, event_fd, &raw_event);
+    if (ret < 0) {
+        ucs_error("epoll_ctl(epfd=%d, ADD, fd=%d) failed: %m", event_set->epfd,
+                  event_fd);
+        return UCS_ERR_IO_ERROR;
+    }
+
+    return UCS_OK;
+}
+
+ucs_status_t ucs_event_set_mod(ucs_sys_event_set_t *event_set, int event_fd,
+                               ucs_event_set_type_t events)
+{
+    struct epoll_event raw_event;
+    int ret;
+
+    memset(&raw_event, 0, sizeof(raw_event));
+    raw_event.events   = ucs_event_set_map_to_raw_events(events);
+    raw_event.data.fd  = event_fd;
+
+    ret = epoll_ctl(event_set->epfd, EPOLL_CTL_MOD, event_fd, &raw_event);
+    if (ret < 0) {
+        ucs_error("epoll_ctl(epfd=%d, MOD, fd=%d) failed: %m", event_set->epfd,
+                  event_fd);
+        return UCS_ERR_IO_ERROR;
+    }
+
+    return UCS_OK;
+}
+
+ucs_status_t ucs_event_set_del(ucs_sys_event_set_t *event_set, int event_fd)
+{
+    int ret;
+
+    ret = epoll_ctl(event_set->epfd, EPOLL_CTL_DEL, event_fd, NULL);
+    if (ret < 0) {
+        ucs_error("epoll_ctl(epfd=%d, DEL, fd=%d) failed: %m", event_set->epfd,
+                  event_fd);
+        return UCS_ERR_IO_ERROR;
+    }
+
+    return UCS_OK;
+}
+
+ucs_status_t ucs_event_set_wait(ucs_sys_event_set_t *event_set, int timeout_ms,
+                                event_set_handler_t event_set_handler)
+{
+    int nready;
+    struct epoll_event ep_events[UCS_EVENT_EPOLL_MAX_EVENTS];
+    int i;
+
+    if (event_set_handler == NULL) {
+        return UCS_ERR_INVALID_PARAM;
+    }
+
+    nready = epoll_wait(event_set->epfd, ep_events, UCS_EVENT_EPOLL_MAX_EVENTS,
+                        timeout_ms);
+    if ((nready < 0) && (errno != EINTR)) {
+        ucs_error("epoll_wait() failed: %m");
+        return UCS_ERR_IO_ERROR;
+    }
+    ucs_trace_data("epoll_wait(epfd=%d, timeout=%d) returned %d",
+                   event_set->epfd, timeout_ms, nready);
+
+    for (i=0; i < nready; i++) {
+        int events = ucs_event_set_map_to_events(ep_events[i].events);
+        event_set_handler(ep_events[i].data.fd, events);
+    }
+    return UCS_OK;
+}
+
+void ucs_event_set_cleanup(ucs_sys_event_set_t *event_set)
+{
+    close(event_set->epfd);
+    ucs_free(event_set);
+}

src/ucs/sys/event_set.h

@@ -0,0 +1,98 @@
+/**
+ * Copyright (C) Hiroyuki Sato. 2019. ALL RIGHTS RESERVED.
+ *
+ * See file LICENSE for terms.
+ */
+
+#ifndef UCS_EVENT_SET_H
+#define UCS_EVENT_SET_H
+
+#include <ucs/type/status.h>
+
+/**
+ * ucs_sys_event_set structure used in ucs_event_set_XXX functions.
+ *
+ */
+typedef struct ucs_sys_event_set ucs_sys_event_set_t;
+
+
+/**
+ * ucs_event_set_handler call this handler for notifying event
+ *
+ */
+typedef void (*event_set_handler_t)(int fd, int event);
+
+/**
+ * ucs_event_set_type_t member is a bit set composed using the following
+ * available event types
+ */
+typedef enum {
+    UCS_EVENT_SET_EVREAD  = UCS_BIT(0),
+    UCS_EVENT_SET_EVWRITE = UCS_BIT(1),
+    UCS_EVENT_SET_EVNONE =  UCS_BIT(2)
+} ucs_event_set_type_t;
+
+/**
+ * Allocate ucs_sys_event_set_t structure.
+ *
+ * @param [out] event_set_p  Event set pointer to initialize.
+ *
+ * @return UCS_OK on success or an error code on failure.
+ */
+ucs_status_t ucs_event_set_create(ucs_sys_event_set_t **event_set_p);
+
+/**
+ * Register the target event.
+ *
+ * @param [in] event_set_p  Event set pointer to initialize.
+ * @param [in] event_fd     Register the target file descriptor fd.
+ * @param [in] events       Operation events.
+ *
+ * @return UCS_OK on success or an error code on failure.
+ */
+ucs_status_t ucs_event_set_add(ucs_sys_event_set_t *event_set, int event_fd,
+                               ucs_event_set_type_t events);
+
+/**
+ * Modify the target event.
+ *
+ * @param [in] event_set    Event set created by ucs_event_set_create.
+ * @param [in] event_fd     Register the target file descriptor fd.
+ * @param [in] events       Operation events.
+ *
+ * @return UCS_OK on success or an error code on failure.
+ */
+ucs_status_t ucs_event_set_mod(ucs_sys_event_set_t *event_set, int event_fd,
+                               ucs_event_set_type_t events);
+
+/**
+ * Remove the target event.
+ *
+ * @param [in] event_set    Event set created by ucs_event_set_create.
+ * @param [in] event_fd     Register the target file descriptor fd.
+ *
+ * @return UCS_OK on success or an error code on failure.
+ */
+ucs_status_t ucs_event_set_del(ucs_sys_event_set_t *event_set, int event_fd);
+
+/**
+ * Wait for an I/O events
+ *
+ * @param [in] event_set          Event set created by ucs_event_set_create.
+ * @param [in] timeout_ms         Timeout period in ms.
+ * @param [in] event_set_handler  Callback functions.
+ *
+ * @return UCS_OK on success or an error code on failure.
+ */
+ucs_status_t ucs_event_set_wait(ucs_sys_event_set_t *event_set, int timeout_ms,
+                                event_set_handler_t event_set_handler);
+
+/**
+ * Cleanup event set
+ *
+ * @param [in] event_set    Event set created by ucs_event_set_create.
+ *
+ */
+void ucs_event_set_cleanup(ucs_sys_event_set_t *event_set);
+
+#endif

test/gtest/Makefile.am

@@ -100,6 +100,7 @@ gtest_SOURCES = \
 	uct/uct_p2p_test.cc \
 	uct/uct_test.cc \
 	uct/test_stats.cc \
+	ucs/test_event_set.cc \
 	ucs/test_stats_filter.cc \
 	uct/test_peer_failure.cc \
 	uct/test_tag.cc \