October 2011 - Triton-commits - mailman3.cels.anl.gov

Triton Repository branch, master, updated. 563e85f9de7dc7aa6a839f4bc890632003592c05
by noreply＠mcs.anl.gov 28 Oct '11

28 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 563e85f9de7dc7aa6a839f4bc890632003592c05 (commit) via 4d637ede4422b6876ef89d1be9d2cef870e149cc (commit) via c355b1a35ac588284ada55b9f15b9b98bb84f7a8 (commit) via 02c3bdd1d8f25a8b82191eb30bc4362f2022bb07 (commit) from ae7a0f41a867d9c8f11bcf792e4c326dfc6639f2 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 563e85f9de7dc7aa6a839f4bc890632003592c05 Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Fri Oct 28 21:30:32 2011 -0500 Rewrite signal resource to use threads - libev only supports signals in the default event loop - libev event loops are not thread safe. Combining the two made it hard to write a correct (i.e. thread safe) signal resource. Rewrote using sigwait/sigqueue. Note: for this resource to work correctly, all other aesop threads must block signals to ensure the signal resource is able to receive the signals. The easiest way to do this is to block signals in main(), *before* creating any other threads. commit 4d637ede4422b6876ef89d1be9d2cef870e149cc Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Fri Oct 28 20:57:58 2011 -0500 Some minor triton_list modifications - Add triton_list_destroy function (optional for now, but not a bad idea to use it!) - Make triton_list_peek_front function public. commit c355b1a35ac588284ada55b9f15b9b98bb84f7a8 Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Fri Oct 28 20:57:26 2011 -0500 Add some documentation to opcache.h commit 02c3bdd1d8f25a8b82191eb30bc4362f2022bb07 Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Fri Oct 28 17:21:27 2011 -0500 Signal resource Old code I forgot to commit. ----------------------------------------------------------------------- Summary of changes: code/configure.ac | 2 + code/src/aesop/opcache.h | 8 +- code/src/common/resources/signal/module.mk.in | 5 + code/src/common/resources/signal/signal.c | 352 ++++++++++++++++++++ code/src/common/resources/signal/signal.hae | 19 + code/src/common/resources/signal/test/module.mk.in | 5 + code/src/common/resources/signal/test/simple.ae | 91 +++++ code/src/common/triton-init-registry-lite.c | 2 + code/src/common/triton-init-registry.c | 2 + code/src/common/triton-list.h | 16 + 10 files changed, 501 insertions(+), 1 deletions(-) create mode 100644 code/src/common/resources/signal/module.mk.in create mode 100644 code/src/common/resources/signal/signal.c create mode 100644 code/src/common/resources/signal/signal.hae create mode 100644 code/src/common/resources/signal/test/module.mk.in create mode 100644 code/src/common/resources/signal/test/simple.ae Diff of changes: diff --git a/code/configure.ac b/code/configure.ac index 020d92a..e029616 100644 --- a/code/configure.ac +++ b/code/configure.ac @@ -399,6 +399,8 @@ src/transactional-osd/module.mk src/transactional-osd/tests/module.mk src/common/resources/bdb/module.mk src/common/resources/file/module.mk +src/common/resources/signal/module.mk +src/common/resources/signal/test/module.mk src/replicated-osd/module.mk src/replicated-osd/tests/module.mk src/mapping/module.mk diff --git a/code/src/aesop/opcache.h b/code/src/aesop/opcache.h index 4e2be67..d7572fc 100644 --- a/code/src/aesop/opcache.h +++ b/code/src/aesop/opcache.h @@ -12,6 +12,8 @@ typedef struct ae_opcache *ae_opcache_t; /* complete an operation by returning it to opcache and invoking callback */ /* note that __error_code is copied on purpose for safety in case it is a * member of the __op structure. + * + * ae_opcache_complete_op is thread-safe with respect to the opcache. */ #define ae_opcache_complete_op(__opcache, __op, __ret_type, __error_code) do { \ void (*__callback)(void *, __ret_type) = (__op)->callback; \ @@ -52,13 +54,17 @@ void ae_opcache_destroy(ae_opcache_t cache); /** * Obtain an ae_op entry. The entry will have a valid - * op->cache_id + * op->cache_id. + * + * ae_opcache_get is thread-safe. */ struct ae_op *ae_opcache_get(ae_opcache_t cache); /** * Return ae_op entry to the cache + * + * ae_opcache_put is thread-safe. */ void ae_opcache_put(ae_opcache_t cache, struct ae_op *op); diff --git a/code/src/common/resources/signal/module.mk.in b/code/src/common/resources/signal/module.mk.in new file mode 100644 index 0000000..ec6d92c --- /dev/null +++ b/code/src/common/resources/signal/module.mk.in @@ -0,0 +1,5 @@ +DIR := src/common/resources/signal + +AESOP_HDR += $(DIR)/signal.hae + +LIBSRC += $(DIR)/signal.c diff --git a/code/src/common/resources/signal/signal.c b/code/src/common/resources/signal/signal.c new file mode 100644 index 0000000..d6424e6 --- /dev/null +++ b/code/src/common/resources/signal/signal.c @@ -0,0 +1,352 @@ +#include "src/aesop/aesop.h" +#include "src/aesop/op.h" +#include "src/aesop/opcache.h" +#include "src/common/resources/signal/signal.h" +#include "src/common/triton-init.h" +#include "src/common/triton-debug.h" + +#include <signal.h> +#include <sys/types.h> +#include <assert.h> +#include <pthread.h> +#include <unistd.h> + +/* Size of initial opcache */ +#define SIGNAL_DEFAULT_SIZE 1 + +/* NOTE: not using libev since libev only supports signals in the default + * event loop. However, since the event loops are not thread safe it is hard + * to share the default loop. + */ + +static ae_opcache_t signal_opcache = NULL; +static int triton_signal_resource_id; + +static pthread_t signal_thread; +static int signal_thread_pid; +static triton_debug_mask_t signal_debug_mask; + +static triton_mutex_t signal_mutex = TRITON_MUTEX_INITIALIZER; + +#define WAKEUP_SIGNAL SIGUSR1 + +typedef struct +{ + ae_op_id_t op_id; + struct ae_op op; + triton_ret_t tret; + int signal; + triton_list_link_t link; +} signal_op_t; + + +static int signal_shutdown; +static triton_list_t signal_completed; +static triton_list_t signal_active; +static sigset_t signal_set; + +// -------------------------------------------------------------------------- + +static void * signal_thread_main (void * data); + +static void wakeup_thread () +{ + union sigval val; + val.sival_ptr = &signal_thread_main; + + // Note: both functions are possible + // -- not sure which is faster + sigqueue (signal_thread_pid, WAKEUP_SIGNAL, val); + // pthread_sigqueue (&signal_thread, WAKEUP_SIGNAL, val); +} + + +ae_define_post(triton_ret_t, triton_signal_wait, int signum) +{ + struct ae_op * op; + signal_op_t * new_op; + + if(!signal_opcache) + { + triton_err(triton_log_default, "Error: signal resource not initialized."); + assert(0); + } + + op = ae_opcache_get(signal_opcache); + ae_op_fill(op); + + new_op = ae_op_entry(op, signal_op_t, op); + new_op->op_id = ae_id_gen(triton_signal_resource_id, (intptr_t) op); + new_op->tret = TRITON_SUCCESS; + + *__ae_op_id = new_op->op_id; + + + triton_mutex_lock (&signal_mutex); + + triton_list_add_back (&new_op->link, &signal_active); + + // Add the signal to the set of watched signals. + sigaddset (&signal_set, signum); + + triton_mutex_unlock (&signal_mutex); + + wakeup_thread (); + + return TRITON_SUCCESS; +} + +static triton_ret_t triton_signal_poll(ae_context_t context) +{ + triton_mutex_lock (&signal_mutex); + + if (triton_list_empty (&signal_completed)) + { + triton_mutex_unlock (&signal_mutex); + return TRITON_SUCCESS; + } + + // Swap the list and release the lock + triton_list_t todo = signal_completed; + triton_list_init (&signal_completed); + + triton_mutex_unlock (&signal_mutex); + + + // Now call all the callbacks + + struct triton_list_link * current; + struct triton_list_link * safe; + + triton_list_for_each (current, safe, &todo) + { + signal_op_t * sig = triton_list_get_entry (current, + signal_op_t, link); + + // calls callback and return op entry to the cache + ae_opcache_complete_op(signal_opcache, &sig->op, triton_ret_t, + sig->tret); + } + + triton_list_destroy (&todo); + + return TRITON_SUCCESS; +} + +static triton_ret_t triton_signal_cancel(ae_context_t triton_ctx, + ae_op_id_t op_id) +{ + int resource_id; + struct ae_op * op; + ae_context_t ctx; + + // Need to lock to prevent completion of the op + triton_mutex_lock(&signal_mutex); + + op = intptr2op (ae_id_lookup(op_id, &resource_id)); + assert(resource_id == triton_signal_resource_id); + signal_op_t * sig = ae_op_entry(op, signal_op_t, op); + + // Check if the operation is still active + if (!triton_list_exists (&signal_active, &sig->link)) + { + triton_mutex_unlock (&signal_mutex); + return TRITON_ERR_NOENT; + } + + // need to make a copy here since as soon as we release the lock this op + // might go away. + ae_context_t context = sig->op.ctx; + + sig->tret = TRITON_ERR_CANCELED; + triton_list_del (&sig->link); + triton_list_add_back (&sig->link, &signal_completed); + + triton_mutex_unlock (&signal_mutex); + + // We don't complete operations (see cancel docs) so request a poll + ae_resource_request_poll(context, triton_signal_resource_id); + + return TRITON_SUCCESS; +} + + +struct ae_resource triton_signal_resource = +{ + .resource_name = "signal", + .poll_context = triton_signal_poll, + .cancel = triton_signal_cancel +}; + +__attribute__((constructor)) void triton_signal_init_register(void); + +__attribute__((constructor)) void triton_signal_init_register(void) +{ + triton_init_register("triton.resource.signal", triton_signal_init, + triton_signal_finalize, NULL, "aesop.control"); +} + + +static void * signal_thread_main (void * d) +{ + const int parentpid = (int) (intptr_t) d; + signal_thread_pid = getpid (); + const int posix_compliant = (signal_thread_pid == parentpid); + + triton_debug (signal_debug_mask, "Signal resource thread starting..."); + + if (!posix_compliant) + { + triton_debug (signal_debug_mask, "Thread implementation is not POSIX " + "compliant!"); + } + + + while (!signal_shutdown) + { + sigset_t currentset; + + triton_mutex_lock (&signal_mutex); + currentset = signal_set; + triton_mutex_unlock (&signal_mutex); + + // Always listen for WAKEUP_SIGNAL, we use it internally + sigaddset (&currentset, WAKEUP_SIGNAL); + + siginfo_t info; + + const int signum = sigwaitinfo (&currentset, &info); + if (signum < 0) + { + assert (errno == EINTR); + // Got interrupted because of a signal we're not monitoring. + // Try again. + continue; + } + + // Check if this was an internal signal + // POSIX says that all threads share the same PID, + // but unfortunately the old LinuxThreads implementation does not + // follow this. (NTPL is OK) + // + // In non-compliant thread implementations, there's a small change + // we'll receive a signal and mistakenly assume it is an internal + // signal. + const int internal_signal = + (signum == WAKEUP_SIGNAL) + && (info.si_ptr == &signal_thread_main) + && (posix_compliant ? info.si_pid == signal_thread_pid : 1); + + if (internal_signal) + { + // Internal signal; Simply restart processing + continue; + } + + // ret contains the signal number + triton_list_link_t * current; + triton_list_link_t * dummy; + + + triton_mutex_lock (&signal_mutex); + + // We'll recalculate the signal set while searching for completed + // operations + sigemptyset (&signal_set); + + triton_list_for_each (current, dummy, &signal_active) + { + signal_op_t * op = triton_list_get_entry (current, signal_op_t, link); + if (op->signal != signum) + { + sigaddset (&signal_set, op->signal); + continue; + } + + // Remove entry and add to completed list + triton_list_del (current); + triton_list_add_back (current, &signal_completed); + } + + const int need_poll = !triton_list_empty (&signal_completed); + + ae_context_t context; + + if (need_poll) + { + const signal_op_t * sig = triton_list_get_entry ( + triton_list_peek_front (&signal_completed), + signal_op_t, + link); + + context = sig->op.ctx; + } + + triton_mutex_unlock (&signal_mutex); + + if (need_poll) + { + // Note: we don't keep track of specific context - poll will complete + // them all + ae_resource_request_poll (context, triton_signal_resource_id); + } + }; + + triton_debug (signal_debug_mask, "Signal resource thread exiting..."); +} + + +triton_ret_t triton_signal_init (void) +{ + triton_ret_t tret; + int ret; + + triton_debug_add_mask ("triton.signal", &signal_debug_mask, + "Signal resource"); + + tret = AE_OPCACHE_INIT (signal_op_t, op, SIGNAL_DEFAULT_SIZE, + &signal_opcache); + if (tret != TRITON_SUCCESS) + { + return tret; + } + + signal_shutdown = 0; + triton_list_init (&signal_active); + triton_list_init (&signal_completed); + + sigemptyset (&signal_set); + + // We pass out pid to enable the new thread to check if the thread + // implementation is posix compliant. + pthread_create (&signal_thread, NULL, &signal_thread_main, + (void*) (intptr_t) getpid ()); + + return ae_resource_register (&triton_signal_resource, + &triton_signal_resource_id); +} + +void triton_signal_finalize (void) +{ + triton_debug (signal_debug_mask, "Waiting for resource thread to exit..."); + + signal_shutdown = 1; + wakeup_thread (); + + pthread_join (signal_thread, NULL); + + triton_list_destroy (&signal_active); + triton_list_destroy (&signal_completed); + + ae_resource_unregister(triton_signal_resource_id); + ae_opcache_destroy(signal_opcache); +} + +/* + * Local variables: + * c-indent-level: 4 + * c-basic-offset: 4 + * End: + * + * vim: ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/common/resources/signal/signal.hae b/code/src/common/resources/signal/signal.hae new file mode 100644 index 0000000..c479888 --- /dev/null +++ b/code/src/common/resources/signal/signal.hae @@ -0,0 +1,19 @@ +#ifndef TRITON_SIGNAL_SIGNAL_HAE +#define TRITON_SIGNAL_SIGNAL_HAE + +#include "src/aesop/aesop.h" + + +/** + * Block until the specified signal is delivered. + * This function can be called multiple times for the same signal number. + * When a signal arrives, all callers waiting in signal_wait will be released. + */ +__blocking triton_ret_t triton_signal_wait (int signum); + +triton_ret_t triton_signal_init (void); + +void triton_signal_finalize (void); + +#endif + diff --git a/code/src/common/resources/signal/test/module.mk.in b/code/src/common/resources/signal/test/module.mk.in new file mode 100644 index 0000000..50e4453 --- /dev/null +++ b/code/src/common/resources/signal/test/module.mk.in @@ -0,0 +1,5 @@ +DIR := src/common/resources/signal/test + +AETESTSRC += $(DIR)/simple.ae + + diff --git a/code/src/common/resources/signal/test/simple.ae b/code/src/common/resources/signal/test/simple.ae new file mode 100644 index 0000000..9539257 --- /dev/null +++ b/code/src/common/resources/signal/test/simple.ae @@ -0,0 +1,91 @@ +#include "src/aesop/aesop.h" +#include "src/common/resources/timer/timer.hae" +#include "src/common/resources/signal/signal.hae" +#include "src/aesop/aesop-support.hae" + +#include <signal.h> +#include <sys/types.h> +#include <unistd.h> + +__blocking int signalself () +{ + int ret; + + ret = 0; + + pwait + { + pbranch + { + printf ("Waiting for SIGINT...\n"); + triton_signal_wait (SIGINT); + aesop_cancel_branches_wait(); + } + + pbranch + { + triton_ret_t tret; + + triton_timer (500); + printf ("Sending SIGINT...\n"); + kill (getpid (), SIGINT); + tret = triton_timer (1000); + if (!triton_error_equal (tret, TRITON_ERR_CANCELED)) + { + fprintf (stderr, "Did *NOT* receive signal! Cancelling test!\n"); + ret = 1; + } + } + } + + printf ("All done!\n"); + return ret; +} + +__blocking int test_cancel () +{ + int ret; + + ret = 1; + + pwait + { + pbranch + { + triton_ret_t tret; + + printf ("Waiting for SIGINT...\n"); + tret = triton_signal_wait (SIGINT); + if (triton_error_equal (tret, TRITON_ERR_CANCELED)) + { + printf ("Signal wait cancelled :-)\n"); + ret = 0; + } + aesop_cancel_branches_wait(); + } + + pbranch + { + triton_ret_t tret; + + triton_timer (500); + printf ("Sending SIGINT...\n"); + aesop_cancel_branches_wait (); + } + } + return ret; +} + +__blocking int aesop_main(int argc, char **argv) +{ + triton_signal_init (); + + int ret; + ret = signalself(); + ret = ret || test_cancel (); + return ret; +} + +aesop_main_set_with_init (NULL, "triton.core", aesop_main); + + diff --git a/code/src/common/triton-init-registry-lite.c b/code/src/common/triton-init-registry-lite.c index 36590d9..e394703 100644 --- a/code/src/common/triton-init-registry-lite.c +++ b/code/src/common/triton-init-registry-lite.c @@ -17,6 +17,7 @@ void triton_debug_init_register(void); void triton_zeroconf_init_register(void); void triton_file_init_register(void); void triton_aesocket_init_register(void); +void triton_signal_init_register (void); void triton_init_register_components(void); @@ -29,4 +30,5 @@ void triton_init_register_components(void) triton_zeroconf_init_register(); triton_file_init_register(); triton_aesocket_init_register(); + triton_signal_init_register (); } diff --git a/code/src/common/triton-init-registry.c b/code/src/common/triton-init-registry.c index fb1ae9e..0ae8877 100644 --- a/code/src/common/triton-init-registry.c +++ b/code/src/common/triton-init-registry.c @@ -36,6 +36,7 @@ void triton_kv_init_register(void); void triton_logical_register(void); void triton_aesocket_init_register(void); void triton_abt_init_register(void); +void triton_signal_init_register (void); void triton_init_register_components(void); @@ -68,4 +69,5 @@ void triton_init_register_components(void) triton_logical_register(); triton_aesocket_init_register(); triton_abt_init_register(); + triton_signal_init_register(); } diff --git a/code/src/common/triton-list.h b/code/src/common/triton-list.h index 650744e..ac2726b 100644 --- a/code/src/common/triton-list.h +++ b/code/src/common/triton-list.h @@ -97,6 +97,16 @@ static inline int triton_list_empty(triton_list_t *list) return list->count == 0; } + +/** + * Destroy the list. + * List must be empty + */ +static inline void triton_list_destroy (triton_list_t * list) +{ + assert (triton_list_empty (list)); +} + static inline uint64_t triton_list_count(triton_list_t *list) { return list->count; @@ -215,6 +225,12 @@ static inline struct triton_list_link *__triton_list_peek_front(triton_list_t *l return list->entries.next; } +static inline struct triton_list_link * triton_list_peek_front ( + triton_list_t * list) +{ + return __triton_list_peek_front (list); +} + #define triton_queue_enqueue(__entry, __list) triton_list_add_back((__entry), (__list)) #define triton_queue_dequeue(__list) __triton_list_del_front(__list) #define triton_queue_peek(__list) __triton_list_peek_front(__list) hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. ae7a0f41a867d9c8f11bcf792e4c326dfc6639f2
by noreply＠mcs.anl.gov 28 Oct '11

28 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via ae7a0f41a867d9c8f11bcf792e4c326dfc6639f2 (commit) via a149173588acfc84a8e2fcfabdb347963c37eb3c (commit) from f6cc376fec7c392d12c02ca16272eae32eabea90 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit ae7a0f41a867d9c8f11bcf792e4c326dfc6639f2 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 28 15:45:58 2011 -0400 added ascidoc for aesop pbranch cancel commit a149173588acfc84a8e2fcfabdb347963c37eb3c Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri Oct 28 14:08:41 2011 -0400 committing incremental progress before task switch ----------------------------------------------------------------------- Summary of changes: code/doc/aesop-pbranch-cancel.txt | 203 ++++++++++++++++++++++++++++++ code/doc/module.mk.in | 3 +- code/src/net/mpi/test/aesop-mpi-bench.ae | 68 +++++++++- code/src/net/mpi/test/mpi-bench.h | 123 ++++++++++++++++++ 4 files changed, 391 insertions(+), 6 deletions(-) create mode 100644 code/doc/aesop-pbranch-cancel.txt Diff of changes: diff --git a/code/doc/aesop-pbranch-cancel.txt b/code/doc/aesop-pbranch-cancel.txt new file mode 100644 index 0000000..712f5f2 --- /dev/null +++ b/code/doc/aesop-pbranch-cancel.txt @@ -0,0 +1,203 @@ +Aesop: how to cancel pbranches +============================== + +== Existing implementation: Oct. 2011 + +for some context, here is an example of how things are done currently +(this is a client doing an RPC): + +[source,c] +---- +pwait +{ + pbranch + { + ret = timer(); + /* if timer triggers, then cancel other branches */ + if(ret == SUCCESS) + aesop_cancel_branches_wait(); + } + pbranch + { + ret = recv(); + /* cancel other branches in any case except when we detect that + * another branch has already started cancelling. + */ + if(ret != CANCELLED) + aesop_cancel_branches_wait(); + } + pbranch + { + ret = send(); + /* cancel branches on send error except when we detect that another + * branch has already started cancelling. + */ + if(ret != CANCELLED && ret != SUCCESS) + aesop_cancel_branches_wait(); + } +} +---- + +The aesop_cancel_branches_wait() works by trying to cancel pbranches in +a loop (with a timer) until the pbranch count hits 1. It has to have a +loop because "signaling" the other pbranches is lossy; there is no +way to guarantee that a pbranch cancel will be honored. It might hit +a pbranch that isn't currently in a blocking call. + +Problems with the current implementation: + +* if two pbranches call cancel simultaneously, it will deadlock +* there are some race conditions re: if a pbranch will detect if it has been + cancelled or not +* the cancel implementation is kludgey and more expensive than it needs to + be because of the timer loop +* semantics are unclear, especially at the aesop resource level +* if the send happens to complete (successfully) after the recv completes, + due to network transport ordering issues, then the + aesop_cancel_branches_wait() call after the recv might be triggered too + early + +== New semantics + +=== Semantics of the aesop-level cancel: +* totally asynchronous (no longer blocks until pbranches exit) +* cancel signal to pbranch is guaranteed to be delivered exactly once +** will affect current resource operation (if present) or the next one to + be posted +** cancel status is cleared for a pbranch once it has impacted one blocking + resource call (so pbranch can issue blocking cleanup functions safely if + it needs to) +* cancel will return true if it succeeded in delivering the cancel request, + false if it detects that another pbranch *at the same scope or higher* + already issued a cancel +** it is Ok if a child pbranch already is already cancelled in its own + scope; just skip that one and return success anyway + +* caller can ignore cancel return code if it doesn't care (most codes will + not) +* illegal to call cancel from a lonely pbranch + +=== Semantics of the resource-level cancel function: +* totally asynchronous +* returns true if it delivered the cancel signal to the operation in + question (and the operation is guaranteed to return ECANCEL or + whatever is appropriate for the resource) +* returns false if it failed to cancel the operation in question +* cancel function can no longer issue op callback directly (this complicates + some possible deadlock scenarios) + +=== Aesop internal implementation details: +* serialize calls to cancel (so that only one pbranch will "win") +* set flag at the pwait level to indicate that the branches have been + cancelled (or return failure if flag already set) +* for each pbranch: +** atomically check for resource op id (and make sure no posts happen + concurrently) +*** if op id present, try resource cancel +**** if success, done +**** if fail, set flag in ctrl structure to say pbranch should cancel +*** if op id not present, set flag in ctrl structure to say pbranch should + cancel +** proceed recursively, setting flag in child pbranches in the same manner + +=== Semantics of the resource-level post functions: +* will have hook to check (and clear) parent ctrl structure cancel flag +* if detect that pbranch has cancelled, the post should immediately fail + with appropriate return code for resource + +=== Misc: +* provide helper functions is_cancelled() and clear_cancelled() so a pbranch + can check and clear it's cancelled status manually if it wants to +* One possible implementation idea is to treat child pwait blocks as + resources, so that they generate op ids and have a cancel function just + like a resource call. Might make the aesop code a little cleaner in + exchange for pushing some complexity elsewhere. Need to look at code to + see if this is worth doing or not. +* A possible extension to this document, would be to name each pbranch and + add the ability to cancel specific pbranches by name. Probably not that + hard to implement, but we don't have a clear use case yet. + +Example using new semantics: + +[source,c] +---- + +while(retry) +{ + /* the send and recv are grouped into a single function + * (do_communication) to make it clearer that we want both to complete + * before cancelling the timer. + * + * The aesop_cancel_branches() call is safe against race conditions, so + * neither pbranch has to check if it has been cancelled before + * initiating an aesop_cancel_branches() of its own. + */ + pwait + { + pbranch + { + ret = timer(); + aesop_cancel_branches(); + } + + pbranch + { + comm_ret = do_communication() + aesop_cancel_branches(); + } + } +} + + +do_communication() +{ + pwait + { + pbranch + { + recv_ret = recv(); + } + pbranch + { + send_ret = send(); + if(ret != SUCCESS) + aesop_cancel_branches(); + } + } + + return(whatever); +} + +---- + +Example resource cancel semantics: + +[source,c] +---- + +triton_ret_t resource_post(void) +{ + /* don't start op if we have a pending, undelivered cancel signal */ + if(test_and_clear_cancelled_state()) + return(ECANCELLED); + + /* do whatever .... */ +} + +triton_ret_t resource_cancel(op_id) +{ + /* lock to protect against post/callback/cancel races */ + + if(queued) + /* move to cancelled queue, return success */ + + if(callback already in progress) + /* return failure */ + + if(cant find operation) + /* return failure */ + +} + +---- + diff --git a/code/doc/module.mk.in b/code/doc/module.mk.in index 46769d8..831fca7 100644 --- a/code/doc/module.mk.in +++ b/code/doc/module.mk.in @@ -5,4 +5,5 @@ DOCSRC += $(DIR)/test.txt \ $(DIR)/fault-injection.txt \ $(DIR)/pipelining.txt \ $(DIR)/compiling-triton.txt \ - $(DIR)/aesop.txt + $(DIR)/aesop.txt \ + $(DIR)/aesop-pbranch-cancel.txt diff --git a/code/src/net/mpi/test/aesop-mpi-bench.ae b/code/src/net/mpi/test/aesop-mpi-bench.ae index 3a9d46b..1c69875 100644 --- a/code/src/net/mpi/test/aesop-mpi-bench.ae +++ b/code/src/net/mpi/test/aesop-mpi-bench.ae @@ -17,7 +17,14 @@ void parse_args (int argc, char **argv); #define NUM_PREPOSTED 64 static __blocking void handleConnection(int client_rank, int num_reqs); -static __blocking int dotest(int argc, char** argv); +static __blocking void dotest(void); + +static int done = 0; +static void cb(void *up) +{ + done = 1; + ae_poll_break(NULL); +} void usage(char *progname) { @@ -144,7 +151,7 @@ static __blocking void handleConnection(int client_rank, int num_reqs) }; } -static __blocking int dotest(int argc, char** argv) +static __blocking void dotest(void) { struct sigaction new_action; int clients; @@ -156,8 +163,6 @@ static __blocking int dotest(int argc, char** argv) sigaction(SIGINT, &new_action, NULL); sigaction(SIGQUIT, &new_action, NULL); - parse_args(argc, argv); - MPI_Comm_size(MPI_COMM_WORLD, &clients); clients--; @@ -174,10 +179,63 @@ static __blocking int dotest(int argc, char** argv) } } - return 0; + return; +} + +int main(int argc, char **argv) +{ + int rc; + int provided; + int rank; + triton_ret_t ret; + MPI_Group client_group, orig_group; + MPI_Comm client_comm; + int comm_size; + int server_rank = 0; + ae_op_id_t op_id; + ae_hints_t hints; + + rc = MPI_Init_thread(NULL, NULL, MPI_THREAD_MULTIPLE, &provided); + assert(rc == MPI_SUCCESS); + + MPI_Comm_rank(MPI_COMM_WORLD, &rank); + MPI_Comm_size(MPI_COMM_WORLD, &comm_size); + + /* TODO: shared routine to parse command line arguments... */ + + parse_args(argc, argv); + + /* set up communicator for clients */ + rc = MPI_Comm_group(MPI_COMM_WORLD, &orig_group); + assert(rc == MPI_SUCCESS); + rc = MPI_Group_excl(orig_group, 1, &server_rank, &client_group); + assert(rc == MPI_SUCCESS); + rc = MPI_Comm_create(MPI_COMM_WORLD, client_group, &client_comm); + assert(rc == MPI_SUCCESS); + + if(rank == 0) + { + /* initialize triton */ + ret = triton_init("triton.tosd.file", "triton.resource.timer", "triton.net.mpi"); + /* start server routine */ + ae_hints_init(&hints); + ret = ae_post_blocking(dotest, cb, NULL, &hints, NULL, &op_id); + while(!done) + { + ae_poll(NULL, 1000); + } + ae_hints_destroy(&hints); + } + else + { + /* clients */ + } + } +#if 0 aesop_main_set_with_init(NULL, "triton.echo-server", dotest, "timer", "aesocket", "file"); +#endif /* * Local variables: diff --git a/code/src/net/mpi/test/mpi-bench.h b/code/src/net/mpi/test/mpi-bench.h index 4dbaa6c..fba0399 100644 --- a/code/src/net/mpi/test/mpi-bench.h +++ b/code/src/net/mpi/test/mpi-bench.h @@ -1,6 +1,9 @@ #ifndef MPI_BENCH_H #define MPI_BENCH_H +#include <unistd.h> +#include "mpi.h" + enum test_type { TEST_READ = 1, @@ -41,6 +44,126 @@ static void handler(int sig) exit(0); } +static void mpi_bench_client(MPI_Comm client_comm, int num_reqs, int size, int type, char* path) +{ + int rank; + void* data; + struct test_request req; + struct test_response resp; + double total_service_time; + double total_service_time_max; + double total_service_time_min; + double latency_list[7]; + double *times; + double *service_times; + double *latencies = NULL; + int rc; + double t1; + double t2; + double tt1; + double tt2; + int svc_count = 0; + int i; + + MPI_Comm_size(MPI_COMM_WORLD, &rank); + + /* wait a bit to make sure server is ready */ + sleep(5); + + data = malloc(size); + assert(data); + memset(data, rank & 0xff, size); + + memset(&req, 0, sizeof(req)); + req.type = type; + req.size = size; + + service_times = malloc(num_reqs*sizeof(double)); + assert(service_times); + memset(service_times, 0, num_reqs*sizeof(double)); + + MPI_Barrier(client_comm); + +#if 0 + for (i=0; i<num_reqs; i++) + { + tt1 = MPI_Wtime(); + switch(type) + { + case TEST_READ: + snprintf(request.filename, + sizeof(request.filename), + "%s/file.%d.%d", + path, rank, i); + + rc = MPI_Send + + rc = send_helper(tdata->socket_client, &tdata->request, sizeof(tdata->request)); + if(rc < 0) + return NULL; + + rc = recv_helper(tdata->socket_client, tdata->data, tdata->size); + if(rc < 0) + return NULL; + + break; + case TEST_WRITE: + snprintf(tdata->request.filename, + sizeof(tdata->request.filename), + "%s/file.%d-%d.%d", + tdata->path, tdata->rank, tdata->thread, i); + + rc = send_helper(tdata->socket_client, &tdata->request, sizeof(tdata->request)); + if(rc < 0) + return NULL; + + rc = send_helper(tdata->socket_client, tdata->data, tdata->size); + if(rc < 0) + return NULL; + break; + case TEST_READ_NULL: + rc = send_helper(tdata->socket_client, &tdata->request, sizeof(tdata->request)); + if(rc < 0) + return NULL; + + rc = recv_helper(tdata->socket_client, tdata->data, tdata->size); + if(rc < 0) + return NULL; + + break; + case TEST_WRITE_NULL: + rc = send_helper(tdata->socket_client, &tdata->request, sizeof(tdata->request)); + if(rc < 0) + return NULL; + + rc = send_helper(tdata->socket_client, tdata->data, tdata->size); + if(rc < 0) + return NULL; + + break; + default: + assert(0); + break; + } + + rc = recv_helper(tdata->socket_client, &response, sizeof(response)); + if(rc < 0) + return NULL; + + tt2 = MPI_Wtime(); + tdata->service_times[svc_count++] = tt2-tt1; + } + t2 = MPI_Wtime(); + + *tdata->total_service_time = t2 - t1; + + shutdown(tdata->socket_client, SHUT_RDWR); + close(tdata->socket_client); + + /* TODO: rest of stuff from old main(int argc, char **argv) */ +#endif +} + #endif /* MPI_BENCH_H */ /* * Local variables: hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. f6cc376fec7c392d12c02ca16272eae32eabea90
by noreply＠mcs.anl.gov 27 Oct '11

27 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via f6cc376fec7c392d12c02ca16272eae32eabea90 (commit) from 4707f2e72b7d5a7f5a01937c6a69ddb0539629af (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit f6cc376fec7c392d12c02ca16272eae32eabea90 Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Thu Oct 27 18:18:10 2011 -0500 Add triton_debug_error_wrap function Like triton_error_wrap, but takes a debug mask and will be compiled out if -DNDEBUG is active. ----------------------------------------------------------------------- Summary of changes: code/src/common/triton-debug.h | 76 +++++++++++++++++++++++++++++++++++---- code/src/common/triton-error.c | 6 ++-- code/src/common/triton-error.h | 18 +++++++-- 3 files changed, 85 insertions(+), 15 deletions(-) Diff of changes: diff --git a/code/src/common/triton-debug.h b/code/src/common/triton-debug.h index 72e17c4..d78db09 100644 --- a/code/src/common/triton-debug.h +++ b/code/src/common/triton-debug.h @@ -31,13 +31,16 @@ typedef struct * just net.send.expected * * The handle to the mask (a 32 bit integer) is handed to future debugging - * calls. If the same mask has already been added, the same mask handle is returned. + * calls. If the same mask has already been added, the same mask handle is + * returned. */ -triton_ret_t triton_debug_add_mask(const char *mask_str, triton_debug_mask_t *mask, const char *description); +triton_ret_t triton_debug_add_mask(const char *mask_str, + triton_debug_mask_t *mask, const char *description); /** * Enable debugging for certain components. The mask string is - * a comma separated list of components, each component can have multiple sub-components. + * a comma separated list of components, each component can have multiple + * sub-components. * For example: * * triton_debug_enable("net,cosd"); @@ -46,8 +49,8 @@ triton_ret_t triton_debug_add_mask(const char *mask_str, triton_debug_mask_t *ma * "all" - enables all masks * "none" - disables all masks * - * Returns TRITON_SUCCESS if debugging was successfully enabled with output going - * to the specified file, or a failure on error. + * Returns TRITON_SUCCESS if debugging was successfully enabled with output + * going to the specified file, or a failure on error. */ triton_ret_t triton_debug_enable(const char *file, const char *masks); @@ -79,11 +82,18 @@ extern TRITON_BITARRAY_VARDECL(triton_debug_enabled_minor); */ #ifndef NDEBUG -static inline void _triton_debug(triton_debug_mask_t mask, const char *format, ...) __attribute__ ((format (printf, 2, 3))); - static inline int triton_debug_enabled(triton_debug_mask_t mask); -static inline void _triton_debug(triton_debug_mask_t mask, const char *format, ...) +static inline void _triton_debug( + triton_debug_mask_t mask, + const char *format, + ...) + __attribute__ ((format (printf, 2, 3))); + +static inline void _triton_debug( + triton_debug_mask_t mask, + const char *format, + ...) { /* if none are enabled, return without doing mask lookup, also, * if major isn't enabled, don't bother with lookup @@ -109,6 +119,44 @@ static inline void _triton_debug(triton_debug_mask_t mask, const char *format, . return; } +static inline triton_ret_t triton_debug_error_wrap ( + triton_debug_mask_t mask, + triton_ret_t parent, + triton_addr_t node, + const char * message, + ...) __attribute__ ((format (printf, 4, 5))); + +/** + * If the debug mask is enabled, add extra information to the error. + * Otherwise, simply return the original error. + * + * Note that there is no triton_debug_error_wrap_with_code, since the error + * code returned should not depend on wether debugging is enabled or not. + */ +static inline triton_ret_t triton_debug_error_wrap ( + triton_debug_mask_t mask, + triton_ret_t parent, + triton_addr_t node, + const char * message, + ...) +{ + if (triton_debug_enabled_none) + return parent; + + if (!triton_debug_enabled (mask)) + return parent; + + triton_ret_t ret; + va_list ap; + + va_start (ap, message); + ret = triton_error_wrap_va (parent, node, parent->error_code, message, ap); + va_end (ap); + + return ret; +} + + #define triton_debug(_mask, _format, ...) _triton_debug(_mask, "[%s]: " _format, (_mask).string , ## __VA_ARGS__) #define triton_debug_brief(_mask, _format, ...) _triton_debug(_mask, _format , ## __VA_ARGS__) @@ -121,9 +169,21 @@ static inline int triton_debug_enabled(triton_debug_mask_t mask) } #else + #define triton_debug(mask, message, ...) do { } while (0) #define triton_debug_brief(mask, message, ...) do { } while (0) #define triton_debug_enabled(mask) 0 + +static inline triton_ret_t triton_debug_error_wrap ( + triton_debug_mask_t mask, + triton_ret_t parent, + triton_addr_t node, + const char * message, + ...) __attribute__ ((format (printf, 4, 5))) +{ + return parent; +} + #endif void triton_debug_disable(void); diff --git a/code/src/common/triton-error.c b/code/src/common/triton-error.c index b1607ed..95ee723 100644 --- a/code/src/common/triton-error.c +++ b/code/src/common/triton-error.c @@ -48,7 +48,7 @@ triton_ret_t triton_error_from_errno(int errno_val) return TRITON_ERR_UNKNOWN; } -static inline triton_ret_t error_wrap_va( +triton_ret_t triton_error_wrap_va( triton_ret_t parent, triton_addr_t node, triton_error_code_t error_code, @@ -94,7 +94,7 @@ triton_ret_t triton_error_wrap(triton_ret_t parent, triton_addr_t node, const ch triton_ret_t ret; va_list ap; va_start(ap, format); - ret = error_wrap_va(parent, node, parent->error_code, format, ap); + ret = triton_error_wrap_va(parent, node, parent->error_code, format, ap); va_end(ap); return ret; } @@ -104,7 +104,7 @@ triton_ret_t triton_error_wrap_with_code(triton_ret_t parent, triton_addr_t node triton_ret_t ret; va_list ap; va_start(ap, format); - ret = error_wrap_va(parent, node, error_code, format, ap); + ret = triton_error_wrap_va(parent, node, error_code, format, ap); va_end(ap); return ret; } diff --git a/code/src/common/triton-error.h b/code/src/common/triton-error.h index 2dfc969..41c3a5c 100644 --- a/code/src/common/triton-error.h +++ b/code/src/common/triton-error.h @@ -41,14 +41,14 @@ struct triton_error #include "src/common/triton-error-decls.h" /** - * Get the error pointer of a given errno. If the errno is not known, just return - * an unknown error pointer. + * Get the error pointer of a given errno. If the errno is not known, just + * return an unknown error pointer. */ triton_ret_t triton_error_from_errno(int errno_val); /** - * Get the base error pointer from an error code. If the error code isn't known, just return - * an unknown error pointer. + * Get the base error pointer from an error code. If the error code isn't + * known, just return an unknown error pointer. */ triton_ret_t triton_error_from_error_code(triton_error_code_t ec); @@ -61,6 +61,9 @@ triton_ret_t triton_error_from_error_code(triton_error_code_t ec); * finger pointer as errors are passed from node to node. * * If no parent error already exists, TRITON_ERR_NULL should be used. + * + * See triton-debug.h for versions of these functions that accept a debug + * mask. */ triton_ret_t triton_error_wrap( triton_ret_t parent, @@ -74,6 +77,13 @@ triton_ret_t triton_error_wrap_with_code( triton_error_code_t error_code, const char *format, ...) __attribute__ ((format (printf, 4, 5))) ; +triton_ret_t triton_error_wrap_va( + triton_ret_t parent, + triton_addr_t node, + triton_error_code_t error_code, + const char *format, + va_list ap); + /** * Free an error that was created with triton_error_wrap. */ hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 4707f2e72b7d5a7f5a01937c6a69ddb0539629af
by noreply＠mcs.anl.gov 24 Oct '11

24 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 4707f2e72b7d5a7f5a01937c6a69ddb0539629af (commit) from 376cb69a2e97984ac235b3be109c6bd3ba8e64ba (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 4707f2e72b7d5a7f5a01937c6a69ddb0539629af Author: Phil Carns <carns(a)mcs.anl.gov> Date: Mon Oct 24 16:26:32 2011 -0500 starting point for mpi/aesop server - needs testing - needs client routine - needs code to split clients/servers ----------------------------------------------------------------------- Summary of changes: code/src/net/mpi/test/aesop-mpi-bench.ae | 189 ++++++++++++++++++++ code/src/net/mpi/test/module.mk.in | 8 +- .../echo-server.h => net/mpi/test/mpi-bench.h} | 6 +- 3 files changed, 199 insertions(+), 4 deletions(-) create mode 100644 code/src/net/mpi/test/aesop-mpi-bench.ae copy code/src/{socket/test/echo-server.h => net/mpi/test/mpi-bench.h} (92%) Diff of changes: diff --git a/code/src/net/mpi/test/aesop-mpi-bench.ae b/code/src/net/mpi/test/aesop-mpi-bench.ae new file mode 100644 index 0000000..3a9d46b --- /dev/null +++ b/code/src/net/mpi/test/aesop-mpi-bench.ae @@ -0,0 +1,189 @@ +#include "src/net/mpi/mpi.hae" +#include "src/aesop/aesop.h" +#include "src/common/resources/timer/timer.hae" +#include "src/common/resources/file/file-resource.hae" +#include "src/common/triton-debug.h" +#include "mpi-bench.h" + +#include <getopt.h> +#include <sys/types.h> +#include <fcntl.h> +#include <unistd.h> +#include <signal.h> + +void usage(char *progname); +void parse_args (int argc, char **argv); + +#define NUM_PREPOSTED 64 + +static __blocking void handleConnection(int client_rank, int num_reqs); +static __blocking int dotest(int argc, char** argv); + +void usage(char *progname) +{ + fprintf(stderr, + "usage: %s [--help]\n", + progname); + exit(1); + return; +} + +void parse_args (int argc, char **argv) +{ + struct option options[] = + { + { "help", no_argument, NULL, 'h' } + }; + int choice; + + while ((choice = getopt_long(argc, argv, "", options, NULL)) != -1) + { + switch(choice) + { + case '?': + case 'h': + default: + usage(argv[0]); + exit(-1); + break; + } + } + + return; +} + + +static __blocking void handleConnection(int client_rank, int num_reqs) +{ + char* buffer; + int len; + int written; + struct test_request req; + struct test_response resp; + int file_fd; + int rc; + MPI_Status status; + int i; + triton_ret_t ret; + + for(i=0; i<num_reqs; i++) + { + /* read request */ + rc = triton_mpi_recv(&req, sizeof(req), MPI_BYTE, client_rank, 0, MPI_COMM_WORLD, &status); + assert(rc == 0); + + /* allocate a buffer for moving data */ + buffer = malloc(req.size); + assert(buffer); + + /* perform whatever operation was requested by client */ + switch(req.type) + { + case TEST_READ_NULL: + rc = triton_mpi_send(buffer, req.size, MPI_BYTE, client_rank, + 0, MPI_COMM_WORLD); + assert(rc == 0); + break; + + case TEST_WRITE_NULL: + rc = triton_mpi_recv(buffer, req.size, MPI_BYTE, client_rank, + 0, MPI_COMM_WORLD, &status); + assert(rc == 0); + break; + + case TEST_READ: + ret = file_open(&file_fd, + req.filename, + (O_RDONLY), + 0644); + assert(ret == TRITON_SUCCESS); + + ret = file_pread(file_fd, + buffer, + req.size, + 0); + assert(ret == TRITON_SUCCESS); + + rc = triton_mpi_send(buffer, req.size, MPI_BYTE, client_rank, + 0, MPI_COMM_WORLD); + assert(rc == 0); + + ret = file_close(file_fd); + assert(ret == TRITON_SUCCESS); + break; + + case TEST_WRITE: + rc = triton_mpi_recv(buffer, req.size, MPI_BYTE, client_rank, + 0, MPI_COMM_WORLD, &status); + assert(rc == 0); + + ret = file_open(&file_fd, + req.filename, + (O_CREAT|O_TRUNC|O_WRONLY), + 0644); + assert(ret == TRITON_SUCCESS); + + ret = file_pwrite(file_fd, + buffer, + req.size, + 0); + assert(ret == TRITON_SUCCESS); + + ret = file_close(file_fd); + assert(ret == TRITON_SUCCESS); + + break; + + default: + assert(0 && "Unknown request type!"); + } + rc = triton_mpi_send(&resp, sizeof(resp), MPI_BYTE, client_rank, + 0, MPI_COMM_WORLD); + assert(rc == 0); + free(buffer); + }; +} + +static __blocking int dotest(int argc, char** argv) +{ + struct sigaction new_action; + int clients; + + new_action.sa_handler = handler; + sigemptyset(&new_action.sa_mask); + new_action.sa_flags = SA_RESETHAND; + sigaction(SIGTERM, &new_action, NULL); + sigaction(SIGINT, &new_action, NULL); + sigaction(SIGQUIT, &new_action, NULL); + + parse_args(argc, argv); + + MPI_Comm_size(MPI_COMM_WORLD, &clients); + clients--; + + pwait + { + pprivate int i; + + for(i=1; i<=clients; i++) + { + pbranch + { + handleConnection(i, 1); /* TODO: fill in num_reqs */ + } + } + } + + return 0; +} + +aesop_main_set_with_init(NULL, "triton.echo-server", dotest, "timer", "aesocket", "file"); + +/* + * Local variables: + * c-indent-level: 4 + * c-basic-offset: 4 + * End: + * + * vim: ts=8 sts=4 sw=4 expandtab + */ diff --git a/code/src/net/mpi/test/module.mk.in b/code/src/net/mpi/test/module.mk.in index 37ccbd3..5b03224 100644 --- a/code/src/net/mpi/test/module.mk.in +++ b/code/src/net/mpi/test/module.mk.in @@ -4,7 +4,8 @@ ifneq (,$(BUILD_MPI)) AETESTSRC += $(DIR)/send-recv.ae \ $(DIR)/pthread-compare.ae \ - $(DIR)/noop-bench.ae + $(DIR)/noop-bench.ae \ + $(DIR)/aesop-mpi-bench.ae MODCFLAGS_$(DIR)/send-recv = $(MPICFLAGS) MODLDFLAGS_$(DIR)/send-recv = $(MPILDFLAGS) @@ -21,4 +22,9 @@ MODLDFLAGS_$(DIR)/noop-bench = $(MPILDFLAGS) MODLIBS_$(DIR)/noop-bench = $(MPILIBS) MODCC_$(DIR)/noop-bench = $(MPICC) +MODCFLAGS_$(DIR)/aesop-mpi-bench = $(MPICFLAGS) +MODLDFLAGS_$(DIR)/aesop-mpi-bench = $(MPILDFLAGS) +MODLIBS_$(DIR)/aesop-mpi-bench = $(MPILIBS) +MODCC_$(DIR)/aesop-mpi-bench = $(MPICC) + endif # BUILD_MPI diff --git a/code/src/socket/test/echo-server.h b/code/src/net/mpi/test/mpi-bench.h similarity index 92% copy from code/src/socket/test/echo-server.h copy to code/src/net/mpi/test/mpi-bench.h index 8cfce28..4dbaa6c 100644 --- a/code/src/socket/test/echo-server.h +++ b/code/src/net/mpi/test/mpi-bench.h @@ -1,5 +1,5 @@ -#ifndef ECHO_SERVER_H -#define ECHO_SERVER_H +#ifndef MPI_BENCH_H +#define MPI_BENCH_H enum test_type { @@ -41,7 +41,7 @@ static void handler(int sig) exit(0); } -#endif /* ECHO_SERVER_H */ +#endif /* MPI_BENCH_H */ /* * Local variables: * c-indent-level: 4 hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 376cb69a2e97984ac235b3be109c6bd3ba8e64ba
by noreply＠mcs.anl.gov 20 Oct '11

20 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 376cb69a2e97984ac235b3be109c6bd3ba8e64ba (commit) from d8077ee8d982bf25d12b271e7723e4d0032d21fa (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 376cb69a2e97984ac235b3be109c6bd3ba8e64ba Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Thu Oct 20 14:56:07 2011 -0500 Fix in triton_error_wrap Copy error code from parent if no new error code is given. ----------------------------------------------------------------------- Summary of changes: code/src/common/triton-error.c | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) Diff of changes: diff --git a/code/src/common/triton-error.c b/code/src/common/triton-error.c index 146b12a..b1607ed 100644 --- a/code/src/common/triton-error.c +++ b/code/src/common/triton-error.c @@ -94,7 +94,7 @@ triton_ret_t triton_error_wrap(triton_ret_t parent, triton_addr_t node, const ch triton_ret_t ret; va_list ap; va_start(ap, format); - ret = error_wrap_va(parent, node, TRITON_ERR_NONROOT_CODE, format, ap); + ret = error_wrap_va(parent, node, parent->error_code, format, ap); va_end(ap); return ret; } hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. d8077ee8d982bf25d12b271e7723e4d0032d21fa
by noreply＠mcs.anl.gov 19 Oct '11

19 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via d8077ee8d982bf25d12b271e7723e4d0032d21fa (commit) from 3e6a550d0c752471928baa39002ca844cc76fe84 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit d8077ee8d982bf25d12b271e7723e4d0032d21fa Author: Dries Kimpe <dkimpe(a)mcs.anl.gov> Date: Wed Oct 19 17:59:44 2011 -0500 Add thread-per-op with blocking read/write/accept Modified triton-socket and the aesocket resource to support executing read/write/accept on a separate thread using blocking calls. Callbacks are called from the same thread that execute the syscall. Enable by setting: triton.socket.thread_per_op (or triton_socket_thread_per_op=1 in environment). ----------------------------------------------------------------------- Summary of changes: code/src/common/resources/aesocket/aesocket.c | 211 ++++++++++++++++++++++- code/src/common/resources/aesocket/aesocket.hae | 13 ++ code/src/socket/test/echo-server.ae | 9 +- code/src/socket/triton-socket.ae | 104 +++++++++--- code/src/socket/triton-socket.hae | 3 + 5 files changed, 304 insertions(+), 36 deletions(-) Diff of changes: diff --git a/code/src/common/resources/aesocket/aesocket.c b/code/src/common/resources/aesocket/aesocket.c index 4392581..f058a14 100644 --- a/code/src/common/resources/aesocket/aesocket.c +++ b/code/src/common/resources/aesocket/aesocket.c @@ -33,6 +33,12 @@ struct event_loop_data int running; }; +enum { TOP_NONE = 0, + TOP_READ, + TOP_WRITE, + TOP_ACCEPT +}; + struct aesocket_op { ae_op_id_t op_id; @@ -42,6 +48,27 @@ struct aesocket_op int flags; ev_io io; struct event_loop_data* event_loop; + + pthread_t tid; + unsigned int top; + union + { + struct { + void * buf; + size_t size; + int * ret; + } read; + struct { + const void * buf; + size_t size; + int * ret; + } write; + struct { + struct sockaddr * addr; + socklen_t * addrlen; + int * newfd; + } accept; + }; }; static ae_ops_t posted_oplist; @@ -82,10 +109,24 @@ static void aesocket_fd_ready( triton_mutex_unlock(mutex_p); ev_io_stop(eloop, io); - ae_opcache_complete_op_threaded_with_affinity(aesocket_opcache, op, &socket_op->fd, sizeof(socket_op->fd)); + + ae_opcache_complete_op_threaded_with_affinity(aesocket_opcache, op, + &socket_op->fd, sizeof(socket_op->fd)); + return; } +static void check_init () +{ + + if (!aesocket_opcache) + { + triton_err(triton_log_default, + "Error: aesocket resource not initialized."); + assert(0); + } +} + ae_define_post(triton_ret_t, triton_aesocket_ready, int fd, int mode) { @@ -98,12 +139,7 @@ ae_define_post(triton_ret_t, triton_aesocket_ready, int fd, ae_ops_t* inflight_p; struct event_loop_data* event_loop_p = NULL; - if (!aesocket_opcache) - { - triton_err(triton_log_default, - "Error: aesocket resource not initialized."); - assert(0); - } + check_init (); op = ae_opcache_get(aesocket_opcache); ae_op_fill(op); @@ -114,6 +150,7 @@ ae_define_post(triton_ret_t, triton_aesocket_ready, int fd, ae_id_gen(triton_aesocket_resource_id, (intptr_t) op); socket_op->tret = TRITON_SUCCESS; socket_op->fd = fd; + socket_op->top = TOP_NONE; socket_op->flags = (mode & AESOCKET_READ ? EV_READ : 0) | (mode & AESOCKET_WRITE ? EV_WRITE : 0); if(nthreads_event_loop > 0) @@ -136,7 +173,8 @@ ae_define_post(triton_ret_t, triton_aesocket_ready, int fd, inflight_p = &inflight_oplist; } - assert(socket_op->flags && "Need to specify one of AESOCKET_READ|AESOCKET_WRITE"); + assert(socket_op->flags && "Need to specify one of " + "AESOCKET_READ|AESOCKET_WRITE"); *__ae_op_id = socket_op->op_id; @@ -175,6 +213,157 @@ ae_define_post(triton_ret_t, triton_aesocket_ready, int fd, return TRITON_SUCCESS; } +static void * thread_op_main (void * data) +{ + struct ae_op * op = (struct ae_op*) data; + struct aesocket_op *socket_op; + socket_op = ae_op_entry(op, struct aesocket_op, + op); + + int ret; + + switch (socket_op->top) + { + case TOP_READ: + *socket_op->read.ret = ret = read (socket_op->fd, socket_op->read.buf, + socket_op->read.size); + break; + case TOP_WRITE: + *socket_op->write.ret = ret = write (socket_op->fd, + socket_op->write.buf, socket_op->write.size); + break; + case TOP_ACCEPT: + *socket_op->accept.newfd = ret = accept (socket_op->fd, + socket_op->accept.addr, socket_op->accept.addrlen); + break; + default: + assert (0 && "Invalid socket_op->top!"); + } + + triton_ret_t tret = + (ret < 0 ? triton_error_from_errno (errno) + : TRITON_SUCCESS); + + ae_opcache_complete_op(aesocket_opcache, op, triton_ret_t, tret); + + return 0; +} + +ae_define_post (triton_ret_t, triton_aesocket_read, int fd, void * buf, + size_t count, int * fret) +{ + struct ae_op *op; + struct aesocket_op *socket_op; + int ret; + + check_init (); + + op = ae_opcache_get(aesocket_opcache); + ae_op_fill(op); + + socket_op = ae_op_entry(op, struct aesocket_op, + op); + socket_op->op_id = + ae_id_gen(triton_aesocket_resource_id, (intptr_t) op); + socket_op->tret = TRITON_SUCCESS; + socket_op->fd = fd; + socket_op->read.buf = buf; + socket_op->read.size = count; + socket_op->read.ret = fret; + socket_op->top = TOP_READ; + + *__ae_op_id = socket_op->op_id; + + // for now, until implementing cancel + pthread_attr_t attr; + pthread_attr_init (&attr); + pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED); + + ret = pthread_create (&socket_op->tid, &attr, thread_op_main, op); + + pthread_attr_destroy (&attr); + + if (ret) + return TRITON_ERR_UNKNOWN; + + return TRITON_SUCCESS; +} + +static triton_ret_t execute_op (struct ae_op * op) +{ + // for now, until implementing cancel + pthread_attr_t attr; + pthread_attr_init (&attr); + pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED); + + struct aesocket_op * socket_op = ae_op_entry(op, struct aesocket_op, op); + + int ret = pthread_create (&socket_op->tid, &attr, thread_op_main, op); + + pthread_attr_destroy (&attr); + + if (ret) + return TRITON_ERR_UNKNOWN; + + return TRITON_SUCCESS; +} + + +ae_define_post (triton_ret_t, triton_aesocket_write, int fd, const void * buf, + size_t count, int * fret) +{ + struct ae_op *op; + struct aesocket_op *socket_op; + int ret; + + check_init (); + + op = ae_opcache_get(aesocket_opcache); + ae_op_fill(op); + + socket_op = ae_op_entry(op, struct aesocket_op, op); + socket_op->op_id = + ae_id_gen(triton_aesocket_resource_id, (intptr_t) op); + socket_op->tret = TRITON_SUCCESS; + socket_op->fd = fd; + socket_op->write.buf = buf; + socket_op->write.size = count; + socket_op->write.ret = fret; + socket_op->top = TOP_WRITE; + + *__ae_op_id = socket_op->op_id; + + return execute_op (op); +} + +ae_define_post (triton_ret_t, triton_aesocket_accept, int sockfd, + struct sockaddr * addr, socklen_t * addrlen, int * newfd) +{ + struct ae_op *op; + struct aesocket_op *socket_op; + int ret; + + check_init (); + + op = ae_opcache_get(aesocket_opcache); + ae_op_fill(op); + + socket_op = ae_op_entry(op, struct aesocket_op, + op); + socket_op->op_id = + ae_id_gen(triton_aesocket_resource_id, (intptr_t) op); + socket_op->tret = TRITON_SUCCESS; + socket_op->fd = sockfd; + socket_op->accept.addr = addr; + socket_op->accept.addrlen = addrlen; + socket_op->accept.newfd = newfd; + socket_op->top = TOP_ACCEPT; + + *__ae_op_id = socket_op->op_id; + + return execute_op (op); +} + static triton_ret_t triton_aesocket_poll( ae_context_t context) { @@ -243,7 +432,11 @@ static triton_ret_t triton_aesocket_cancel( op); ctx = op->ctx; - if(ae_ops_exists(&posted_oplist, &op->link)) + if (socket_op->top != TOP_NONE) + { + assert (0 && "Cancel not supported for threads for now"); + } + else if (ae_ops_exists(&posted_oplist, &op->link)) { /* this op hasn't been processed by the poll function yet; pull it * out of the queue diff --git a/code/src/common/resources/aesocket/aesocket.hae b/code/src/common/resources/aesocket/aesocket.hae index a4d30cb..ac97835 100644 --- a/code/src/common/resources/aesocket/aesocket.hae +++ b/code/src/common/resources/aesocket/aesocket.hae @@ -3,6 +3,10 @@ #include "src/aesop/aesop.h" +#include <stddef.h> +#include <sys/types.h> +#include <sys/socket.h> + enum { AESOCKET_READ = 0x01, @@ -13,6 +17,15 @@ __blocking triton_ret_t triton_aesocket_ready( int fd, int mode); +__blocking triton_ret_t triton_aesocket_read (int fd, void * buf, + size_t count, int * ret); + +__blocking triton_ret_t triton_aesocket_write (int fd, const void * buf, + size_t count, int * ret); + +__blocking triton_ret_t triton_aesocket_accept (int sockfd, + struct sockaddr * addr, socklen_t * addrlen, int * newfd); + triton_ret_t triton_aesocket_init( void); diff --git a/code/src/socket/test/echo-server.ae b/code/src/socket/test/echo-server.ae index e3cfe02..27e7f21 100644 --- a/code/src/socket/test/echo-server.ae +++ b/code/src/socket/test/echo-server.ae @@ -73,8 +73,8 @@ static __blocking void handleConnection( resp.error_code = 42; - /* mark socket as nonblocking */ - fcntl(fd, F_SETFL, O_NONBLOCK); + triton_socket_prepare (fd); + pbranch { @@ -148,7 +148,7 @@ static __blocking void handleConnection( break; default: - assert(0); + assert(0 && "Unknown request type!"); } ret = triton_write(fd, &resp, sizeof(resp), &written); assert(ret == TRITON_SUCCESS); @@ -182,7 +182,8 @@ static __blocking int dotest(int argc, char** argv) listensocket = socket(AF_INET, SOCK_STREAM, 0); assert(listensocket >= 0); - fcntl(listensocket, F_SETFL, O_NONBLOCK); + // Prepare socket for use with the triton-socket module + triton_socket_prepare (listensocket); memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; diff --git a/code/src/socket/triton-socket.ae b/code/src/socket/triton-socket.ae index 961b40f..e915e60 100644 --- a/code/src/socket/triton-socket.ae +++ b/code/src/socket/triton-socket.ae @@ -1,18 +1,20 @@ #include "triton-socket.hae" #include "src/common/resources/aesocket/aesocket.hae" #include "src/zeroconf/zeroconf.h" +#include "src/common/triton-error.h" #include <unistd.h> #include <assert.h> +#include <fcntl.h> -static triton_ret_t immed_completion_updater( - const char* key, const char* value); -static triton_ret_t triton_socket_init( - void); -static void triton_socket_finalize( - void); +static triton_ret_t immed_completion_updater (const char* key, const char* value); +static triton_ret_t thread_updater (const char * key, const char * value); + +static triton_ret_t triton_socket_init (void); +static void triton_socket_finalize (void); static int immediate_completion = 1; +static int thread_per_op = 0; /* Number of zero byte reads in a row that we will observe before assuming * that the peer has closed the socket; this is a hack to weed out @@ -20,12 +22,25 @@ static int immediate_completion = 1; */ #define ZERO_BYTE_READ_THRESHOLD 0 + +int triton_socket_prepare (int fd) +{ + if (thread_per_op) + return 0; + + /* mark socket as nonblocking */ + return fcntl(fd, F_SETFL, O_NONBLOCK); +} + __blocking triton_ret_t triton_accept( int sockfd, struct sockaddr *addr, socklen_t * addrlen, int *newfd) { + if (thread_per_op) + return triton_aesocket_accept (sockfd, addr, addrlen, newfd); + triton_ret_t ret; int r; @@ -80,14 +95,28 @@ __blocking triton_ret_t triton_read( do { - if(!immediate_completion) + + if (thread_per_op) { - ret = triton_aesocket_ready(fd, AESOCKET_READ); - if (ret != TRITON_SUCCESS) - return ret; + triton_ret_t tret; + tret = triton_aesocket_read (fd, tmp_buf, tmp_count, &r); + if (!triton_error_equal (tret, TRITON_SUCCESS)) + { + // need to map errors instead of assuming err_IO + return TRITON_ERR_IO; + } + } + else + { + if(!immediate_completion) + { + ret = triton_aesocket_ready(fd, AESOCKET_READ); + if (ret != TRITON_SUCCESS) + return ret; + } + r = read(fd, tmp_buf, tmp_count); } - r = read(fd, tmp_buf, tmp_count); if(r > 0) { tmp_buf += r; @@ -108,7 +137,7 @@ __blocking triton_ret_t triton_read( return triton_error_from_errno(errno); } - if(immediate_completion && tmp_count > 0) + if(!thread_per_op && immediate_completion && tmp_count > 0) { ret = triton_aesocket_ready(fd, AESOCKET_READ); if (ret != TRITON_SUCCESS) @@ -137,14 +166,26 @@ __blocking triton_ret_t triton_write( do { - if(!immediate_completion) + if (thread_per_op) { - ret = triton_aesocket_ready(fd, AESOCKET_WRITE); - if (ret != TRITON_SUCCESS) - return ret; + triton_ret_t tret; + tret = triton_aesocket_write (fd, tmp_buf, tmp_count, &r); + if (!triton_error_equal (tret, TRITON_SUCCESS)) + { + return TRITON_ERR_IO; + } + } + else + { + if(!immediate_completion) + { + ret = triton_aesocket_ready(fd, AESOCKET_WRITE); + if (ret != TRITON_SUCCESS) + return ret; + } + + r = write(fd, tmp_buf, tmp_count); } - - r = write(fd, buf, count); if(r >= 0) { tmp_buf += r; @@ -186,13 +227,20 @@ static triton_ret_t triton_socket_init( { triton_ret_t tret; - tret = triton_zeroconf_register("triton.socket.immed_completion", "1", + tret = triton_zeroconf_register("triton.socket.immed_completion", "1", immed_completion_updater, "Enable immediate completion (1|0)"); if(tret != TRITON_SUCCESS) { return(tret); } + tret = triton_zeroconf_register("triton.socket.thread_per_op", "0", + thread_updater, "Use blocking calls and threads (1|0)"); + if(tret != TRITON_SUCCESS) + { + return(tret); + } + return(TRITON_SUCCESS); } @@ -202,13 +250,13 @@ static void triton_socket_finalize( return; } - -static triton_ret_t immed_completion_updater(const char* key, const char* value) +static triton_ret_t bool_updater (const char * key, const char * val, + int * option) { int ret; int tmp; - ret = sscanf(value, "%d", &tmp); + ret = sscanf(val, "%d", &tmp); if(ret != 1) { return(TRITON_ERR_INVAL); @@ -218,11 +266,21 @@ static triton_ret_t immed_completion_updater(const char* key, const char* value) return(TRITON_ERR_INVAL); } - immediate_completion = tmp; + *option = tmp; return(TRITON_SUCCESS); } +static triton_ret_t thread_updater (const char * key, const char * value) +{ + return bool_updater (key, value, &thread_per_op); +} + +static triton_ret_t immed_completion_updater(const char* key, const char* value) +{ + return bool_updater (key, value, &immediate_completion); +} + /* * Local variables: * c-indent-level: 4 diff --git a/code/src/socket/triton-socket.hae b/code/src/socket/triton-socket.hae index e030f20..e13aac2 100644 --- a/code/src/socket/triton-socket.hae +++ b/code/src/socket/triton-socket.hae @@ -7,6 +7,9 @@ #include "src/aesop/aesop.h" +/** Prepare fd for use with the functions in the triton-socket module. */ +int triton_socket_prepare (int fd); + __blocking triton_ret_t triton_accept( int sockfd, struct sockaddr *addr, hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 3e6a550d0c752471928baa39002ca844cc76fe84
by noreply＠mcs.anl.gov 13 Oct '11

13 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 3e6a550d0c752471928baa39002ca844cc76fe84 (commit) from 494a78a6f5e420dc58620e3b1afef9ca8056568f (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 3e6a550d0c752471928baa39002ca844cc76fe84 Author: Philip Carns <carns(a)flogin3.lcrc.anl.gov> Date: Thu Oct 13 12:34:29 2011 -0500 update build rule for echo-server-lite ----------------------------------------------------------------------- Summary of changes: code/src/socket/test/module.mk.in | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) Diff of changes: diff --git a/code/src/socket/test/module.mk.in b/code/src/socket/test/module.mk.in index 1efbce8..d0b40ec 100644 --- a/code/src/socket/test/module.mk.in +++ b/code/src/socket/test/module.mk.in @@ -2,5 +2,5 @@ DIR := src/socket/test AETESTSRC += $(DIR)/echo-server.ae -src/socket/test/echo-server-lite: src/socket/test/echo-server.o src/socket/triton-socket.o src/versioned-osd/prototype/file-resource/file-resource.o src/common/libev/ev.o src/common/resources/aesocket/aesocket.o src/common/resources/timer/timer.o src/common/triton-error.o src/common/triton-debug.o src/common/triton-log.o src/common/triton-string.o src/common/lookup3.o src/common/triton-init.o src/common/triton-init-registry-lite.o src/zeroconf/zeroconf.o src/aesop/opcache.o src/aesop/aesop.o src/aesop/resource.o src/aesop/hints.o src/aesop/ae-ctl.o src/common/triton-error-defs.o src/net/triton-message.o src/remote/message.o +src/socket/test/echo-server-lite: src/socket/test/echo-server.o src/socket/triton-socket.o src/common/resources/file/file-resource.o src/common/libev/ev.o src/common/resources/aesocket/aesocket.o src/common/resources/timer/timer.o src/common/triton-error.o src/common/triton-debug.o src/common/triton-log.o src/common/triton-string.o src/common/lookup3.o src/common/triton-init.o src/common/triton-init-registry-lite.o src/zeroconf/zeroconf.o src/aesop/opcache.o src/aesop/aesop.o src/aesop/resource.o src/aesop/hints.o src/aesop/ae-ctl.o src/common/triton-error-defs.o src/net/triton-message.o src/remote/message.o gcc $(LDFLAGS) $^ -lpthread -lm -o $@ hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 494a78a6f5e420dc58620e3b1afef9ca8056568f
by noreply＠mcs.anl.gov 13 Oct '11

13 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 494a78a6f5e420dc58620e3b1afef9ca8056568f (commit) from f53604d130acb2aea93f35ae1e0d35f274f86054 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 494a78a6f5e420dc58620e3b1afef9ca8056568f Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu Oct 13 11:22:52 2011 -0400 experimental change to opcache thread pool - instead of a single work queue, make a separate queue for each consumer (thread) - add mechanism for producer to have similar operations (ie, operations on the same socket) to get consistently routed to the same consumer ----------------------------------------------------------------------- Summary of changes: code/src/aesop/opcache.c | 100 +++++++++++++++++++++++++ code/src/aesop/opcache.h | 6 ++ code/src/common/resources/aesocket/aesocket.c | 4 +- 3 files changed, 108 insertions(+), 2 deletions(-) Diff of changes: diff --git a/code/src/aesop/opcache.c b/code/src/aesop/opcache.c index c51217d..a0fbe14 100644 --- a/code/src/aesop/opcache.c +++ b/code/src/aesop/opcache.c @@ -1,6 +1,7 @@ #include "src/aesop/aesop.h" #include "src/aesop/opcache.h" #include "src/common/triton-error.h" +#include "src/common/jenkins-hash.h" #include <errno.h> #include <opa_primitives.h> @@ -17,6 +18,15 @@ */ #define THREAD_WORK_THRESHOLD 4 +struct thread_data +{ + ae_ops_t thread_queue; + int active; + triton_mutex_t mutex; + triton_cond_t cond; + void* parent; +}; + struct ae_opcache { #ifndef TRITON_OPCACHE_MALLOC @@ -36,9 +46,33 @@ struct ae_opcache ae_ops_t thread_queue; int typesize; int member_offset; + struct thread_data* thread_data_array; }; static void* thread_pool_fn(void* foo); +static void* thread_pool_fn_with_affinity(void* foo); + +static void* thread_pool_fn_with_affinity(void* foo) +{ + struct thread_data* tdata = (struct thread_data*)foo; + ae_opcache_t cache = (ae_opcache_t)tdata->parent; + struct ae_op* op; + + while(OPA_load_int(&cache->num_threads) > 0) + { + triton_mutex_lock(&tdata->mutex); + while((op = ae_ops_dequeue(&tdata->thread_queue)) == NULL) + { + tdata->active = 0; + pthread_cond_wait(&tdata->cond, &tdata->mutex); + tdata->active = 1; + } + triton_mutex_unlock(&tdata->mutex); + cache->completion_fn(cache, op); + } + return(NULL); +} + static void* thread_pool_fn(void* foo) { @@ -70,6 +104,43 @@ static void* thread_pool_fn(void* foo) return(NULL); } +triton_ret_t ae_opcache_set_threads_with_affinity(ae_opcache_t cache, + void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), + int num_threads) +{ + int i; + int ret; + + OPA_store_int (&cache->num_threads, num_threads); + cache->num_threads_active = num_threads; + cache->tids = (pthread_t*)malloc(num_threads*sizeof(pthread_t)); + if(!cache->tids) + return(TRITON_ERR_NOMEM); + cache->completion_fn = completion_fn; + + cache->thread_data_array = (struct thread_data*)malloc(num_threads*sizeof(*cache->thread_data_array)); + if(!cache->thread_data_array) + return(TRITON_ERR_NOMEM); + + for(i=0; i<num_threads; i++) + { + triton_mutex_init(&cache->thread_data_array[i].mutex, NULL); + triton_cond_init(&cache->thread_data_array[i].cond, NULL); + ae_ops_init(&cache->thread_data_array[i].thread_queue); + cache->thread_data_array[i].parent = cache; + cache->thread_data_array[i].active = 1; + + ret = pthread_create(&cache->tids[i], NULL, thread_pool_fn_with_affinity, &cache->thread_data_array[i]); + if(ret != 0) + { + return(TRITON_ERR_UNKNOWN); + } + } + + return(TRITON_SUCCESS); +} + + triton_ret_t ae_opcache_set_threads(ae_opcache_t cache, void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), int num_threads) @@ -96,6 +167,35 @@ triton_ret_t ae_opcache_set_threads(ae_opcache_t cache, return(TRITON_SUCCESS); } +void ae_opcache_complete_op_threaded_with_affinity(ae_opcache_t cache, struct ae_op* op, void* affinity_data, int affinity_data_size) +{ + uint32_t pc = 0, pb = 0; + int thread_index = 0; + int num_threads = OPA_load_int(&cache->num_threads); + + assert(affinity_data); /* we could pick thread randomly if not set... */ + + if(num_threads == 0) + { + cache->completion_fn(cache, op); + return; + } + + /* hash the affinity data to get a consistent thread selection */ + bj_hashlittle2(affinity_data, affinity_data_size, &pc, &pb); + thread_index = pc % num_threads; + + triton_mutex_lock(&cache->thread_data_array[thread_index].mutex); + ae_ops_enqueue(op, &cache->thread_data_array[thread_index].thread_queue); + if(!cache->thread_data_array[thread_index].active) + { + triton_cond_signal(&cache->thread_data_array[thread_index].cond); + } + triton_mutex_unlock(&cache->thread_data_array[thread_index].mutex); + + return; +} + void ae_opcache_complete_op_threaded(ae_opcache_t cache, struct ae_op* op) { if(OPA_load_int(&cache->num_ops_in_use) > 1) diff --git a/code/src/aesop/opcache.h b/code/src/aesop/opcache.h index 72cadb6..4e2be67 100644 --- a/code/src/aesop/opcache.h +++ b/code/src/aesop/opcache.h @@ -23,6 +23,8 @@ typedef struct ae_opcache *ae_opcache_t; void ae_opcache_complete_op_threaded(ae_opcache_t cache, struct ae_op* op); +void ae_opcache_complete_op_threaded_with_affinity(ae_opcache_t cache, struct ae_op* op, void* affinity_data, int affinity_data_size); + /** * Create an opcache. * init_size is a hint and may be ignored. @@ -37,6 +39,10 @@ triton_ret_t ae_opcache_set_threads(ae_opcache_t cache, void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), int num_threads); +triton_ret_t ae_opcache_set_threads_with_affinity(ae_opcache_t cache, + void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), + int num_threads); + /** * Destroy the given opcache. * Note that all entries obtained from this cache are released and diff --git a/code/src/common/resources/aesocket/aesocket.c b/code/src/common/resources/aesocket/aesocket.c index 2691701..4392581 100644 --- a/code/src/common/resources/aesocket/aesocket.c +++ b/code/src/common/resources/aesocket/aesocket.c @@ -82,7 +82,7 @@ static void aesocket_fd_ready( triton_mutex_unlock(mutex_p); ev_io_stop(eloop, io); - ae_opcache_complete_op_threaded(aesocket_opcache, op); + ae_opcache_complete_op_threaded_with_affinity(aesocket_opcache, op, &socket_op->fd, sizeof(socket_op->fd)); return; } @@ -357,7 +357,7 @@ triton_ret_t triton_aesocket_init( return tret; } - tret = ae_opcache_set_threads(aesocket_opcache, aesocket_completion_fn, + tret = ae_opcache_set_threads_with_affinity(aesocket_opcache, aesocket_completion_fn, nthreads); if (tret != TRITON_SUCCESS) { hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. f53604d130acb2aea93f35ae1e0d35f274f86054
by noreply＠mcs.anl.gov 11 Oct '11

11 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via f53604d130acb2aea93f35ae1e0d35f274f86054 (commit) from 3fa3bc07ba5336e16f2756c838abf5d404240221 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit f53604d130acb2aea93f35ae1e0d35f274f86054 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Tue Oct 11 13:24:21 2011 -0400 more cancel test cases ----------------------------------------------------------------------- Summary of changes: .../{cancel-pbranch3.ae => cancel-pbranch4.ae} | 7 +------ code/src/aesop/parser/tests/blocking/module.mk.in | 3 ++- 2 files changed, 3 insertions(+), 7 deletions(-) copy code/src/aesop/parser/tests/blocking/{cancel-pbranch3.ae => cancel-pbranch4.ae} (81%) Diff of changes: diff --git a/code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae b/code/src/aesop/parser/tests/blocking/cancel-pbranch4.ae similarity index 81% copy from code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae copy to code/src/aesop/parser/tests/blocking/cancel-pbranch4.ae index 638c087..e46fdca 100644 --- a/code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae +++ b/code/src/aesop/parser/tests/blocking/cancel-pbranch4.ae @@ -14,10 +14,6 @@ static __blocking void run_test(void) { pbranch { - printf("pbranch1: calling triton_timer().\n"); - triton_timer(5); - printf("pbranch1: finished triton_timer().\n"); - sleep(5); } @@ -32,8 +28,7 @@ static __blocking void run_test(void) printf("pwait done\n"); - - + return; } __blocking int aesop_main(int argc, char **argv) diff --git a/code/src/aesop/parser/tests/blocking/module.mk.in b/code/src/aesop/parser/tests/blocking/module.mk.in index 54d9bfd..c0a8ed4 100644 --- a/code/src/aesop/parser/tests/blocking/module.mk.in +++ b/code/src/aesop/parser/tests/blocking/module.mk.in @@ -56,7 +56,8 @@ AETESTSRC += $(DIR)/icpfor.ae \ $(DIR)/pbranch-completion.ae \ $(DIR)/cancel-pbranch1.ae \ $(DIR)/cancel-pbranch2.ae \ - $(DIR)/cancel-pbranch3.ae + $(DIR)/cancel-pbranch3.ae \ + $(DIR)/cancel-pbranch4.ae # This should not compile # $(DIR)/pbranch-with-return.ae hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 3fa3bc07ba5336e16f2756c838abf5d404240221
by noreply＠mcs.anl.gov 11 Oct '11

11 Oct '11

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 3fa3bc07ba5336e16f2756c838abf5d404240221 (commit) from 214782485d46358e4dcce2dca0e608b7f11a46ff (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 3fa3bc07ba5336e16f2756c838abf5d404240221 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Tue Oct 11 13:11:13 2011 -0400 some pbranch cancel test cases ----------------------------------------------------------------------- Summary of changes: .../aesop/parser/tests/blocking/cancel-pbranch1.ae | 162 ++++++++++++++++++++ .../aesop/parser/tests/blocking/cancel-pbranch2.ae | 57 +++++++ .../aesop/parser/tests/blocking/cancel-pbranch3.ae | 45 ++++++ code/src/aesop/parser/tests/blocking/module.mk.in | 5 +- 4 files changed, 268 insertions(+), 1 deletions(-) create mode 100644 code/src/aesop/parser/tests/blocking/cancel-pbranch1.ae create mode 100644 code/src/aesop/parser/tests/blocking/cancel-pbranch2.ae create mode 100644 code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae Diff of changes: diff --git a/code/src/aesop/parser/tests/blocking/cancel-pbranch1.ae b/code/src/aesop/parser/tests/blocking/cancel-pbranch1.ae new file mode 100644 index 0000000..cc08ec0 --- /dev/null +++ b/code/src/aesop/parser/tests/blocking/cancel-pbranch1.ae @@ -0,0 +1,162 @@ +#include <unistd.h> +#include "src/aesop/aesop.h" +#include "src/aesop/parser/tests/blocking/btest.hae" +#include <assert.h> +#include "src/aesop/aesop-support.hae" + +static __blocking void foo_fn(void) +{ + int ret1; + int a = 0; + + ret1 = tctest1(&a); + + sleep(5); + return; +} + +static __blocking void foo_fn2(void) +{ + int ret1; + int a = 0; + + ret1 = tctest1(&a); + + return; +} + +static __blocking void run_test(void) +{ + int ret1; + int ret2; + int a = 0; + + pwait + { + pbranch + { + printf("pbranch1: calling tctest1.\n"); + ret1 = tctest1(&a); + printf("pbranch1: finished tctest1.\n"); + /* tctest1 runs the callback in a thread. Sleep here to let + * the other pbranch try to cancel before we finish this + * pbranch. + */ + printf("pbranch1: about to sleep for 5 seconds.\n"); + sleep(5); + printf("pbranch1: done.\n"); + } + + pbranch + { + /* wait 1 second to let the tctest1 call finish */ + printf("pbranch2: setting timer for 1 second.\n"); + triton_timer(1000); + printf("pbranch2: timer done.\n"); + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + pwait + { + pbranch + { + printf("pbranch1: calling foo_fn.\n"); + foo_fn(); + printf("pbranch1: finished foo_fn.\n"); + printf("pbranch1: done.\n"); + } + + pbranch + { + /* wait 1 second to let the tctest1 call finish */ + printf("pbranch2: setting timer for 1 second.\n"); + triton_timer(1000); + printf("pbranch2: timer done.\n"); + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + pwait + { + pbranch + { + printf("pbranch1: calling foo_fn2.\n"); + foo_fn2(); + printf("pbranch1: finished foo_fn2.\n"); + printf("pbranch1: about to sleep for 5 seconds.\n"); + sleep(5); + printf("pbranch1: done.\n"); + printf("pbranch1: done.\n"); + } + + pbranch + { + /* wait 1 second to let the tctest1 call finish */ + printf("pbranch2: setting timer for 1 second.\n"); + triton_timer(1000); + printf("pbranch2: timer done.\n"); + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + pwait + { + pbranch + { + printf("pbranch1: calling tctest1.\n"); + ret1 = tctest1(&a); + printf("pbranch1: finished tctest1.\n"); + /* tctest1 runs the callback in a thread. Sleep here to let + * the other pbranch try to cancel before we finish this + * pbranch. + */ + printf("pbranch1: about to sleep for 5 seconds.\n"); + sleep(5); + printf("pbranch1: done.\n"); + printf("pbranch1: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch1: done.\n"); + } + + pbranch + { + /* wait 1 second to let the tctest1 call finish */ + printf("pbranch2: setting timer for 1 second.\n"); + triton_timer(1000); + printf("pbranch2: timer done.\n"); + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + + +} + +__blocking int aesop_main(int argc, char **argv) +{ + run_test(); + return 0; +} +btest_main_set(aesop_main, "btest", "timer"); + diff --git a/code/src/aesop/parser/tests/blocking/cancel-pbranch2.ae b/code/src/aesop/parser/tests/blocking/cancel-pbranch2.ae new file mode 100644 index 0000000..850783e --- /dev/null +++ b/code/src/aesop/parser/tests/blocking/cancel-pbranch2.ae @@ -0,0 +1,57 @@ +#include <unistd.h> +#include "src/aesop/aesop.h" +#include "src/aesop/parser/tests/blocking/btest.hae" +#include <assert.h> +#include "src/aesop/aesop-support.hae" + +static __blocking void run_test(void) +{ + int ret1; + int ret2; + int a = 0; + + pwait + { + pbranch + { + printf("pbranch1: calling tctest1.\n"); + ret1 = tctest1(&a); + printf("pbranch1: finished tctest1.\n"); + /* tctest1 runs the callback in a thread. Sleep here to let + * the other pbranch try to cancel before we finish this + * pbranch. + */ + printf("pbranch1: about to sleep for 5 seconds.\n"); + sleep(5); + printf("pbranch1: done.\n"); + printf("pbranch1: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch1: done.\n"); + } + + pbranch + { + /* wait 1 second to let the tctest1 call finish */ + printf("pbranch2: setting timer for 1 second.\n"); + triton_timer(1000); + printf("pbranch2: timer done.\n"); + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + + +} + +__blocking int aesop_main(int argc, char **argv) +{ + run_test(); + return 0; +} +btest_main_set(aesop_main, "btest", "timer"); + diff --git a/code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae b/code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae new file mode 100644 index 0000000..638c087 --- /dev/null +++ b/code/src/aesop/parser/tests/blocking/cancel-pbranch3.ae @@ -0,0 +1,45 @@ +#include <unistd.h> +#include "src/aesop/aesop.h" +#include "src/aesop/parser/tests/blocking/btest.hae" +#include <assert.h> +#include "src/aesop/aesop-support.hae" + +static __blocking void run_test(void) +{ + int ret1; + int ret2; + int a = 0; + + pwait + { + pbranch + { + printf("pbranch1: calling triton_timer().\n"); + triton_timer(5); + printf("pbranch1: finished triton_timer().\n"); + + sleep(5); + } + + pbranch + { + /* cancel the other pbranch */ + printf("pbranch2: calling aesop_cancel_branches_wait().\n"); + aesop_cancel_branches_wait(); + printf("pbranch2: done.\n"); + } + } + + printf("pwait done\n"); + + + +} + +__blocking int aesop_main(int argc, char **argv) +{ + run_test(); + return 0; +} +btest_main_set(aesop_main, "btest", "timer"); + diff --git a/code/src/aesop/parser/tests/blocking/module.mk.in b/code/src/aesop/parser/tests/blocking/module.mk.in index 2f7cba5..54d9bfd 100644 --- a/code/src/aesop/parser/tests/blocking/module.mk.in +++ b/code/src/aesop/parser/tests/blocking/module.mk.in @@ -53,7 +53,10 @@ AETESTSRC += $(DIR)/icpfor.ae \ $(DIR)/lonely-pbranch-variable-scope.ae \ $(DIR)/pwait-in-while.ae \ $(DIR)/pwait-in-for.ae \ - $(DIR)/pbranch-completion.ae + $(DIR)/pbranch-completion.ae \ + $(DIR)/cancel-pbranch1.ae \ + $(DIR)/cancel-pbranch2.ae \ + $(DIR)/cancel-pbranch3.ae # This should not compile # $(DIR)/pbranch-with-return.ae hooks/post-receive -- Triton Repository

1 0