<commit_before>
#include <gflags/gflags.h>

#include "Aggregator.hpp"
#include "SoftXMT.hpp"
#include <csignal>

#include "PerformanceTools.hpp"

/// command line options for Aggregator
DEFINE_int64( aggregator_autoflush_ticks, 1000, "number of ticks to wait before autoflushing aggregated active messages");

/// global Aggregator instance
Aggregator global_aggregator;

#ifdef STL_DEBUG_ALLOCATOR
DEFINE_int64( aggregator_access_control_signal, SIGUSR2, "signal used to toggle aggregator queue access control");
bool aggregator_access_control_active = false;
static void aggregator_toggle_access_control_sighandler( int signum ) {
  aggregator_access_control_active = ~aggregator_access_control_active;
}

#endif

// Construct Aggregator.
Aggregator::Aggregator( ) 
  : max_nodes_( -1 )
  , buffers_( )
  , route_map_( )
  , previous_timestamp_( 0L )
  , least_recently_sent_( )
  , aggregator_deaggregate_am_handle_( -1 )
#ifdef VTRACE_FULL
  , tag_( -1 )
  , vt_agg_commid_( VT_COMM_DEF( "Aggregator" ) )
#endif
  , stats()
{ 
#ifdef STL_DEBUG_ALLOCATOR
  struct sigaction access_control_toggle_sa;
  sigemptyset( &access_control_toggle_sa.sa_mask );
  access_control_toggle_sa.sa_flags = 0;
  access_control_toggle_sa.sa_handler = &aggregator_toggle_access_control_sighandler;
  CHECK_EQ( 0, sigaction( FLAGS_aggregator_access_control_signal, &access_control_toggle_sa, 0 ) ) 
    << "Aggregator access control signal handler installation failed.";
  if( aggregator_access_control_active ) STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadOnly);
#endif
}

void Aggregator_deaggregate_am( gasnet_token_t token, void * buf, size_t size );

void Aggregator::init() {
  max_nodes_ = global_communicator.nodes();
  aggregator_deaggregate_am_handle_ = global_communicator.register_active_message_handler( &Aggregator_deaggregate_am );
  buffers_.resize( max_nodes_ - buffers_.size() );
  route_map_.resize( max_nodes_ - route_map_.size() );
  // initialize route map
  for( Node i = 0; i < max_nodes_; ++i ) {
    route_map_[i] = i;
  }
#ifdef VTRACE_FULL
  tag_ = global_communicator.mynode();
#endif
}

Aggregator::~Aggregator() {
#ifdef STL_DEBUG_ALLOCATOR
  STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadWrite);
#endif
}

void Aggregator::deaggregate( ) {
  GRAPPA_FUNCTION_PROFILE( GRAPPA_COMM_GROUP );
#ifdef VTRACE_FULL
  VT_TRACER("deaggregate");
#endif
  StateTimer::enterState_deaggregation();
  while( !received_AM_queue_.empty() ) {
    DVLOG(5) << "deaggregating";
    // TODO: too much copying
    ReceivedAM amp = received_AM_queue_.front();

#ifdef STL_DEBUG_ALLOCATOR
    if( aggregator_access_control_active ) STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadWrite);<commit_msg>Missing change to support new aggregator queue.<commit_after>
#include <gflags/gflags.h>

#include "Aggregator.hpp"
#include "SoftXMT.hpp"
#include <csignal>

#include "PerformanceTools.hpp"

/// command line options for Aggregator
DEFINE_int64( aggregator_autoflush_ticks, 1000, "number of ticks to wait before autoflushing aggregated active messages");

/// global Aggregator instance
Aggregator global_aggregator;

#ifdef STL_DEBUG_ALLOCATOR
DEFINE_int64( aggregator_access_control_signal, SIGUSR2, "signal used to toggle aggregator queue access control");
bool aggregator_access_control_active = false;
static void aggregator_toggle_access_control_sighandler( int signum ) {
  aggregator_access_control_active = ~aggregator_access_control_active;
}

#endif

// Construct Aggregator.
Aggregator::Aggregator( ) 
  : max_nodes_( -1 )
  , buffers_( )
  , route_map_( )
  , previous_timestamp_( 0L )
  , least_recently_sent_( )
  , aggregator_deaggregate_am_handle_( -1 )
#ifdef VTRACE_FULL
  , tag_( -1 )
  , vt_agg_commid_( VT_COMM_DEF( "Aggregator" ) )
#endif
  , stats()
{ 
#ifdef STL_DEBUG_ALLOCATOR
  struct sigaction access_control_toggle_sa;
  sigemptyset( &access_control_toggle_sa.sa_mask );
  access_control_toggle_sa.sa_flags = 0;
  access_control_toggle_sa.sa_handler = &aggregator_toggle_access_control_sighandler;
  CHECK_EQ( 0, sigaction( FLAGS_aggregator_access_control_signal, &access_control_toggle_sa, 0 ) ) 
    << "Aggregator access control signal handler installation failed.";
  if( aggregator_access_control_active ) STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadOnly);
#endif
}

void Aggregator_deaggregate_am( gasnet_token_t token, void * buf, size_t size );

void Aggregator::init() {
  max_nodes_ = global_communicator.nodes();
  least_recently_sent_.resize( global_communicator.nodes() );
  aggregator_deaggregate_am_handle_ = global_communicator.register_active_message_handler( &Aggregator_deaggregate_am );
  buffers_.resize( max_nodes_ - buffers_.size() );
  route_map_.resize( max_nodes_ - route_map_.size() );
  // initialize route map
  for( Node i = 0; i < max_nodes_; ++i ) {
    route_map_[i] = i;
  }
#ifdef VTRACE_FULL
  tag_ = global_communicator.mynode();
#endif
}

Aggregator::~Aggregator() {
#ifdef STL_DEBUG_ALLOCATOR
  STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadWrite);
#endif
}

void Aggregator::deaggregate( ) {
  GRAPPA_FUNCTION_PROFILE( GRAPPA_COMM_GROUP );
#ifdef VTRACE_FULL
  VT_TRACER("deaggregate");
#endif
  StateTimer::enterState_deaggregation();
  while( !received_AM_queue_.empty() ) {
    DVLOG(5) << "deaggregating";
    // TODO: too much copying
    ReceivedAM amp = received_AM_queue_.front();

#ifdef STL_DEBUG_ALLOCATOR
    if( aggregator_access_control_active ) STLMemDebug::BaseAllocator::getMemMgr().setAccessMode(STLMemDebug::memReadWrite);
{'fraction_non_alphanumeric': 0.10173697270471464, 'fraction_numerical': 0.006380716058135413, 'mean_word_length': 6.357235984354628, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2843981', 'n_tokens_mistral': 1981, 'n_tokens_neox': 1924, 'n_words': 357}