<commit_before>#include <boost/test/unit_test.hpp>

#include "SoftXMT.hpp"
#include "GlobalTaskJoiner.hpp"
#include "ForkJoin.hpp"
#include "AsyncParallelFor.hpp"
#include "Delegate.hpp"
#include "GlobalAllocator.hpp"

#include "PerformanceTools.hpp"


BOOST_AUTO_TEST_SUITE( AsyncParallelFor_Global_tests );

#define MAX_NODES 8
#define SIZE 1000000
int64_t done[SIZE*MAX_NODES] = {0};
int64_t count2 = 0;
int64_t global_count = 0;
int64_t local_count = 0;

#define N (1L<<10)

void loop_body(int64_t start, int64_t num ) {
  DVLOG(5) << "iters " << start << ", " << num;
  /*BOOST_MESSAGE( "execute [" << start << ","
            << start+num << ") with thread " << CURRENT_THREAD->id );*/
  BOOST_CHECK( num <= FLAGS_async_par_for_threshold );
  for (int i=start; i<start+num; i++) {
    int64_t marked = SoftXMT_delegate_fetch_and_add_word( make_global( &done[i], 0 ), 1 );
    BOOST_CHECK( marked == 0 );

    SoftXMT_delegate_fetch_and_add_word( make_global( &global_count, 0 ), 1 ); 
    local_count++;
  }
}

struct parallel_func : public ForkJoinIteration {
  void operator()(int64_t nid) const {
    int64_t start = nid * SIZE;
    int64_t iters = SIZE;

    global_joiner.reset();
    async_parallel_for<loop_body, joinerSpawn<loop_body, ASYNC_PAR_FOR_DEFAULT>, ASYNC_PAR_FOR_DEFAULT > ( start, iters ); 
    global_joiner.wait();
  }
};


// don't care what type goes into the GlobalAddress
typedef int64_t dummy_t;

void loop_body2(int64_t start, int64_t num, GlobalAddress<dummy_t> shared_arg_packed) {
  /*BOOST_MESSAGE( "execute [" << start << ","
            << start+num << ") with thread " << CURRENT_THREAD->id );*/
  BOOST_CHECK( num <= FLAGS_async_par_for_threshold );
  int64_t origin = reinterpret_cast<int64_t>(shared_arg_packed.pointer());
  for (int i=start; i<start+num; i++) {
    SoftXMT_delegate_fetch_and_add_word( make_global( &count2, origin ), 1 );

    SoftXMT_delegate_fetch_and_add_word( make_global( &global_count, 0 ), 1 ); 
    local_count++;
  }
}

LOOP_FUNCTION(check_count2, nid) {
  BOOST_CHECK( count2 == SIZE );
}

struct parallel_func2 : public ForkJoinIteration {
  void operator()(int64_t nid) const {
    local_count = 0;

    int64_t start = nid * SIZE;
    int64_t iters = SIZE;

    // pack an argument into the global address (hack).
    // This is useful if we need just the node() from the global address and want to use the other
    // bits for something else.
    int64_t shared_arg = SoftXMT_mynode(); 
    GlobalAddress<dummy_t> shared_arg_packed = make_global( reinterpret_cast<dummy_t*>(shared_arg) );

    global_joiner.reset();
    async_parallel_for<dummy_t,loop_body2,joinerSpawn_hack<dummy_t,loop_body2,ASYNC_PAR_FOR_DEFAULT>,ASYNC_PAR_FOR_DEFAULT >(start,iters,shared_arg_packed);
    global_joiner.wait();
  }
};


  
LOOP_FUNCTION(func_enable_google_profiler, nid) {
  SoftXMT_start_profiling();
}
void profile_start() {
#ifdef GOOGLE_PROFILER
  func_enable_google_profiler g;
  fork_join_custom(&g);
#endif
}
LOOP_FUNCTION(func_disable_google_profiler, nid) {
  SoftXMT_stop_profiling();
}
void profile_stop() {
#ifdef GOOGLE_PROFILER
  func_disable_google_profiler g;
  fork_join_custom(&g);
#endif
}

LOOP_FUNCTION( no_work_func, nid ) {
    global_joiner.reset();
    /* no work */
    global_joiner.wait();
}

void signaling_task( int64_t * arg ) {
  global_joiner.signal();
}

LOOP_FUNCTION( one_work_func, nid ) {
    global_joiner.reset();
    global_joiner.registerTask();
    int64_t ignore;
    SoftXMT_privateTask( &signaling_task, &ignore );
    global_joiner.wait();
}

LOOP_FUNCTION( one_self_work_func, nid ) {
    global_joiner.reset();

    // register and signal
    global_joiner.registerTask();
    global_joiner.signal();

    global_joiner.wait();
}

LOOP_FUNCTOR( ff_test_func, nid, ((GlobalAddress<double>,array)) ) {
  global_joiner.reset();

  range_t r = blockDist(0, N, nid, SoftXMT_nodes());

  for (int64_t i=r.start; i<r.end; i++) {
    ff_delegate_write(array+i, (double)i);
  }

  global_joiner.wait();
}

void user_main( void * args ) {
  
  BOOST_CHECK( SoftXMT_nodes() <= MAX_NODES );
  
  int64_t total_iters = SoftXMT_nodes() * SIZE;
  BOOST_MESSAGE( total_iters << " iterations over " <<
                 SoftXMT_nodes() << " nodes" );

  BOOST_MESSAGE( "Test default" );
  {

  // add sampling of global_count
  SoftXMT_add_profiling_counter( (uint64_t*)&global_count, "global count", "globcnt", false, 0 );
  
  parallel_func f;
  
  SoftXMT_reset_stats_all_nodes();
  profile_start();<commit_msg>change UTS and Async Global test to use initialize global queue<commit_after>#include <boost/test/unit_test.hpp>

#include "SoftXMT.hpp"
#include "GlobalTaskJoiner.hpp"
#include "ForkJoin.hpp"
#include "AsyncParallelFor.hpp"
#include "Delegate.hpp"
#include "GlobalAllocator.hpp"

#include "PerformanceTools.hpp"


BOOST_AUTO_TEST_SUITE( AsyncParallelFor_Global_tests );

#define MAX_NODES 8
#define SIZE 1000000
int64_t done[SIZE*MAX_NODES] = {0};
int64_t count2 = 0;
int64_t global_count = 0;
int64_t local_count = 0;

#define N (1L<<10)

void loop_body(int64_t start, int64_t num ) {
  DVLOG(5) << "iters " << start << ", " << num;
  /*BOOST_MESSAGE( "execute [" << start << ","
            << start+num << ") with thread " << CURRENT_THREAD->id );*/
  BOOST_CHECK( num <= FLAGS_async_par_for_threshold );
  for (int i=start; i<start+num; i++) {
    int64_t marked = SoftXMT_delegate_fetch_and_add_word( make_global( &done[i], 0 ), 1 );
    BOOST_CHECK( marked == 0 );

    SoftXMT_delegate_fetch_and_add_word( make_global( &global_count, 0 ), 1 ); 
    local_count++;
  }
}

struct parallel_func : public ForkJoinIteration {
  void operator()(int64_t nid) const {
    int64_t start = nid * SIZE;
    int64_t iters = SIZE;

    global_joiner.reset();
    async_parallel_for<loop_body, joinerSpawn<loop_body, ASYNC_PAR_FOR_DEFAULT>, ASYNC_PAR_FOR_DEFAULT > ( start, iters ); 
    global_joiner.wait();
  }
};


// don't care what type goes into the GlobalAddress
typedef int64_t dummy_t;

void loop_body2(int64_t start, int64_t num, GlobalAddress<dummy_t> shared_arg_packed) {
  /*BOOST_MESSAGE( "execute [" << start << ","
            << start+num << ") with thread " << CURRENT_THREAD->id );*/
  BOOST_CHECK( num <= FLAGS_async_par_for_threshold );
  int64_t origin = reinterpret_cast<int64_t>(shared_arg_packed.pointer());
  for (int i=start; i<start+num; i++) {
    SoftXMT_delegate_fetch_and_add_word( make_global( &count2, origin ), 1 );

    SoftXMT_delegate_fetch_and_add_word( make_global( &global_count, 0 ), 1 ); 
    local_count++;
  }
}

LOOP_FUNCTION(check_count2, nid) {
  BOOST_CHECK( count2 == SIZE );
}

struct parallel_func2 : public ForkJoinIteration {
  void operator()(int64_t nid) const {
    local_count = 0;

    int64_t start = nid * SIZE;
    int64_t iters = SIZE;

    // pack an argument into the global address (hack).
    // This is useful if we need just the node() from the global address and want to use the other
    // bits for something else.
    int64_t shared_arg = SoftXMT_mynode(); 
    GlobalAddress<dummy_t> shared_arg_packed = make_global( reinterpret_cast<dummy_t*>(shared_arg) );

    global_joiner.reset();
    async_parallel_for<dummy_t,loop_body2,joinerSpawn_hack<dummy_t,loop_body2,ASYNC_PAR_FOR_DEFAULT>,ASYNC_PAR_FOR_DEFAULT >(start,iters,shared_arg_packed);
    global_joiner.wait();
  }
};


  
LOOP_FUNCTION(func_enable_google_profiler, nid) {
  SoftXMT_start_profiling();
}
void profile_start() {
#ifdef GOOGLE_PROFILER
  func_enable_google_profiler g;
  fork_join_custom(&g);
#endif
}
LOOP_FUNCTION(func_disable_google_profiler, nid) {
  SoftXMT_stop_profiling();
}
void profile_stop() {
#ifdef GOOGLE_PROFILER
  func_disable_google_profiler g;
  fork_join_custom(&g);
#endif
}

LOOP_FUNCTION( no_work_func, nid ) {
    global_joiner.reset();
    /* no work */
    global_joiner.wait();
}

void signaling_task( int64_t * arg ) {
  global_joiner.signal();
}

LOOP_FUNCTION( one_work_func, nid ) {
    global_joiner.reset();
    global_joiner.registerTask();
    int64_t ignore;
    SoftXMT_privateTask( &signaling_task, &ignore );
    global_joiner.wait();
}

LOOP_FUNCTION( one_self_work_func, nid ) {
    global_joiner.reset();

    // register and signal
    global_joiner.registerTask();
    global_joiner.signal();

    global_joiner.wait();
}

LOOP_FUNCTOR( ff_test_func, nid, ((GlobalAddress<double>,array)) ) {
  global_joiner.reset();

  range_t r = blockDist(0, N, nid, SoftXMT_nodes());

  for (int64_t i=r.start; i<r.end; i++) {
    ff_delegate_write(array+i, (double)i);
  }

  global_joiner.wait();
}

void user_main( void * args ) {
  SoftXMT_global_queue_initialize();
  
  BOOST_CHECK( SoftXMT_nodes() <= MAX_NODES );
  
  int64_t total_iters = SoftXMT_nodes() * SIZE;
  BOOST_MESSAGE( total_iters << " iterations over " <<
                 SoftXMT_nodes() << " nodes" );

  BOOST_MESSAGE( "Test default" );
  {

  // add sampling of global_count
  SoftXMT_add_profiling_counter( (uint64_t*)&global_count, "global count", "globcnt", false, 0 );
  
  parallel_func f;
  
  SoftXMT_reset_stats_all_nodes();
  profile_start();
{'fraction_non_alphanumeric': 0.1252607311450214, 'fraction_numerical': 0.018223734767812055, 'mean_word_length': 4.474759615384615, 'pattern_counts': {'":': 0, '<': 107, '<?xml version=': 0, '>': 27, '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': '1998319', 'n_tokens_mistral': 3516, 'n_tokens_neox': 3339, 'n_words': 654}