From 6ade2a5481044710efe098e6b8665efc1b73fc91 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Fri, 19 Jul 2013 23:53:21 -0700 Subject: [PATCH 1/9] local delivery performance counter --- system/Makefile | 2 +- system/MessageBase.cpp | 2 ++ system/MessageBase.hpp | 3 +++ system/RDMAAggregator.cpp | 12 ++++++++++++ 4 files changed, 18 insertions(+), 1 deletion(-) diff --git a/system/Makefile b/system/Makefile index 9eacd8a50..a6128a244 100644 --- a/system/Makefile +++ b/system/Makefile @@ -250,7 +250,7 @@ ifdef GOOGLE_PROFILER #NOTE: no output is generated by default unless you call the API or set this variable: #ENV_VARIABLES+= CPUPROFILE="\$$SLURM_JOB_NAME.\$$SLURM_JOB_ID.\$$SLURM_PROCID.prof" # how often should we sample? (interrupts/sec) -ENV_VARIABLES+= CPUPROFILE_FREQUENCY=100 +ENV_VARIABLES+= CPUPROFILE_FREQUENCY=10 CFLAGS+= -fno-omit-frame-pointer -DGOOGLE_PROFILER LIBS+= -lprofiler endif diff --git a/system/MessageBase.cpp b/system/MessageBase.cpp index 9d43a37dd..9dabc34bf 100644 --- a/system/MessageBase.cpp +++ b/system/MessageBase.cpp @@ -6,6 +6,8 @@ #include "ConditionVariable.hpp" +GRAPPA_DEFINE_STAT( SimpleStatistic, mark_sent_enqueues, 0 ); + namespace Grappa { /// Internal messaging functions diff --git a/system/MessageBase.hpp b/system/MessageBase.hpp index af10a4768..75568fc64 100644 --- a/system/MessageBase.hpp +++ b/system/MessageBase.hpp @@ -20,6 +20,8 @@ typedef int16_t Core; +GRAPPA_DECLARE_STAT( SimpleStatistic, mark_sent_enqueues ); + namespace Grappa { /// Internal messaging functions @@ -74,6 +76,7 @@ namespace Grappa { if( (is_delivered_ == true) && (Grappa::mycore() != source_) ) { DVLOG(5) << __func__ << ": " << this << " Re-enqueuing to " << source_; DCHECK_EQ( this->is_sent_, false ); + mark_sent_enqueues++; enqueue( source_ ); } else { diff --git a/system/RDMAAggregator.cpp b/system/RDMAAggregator.cpp index 2a3ca32fa..d25b05b52 100644 --- a/system/RDMAAggregator.cpp +++ b/system/RDMAAggregator.cpp @@ -236,6 +236,18 @@ namespace Grappa { enqueue_buffer_handle_ = global_communicator.register_active_message_handler( &enqueue_buffer_am ); enqueue_buffer_async_handle_ = global_communicator.register_active_message_handler( &enqueue_buffer_async_am ); copy_enqueue_buffer_handle_ = global_communicator.register_active_message_handler( ©_enqueue_buffer_am ); + + // make sure buffer count and workers per core are sufficient + if( FLAGS_rdma_workers_per_core < remote_buffer_pool_size ) { + LOG(WARNING) << "Updating workers_per_core to match remote_buffer_pool_size" + << " (was " << FLAGS_rdma_workers_per_core << ", now " << remote_buffer_pool_size << ")"; + FLAGS_rdma_workers_per_core = remote_buffer_pool_size; + } + if( FLAGS_rdma_buffers_per_core < remote_buffer_pool_size * 2 ) { + LOG(WARNING) << "Updating buffers_per_core to match remote_buffer_pool_size" + << " (was " << FLAGS_rdma_buffers_per_core << ", now " << remote_buffer_pool_size*2 << ")"; + FLAGS_rdma_buffers_per_core = remote_buffer_pool_size * 2; + } #endif } From 6de608ce15e6744cd7ca32e71a7578381697a860 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Sat, 20 Jul 2013 01:57:51 -0700 Subject: [PATCH 2/9] do not overwrite trace files --- system/Makefile | 1 + 1 file changed, 1 insertion(+) diff --git a/system/Makefile b/system/Makefile index a6128a244..eefa66a77 100644 --- a/system/Makefile +++ b/system/Makefile @@ -266,6 +266,7 @@ endif ifdef VTRACE_SAMPLED #ENV_VARIABLES+= VT_VERBOSE=10 #ENV_VARIABLES+= VT_BUFFER_SIZE=512M +ENV_VARIABLES+= VT_FILE_UNIQUE=yes ENV_VARIABLES+= VT_MAX_FLUSHES=0 ENV_VARIABLES+= VT_PFORM_GDIR=. ENV_VARIABLES+= VT_PFORM_LDIR=/scratch From 291f44346aed8f281270c8a64c2d2dbf71fa4d3a Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Sat, 20 Jul 2013 03:16:26 -0700 Subject: [PATCH 3/9] add high-priority receive tasks --- system/ConditionVariableLocal.hpp | 11 +++++++++++ system/RDMAAggregator.hpp | 2 ++ system/ReuseList.hpp | 2 +- system/tasks/TaskingScheduler.cpp | 2 ++ system/tasks/TaskingScheduler.hpp | 28 ++++++++++++++++++++++++++++ 5 files changed, 44 insertions(+), 1 deletion(-) diff --git a/system/ConditionVariableLocal.hpp b/system/ConditionVariableLocal.hpp index 80d65dbdc..7fb2af69b 100644 --- a/system/ConditionVariableLocal.hpp +++ b/system/ConditionVariableLocal.hpp @@ -80,6 +80,17 @@ namespace Grappa { } } + /// Wake one waiter on a condition variable. + template< typename ConditionVariable > + inline void signal_hip( ConditionVariable * cv ) { + Worker * to_wake = Grappa::impl::get_waiters( cv ); + if( to_wake != NULL ) { + Grappa::impl::set_waiters( cv, to_wake->next ); + to_wake->next = NULL; + impl::global_scheduler.thread_hip_wake( to_wake ); + } + } + /// Wake all waiters on a condition variable. template< typename ConditionVariable > inline void broadcast( ConditionVariable * cv ) { diff --git a/system/RDMAAggregator.hpp b/system/RDMAAggregator.hpp index 22e31a615..7e8c560fb 100644 --- a/system/RDMAAggregator.hpp +++ b/system/RDMAAggregator.hpp @@ -525,6 +525,7 @@ namespace Grappa { Core core = locale * Grappa::locale_cores(); if( check_for_work_on( locale, FLAGS_target_size ) ) { useful = true; + //Grappa::signal_hip( &(localeCoreData(core)->send_cv_) ); Grappa::signal( &(localeCoreData(core)->send_cv_) ); //send_locale_medium( locale ); } @@ -737,6 +738,7 @@ namespace Grappa { rdma_requested_flushes++; Locale locale = Grappa::locale_of(c); if( source_core_for_locale_[ locale ] == Grappa::mycore() ) { + //Grappa::signal_hip( &(localeCoreData( locale * Grappa::locale_cores() )->send_cv_) ); Grappa::signal( &(localeCoreData( locale * Grappa::locale_cores() )->send_cv_) ); } else { // not on our core, so we can't signal it. instead, cause diff --git a/system/ReuseList.hpp b/system/ReuseList.hpp index 61290a11a..90360ebdd 100644 --- a/system/ReuseList.hpp +++ b/system/ReuseList.hpp @@ -83,7 +83,7 @@ class ReuseList { b->set_next( list_ ); list_ = b; - Grappa::signal( &cv ); + Grappa::signal_hip( &cv ); // record the put count_++; diff --git a/system/tasks/TaskingScheduler.cpp b/system/tasks/TaskingScheduler.cpp index 8c85c98a4..9bb623681 100644 --- a/system/tasks/TaskingScheduler.cpp +++ b/system/tasks/TaskingScheduler.cpp @@ -34,6 +34,7 @@ TaskingScheduler global_scheduler; /// init() must subsequently be called before fully initialized. TaskingScheduler::TaskingScheduler ( ) : readyQ ( ) + , hipQ ( ) , periodicQ ( ) , unassignedQ ( ) , master ( NULL ) @@ -58,6 +59,7 @@ TaskingScheduler global_scheduler; void TaskingScheduler::init ( Thread * master_arg, TaskManager * taskman ) { master = master_arg; readyQ.init( FLAGS_readyq_prefetch_distance ); + hipQ.init( FLAGS_readyq_prefetch_distance ); current_thread = master; task_manager = taskman; work_args = new task_worker_args( taskman, this ); diff --git a/system/tasks/TaskingScheduler.hpp b/system/tasks/TaskingScheduler.hpp index b34f32040..5ac4d7674 100644 --- a/system/tasks/TaskingScheduler.hpp +++ b/system/tasks/TaskingScheduler.hpp @@ -70,6 +70,7 @@ class TaskingScheduler : public Scheduler { private: /// Queue for Threads that are ready to run PrefetchingThreadQueue readyQ; + PrefetchingThreadQueue hipQ; /// Queue for Threads that are to run periodically ThreadQueue periodicQ; @@ -178,6 +179,16 @@ class TaskingScheduler : public Scheduler { } + // check ready tasks + result = hipQ.dequeue(); + if (result != NULL) { + // DVLOG(5) << current_thread->id << " scheduler: pick ready"; + stats.state_timers[ stats.prev_state ] += (current_ts - prev_ts) / tick_scale; + stats.prev_state = TaskingSchedulerStatistics::StateReady; + prev_ts = current_ts; + return result; + } + // check ready tasks result = readyQ.dequeue(); if (result != NULL) { @@ -464,6 +475,7 @@ class TaskingScheduler : public Scheduler { bool thread_yield_periodic( ); void thread_suspend( ); void thread_wake( Thread * next ); + void thread_hip_wake( Thread * next ); void thread_yield_wake( Thread * next ); void thread_suspend_wake( Thread * next ); bool thread_idle( uint64_t total_idle ); @@ -577,6 +589,22 @@ inline void TaskingScheduler::thread_wake( Thread * next ) { ready( next ); } +/// Wake a suspended Thread by putting it on the run queue. +/// For now, waking a running Thread is a fatal error. +/// For now, waking a queued Thread is also a fatal error. +/// For now, can only wake a Thread on your scheduler +inline void TaskingScheduler::thread_hip_wake( Thread * next ) { + CHECK( next->sched == this ) << "can only wake a Thread on your scheduler (next="<<(void*) next << " next->sched="<<(void*)next->sched <<" this="<<(void*)this; + CHECK( next->next == NULL ) << "woken Thread should not be on any queue"; + CHECK( !next->running ) << "woken Thread should not be running"; + + next->suspended = 0; + + DVLOG(5) << "Thread " << current_thread->id << " wakes thread " << next->id; + + hipQ.enqueue( next ); +} + /// Yield the current Thread and wake a suspended thread. /// For now, waking a running Thread is a fatal error. /// For now, waking a queued Thread is also a fatal error. From ca61bfa615fd35cf07a17cad509d35de8d267d9e Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Mon, 22 Jul 2013 23:38:01 -0700 Subject: [PATCH 4/9] add support for replies. works with per-core aggregation, but does not scale beyond 6 cores. --- system/Communicator.cpp | 3 ++ system/Communicator.hpp | 24 +++++++++ system/MessageBase.hpp | 3 ++ system/MessageBaseImpl.hpp | 15 ++++++ system/RDMAAggregator.cpp | 104 +++++++++++++++++++++++++++++++++---- system/RDMAAggregator.hpp | 29 +++++++++++ 6 files changed, 167 insertions(+), 11 deletions(-) diff --git a/system/Communicator.cpp b/system/Communicator.cpp index 8f8e6bdb5..06025d097 100644 --- a/system/Communicator.cpp +++ b/system/Communicator.cpp @@ -21,6 +21,9 @@ extern HeapLeakChecker * Grappa_heapchecker; /// Global communicator instance Communicator global_communicator; +gasnet_token_t global_gasnet_token = 0; + + /// declare labels for histogram std::string CommunicatorStatistics::hist_labels[16] = { "\"comm_0_to_255_bytes\"", diff --git a/system/Communicator.hpp b/system/Communicator.hpp index 3e9aa1be3..a18e8f912 100644 --- a/system/Communicator.hpp +++ b/system/Communicator.hpp @@ -71,6 +71,9 @@ typedef Core Node; /// some variables from gasnet extern size_t gasnetc_inline_limit; +//typedef void *gasnet_token_t; +extern gasnet_token_t global_gasnet_token; + /* from gasnet_internal.c: * gasneti_nodemap_local_count = number of GASNet nodes collocated w/ gasneti_mynode * gasneti_nodemap_local_rank = rank of gasneti_mynode among gasneti_nodemap_local_count @@ -440,6 +443,27 @@ class Communicator { GASNET_CHECK( gasnet_AMRequestMedium0( node, handler, buf, size ) ); } + /// Reply with no-argment active message with payload + inline void reply( gasnet_token_t token, int handler, void * buf, size_t size ) { + assert( communication_is_allowed_ ); + assert( size < maximum_message_payload_size ); // make sure payload isn't too big + stats.record_message( size ); +#ifdef VTRACE_FULL + VT_COUNT_UNSIGNED_VAL( send_ev_vt, size ); +#endif + GASNET_CHECK( gasnet_AMReplyMedium0( token, handler, buf, size ) ); + } + + /// Reply with no-argment active message with payload + inline void reply( gasnet_token_t token, int handler ) { + assert( communication_is_allowed_ ); + stats.record_message( 0 ); +#ifdef VTRACE_FULL + VT_COUNT_UNSIGNED_VAL( send_ev_vt, size ); +#endif + GASNET_CHECK( gasnet_AMReplyShort0( token, handler ) ); + } + /// Send no-argment active message with payload. This only allows /// messages will be immediately copied to the HCA. /// TODO: can we avoid the copy onto gasnet's buffer? This is so small it probably doesn't matter. diff --git a/system/MessageBase.hpp b/system/MessageBase.hpp index 75568fc64..6e106edb1 100644 --- a/system/MessageBase.hpp +++ b/system/MessageBase.hpp @@ -237,6 +237,9 @@ namespace Grappa { inline void send_immediate(); inline void send_immediate( Core c ); + inline void reply_immediate( gasnet_token_t token ); + inline void reply_immediate( gasnet_token_t token, Core c ); + virtual void deliver_locally() = 0; inline void delete_after_send() { diff --git a/system/MessageBaseImpl.hpp b/system/MessageBaseImpl.hpp index 28135944a..fc7dc7bfd 100644 --- a/system/MessageBaseImpl.hpp +++ b/system/MessageBaseImpl.hpp @@ -98,6 +98,21 @@ namespace Grappa { send_immediate(); } + inline void Grappa::impl::MessageBase::reply_immediate( gasnet_token_t token ) { + CHECK( !is_moved_ ) << "Shouldn't be sending a message that has been moved!" + << " Your compiler's return value optimization failed you here."; + if( !is_enqueued_ ) source_ = global_communicator.mycore(); + is_enqueued_ = true; + is_delivered_ = true; +#ifndef LEGACY_SEND + Grappa::impl::global_rdma_aggregator.reply_immediate( this, token ); +#endif +#ifdef LEGACY_SEND + CHECK_EQ( true, false ) << "not supported"; + legacy_send(); +#endif + } + /// @} } } diff --git a/system/RDMAAggregator.cpp b/system/RDMAAggregator.cpp index d25b05b52..aba04355c 100644 --- a/system/RDMAAggregator.cpp +++ b/system/RDMAAggregator.cpp @@ -234,6 +234,7 @@ namespace Grappa { deserialize_buffer_handle_ = global_communicator.register_active_message_handler( &deserialize_buffer_am ); deserialize_first_handle_ = global_communicator.register_active_message_handler( &deserialize_first_am ); enqueue_buffer_handle_ = global_communicator.register_active_message_handler( &enqueue_buffer_am ); + null_reply_handle_ = global_communicator.register_active_message_handler( &null_reply_am ); enqueue_buffer_async_handle_ = global_communicator.register_active_message_handler( &enqueue_buffer_async_am ); copy_enqueue_buffer_handle_ = global_communicator.register_active_message_handler( ©_enqueue_buffer_am ); @@ -425,6 +426,7 @@ namespace Grappa { #endif app_messages_deserialized++; Grappa::impl::global_scheduler.set_no_switch_region( true ); + global_gasnet_token = token; Grappa::impl::MessageBase::deserialize_and_call( static_cast< char * >( buf ) ); Grappa::impl::global_scheduler.set_no_switch_region( false ); } @@ -446,24 +448,52 @@ namespace Grappa { global_rdma_aggregator.received_buffer_list_.push( b ); + + } + + + void RDMAAggregator::null_reply_am( gasnet_token_t token ) { + ; } void RDMAAggregator::enqueue_buffer_async_am( gasnet_token_t token, void * buf, size_t size ) { -#ifdef DEBUG - gasnet_node_t src = -1; - gasnet_AMGetMsgSource(token,&src); - DVLOG(5) << __func__ << ": Receiving buffer of size " << size << " from " << src << " through gasnet; enqueuing for deserialization"; -#endif RDMABuffer * b = reinterpret_cast< RDMABuffer * >( buf ); + Core source = b->get_source(); + Core dest = b->get_dest(); RDMABuffer * ack = b->get_ack(); + DVLOG(5) << __func__ << ": Receiving buffer of size " << size + << " from " << b->get_source() << " to " << b->get_dest() + << " at " << buf << " with ack " << b->get_ack() + << " core0 count " << (b->get_counts())[0] + << " core1 count " << (b->get_counts())[1] + << "; enqueuing for deserialization"; + + global_rdma_aggregator.received_buffer_list_.push( b ); + auto n = Grappa::message( source, + [ack] { + global_rdma_aggregator.free_buffer_list_.push(ack); - // send reply as required by spec - auto reply = Grappa::message( b->get_source(), [ack] { - global_rdma_aggregator.free_buffer_list_.push( ack ); - } ); + } ); + n.reply_immediate( token ); } +// void RDMAAggregator::enqueue_buffer_async_am( gasnet_token_t token, void * buf, size_t size ) { +// #ifdef DEBUG +// gasnet_node_t src = -1; +// gasnet_AMGetMsgSource(token,&src); +// DVLOG(5) << __func__ << ": Receiving buffer of size " << size << " from " << src << " through gasnet; enqueuing for deserialization"; +// #endif +// RDMABuffer * b = reinterpret_cast< RDMABuffer * >( buf ); +// RDMABuffer * ack = b->get_ack(); +// global_rdma_aggregator.received_buffer_list_.push( b ); + +// // send reply as required by spec +// auto reply = Grappa::message( b->get_source(), [ack] { +// global_rdma_aggregator.free_buffer_list_.push( ack ); +// } ); +// } + void RDMAAggregator::copy_enqueue_buffer_am( gasnet_token_t token, void * buf, size_t size ) { #ifdef DEBUG gasnet_node_t src = -1; @@ -507,8 +537,6 @@ namespace Grappa { - - void RDMAAggregator::draw_routing_graph() { { if( Grappa::mycore() == 0 ) { @@ -907,6 +935,59 @@ void RDMAAggregator::draw_routing_graph() { } + // void RDMAAggregator::receive_buffer( RDMABuffer * buf ) { + + // uint64_t sequence_number = reinterpret_cast< uint64_t >( buf->get_ack() ); + + // DVLOG(4) << __func__ << "/" << sequence_number << ": Now deaggregating buffer " << buf << " sequence " << sequence_number + // << " core0 count " << (buf->get_counts())[0] + // << " core1 count " << (buf->get_counts())[1] + // << " from core " << buf->get_source(); + + // // send messages to deaggregate everybody's chunks + + // // index array from buffer + // uint16_t * counts = buf->get_counts(); + // char * current_buf = buf->get_payload(); + // char * my_buf = NULL; + // int outstanding = 0; + + // Core mycore = Grappa::mycore(); + // Grappa::CompletionEvent ce( Grappa::locale_cores() ); + + // // fill and send deaggregate messages + // for( Core locale_core = 0; locale_core < Grappa::locale_cores(); ++locale_core ) { + // DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf + // << ": found " << counts[ locale_core ] << " bytes for core " << locale_core << " at " << (void*) current_buf; + + // if( counts[locale_core] > 0 ) { + + // char * buf = current_buf; + // size_t size = counts[locale_core]; + + // auto m = Grappa::message( locale_core + Grappa::mylocale() * Grappa::locale_cores(), + // [&ce,mycore,buf,size] { + // Grappa::impl::global_scheduler.set_no_switch_region( true ); + // global_rdma_aggregator.deaggregate_buffer( buf, size ); + // Grappa::impl::global_scheduler.set_no_switch_region( false ); + // auto n = Grappa::message( mycore, + // [&ce] { + // ce.complete(); + // } ); + // n.reply_immediate( global_gasnet_token ); + // } ); + // m.send_immediate(); + + // current_buf += counts[locale_core]; + // } else { + // ce.complete(); + // } + // } + + // ce.wait(); + // } + + @@ -1277,6 +1358,7 @@ void RDMAAggregator::draw_routing_graph() { if( aggregated_size > 0 ) { // we have a buffer. send. global_communicator.send( dest_core, enqueue_buffer_handle_, b->get_base(), aggregated_size + b->get_base_size(), dest_buf ); + // global_communicator.send_async( dest_core, enqueue_buffer_handle_, b->get_base(), aggregated_size + b->get_base_size(), dest_buf ); rdma_message_bytes += aggregated_size + b->get_base_size(); bytes_sent += aggregated_size + b->get_base_size(); diff --git a/system/RDMAAggregator.hpp b/system/RDMAAggregator.hpp index 7e8c560fb..344c2235f 100644 --- a/system/RDMAAggregator.hpp +++ b/system/RDMAAggregator.hpp @@ -237,6 +237,10 @@ namespace Grappa { static void enqueue_buffer_am( gasnet_token_t token, void * buf, size_t size ); int enqueue_buffer_handle_; + /// Null reply + static void null_reply_am( gasnet_token_t token ); + int null_reply_handle_; + /// Active message to enqueue a buffer to be received and send a reply to meet the spec static void enqueue_buffer_async_am( gasnet_token_t token, void * buf, size_t size ); int enqueue_buffer_async_handle_; @@ -447,6 +451,7 @@ namespace Grappa { , cores_(NULL) , deserialize_buffer_handle_( -1 ) , deserialize_first_handle_( -1 ) + , null_reply_handle_( -1 ) , enqueue_buffer_async_handle_( -1 ) , enqueue_buffer_handle_( -1 ) , copy_enqueue_buffer_handle_( -1 ) @@ -733,6 +738,30 @@ namespace Grappa { } } + /// send a message that will be run in active message context. This requires very limited messages. + void reply_immediate( Grappa::impl::MessageBase * m, gasnet_token_t token ) { + app_messages_immediate++; + + // create temporary buffer + const size_t size = m->serialized_size(); + char buf[ size ] __attribute__ ((aligned (16))); + + // serialize to buffer + Grappa::impl::MessageBase * tmp = m; + while( tmp != nullptr ) { + DVLOG(5) << __func__ << ": Serializing message from " << tmp; + char * end = aggregate_to_buffer( &buf[0], &tmp, size ); + DVLOG(5) << __func__ << ": After serializing, pointer was " << tmp; + DCHECK_EQ( end - buf, size ) << __func__ << ": Whoops! Aggregated message was too long to send as immediate"; + + DVLOG(5) << __func__ << ": Sending " << end - buf + << " bytes of aggregated messages to " << m->destination_; + + // send + global_communicator.reply( token, deserialize_first_handle_, buf, size ); + } + } + /// Flush one destination. void flush( Core c ) { rdma_requested_flushes++; From 5aa648d794419e05eecc28e05c2324adc6b17686 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Mon, 22 Jul 2013 23:41:13 -0700 Subject: [PATCH 5/9] add compiler memory barrier --- system/common.hpp | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/system/common.hpp b/system/common.hpp index 4345e5d9c..16802e5f8 100644 --- a/system/common.hpp +++ b/system/common.hpp @@ -26,6 +26,10 @@ #define BILLION 1000000000 #define MILLION 1000000 +// compiler memory barrier +#define CMB asm volatile("" ::: "memory"); + + /// "Universal" wallclock time (works at least for Mac, MTA, and most Linux) inline double Grappa_walltime(void) { #if defined(__MTA__) From fda37ad26088b6dfdbf9dfa952bc475a59e27b2a Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Tue, 23 Jul 2013 00:33:38 -0700 Subject: [PATCH 6/9] non-functional per-source-core aggregation --- system/MessageBase.hpp | 4 +- system/MessageBaseImpl.hpp | 2 +- system/RDMAAggregator.cpp | 178 +++++++++++++++---------------------- system/RDMAAggregator.hpp | 91 ++++++++++--------- 4 files changed, 124 insertions(+), 151 deletions(-) diff --git a/system/MessageBase.hpp b/system/MessageBase.hpp index 6e106edb1..b6718abf8 100644 --- a/system/MessageBase.hpp +++ b/system/MessageBase.hpp @@ -77,8 +77,8 @@ namespace Grappa { DVLOG(5) << __func__ << ": " << this << " Re-enqueuing to " << source_; DCHECK_EQ( this->is_sent_, false ); mark_sent_enqueues++; - enqueue( source_ ); - + // should only happen for deaggregation now, so enqueue to locale + locale_enqueue( source_ ); } else { DVLOG(5) << __func__ << ": " << this << " Final mark_sent"; DCHECK_EQ( Grappa::mycore(), this->source_ ); diff --git a/system/MessageBaseImpl.hpp b/system/MessageBaseImpl.hpp index fc7dc7bfd..c1a782760 100644 --- a/system/MessageBaseImpl.hpp +++ b/system/MessageBaseImpl.hpp @@ -65,7 +65,7 @@ namespace Grappa { is_enqueued_ = true; DVLOG(5) << this << " on " << Grappa::impl::global_scheduler.get_current_thread() << ": " << this->typestr() - << " enqueuing to " << destination_ << " with is_enqueued_=" << is_enqueued_ << " and is_sent_= " << is_sent_; + << " enqueuing to " << destination_ << " from " << source_ << " with is_enqueued_=" << is_enqueued_ << " and is_sent_= " << is_sent_; #ifndef LEGACY_SEND Grappa::impl::global_rdma_aggregator.enqueue( this, true ); #endif diff --git a/system/RDMAAggregator.cpp b/system/RDMAAggregator.cpp index aba04355c..1629bff7a 100644 --- a/system/RDMAAggregator.cpp +++ b/system/RDMAAggregator.cpp @@ -11,7 +11,7 @@ #include "RDMAAggregator.hpp" #include "Message.hpp" - +#include "CompletionEvent.hpp" namespace Grappa { @@ -26,6 +26,8 @@ void failure_function(); // prefetch two cache lines per message #define DOUBLE_LINE_PREFETCH +DECLARE_bool( flush_on_idle ); + DEFINE_int64( target_size, 1 << 12, "Target size for aggregated messages" ); DEFINE_int64( rdma_workers_per_core, 1 << 4, "Number of RDMA deaggregation worker threads" ); @@ -84,6 +86,8 @@ GRAPPA_DEFINE_STAT( SimpleStatistic, rdma_flush_receive, 0 ); GRAPPA_DEFINE_STAT( SummarizingStatistic, rdma_local_delivery_time, 0 ); +GRAPPA_DEFINE_STAT( SimpleStatistic, send_blocked_on_remote_buffer, 0 ); +GRAPPA_DEFINE_STAT( SimpleStatistic, send_blocked_on_local_buffer, 0 ); GRAPPA_DEFINE_STAT(HistogramStatistic, app_bytes_sent_histogram, 0); @@ -115,6 +119,7 @@ namespace Grappa { while( !Grappa_done_flag ) { Grappa::wait( &flush_cv_ ); rdma_idle_flushes++; + DVLOG(5) << "Idle flusher running...."; Core c = Grappa::mycore(); @@ -123,16 +128,14 @@ namespace Grappa { // receive_poll(); if( !disable_everything_ ) { - - // see if we have anything at all to send + // see if we have anything to send for( int i = 0; i < core_partner_locale_count_; ++i ) { Locale locale = core_partner_locales_[i]; - // if( check_for_any_work_on( locale ) ) { - Grappa::signal( &cores_[ locale * Grappa::locale_cores() ].send_cv_ ); - // } + Core core = locale * Grappa::locale_cores(); + Grappa::signal( &(coreData(core)->send_cv_) ); } - } + DVLOG(5) << "Idle flusher done."; } } @@ -146,49 +149,14 @@ namespace Grappa { // (assume all locales have same core count) // (make we have at least one locale per core) // (round up) - Locale locales_per_core = 1 + ((Grappa::locales() - 1) / Grappa::locale_cores()); - - - // initialize source cores - //source_core_for_locale_ = new Core[ Grappa::locales() ]; - for( int i = 0; i < Grappa::locales(); ++i ) { - if( i == Grappa::mylocale() ) { - // locally, cores are responsible for their own messages - source_core_for_locale_[i] = -1; - } else { - // guess at correct source core on the node - Core offset = i / locales_per_core; - if( offset >= Grappa::locale_cores() ) { // if we guess too high - // give it to the core that would have been responsible for the local locale. - offset = Grappa::mylocale() / locales_per_core; - } - source_core_for_locale_[i] = Grappa::mylocale() * Grappa::locale_cores() + offset; - } - } + Locale locales_per_core = Grappa::locale_cores(); - // initialize destination cores. - // this should match up with source assignments. - //dest_core_for_locale_ = new Core[ Grappa::locales() ]; - for( int i = 0; i < Grappa::locales(); ++i ) { - if( i == Grappa::mylocale() ) { - dest_core_for_locale_[i] = -1; - } else { - // guess at correct destination core on the node - Core offset = Grappa::mylocale() / locales_per_core; - if( offset >= Grappa::locale_cores() ) { // if we guess too high - // give it to the core that would have been responsible for - // the destination's locale. - offset = i / locales_per_core; - } - dest_core_for_locale_[i] = i * Grappa::locale_cores() + offset; - - } - } + // lots of code deleted; this is mostly useless now. for( int i = 0; i < Grappa::locales(); ++i ) { - DVLOG(2) << "From locale " << Grappa::mylocale() << " to locale " << i - << " source core " << source_core_for_locale_[i] - << " dest core " << dest_core_for_locale_[i]; + DVLOG(2) << "From locale " << Grappa::mylocale() << " to locale " << i ; + // << " source core " << source_core_for_locale_[i] + // << " dest core " << dest_core_for_locale_[i]; } @@ -196,7 +164,7 @@ namespace Grappa { for( int i = 0; i < Grappa::cores(); ++i ) { Locale l = Grappa::locale_of(i); cores_[ l ].mylocale_ = l; - cores_[ l ].representative_core_ = source_core_for_locale_[ l ]; + cores_[ l ].representative_core_ = Grappa::mycore(); } } @@ -242,7 +210,7 @@ namespace Grappa { if( FLAGS_rdma_workers_per_core < remote_buffer_pool_size ) { LOG(WARNING) << "Updating workers_per_core to match remote_buffer_pool_size" << " (was " << FLAGS_rdma_workers_per_core << ", now " << remote_buffer_pool_size << ")"; - FLAGS_rdma_workers_per_core = remote_buffer_pool_size; + FLAGS_rdma_workers_per_core = Grappa::locales() * remote_buffer_pool_size; } if( FLAGS_rdma_buffers_per_core < remote_buffer_pool_size * 2 ) { LOG(WARNING) << "Updating buffers_per_core to match remote_buffer_pool_size" @@ -261,10 +229,6 @@ namespace Grappa { // one for each core on this locale + one for communication within the locale. cores_ = Grappa::impl::locale_shared_memory.segment.construct("Cores")[global_communicator.cores() * (global_communicator.locale_cores() + 1 )](); - - // allocate routing info - source_core_for_locale_ = Grappa::impl::locale_shared_memory.segment.construct("SourceCores")[global_communicator.locales()](); - dest_core_for_locale_ = Grappa::impl::locale_shared_memory.segment.construct("DestCores")[global_communicator.locales()](); } catch(...){ failure_function(); @@ -273,6 +237,13 @@ namespace Grappa { compute_route_map(); } + source_core_for_locale_ = new Core[global_communicator.locales()]; + dest_core_for_locale_ = new Core[global_communicator.locales()]; + for( int i = 0; i < global_communicator.locales(); ++i ) { + source_core_for_locale_[i] = Grappa::mycore(); + dest_core_for_locale_[i] = i * Grappa::locale_cores() + Grappa::locale_mycore(); + } + // make sure everything is allocated before other cores try to attach global_communicator.barrier(); @@ -284,15 +255,6 @@ namespace Grappa { p = Grappa::impl::locale_shared_memory.segment.find("Cores"); CHECK_EQ( p.second, global_communicator.cores() * (global_communicator.locale_cores() + 1) ); cores_ = p.first; - - // attach to routing info - std::pair< Core *, boost::interprocess::managed_shared_memory::size_type > q; - q = Grappa::impl::locale_shared_memory.segment.find("SourceCores"); - CHECK_EQ( q.second, global_communicator.locales() ); - source_core_for_locale_ = q.first; - q = Grappa::impl::locale_shared_memory.segment.find("DestCores"); - CHECK_EQ( q.second, global_communicator.locales() ); - dest_core_for_locale_ = q.first; } catch(...){ failure_function(); @@ -305,18 +267,17 @@ namespace Grappa { // // draw route map if enabled - draw_routing_graph(); + //draw_routing_graph(); // spread responsibility for locales between our cores, ensuring // that all locales get assigned // (round up) - Locale locales_per_core = 1 + ((Grappa::locales() - 1) / Grappa::locale_cores()); + Locale locales_per_core = Grappa::locale_cores(); // generate list of locales this core is responsible for core_partner_locales_ = new Locale[ locales_per_core ]; for( int i = 0; i < Grappa::locales(); ++i ) { if( source_core_for_locale_[i] == Grappa::mycore() ) { - CHECK_LT( core_partner_locale_count_, locales_per_core ) << "this core is responsible for more locales than expected"; core_partner_locales_[ core_partner_locale_count_++ ] = i; DVLOG(2) << "Core " << Grappa::mycore() << " responsible for locale " << i; } @@ -351,25 +312,24 @@ namespace Grappa { // spawn flusher #ifndef LEGACY_SEND - Grappa::spawn_worker( [this] { - idle_flusher(); - }); + if( FLAGS_flush_on_idle ) { + Grappa::spawn_worker( [this] { + idle_flusher(); + }); + } #endif // // precache buffers // - // for now, give each core 1 buffer/token - size_t expected_buffers = core_partner_locale_count_; // * remote_buffer_pool_size; - for( int i = 0; i < core_partner_locale_count_; ++i ) { Locale locale = core_partner_locales_[i]; Core dest = dest_core_for_locale_[ locale ]; Core mylocale_core = Grappa::mylocale() * Grappa::locale_cores(); Core mycore = Grappa::mycore(); - DVLOG(3) << __PRETTY_FUNCTION__ << ": sending " << expected_buffers << " buffers to core " << dest << " on locale " << locale; + DVLOG(3) << __PRETTY_FUNCTION__ << ": sending " << remote_buffer_pool_size << " buffers to core " << dest << " on locale " << locale; for( int j = 0; j < remote_buffer_pool_size; ++j ) { RDMABuffer * b = global_rdma_aggregator.free_buffer_list_.try_pop(); @@ -377,7 +337,7 @@ namespace Grappa { // (make heap message since we're not on a shared stack) auto request = Grappa::message( dest, [mycore, b] { - auto p = global_rdma_aggregator.localeCoreData(mycore); + auto p = global_rdma_aggregator.coreData(mycore); DVLOG(3) << __PRETTY_FUNCTION__ << " initializing by pushing buffer " << b << " into " << p << " for " << mycore; p->remote_buffers_.push( b ); rdma_buffers_inuse += remote_buffer_pool_size - p->remote_buffers_.count(); @@ -395,9 +355,9 @@ namespace Grappa { global_communicator.barrier(); if( global_communicator.locale_mycore() == 0 ) { Grappa::impl::locale_shared_memory.segment.destroy("Cores"); - Grappa::impl::locale_shared_memory.segment.destroy("SourceCores"); - Grappa::impl::locale_shared_memory.segment.destroy("DestCores"); } + delete [] source_core_for_locale_; + delete [] dest_core_for_locale_; cores_ = NULL; source_core_for_locale_ = NULL; dest_core_for_locale_ = NULL; @@ -703,26 +663,30 @@ void RDMAAggregator::draw_routing_graph() { // block until we're ready to send to a locale and do so void RDMAAggregator::send_worker( Locale locale ) { CoreData * locale_core = localeCoreData( locale * Grappa::locale_cores() ); + CoreData * core = coreData( locale * Grappa::locale_cores() ); while( !Grappa_done_flag ) { // block until it's time to send to this locale Grappa::impl::global_scheduler.assign_time_to_networking(); - Grappa::wait( &(locale_core->send_cv_) ); + DVLOG(5) << "Sender for locale " << locale << " blocking on CV."; + Grappa::wait( &(core->send_cv_) ); + DVLOG(5) << "Sender for locale " << locale << " woke."; active_send_workers_++; //CHECK_EQ( disable_everything_, false ) << "Whoops! Why are we sending when disabled?"; if( disable_everything_ ) LOG(WARNING) << "Sending while disabled..."; - // send to locale - send_locale( locale ); - // record when we last sent // TODO: should this go earlier? probably not. // TODO: should we tick here? Grappa_tick(); - locale_core->last_sent_ = Grappa_get_timestamp(); + core->last_sent_ = Grappa_get_timestamp(); + core->locale_byte_count_ = 0; + + // send to locale + send_locale( locale ); // send done! loop! active_send_workers_--; @@ -757,13 +721,15 @@ void RDMAAggregator::draw_routing_graph() { // once we're done, send ack to give permission to send again, // unless buffer is from the local core (for debugging) - if( c != Grappa::mycore() ) { + //if( c != Grappa::mycore() ) { + if( true ) { Core mylocale_core = Grappa::mylocale() * Grappa::locale_cores(); Core mycore = Grappa::mycore(); + buf->override_dest(mycore); DVLOG(3) << __PRETTY_FUNCTION__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << " finished with " << buf << "; acking with buffer " << buf << " for core " << mycore << " on core " << c; auto request = Grappa::message( c, [mycore, buf] { - auto p = global_rdma_aggregator.localeCoreData(mycore); + auto p = global_rdma_aggregator.coreData(mycore); DVLOG(3) << __PRETTY_FUNCTION__ << " acking by pushing buffer " << buf << " into " << p << " for " << mycore; p->remote_buffers_.push( buf ); rdma_buffers_inuse += remote_buffer_pool_size - p->remote_buffers_.count(); @@ -818,7 +784,7 @@ void RDMAAggregator::draw_routing_graph() { DVLOG(3) << __PRETTY_FUNCTION__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << " finished with " << buf << "; acking with buffer " << buf << " for core " << mycore << " on core " << c; auto request = Grappa::message( c, [mycore, buf] { - auto p = global_rdma_aggregator.localeCoreData(mycore); + auto p = global_rdma_aggregator.coreData(mycore); DVLOG(3) << __PRETTY_FUNCTION__ << " acking by pushing buffer " << buf << " into " << p << " for " << mycore; p->remote_buffers_.push( buf ); rdma_buffers_inuse += remote_buffer_pool_size - p->remote_buffers_.count(); @@ -867,7 +833,9 @@ void RDMAAggregator::draw_routing_graph() { DVLOG(5) << __PRETTY_FUNCTION__ // << "/" << sequence_number << ": received " << size << "-byte buffer slice at " << (void*) buf << " to deaggregate"; + Grappa::impl::global_scheduler.set_no_switch_region( true ); global_rdma_aggregator.deaggregate_buffer( buf, size ); + Grappa::impl::global_scheduler.set_no_switch_region( false ); DVLOG(5) << __PRETTY_FUNCTION__ // << "/" << sequence_number << ": done deaggregating " << size << "-byte buffer slice at " << (void*) buf; @@ -897,7 +865,7 @@ void RDMAAggregator::draw_routing_graph() { //msgs[locale_core]->buf_base = buf; if( locale_core != Grappa::locale_mycore() ) { - DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf + DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf << " message " << (void*) &(msgs[locale_core]) << " enqueuing " << counts[locale_core] << " bytes to locale_core " << locale_core << " core " << locale_core + Grappa::mylocale() * Grappa::locale_cores() << " my locale_core " << Grappa::locale_mycore() << " my core " << Grappa::mycore(); msgs[locale_core].locale_enqueue( locale_core + Grappa::mylocale() * Grappa::locale_cores() ); outstanding++; @@ -1020,11 +988,10 @@ void RDMAAggregator::draw_routing_graph() { #endif } - Grappa::impl::global_scheduler.set_no_switch_region( true ); while( messages_to_send != NULL ) { - // DVLOG(4) << __func__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << ": Delivered message " << messages_to_send - // << " with is_delivered_=" << messages_to_send->is_delivered_ - // << ": " << messages_to_send->typestr(); + DVLOG(4) << __func__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << ": Delivered message " << messages_to_send + << " with is_delivered_=" << messages_to_send->is_delivered_ + << ": " << messages_to_send->typestr(); MessageBase * next = messages_to_send->next_; DVLOG(5) << __PRETTY_FUNCTION__ << ": Processing " << messages_to_send << ", prefetching " << messages_to_send->prefetch_; char * pf = reinterpret_cast< char* >( messages_to_send->prefetch_ ); @@ -1041,7 +1008,6 @@ void RDMAAggregator::draw_routing_graph() { messages_to_send = next; } - Grappa::impl::global_scheduler.set_no_switch_region( false ); DVLOG(5) << __PRETTY_FUNCTION__ << ": Done."; @@ -1190,6 +1156,8 @@ void RDMAAggregator::draw_routing_graph() { active_send_workers_++; static uint64_t sequence_number = Grappa::mycore(); + DVLOG(5) << "Sending to locale " << locale; + int buffers_used_for_send = 0; int64_t bytes_sent = 0; @@ -1202,7 +1170,9 @@ void RDMAAggregator::draw_routing_graph() { Core max_core = first_core + Grappa::locale_cores(); Core current_dest_core = -1; - MessageListChooser mlc( first_core, max_core, 0, Grappa::locale_cores() ); + // cross-dest-node only + MessageListChooser mlc( first_core, max_core, Grappa::locale_mycore(), Grappa::locale_mycore()+1 ); + DVLOG(3) << __PRETTY_FUNCTION__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << " MessageListChooser constructed at " << &mlc; @@ -1217,16 +1187,20 @@ void RDMAAggregator::draw_routing_graph() { // by doing this first, we limit the rate at which we consume buffers from the local free list // as well as allowing the remote node to limit the rate we send Grappa::impl::global_scheduler.assign_time_to_networking(); - RDMABuffer * dest_buf = localeCoreData( dest_core )->remote_buffers_.block_until_pop(); + + send_blocked_on_remote_buffer++; + RDMABuffer * dest_buf = coreData( dest_core )->remote_buffers_.block_until_pop(); + send_blocked_on_remote_buffer--; CHECK_NOTNULL( dest_buf ); - rdma_buffers_inuse += remote_buffer_pool_size - localeCoreData( dest_core )->remote_buffers_.count(); + rdma_buffers_inuse += remote_buffer_pool_size - coreData( dest_core )->remote_buffers_.count(); // now, grab a temporary buffer for local serialization Grappa::impl::global_scheduler.assign_time_to_networking(); + send_blocked_on_local_buffer++; RDMABuffer * b = free_buffer_list_.block_until_pop(); + send_blocked_on_local_buffer--; - // but only use enough bytes that we can fit in a medium AM. const size_t max_size = b->get_max_size(); DVLOG(5) << "Max buffer size is " << max_size; @@ -1237,6 +1211,7 @@ void RDMAAggregator::draw_routing_graph() { // mark as being from this core b->set_source( Grappa::mycore() ); + b->set_dest( dest_core ); char * current_buf = b->get_payload(); int64_t remaining_size = std::min( max_size, b->get_max_size() ); @@ -1330,17 +1305,6 @@ void RDMAAggregator::draw_routing_graph() { << " for locale " << locale << " core " << current_dest_core; } - if( aggregated_size >= 0 ) { - DVLOG(4) << __func__ << "/" << sequence_number << ": " - << "Ready to send buffer " << b << " size " << aggregated_size - << " to locale " << locale << " core " << dest_core; - } else { - DVLOG(4) << __func__ << "/" << sequence_number << ": " - << "Nothing to send in buffer " << b << " size " << aggregated_size - << " to locale " << locale << " core " << dest_core; - break; - } - DVLOG(3) << __func__ << "/" << sequence_number << ": " << "Sending buffer " << b << " sequence " << sequence_number << " size " << aggregated_size << " core0 count " << (b->get_counts())[0] @@ -1375,8 +1339,8 @@ void RDMAAggregator::draw_routing_graph() { DVLOG(4) << __func__ << "/" << sequence_number << ": No message to send; returning dest buffer " << dest_buf << " to pool"; // return buffer to pool - localeCoreData( dest_core )->remote_buffers_.push( dest_buf ); - rdma_buffers_inuse += remote_buffer_pool_size - localeCoreData( dest_core )->remote_buffers_.count(); + coreData( dest_core )->remote_buffers_.push( dest_buf ); + rdma_buffers_inuse += remote_buffer_pool_size - coreData( dest_core )->remote_buffers_.count(); } sequence_number += Grappa::cores(); @@ -1391,6 +1355,8 @@ void RDMAAggregator::draw_routing_graph() { rdma_bytes_sent_histogram = bytes_sent; + DVLOG(5) << "Done sending to locale " << locale; + active_send_workers_--; rdma_send_end++; } diff --git a/system/RDMAAggregator.hpp b/system/RDMAAggregator.hpp index 344c2235f..2a192618b 100644 --- a/system/RDMAAggregator.hpp +++ b/system/RDMAAggregator.hpp @@ -286,7 +286,8 @@ namespace Grappa { /// Grab a list of messages to send inline Grappa::impl::MessageList grab_messages( Core c, Core sender ) { - Grappa::impl::MessageList * dest_ptr = &(coreData( c, sender )->messages_); + CoreData * cd = coreData( c, sender ); + Grappa::impl::MessageList * dest_ptr = &(cd->messages_); Grappa::impl::MessageList old_ml, new_ml; do { @@ -296,11 +297,14 @@ namespace Grappa { // insert current message } while( !__sync_bool_compare_and_swap( &(dest_ptr->raw_), old_ml.raw_, new_ml.raw_ ) ); + cd->locale_byte_count_ = 0; + return old_ml; } inline Grappa::impl::MessageList grab_locale_messages( Core c ) { - Grappa::impl::MessageList * dest_ptr = &(localeCoreData( c )->messages_); + CoreData * lcd = localeCoreData( c ); + Grappa::impl::MessageList * dest_ptr = &(lcd->messages_); Grappa::impl::MessageList old_ml, new_ml; do { @@ -310,6 +314,8 @@ namespace Grappa { // insert current message } while( !__sync_bool_compare_and_swap( &(dest_ptr->raw_), old_ml.raw_, new_ml.raw_ ) ); + lcd->locale_byte_count_ = 0; + return old_ml; } @@ -324,6 +330,8 @@ namespace Grappa { // insert current message } while( !__sync_bool_compare_and_swap( &(dest_ptr->raw_), old_ml.raw_, new_ml.raw_ ) ); + cd->locale_byte_count_ = 0; + return old_ml; } @@ -372,19 +380,30 @@ namespace Grappa { // return ( byte_count > size ); Core c = locale * Grappa::locale_cores(); - // - CHECK_NE( Grappa::mycore(), c ); + bool retval = false; + CoreData * core_data = coreData( c ); + + // ignore if empty + if( core_data->locale_byte_count_ > 0 ) { + DVLOG(5) << "Found " << core_data->locale_byte_count_ << " bytes for locale " << locale; - // have we timed out? - Grappa_Timestamp current_ts = Grappa_get_timestamp(); - if( current_ts - localeCoreData(c)->last_sent_ > FLAGS_aggregator_autoflush_ticks ) { - return true; + // have we timed out? + Grappa_Timestamp current_ts = Grappa_get_timestamp(); + if( current_ts - core_data->last_sent_ > FLAGS_aggregator_autoflush_ticks ) { + DVLOG(5) << "Found timeout for locale " << locale; + retval = true; + } + + // have we reached sufficient size? + // TODO: make better + if( core_data->locale_byte_count_ > FLAGS_target_size ) { + DVLOG(5) << "Found target size for locale " << locale; + retval = true; + } + } - // // have we reached a size limit? - // size_t byte_count = localeCoreData(c)->locale_byte_count_; - // return ( byte_count > size ); - return false; + return retval; } bool check_for_any_work_on( Locale locale ) { @@ -488,34 +507,22 @@ namespace Grappa { if( localeCoreData(c)->messages_.raw_ != 0 ) { useful = true; - DVLOG(4) << "Polling found messages.raw " << (void*) localeCoreData(c)->messages_.raw_ + int64_t mp = localeCoreData(c)->messages_.pointer_ ; + DVLOG(5) << "Polling found messages at " << (void*) mp << " count " << localeCoreData(c)->messages_.count_ ; //Grappa_Timestamp start = Grappa_force_tick(); Grappa::impl::MessageList ml = grab_locale_messages( c ); + + Grappa::impl::global_scheduler.set_no_switch_region( true ); size_t count = deliver_locally( c, ml, localeCoreData( c ) ); + Grappa::impl::global_scheduler.set_no_switch_region( false ); //Grappa_Timestamp elapsed = Grappa_force_tick() - start; //rdma_local_delivery_time += (double) elapsed / tick_rate; } - for( Core locale_source = 0; locale_source < Grappa::locale_cores(); ++locale_source ) { - if( coreData(c, locale_source)->messages_.raw_ != 0 ) { - useful = true; - - DVLOG(4) << "Polling found messages.raw " << (void*) coreData(c,locale_source)->messages_.raw_ - << " count " << coreData(c,locale_source)->messages_.count_ ; - - //Grappa_Timestamp start = Grappa_force_tick(); - - Grappa::impl::MessageList ml = grab_messages( c, locale_source ); - size_t count = deliver_locally( c, ml, coreData( c, locale_source ) ); - - //Grappa_Timestamp elapsed = Grappa_force_tick() - start; - //rdma_local_delivery_time += (double) elapsed / tick_rate; - } - } if( useful ) rdma_poll_receive_success++; return useful; } @@ -529,10 +536,10 @@ namespace Grappa { Locale locale = core_partner_locales_[i]; Core core = locale * Grappa::locale_cores(); if( check_for_work_on( locale, FLAGS_target_size ) ) { + DVLOG(5) << "Looks like there's work on locale " << locale << " (core " << core << ")"; useful = true; - //Grappa::signal_hip( &(localeCoreData(core)->send_cv_) ); - Grappa::signal( &(localeCoreData(core)->send_cv_) ); - //send_locale_medium( locale ); + //Grappa::signal_hip( &(coreData(core)->send_cv_) ); + Grappa::signal( &(coreData(core)->send_cv_) ); } } } @@ -575,9 +582,12 @@ namespace Grappa { // freq = 10; // } + size_t size = m->serialized_size(); + DVLOG(5) << __func__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << ": Enqueued message " << m - << " with is_delivered_=" << m->is_delivered_ + << " with locale_enqueue=" << locale_enqueue + << " is_delivered_=" << m->is_delivered_ << ": " << m->typestr(); // get destination pointer @@ -592,6 +602,7 @@ namespace Grappa { //CoreData * sender = &cores_[ dest->representative_core_ ]; CoreData * locale_core = localeCoreData( Grappa::locale_of( m->destination_ ) * Grappa::locale_cores() ); + CoreData * core_data = coreData( Grappa::locale_of( m->destination_ ) * Grappa::locale_cores() ); //Grappa::impl::MessageBase ** dest_ptr = &dest->messages_; Grappa::impl::MessageList * dest_ptr = &(dest->messages_); @@ -599,7 +610,6 @@ namespace Grappa { // new values computed from previous totals int count = 0; - size_t size = 0; swap_ml.raw_ = 0; bool spawn_send = false; @@ -692,6 +702,9 @@ namespace Grappa { // // warning: racy // locale_core->locale_byte_count_ += size; + if( !locale_enqueue ) { + core_data->locale_byte_count_ += size; + } dest->prefetch_queue_[ count % prefetch_dist ].size_ = size < max_size_ ? size : max_size_-1; set_pointer( &(dest->prefetch_queue_[ count % prefetch_dist ]), m ); @@ -766,14 +779,8 @@ namespace Grappa { void flush( Core c ) { rdma_requested_flushes++; Locale locale = Grappa::locale_of(c); - if( source_core_for_locale_[ locale ] == Grappa::mycore() ) { - //Grappa::signal_hip( &(localeCoreData( locale * Grappa::locale_cores() )->send_cv_) ); - Grappa::signal( &(localeCoreData( locale * Grappa::locale_cores() )->send_cv_) ); - } else { - // not on our core, so we can't signal it. instead, cause - // polling thread to detect timeout. - localeCoreData( locale * Grappa::locale_cores() )->last_sent_ = 0; - } + //Grappa::signal_hip( &(coreData( locale * Grappa::locale_cores() )->send_cv_) ); + Grappa::signal( &(coreData( locale * Grappa::locale_cores() )->send_cv_) ); } /// Initiate an idle flush. From 95071cc742d4ee6c4b18dd4b0b964a22959a5d19 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Tue, 23 Jul 2013 00:34:11 -0700 Subject: [PATCH 7/9] first stab at porting GDB macros to new Worker model; not working yet --- system/grappa_gdb.macros | 165 ++++++++++++++++++--------------------- 1 file changed, 77 insertions(+), 88 deletions(-) diff --git a/system/grappa_gdb.macros b/system/grappa_gdb.macros index f05151cf9..122ae142e 100644 --- a/system/grappa_gdb.macros +++ b/system/grappa_gdb.macros @@ -29,31 +29,31 @@ end document grappa These are the Grappa gdb macros. - queues print both task and coroutine queues + queues print both task and worker queues tasks print public and private task queues public_tasks private_tasks - coroutines print all coroutines - current_coroutine + workers print all workers + current_worker periodic_queue ready_queue unassigned_queue - suspended_coroutines + suspended_workers - backtrace_coroutine backtrace for specific coroutine - btc backtrace for specific coroutine + backtrace_worker backtrace for specific worker + btc backtrace for specific worker - backtrace_coroutine_full backtrace for specific coroutine with locals - btcf backtrace for specific coroutine with locals + backtrace_worker_full backtrace for specific worker with locals + btcf backtrace for specific worker with locals - switch_to_coroutine change $rip and $rsp to coroutine's. + switch_to_worker change $rip and $rsp to worker's. DO NOT CONTINUE UNTIL RESTORE grappa_save_state save real $rip and $rsp grappa_restore_state restore real $rip and $rsp - dumpstack dump the stack (as raw int64_t's) + dumpstack dump the stack (as raw int64_t's) Type "help " for more specific help on a particular macro. Type "show user " to see what the macro is really doing. @@ -74,7 +74,7 @@ define grappa_save_state set $grappa_saved_rsp = $rsp end document grappa_save_state -Save current coroutine $rip and $rsp for later. +Save current worker $rip and $rsp for later. Syntax: (gdb) grappa_save_state end @@ -88,7 +88,7 @@ define grappa_restore_state end end document grappa_restore_state -Restore saved coroutine $rip and $rsp. +Restore saved worker $rip and $rsp. p $Syntax: (gdb) grappa_restore_state end @@ -98,53 +98,42 @@ end -define grappa_coroutine_print - set $grappa_coroutine_print_elem = (coro*) $arg0 - printf " Coroutine %p running=%d suspended=%d IP ", $grappa_coroutine_print_elem, $grappa_coroutine_print_elem->running, $grappa_coroutine_print_elem->suspended - output/a *(int*)($grappa_coroutine_print_elem->stack + $grappa_stack_offset) +define grappa_worker_print + set $grappa_worker_print_elem = (Worker*) $arg0 + printf " Worker %p running=%d suspended=%d IP ", $grappa_worker_print_elem, $grappa_worker_print_elem->running, $grappa_worker_print_elem->suspended + output/a *(int*)($grappa_worker_print_elem->stack + $grappa_stack_offset) printf "\n" end -document grappa_coroutine_print -Prints coroutine data +document grappa_worker_print +Prints worker data Syntax: - (gdb) grappa_coroutine_print + (gdb) grappa_worker_print end -define grappa_coroutines - set $grappa_coroutines_elem = all_coros - while $grappa_coroutines_elem != 0 - grappa_coroutine_print $grappa_coroutines_elem - set $grappa_coroutines_elem = $grappa_coroutines_elem->next +define grappa_workers + set $grappa_workers_elem = all_coros + while $grappa_workers_elem != 0 + grappa_worker_print $grappa_workers_elem + set $grappa_workers_elem = $grappa_workers_elem->next end end -document grappa_coroutines -Prints simple state of all coroutines +document grappa_workers +Prints simple state of all workers Syntax: - (gdb) grappa_coroutines -end - - -define grappa_thread_print - set $grappa_thread_print_elem = (Thread*)($arg0) - grappa_coroutine_print $grappa_thread_print_elem->co -end -document grappa_thread_print -Prints coroutine data from thread. -Syntax: - (gdb) grappa_thread_print + (gdb) grappa_workers end define grappa_queue_print set $grappa_queue = $arg1 - if $grappa_queue->head + if $grappa_queue->queues->head set $grappa_queue_length = 0 else - set $grappa_queue_length = $grappa_queue->len + set $grappa_queue_length = $grappa_queue->queues->len end - printf "%s queue has length %d:\n", $arg0, $grappa_queue->len - set $grappa_queue_print_elem = $grappa_queue->head + printf "%s queue has length %d:\n", $arg0, $grappa_queue->queues->len + set $grappa_queue_print_elem = $grappa_queue->queues->head while $grappa_queue_print_elem != 0 - grappa_coroutine_print $grappa_queue_print_elem->co + grappa_worker_print $grappa_queue_print_elem set $grappa_queue_print_elem = $grappa_queue_print_elem->next end end @@ -156,10 +145,10 @@ end define grappa_queue_length set $grappa_queue = $arg1 - if $grappa_queue->head + if $grappa_queue->queues->head set $grappa_queue_length = 0 else - set $grappa_queue_length = $grappa_queue->len + set $grappa_queue_length = $grappa_queue->queues->len end printf "%s queue has length %d. (Contents not shown.)\n", $arg0, $grappa_queue->len end @@ -208,45 +197,45 @@ Syntax: (gdb) unassigned_queue_length end -define suspended_coroutines - printf "Suspended non-idle coroutines:\n" +define suspended_workers + printf "Suspended non-idle workers:\n" set $grappa_suspended_coros_elem = all_coros while $grappa_suspended_coros_elem != 0 if $grappa_suspended_coros_elem->suspended if !($grappa_suspended_coros_elem->idle) - grappa_coroutine_print $grappa_suspended_coros_elem + grappa_worker_print $grappa_suspended_coros_elem end end set $grappa_suspended_coros_elem = $grappa_suspended_coros_elem->next end end -document suspended_coroutines -Prints all suspended coroutines +document suspended_workers +Prints all suspended workers Syntax: - (gdb) suspended_coroutines + (gdb) suspended_workers end -define current_coroutine - printf "Current coroutine:\n" - grappa_coroutine_print Grappa::impl::global_scheduler.current_thread->co +define current_worker + printf "Current worker:\n" + grappa_worker_print Grappa::impl::global_scheduler.current_thread end -document current_coroutine -Prints currently scheduled Grappa coroutine. +document current_worker +Prints currently scheduled Grappa worker. Syntax: - (gdb) current_coroutine + (gdb) current_worker end -define coroutines - current_coroutine +define workers + current_worker periodic_queue ready_queue unassigned_queue_length - suspended_coroutines + suspended_workers end -document coroutines -Prints contents of Grappa coroutine queues. +document workers +Prints contents of Grappa worker queues. Syntax: - (gdb) coroutines + (gdb) workers end @@ -298,11 +287,11 @@ Syntax: end define queues - coroutines + workers tasks end document queues -Prints contents of Grappa coroutine/task queues. +Prints contents of Grappa worker/task queues. Syntax: (gdb) queues end @@ -313,71 +302,71 @@ end -define switch_to_coroutine +define switch_to_worker grappa_save_state - set $grappa_coro = (coro*) $arg0 + set $grappa_coro = (Worker*) $arg0 set $rsp = $grappa_coro->stack + $grappa_stack_offset # select-frame ($grappa_coro->stack + $grappa_stack_offset) set $rip = *(int*)$rsp # set $rip = *(int*)($grappa_coro->stack + $grappa_stack_offset) end -document switch_to_coroutine -Switch to a Grappa coroutine for backtracing. Do not continue or bad things will happen. +document switch_to_worker +Switch to a Grappa worker for backtracing. Do not continue or bad things will happen. Use grappa_restore_state to switch back to the original context. Syntax: - (gdb) switch_to_coroutine + (gdb) switch_to_worker end -define backtrace_coroutine - switch_to_coroutine $arg0 +define backtrace_worker + switch_to_worker $arg0 backtrace grappa_restore_state end -document backtrace_coroutine -Print backtrace of Grappa coroutine. +document backtrace_worker +Print backtrace of Grappa worker. Syntax: - (gdb) backtrace_coroutine + (gdb) backtrace_worker end -define backtrace_coroutine_full - switch_to_coroutine $arg0 +define backtrace_worker_full + switch_to_worker $arg0 backtrace full grappa_restore_state end -document backtrace_coroutine_full -Print backtrace of Grappa coroutine with locals. +document backtrace_worker_full +Print backtrace of Grappa worker with locals. Syntax: - (gdb) backtrace_coroutine_full + (gdb) backtrace_worker_full end define btc - switch_to_coroutine $arg0 + switch_to_worker $arg0 backtrace grappa_restore_state end document btc -Print backtrace of Grappa coroutine. +Print backtrace of Grappa worker. Syntax: - (gdb) btc + (gdb) btc end define btcf - switch_to_coroutine $arg0 + switch_to_worker $arg0 backtrace full grappa_restore_state end document btcf -Print backtrace of Grappa coroutine with locals. +Print backtrace of Grappa worker with locals. Syntax: - (gdb) btcf + (gdb) btcf end define dumpstack - p ((int64_t*)((coro*)$arg0)->stack) @ (((coro*)$arg0)->base+((coro*)$arg0)->ssize+4096-((coro*)$arg0)->stack)/sizeof(int64_t) + p ((int64_t*)((Worker*)$arg0)->stack) @ (((Worker*)$arg0)->base+((Worker*)$arg0)->ssize+4096-((Worker*)$arg0)->stack)/sizeof(int64_t) end document dumpstack Dump stack (as int64_t's). Syntax: - (gdb) dumpstack + (gdb) dumpstack end From 6079fce059f7fdafc3e10a437a96f1a24eccba32 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Tue, 23 Jul 2013 00:35:29 -0700 Subject: [PATCH 8/9] fix profiler filenames --- system/PerformanceTools.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/system/PerformanceTools.cpp b/system/PerformanceTools.cpp index d0123cf64..b2645dead 100644 --- a/system/PerformanceTools.cpp +++ b/system/PerformanceTools.cpp @@ -61,7 +61,7 @@ char * Grappa_get_next_profiler_filename( ) { char * jobid = getenv("SLURM_JOB_ID"); char * procid = getenv("SLURM_PROCID"); if( /*jobname != NULL &&*/ jobid != NULL && procid != NULL ) { - sprintf( profiler_filename, "%s.%s.rank%s.phase%d.prof", "exe", jobid, procid, profiler_phase ); + sprintf( profiler_filename, "%s.%s.rank%s.phase%d.prof", argv0_for_profiler, jobid, procid, profiler_phase ); } else { sprintf( profiler_filename, "%s.%d.pid%d.phase%d.prof", argv0_for_profiler, time_for_profiler, getpid(), profiler_phase ); } From 47c95be850e51105b638e9e1c6551f1f0711a570 Mon Sep 17 00:00:00 2001 From: Jacob Nelson Date: Tue, 23 Jul 2013 00:50:19 -0700 Subject: [PATCH 9/9] a further try at doing local message delivery properly --- system/MessageBase.hpp | 10 +-- system/RDMAAggregator.cpp | 141 +++++++++++++++++++++----------------- 2 files changed, 83 insertions(+), 68 deletions(-) diff --git a/system/MessageBase.hpp b/system/MessageBase.hpp index b6718abf8..1f9fcbac7 100644 --- a/system/MessageBase.hpp +++ b/system/MessageBase.hpp @@ -74,11 +74,11 @@ namespace Grappa { //if( Grappa::mycore() != source_ ) { if( (is_delivered_ == true) && (Grappa::mycore() != source_) ) { - DVLOG(5) << __func__ << ": " << this << " Re-enqueuing to " << source_; - DCHECK_EQ( this->is_sent_, false ); - mark_sent_enqueues++; - // should only happen for deaggregation now, so enqueue to locale - locale_enqueue( source_ ); + // DVLOG(5) << __func__ << ": " << this << " Re-enqueuing to " << source_; + // DCHECK_EQ( this->is_sent_, false ); + // mark_sent_enqueues++; + // // should only happen for deaggregation now, so enqueue to locale + // locale_enqueue( source_ ); } else { DVLOG(5) << __func__ << ": " << this << " Final mark_sent"; DCHECK_EQ( Grappa::mycore(), this->source_ ); diff --git a/system/RDMAAggregator.cpp b/system/RDMAAggregator.cpp index 1629bff7a..12eb6f26a 100644 --- a/system/RDMAAggregator.cpp +++ b/system/RDMAAggregator.cpp @@ -725,7 +725,6 @@ void RDMAAggregator::draw_routing_graph() { if( true ) { Core mylocale_core = Grappa::mylocale() * Grappa::locale_cores(); Core mycore = Grappa::mycore(); - buf->override_dest(mycore); DVLOG(3) << __PRETTY_FUNCTION__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << " finished with " << buf << "; acking with buffer " << buf << " for core " << mycore << " on core " << c; auto request = Grappa::message( c, [mycore, buf] { @@ -820,83 +819,99 @@ void RDMAAggregator::draw_routing_graph() { << " core1 count " << (buf->get_counts())[1] << " from core " << buf->get_source(); - { - // now send messages to deaggregate everybody's chunks - - // message to send - struct ReceiveBuffer { - char * buf; - uint16_t size; - //uint64_t sequence_number; - //Grappa::impl::RDMABuffer * buf_base; - void operator()() { - DVLOG(5) << __PRETTY_FUNCTION__ // << "/" << sequence_number - << ": received " << size << "-byte buffer slice at " << (void*) buf << " to deaggregate"; - - Grappa::impl::global_scheduler.set_no_switch_region( true ); - global_rdma_aggregator.deaggregate_buffer( buf, size ); - Grappa::impl::global_scheduler.set_no_switch_region( false ); - - DVLOG(5) << __PRETTY_FUNCTION__ // << "/" << sequence_number - << ": done deaggregating " << size << "-byte buffer slice at " << (void*) buf; - } - }; + // now send messages to deaggregate everybody's chunks - // deaggregate messages to send - Grappa::Message< ReceiveBuffer > msgs[ Grappa::locale_cores() ]; + Grappa::CompletionEvent ce( Grappa::locale_cores() ); + + struct ReceiveBufferReply { + Grappa::CompletionEvent * ce_p; + void operator()() { + VLOG(1) << __PRETTY_FUNCTION__ // << "/" << sequence_number + << ": signaling completion"; + ce_p->complete(); + } + }; - // index array from buffer - uint16_t * counts = buf->get_counts(); - char * current_buf = buf->get_payload(); - char * my_buf = NULL; - int outstanding = 0; + struct ReceiveBuffer { + char * buf; + uint16_t size; + Grappa::Message< ReceiveBufferReply > * reply; + void operator()() { + VLOG(1) << __PRETTY_FUNCTION__ // << "/" << sequence_number + << ": received " << size << "-byte buffer slice at " << (void*) buf << " to deaggregate"; + Grappa::impl::global_scheduler.set_no_switch_region( true ); + global_rdma_aggregator.deaggregate_buffer( buf, size ); + Grappa::impl::global_scheduler.set_no_switch_region( false ); - // fill and send deaggregate messages - for( Core locale_core = 0; locale_core < Grappa::locale_cores(); ++locale_core ) { - DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf - << ": found " << counts[ locale_core ] << " bytes for core " << locale_core << " at " << (void*) current_buf; + VLOG(1) << __PRETTY_FUNCTION__ // << "/" << sequence_number + << ": done deaggregating " << size << "-byte buffer slice at " << (void*) buf; + reply->locale_enqueue(); + } + }; + + // deaggregate messages to send + Grappa::Message< ReceiveBuffer > msgs[ Grappa::locale_cores() ]; + Grappa::Message< ReceiveBufferReply > replies[ Grappa::locale_cores() ]; - if( counts[locale_core] > 0 ) { + // index array from buffer + uint16_t * counts = buf->get_counts(); + char * current_buf = buf->get_payload(); + char * my_buf = NULL; - msgs[locale_core]->buf = current_buf; - msgs[locale_core]->size = counts[locale_core]; - //msgs[locale_core]->sequence_number = sequence_number; - //msgs[locale_core]->buf_base = buf; + // fill and send deaggregate messages + for( Core locale_core = 0; locale_core < Grappa::locale_cores(); ++locale_core ) { + DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf + << ": found " << counts[ locale_core ] << " bytes for core " << locale_core << " at " << (void*) current_buf; - if( locale_core != Grappa::locale_mycore() ) { - DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf << " message " << (void*) &(msgs[locale_core]) - << " enqueuing " << counts[locale_core] << " bytes to locale_core " << locale_core << " core " << locale_core + Grappa::mylocale() * Grappa::locale_cores() << " my locale_core " << Grappa::locale_mycore() << " my core " << Grappa::mycore(); - msgs[locale_core].locale_enqueue( locale_core + Grappa::mylocale() * Grappa::locale_cores() ); - outstanding++; - } else { - my_buf = current_buf; - } + if( counts[locale_core] > 0 ) { - current_buf += counts[locale_core]; - } - } + msgs[locale_core]->buf = current_buf; + msgs[locale_core]->size = counts[locale_core]; + msgs[locale_core]->reply = &replies[locale_core]; - // deaggregate my messages - if( counts[ Grappa::locale_mycore() ] > 0 ) { - DVLOG(5) << __func__ << "/" << sequence_number - << ": deaggregating my own " << counts[ Grappa::locale_mycore() ] << "-byte buffer slice at " << (void*) buf; + replies[locale_core]->ce_p = &ce; + replies[locale_core].destination_ = Grappa::mycore(); - Grappa::impl::global_scheduler.set_no_switch_region( true ); - deaggregate_buffer( my_buf, counts[ Grappa::locale_mycore() ] ); - Grappa::impl::global_scheduler.set_no_switch_region( false ); + if( locale_core != Grappa::locale_mycore() ) { + + DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf << " message " << (void*) &(msgs[locale_core]) + << " enqueuing " << counts[locale_core] << " bytes to locale_core " << locale_core + << " core " << locale_core + Grappa::mylocale() * Grappa::locale_cores() + << " my locale_core " << Grappa::locale_mycore() << " my core " << Grappa::mycore(); + msgs[locale_core].locale_enqueue( locale_core + Grappa::mylocale() * Grappa::locale_cores() ); + + } else { + my_buf = current_buf; + } - DVLOG(5) << __func__ << "/" << sequence_number - << ": done deaggregating my own " << counts[ Grappa::locale_mycore() ] << "-byte buffer slice at " << (void*) buf; + current_buf += counts[locale_core]; + } else { + ce.complete(); } + } + + // deaggregate my messages + if( counts[ Grappa::locale_mycore() ] > 0 ) { + DVLOG(5) << __func__ << "/" << sequence_number + << ": deaggregating my own " << counts[ Grappa::locale_mycore() ] << "-byte buffer slice at " << (void*) buf; - // // if we're lucky, responses are now waiting - // receive_poll(); + Grappa::impl::global_scheduler.set_no_switch_region( true ); + deaggregate_buffer( my_buf, counts[ Grappa::locale_mycore() ] ); + Grappa::impl::global_scheduler.set_no_switch_region( false ); - // block here until messages are sent (i.e., delivered and deaggregated) - DVLOG(5) << __func__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << "/" << sequence_number - << ": maybe blocking until my " << outstanding << " outstanding messages are delivered"; + DVLOG(5) << __func__ << "/" << sequence_number + << ": done deaggregating my own " << counts[ Grappa::locale_mycore() ] << "-byte buffer slice at " << (void*) buf; } + ce.complete(); + + // // if we're lucky, responses are now waiting + // receive_poll(); + + // block here until messages are sent (i.e., delivered and deaggregated) + VLOG(1) << __func__ << "/" << Grappa::impl::global_scheduler.get_current_thread() << "/" << sequence_number + << ": maybe blocking"; + ce.wait(); DVLOG(5) << __func__ << "/" << (void*) sequence_number << "/" << (void*) buf << ": continuting; my outstanding messages are all delivered";