sched: add a registration-based scheduler to argobots

examples/scheduling/Makefile.am

@@ -8,7 +8,8 @@ TESTS = \
 	sched_predef \
 	sched_shared_pool \
 	sched_stack \
-	sched_user
+	sched_user \
+	sched_reg
 
 check_PROGRAMS = $(TESTS)
 noinst_PROGRAMS = $(TESTS)
@@ -20,3 +21,4 @@ sched_predef_SOURCES = sched_predef.c
 sched_shared_pool_SOURCES = sched_shared_pool.c
 sched_stack_SOURCES = sched_stack.c
 sched_user_SOURCES = sched_user.c
+sched_reg_SOURCES = sched_reg.c

examples/scheduling/sched_reg.c

@@ -0,0 +1,146 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include "abt.h"
+
+#define NUM_XSTREAMS 4
+#define NUM_THREADS 4
+
+
+static void create_threads(void *arg);
+static void thread_hello(void *arg);
+
+int main(int argc, char *argv[])
+{
+    ABT_xstream xstreams[NUM_XSTREAMS];
+    ABT_sched scheds[NUM_XSTREAMS];
+    ABT_pool pools[NUM_XSTREAMS];
+    ABT_thread threads[NUM_XSTREAMS];
+    int i;
+
+    ABT_init(argc, argv);
+
+    /* Create pools */
+    for (i = 0; i < NUM_XSTREAMS; i++) {
+        ABT_pool_create_basic(ABT_POOL_FIFO, ABT_POOL_ACCESS_MPMC, ABT_TRUE,
+                              &pools[i]);
+    }
+
+    /* Create schedulers */
+    //create_scheds(NUM_XSTREAMS, pools, scheds);
+    ABT_sched_policy policies[NUM_XSTREAMS];
+    for (int i = 0; i < NUM_XSTREAMS; i++) {
+      policies[i].ready_at = 0;
+      policies[i].ready_at_wt = ABT_get_wtime();
+      policies[i].priority = i;
+      policies[i].max_successes = 10;
+      policies[i].min_successes = 2 * i + 2;
+      policies[i].success_timeout = 5;
+      policies[i].fail_timeout = 100;
+      policies[i].success_timeout_wt = 0.0;
+      policies[i].fail_timeout_wt = 5.0;
+    }
+
+    policies[0].max_successes = 100;
+    policies[0].fail_timeout_wt=10.0;
+    for (int i = 0; i < NUM_XSTREAMS; i++) {
+      for (int j = 0; j < NUM_XSTREAMS;j++) {
+        policies[(i+j) % NUM_XSTREAMS].priority = j;
+      }
+      ABT_sched_create_reg(NUM_XSTREAMS, pools, NUM_XSTREAMS, policies, &scheds[i]);
+    }
+
+    /* Create ESs */
+    ABT_xstream_self(&xstreams[0]);
+    ABT_xstream_set_main_sched(xstreams[0], scheds[0]);
+    for (i = 1; i < NUM_XSTREAMS; i++) {
+        ABT_xstream_create(scheds[i], &xstreams[i]);
+    }
+
+    /* Create ULTs */
+    for (i = 0; i < NUM_XSTREAMS; i++) {
+        size_t tid = (size_t)i;
+        ABT_thread_create(pools[i], create_threads, (void *)tid,
+                          ABT_THREAD_ATTR_NULL, &threads[i]);
+    }
+    /* Join & Free */
+    for (i = 0; i < NUM_XSTREAMS; i++) {
+        ABT_thread_join(threads[i]);
+        ABT_thread_free(&threads[i]);
+    }
+    for (i = 1; i < NUM_XSTREAMS; i++) {
+        ABT_xstream_join(xstreams[i]);
+        ABT_xstream_free(&xstreams[i]);
+    }
+    /* Free schedulers */
+    /* Note that we do not need to free the scheduler for the primary ES,
+     * i.e., xstreams[0], because its scheduler will be automatically freed in
+     * ABT_finalize(). */
+    for (i = 1; i < NUM_XSTREAMS; i++) {
+        ABT_sched_free(&scheds[i]);
+    }
+
+    /* Finalize */
+    ABT_finalize();
+
+    return 0;
+}
+
+static void create_threads(void *arg)
+{
+    int i, rank, tid = (int)(size_t)arg;
+    ABT_xstream xstream;
+    ABT_pool pools[NUM_XSTREAMS];
+    ABT_thread *threads;
+
+    ABT_xstream_self(&xstream);
+    ABT_xstream_get_main_pools(xstream, NUM_XSTREAMS, pools);
+
+    ABT_xstream_get_rank(xstream, &rank);
+    printf("[U%d:E%d] creating ULTs\n", tid, rank);
+
+    threads = (ABT_thread *)malloc(sizeof(ABT_thread) * NUM_THREADS);
+    for (i = 0; i < NUM_THREADS; i++) {
+        size_t id = (rank + 1) * 10 + i;
+        ABT_thread_create(pools[rank], thread_hello, (void *)id, ABT_THREAD_ATTR_NULL,
+                          &threads[i]);
+    }
+
+    ABT_xstream_get_rank(xstream, &rank);
+    printf("[U%d:E%d] freeing ULTs\n", tid, rank);
+    for (i = 0; i < NUM_THREADS; i++) {
+        ABT_thread_free(&threads[i]);
+    }
+    free(threads);
+}
+
+static void thread_hello(void *arg)
+{
+    int tid = (int)(size_t)arg;
+    int old_rank, cur_rank;
+    char *msg;
+
+    ABT_xstream_self_rank(&cur_rank);
+
+    printf("  [U%d:E%d] Hello, world!\n", tid, cur_rank);
+
+    ABT_thread_yield();
+
+    old_rank = cur_rank;
+    ABT_xstream_self_rank(&cur_rank);
+    msg = (cur_rank == old_rank) ? "" : " (stolen)";
+    printf("  [U%d:E%d] Hello again.%s\n", tid, cur_rank, msg);
+
+    ABT_thread_yield();
+
+    old_rank = cur_rank;
+    ABT_xstream_self_rank(&cur_rank);
+    msg = (cur_rank == old_rank) ? "" : " (stolen)";
+    printf("  [U%d:E%d] Goodbye, world!%s\n", tid, cur_rank, msg);
+}

src/include/abt.h.in

@@ -2309,6 +2309,87 @@ typedef void (*ABT_tool_thread_callback_fn)(ABT_thread, ABT_xstream, uint64_t ev
 typedef void (*ABT_tool_task_callback_fn)(ABT_task, ABT_xstream, uint64_t event,
                                           ABT_tool_context context, void *user_arg);
 
+/**
+ * @ingroup SCHED
+ * @brief   A struct that defines the scheduling policy for an associated pool.
+ */
+typedef struct {
+  /**
+   * @brief Pool priority.
+   *
+   * This value determines the relative priority of the associated pool when
+   * the scheduler is selecting between multiple ready pools. The scheduler
+   * will favor pools with lower priority. Value must be non-negative.
+   */
+  int priority;
+  /**
+   * @brief Minimum number of successful attempts.
+   *
+   * This value determines how many work units will be pulled from this pool
+   * before returning to the scheduler to check for interrupts and/or higher
+   * priority pools that may have become ready.
+   */
+  int min_successes;
+  /**
+   * @brief Maximum number of successful attempts.
+   *
+   * This value determines the maximum number of work units will be executed
+   * from this pool before the pool is timed-out to allow for the scheduler
+   * to check other pools.
+   */
+  int max_successes;
+  /**
+   * @brief Scheduler event number at which this pool becomes ready.
+   *
+   * This value determines the point in time that this pool becomes ready
+   * after being timed-out. It is specified in terms of number of work units
+   * that the scheduler has executed.
+   */
+  int ready_at;
+  /**
+   * @brief Timeout value when the scheduler fails to pop a work unit.
+   *
+   * This value determines how many work units the scheduler should execute
+   * from OTHER pools before this pool becomes ready again in the case where
+   * the scheduler fails to execute a work unit from this pool (because the
+   * pool has become empty).
+   */
+  int fail_timeout;
+  /**
+   * @brief Timeout value after the maximum number of successful pops.
+   *
+   * This value determines how many work units the scheduler should execute
+   * from OTHER pools before this pool becomes ready again in the case where
+   * the scheduler has executed the maximum number of events from this pool,
+   * as specified by max_successes.
+   */
+  int success_timeout;
+  /**
+   * @brief Walltime at which this pool becomes ready after a timeout.
+   *
+   * This value determines the walltime at which this pool becomes ready
+   * again after a timeout. It works in conjunction with ready_at. For a pool
+   * to be ready, it must have reached both the ready_at_wt walltime and the
+   * ready_at scheduler event number.
+   */
+  double ready_at_wt;
+  /**
+   * @brief Timeout value (in seconds) when the scheduler fails to pop.
+   *
+   * This value determines how many seconds this pool will be timed out
+   * in the case where the scheduler fails to pop an work unit from this pool.
+   */
+  double fail_timeout_wt;
+  /**
+   * @brief Timeout value (in seconds) after the maximum successful pops.
+   *
+   * This value determines how many seconds this pool will be timed out
+   * in the case where the scheduler fails pops the maximum number of work
+   * units from this pool as determined by max_successes.
+   */
+  double success_timeout_wt;
+} ABT_sched_policy;
+
 /* Init & Finalize */
 int ABT_init(int argc, char **argv) ABT_API_PUBLIC;
 int ABT_finalize(void) ABT_API_PUBLIC;
@@ -2363,6 +2444,9 @@ int ABT_sched_create(ABT_sched_def *def, int num_pools, ABT_pool *pools,
 int ABT_sched_create_basic(ABT_sched_predef predef, int num_pools,
                            ABT_pool *pools, ABT_sched_config config,
                            ABT_sched *newsched) ABT_API_PUBLIC;
+void ABT_sched_create_reg(int num_pools, ABT_pool *pools,
+                          int num_policies, ABT_sched_policy *policies,
+                          ABT_sched *newsched) ABT_API_PUBLIC;
 int ABT_sched_free(ABT_sched *sched) ABT_API_PUBLIC;
 int ABT_sched_set_data(ABT_sched sched, void *data) ABT_API_PUBLIC;
 int ABT_sched_get_data(ABT_sched sched, void **data) ABT_API_PUBLIC;

src/sched/Makefile.mk

@@ -8,6 +8,7 @@ abt_sources += \
 	sched/basic_wait.c \
 	sched/prio.c \
 	sched/randws.c \
+	sched/reg.c \
 	sched/sched.c \
 	sched/sched_config.c