<commit_before>/* Copyright (C) 2010-2011 The Trustees of Indiana University.             */
/*                                                                         */
/* Use, modification and distribution is subject to the Boost Software     */
/* License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at */
/* http://www.boost.org/LICENSE_1_0.txt)                                   */
/*                                                                         */
/*  Authors: <NAME>                                             */
/*           <NAME>                                              */

#include "common.h"
#include "oned_csr.h"
#include <stdint.h>
#include <inttypes.h>
#include <stdlib.h>
#include <stddef.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>

#include "Grappa.hpp"
#include "GlobalAllocator.hpp"
#include "ForkJoin.hpp"
#include "Cache.hpp"
#include "Collective.hpp"
#include "Delegate.hpp"
#include "AsyncDelegate.hpp"
#include "GlobalTaskJoiner.hpp"
#include <Array.hpp>
#include "GlobalCompletionEvent.hpp"

using namespace Grappa;

#include "timer.h"

#include <sstream>
#include <functional>

#define MINVECT_SIZE 2

#define XOFF(k) (xoff+2*(k))
#define XENDOFF(k) (xoff+1+2*(k))

GlobalAddress<int64_t> xoff;

static int64_t maxvtx, nv;

LOOP_FUNCTOR( max_func, index,
             (( GlobalAddress<packed_edge>,edges))
             ((int64_t *, max)) )
{
  Incoherent<packed_edge>::RO cedge(edges+index, 1);
  if (cedge[0].v0 > *max) {
    *max = cedge[0].v0;
  }
  if (cedge[0].v1 > *max) {
    *max = cedge[0].v1;
  }
}

LOOP_FUNCTOR( node_max_func, mynode, 
  ((GlobalAddress<packed_edge>, edges))
  ((int64_t,start)) (( int64_t,end )) ((int64_t,init_max)) )
{
  int64_t max = init_max;
  range_t myblock = blockDist(start, end, mynode, Grappa_nodes());
  max_func f;
  f.edges = edges;
  f.max = &max;
  fork_join_onenode(&f, myblock.start, myblock.end);
    
  maxvtx = Grappa_allreduce<int64_t, collective_max, 0>( max );
  
  nv = maxvtx+1;
}

static void find_nv(const tuple_graph* const tg) {
  node_max_func f(tg->edges, 0, tg->nedge, 0);
  fork_join_custom(&f);
}

// note: this corresponds to how Graph500 counts 'degree' (both in- and outgoing edges to each vertex)
LOOP_FUNCTOR( degree_func, index, (( GlobalAddress<packed_edge>, edges )) ((GlobalAddress<int64_t>,xoff)) )
{ 
  Incoherent<packed_edge>::RO cedge(edges+index, 1);
  int64_t i = cedge[0].v0;
  int64_t j = cedge[0].v1;
  if (i != j) { //skip self-edges
    // TODO: these should be overlapping
    Grappa_delegate_fetch_and_add_word(XOFF(i), 1);
    Grappa_delegate_fetch_and_add_word(XOFF(j), 1);
  }
}

LOOP_FUNCTOR( local_reduce_sum, index,
    (( GlobalAddress<int64_t>, xoff )) (( int64_t *, sum )) )
{
  int64_t v = Grappa_delegate_read_word(XOFF(index));
  *sum += v;
}

#define minint(A,B) ((A)<(B)) ? (A) : (B)
int64_t local_prefix;
// ideally this would be forall local with contiguous addresses
LOOP_FUNCTOR( prefix_sum_node, nid,
    ((GlobalAddress<int64_t>,xoff))
    ((int64_t,nv)) ) {
  range_t myblock = blockDist(0, nv, nid, Grappa_nodes());
  
  local_prefix = 0;
  local_reduce_sum fr;
  fr.xoff = xoff;
  fr.sum = &local_prefix;
  fork_join_onenode(&fr, myblock.start, myblock.end);
 
  Grappa_barrier_suspending(); // "phaser style", alternative is replace with forjoins 

  // node 0 sets buf[i] to prefix sum
  // TODO: do this as parallel prefix
  if (nid == 0) {
    int64_t total = local_prefix;
    for (int64_t i=1; i<Grappa_nodes(); i++) {
      //LOG(INFO) << "will add total=" << total << " to node " << i;
      total += Grappa_delegate_fetch_and_add_word(make_global(&local_prefix,i), total);
    }
  }
  
  Grappa_barrier_suspending(); // alternative: replace with forkjoins
  
  int64_t prev_sum = (nid > 0) ? Grappa_delegate_read_word(make_global(&local_prefix,nid-1)) : 0;
  //LOG(INFO) << "prev_sum= " << prev_sum;
 
  // do prefix sum within the slice 
  // Cache a part at a time, doing the sum locally and writing back
  size_t bufsize = 1024;
  int64_t * buf = new int64_t[2*bufsize];
  for (int64_t i=myblock.start; i<myblock.end; i+=bufsize) {
    size_t end = minint(myblock.end, i+bufsize);
    int64_t len = end - i;
    Incoherent<int64_t>::RW cv(XOFF(i), XOFF(end)-XOFF(i), buf);
    for (int64_t j = 0; j<len; j++) {
      int64_t tmp = cv[2*j]; // 2*j because XOFF entries are even
      cv[2*j] = prev_sum;
      prev_sum += tmp;
    } 
  }
  delete buf;
}

static int64_t prefix_sum(GlobalAddress<int64_t> xoff, int64_t nv) {
  prefix_sum_node fps;
  fps.xoff = xoff;
  fps.nv = nv;
  fork_join_custom(&fps);
 
  // return total sum 
  return Grappa_delegate_read_word(make_global(&local_prefix, Grappa_nodes()-1));
}

LOOP_FUNCTOR( minvect_func, index,
          (( GlobalAddress<int64_t>,xoff )) )
{
  int64_t v = Grappa_delegate_read_word(XOFF(index));
  if (v < MINVECT_SIZE) {
    Grappa_delegate_write_word(XOFF(index), MINVECT_SIZE);
  }
}

LOOP_FUNCTOR( init_xendoff_func, index,
             (( GlobalAddress<int64_t>, xoff )) )
{
  Grappa_delegate_write_word(XENDOFF(index), Grappa_delegate_read_word(XOFF(index)));
}

static void setup_deg_off(const tuple_graph * const tg, csr_graph * g) {
  // note: this corresponds to how Graph500 counts 'degree' (both in- and outgoing edges to each vertex)
  auto _xoff = g->xoff;
  call_on_all_cores([_xoff]{ xoff = _xoff; });
  <commit_msg>Ported prefix_sum to be more parallel.<commit_after>/* Copyright (C) 2010-2011 The Trustees of Indiana University.             */
/*                                                                         */
/* Use, modification and distribution is subject to the Boost Software     */
/* License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at */
/* http://www.boost.org/LICENSE_1_0.txt)                                   */
/*                                                                         */
/*  Authors: <NAME>                                             */
/*           <NAME>                                              */

#include "common.h"
#include "oned_csr.h"
#include <stdint.h>
#include <inttypes.h>
#include <stdlib.h>
#include <stddef.h>
#include <string.h>
#include <stdio.h>
#include <assert.h>

#include "Grappa.hpp"
#include "GlobalAllocator.hpp"
#include "ForkJoin.hpp"
#include "Cache.hpp"
#include "Collective.hpp"
#include "Delegate.hpp"
#include "AsyncDelegate.hpp"
#include "GlobalTaskJoiner.hpp"
#include <Array.hpp>
#include "GlobalCompletionEvent.hpp"

using namespace Grappa;

#include "timer.h"

#include <sstream>
#include <functional>

#define MINVECT_SIZE 2

#define XOFF(k) (xoff+2*(k))
#define XENDOFF(k) (xoff+1+2*(k))

GlobalAddress<int64_t> xoff;

static int64_t maxvtx, nv;

LOOP_FUNCTOR( max_func, index,
             (( GlobalAddress<packed_edge>,edges))
             ((int64_t *, max)) )
{
  Incoherent<packed_edge>::RO cedge(edges+index, 1);
  if (cedge[0].v0 > *max) {
    *max = cedge[0].v0;
  }
  if (cedge[0].v1 > *max) {
    *max = cedge[0].v1;
  }
}

LOOP_FUNCTOR( node_max_func, mynode, 
  ((GlobalAddress<packed_edge>, edges))
  ((int64_t,start)) (( int64_t,end )) ((int64_t,init_max)) )
{
  int64_t max = init_max;
  range_t myblock = blockDist(start, end, mynode, Grappa_nodes());
  max_func f;
  f.edges = edges;
  f.max = &max;
  fork_join_onenode(&f, myblock.start, myblock.end);
    
  maxvtx = Grappa_allreduce<int64_t, collective_max, 0>( max );
  
  nv = maxvtx+1;
}

static void find_nv(const tuple_graph* const tg) {
  node_max_func f(tg->edges, 0, tg->nedge, 0);
  fork_join_custom(&f);
}

// note: this corresponds to how Graph500 counts 'degree' (both in- and outgoing edges to each vertex)
LOOP_FUNCTOR( degree_func, index, (( GlobalAddress<packed_edge>, edges )) ((GlobalAddress<int64_t>,xoff)) )
{ 
  Incoherent<packed_edge>::RO cedge(edges+index, 1);
  int64_t i = cedge[0].v0;
  int64_t j = cedge[0].v1;
  if (i != j) { //skip self-edges
    // TODO: these should be overlapping
    Grappa_delegate_fetch_and_add_word(XOFF(i), 1);
    Grappa_delegate_fetch_and_add_word(XOFF(j), 1);
  }
}

LOOP_FUNCTOR( local_reduce_sum, index,
    (( GlobalAddress<int64_t>, xoff )) (( int64_t *, sum )) )
{
  int64_t v = Grappa_delegate_read_word(XOFF(index));
  *sum += v;
}

#define minint(A,B) ((A)<(B)) ? (A) : (B)
int64_t local_prefix;
// ideally this would be forall local with contiguous addresses
LOOP_FUNCTOR( prefix_sum_node, nid,
    ((GlobalAddress<int64_t>,xoff))
    ((int64_t,nv)) ) {
  range_t myblock = blockDist(0, nv, nid, Grappa_nodes());
  
  local_prefix = 0;
  local_reduce_sum fr;
  fr.xoff = xoff;
  fr.sum = &local_prefix;
  fork_join_onenode(&fr, myblock.start, myblock.end);
 
  Grappa_barrier_suspending(); // "phaser style", alternative is replace with forjoins 

  // node 0 sets buf[i] to prefix sum
  // TODO: do this as parallel prefix
  if (nid == 0) {
    int64_t total = local_prefix;
    for (int64_t i=1; i<Grappa_nodes(); i++) {
      //LOG(INFO) << "will add total=" << total << " to node " << i;
      total += Grappa_delegate_fetch_and_add_word(make_global(&local_prefix,i), total);
    }
  }
  
  Grappa_barrier_suspending(); // alternative: replace with forkjoins
  
  int64_t prev_sum = (nid > 0) ? Grappa_delegate_read_word(make_global(&local_prefix,nid-1)) : 0;
  //LOG(INFO) << "prev_sum= " << prev_sum;
 
  // do prefix sum within the slice 
  // Cache a part at a time, doing the sum locally and writing back
  size_t bufsize = 1024;
  int64_t * buf = new int64_t[2*bufsize];
  for (int64_t i=myblock.start; i<myblock.end; i+=bufsize) {
    size_t end = minint(myblock.end, i+bufsize);
    int64_t len = end - i;
    Incoherent<int64_t>::RW cv(XOFF(i), XOFF(end)-XOFF(i), buf);
    for (int64_t j = 0; j<len; j++) {
      int64_t tmp = cv[2*j]; // 2*j because XOFF entries are even
      cv[2*j] = prev_sum;
      prev_sum += tmp;
    } 
  }
  delete buf;
}

int64_t * prefix_temp_base;
int64_t prefix_count;

static int64_t prefix_sum(GlobalAddress<int64_t> xoff, int64_t nv) {
  auto prefix_temp = Grappa_typed_malloc<int64_t>(nv);
  call_on_all_cores([prefix_temp,nv]{ prefix_temp_base = prefix_temp.localize(); });
  
  struct XoffT { int64_t start, end; };
  auto offsets = static_cast<GlobalAddress<XoffT>>(xoff);
  
  forall_localized(offsets, nv, [offsets,nv](int64_t s, int64_t n, XoffT * x){
    char buf[n * delegate::write_msg_size<int64_t>()];
    MessagePoolBase pool(buf, sizeof(buf));
    
    for (int i=0; i<n; i++) {
      int64_t index = make_linear(x+i)-offsets;
      block_offset_t b = indexToBlock(index, nv, cores());
      CHECK_LT(b.block, cores());
      delegate::write_async(pool, make_global(prefix_temp_base+b.offset, b.block), x[i].start);
    }
  });
  
  VLOG(1) << "after moving to block-distribution";
  
  int64_t total_sum = 0;
  auto total_addr = make_global(&total_sum);
  
  on_all_cores([xoff,nv,total_addr] {
    // compute reduction locally
    range_t r = blockDist(0, nv, mycore(), cores());
    auto nlocal = r.end - r.start;
    
    int64_t local_sum = 0;
    for (int64_t i=0; i<nlocal; i++) { local_sum += prefix_temp_base[i]; }
    
    // compute global scan across cores
    Core cafter = cores()-mycore();
    for (Core c = mycore()+1; c < cores(); c++) {
      // TODO: overlap delegates
      delegate::fetch_and_add(make_global(&prefix_count,c), local_sum);
    }

    barrier();
    
    if (mycore() == cores()-1) {
      auto accum = prefix_count + local_sum;
      VLOG(1) << "nadj = " << accum + MINVECT_SIZE;
      delegate::write(xoff+2*nv, accum);
      delegate::write(total_addr, accum + MINVECT_SIZE);
    }
    
    DVLOG(4) << "prefix_count = " << prefix_count << ", local_sum = " << local_sum;
    // do prefix sum locally
    local_sum = prefix_count;
    for (int64_t i=0; i<nlocal; i++) {
      int64_t tmp = prefix_temp_base[i];
      prefix_temp_base[i] = local_sum;
      local_sum += tmp;
    }
    
    // put back into original array
    forall_here(0, nlocal, [xoff,r](int64_t s, int64_t n){
      char buf[n * delegate::write_msg_size<int64_t>()];
      MessagePoolBase pool(buf, sizeof(buf));
      for (int64_t i=s; i<s+n; i++) {
        delegate::write_async(pool, xoff+2*(r.start+i), prefix_temp_base[i]);
      }
    });
    impl::local_gce.wait();
  });

  // return total sum
  return total_sum;
}

LOOP_FUNCTOR( minvect_func, index,
          (( GlobalAddress<int64_t>,xoff )) )
{
  int64_t v = Grappa_delegate_read_word(XOFF(index));
  if (v < MINVECT_SIZE) {
    Grappa_delegate_write_word(XOFF(index), MINVECT_SIZE);
  }
}

LOOP_FUNCTOR( init_xendoff_func, index,
             (( GlobalAddress<int64_t>, xoff )) )
{
  Grappa_delegate_write_word(XENDOFF(index), Grappa_delegate_read_word(XOFF(index)));
}

static void setup_deg_off(const tuple_graph * const tg, csr_graph * g) {
  // note: this corresponds to how Graph500 counts 'degree' (both in- and outgoing edges to each vertex)
  auto _xoff = g->xoff;
  call_on_all_cores([_xoff]{ xoff = _xoff; });
  
{'fraction_non_alphanumeric': 0.1434186189889026, 'fraction_numerical': 0.023350801479654747, 'mean_word_length': 3.1579621916052547, 'pattern_counts': {'":': 0, '<': 113, '<?xml version=': 0, '>': 72, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '514466', 'n_tokens_mistral': 5004, 'n_tokens_neox': 4785, 'n_words': 1055}