<commit_before>/* ----------------------------------------------------------------------
   MR-MPI = MapReduce-MPI library
   http://www.cs.sandia.gov/~sjplimp/mapreduce.html
   <NAME>, <EMAIL>, Sandia National Laboratories

   Copyright (2009) Sandia Corporation.  Under the terms of Contract
   DE-AC04-94AL85000 with Sandia Corporation, the U.S. Government retains
   certain rights in this software.  This software is distributed under 
   the modified Berkeley Software Distribution (BSD) License.

   See the README file in the top-level MapReduce directory.
------------------------------------------------------------------------- */

#include "mpi.h"
#include "ctype.h"
#include "string.h"
#include "stdio.h"
#include "stdlib.h"
#include "vtkType.h"
#include "sys/types.h"
#include "sys/stat.h"
#include "mapreduce.h"
#include "keyvalue.h"
#include "keymultivalue.h"
#include "irregular.h"
#include "hash.h"
#include "memory.h"
#include "error.h"

using namespace MAPREDUCE_NS;

// allocate space for and initialize static class variables

MapReduce *MapReduce::mrptr;
int MapReduce::instance_count = 0;
int MapReduce::mpi_finalize_flag = 0;

// prototypes for non-class functions

void map_file_standalone(int nmap, KeyValue *ptr, void *data);

extern "C" {
  int compare_keys_standalone(const void *, const void *);
  int compare_values_standalone(const void *, const void *);
  int compare_multivalues_standalone(const void *, const void *);
};

#ifdef MIN
#  undef MIN
#endif
#define MIN(A,B) ((A) < (B)) ? (A) : (B)

#ifdef MAX
#  undef MAX
#endif
#define MAX(A,B) ((A) > (B)) ? (A) : (B)

#define FILECHUNK 1
#define MAXLINE 1024

/* ----------------------------------------------------------------------
   construct using caller's MPI communicator
   perform no MPI_init() and no MPI_Finalize()
------------------------------------------------------------------------- */

MapReduce::MapReduce(MPI_Comm caller)
{
  instance_count++;

  comm = caller;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   construct without MPI communicator, use MPI_COMM_WORLD
   perform MPI_Init() if not already initialized
   perform no MPI_Finalize()
------------------------------------------------------------------------- */

MapReduce::MapReduce()
{
  instance_count++;

  int flag;
  MPI_Initialized(&flag);

  if (!flag) {
    int argc = 0;
    char **argv = NULL;
    MPI_Init(&argc,&argv);
  }

  comm = MPI_COMM_WORLD;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   construct without MPI communicator, use MPI_COMM_WORLD
   perform MPI_Init() if not already initialized
   perform MPI_Finalize() if final instance is destructed
------------------------------------------------------------------------- */

MapReduce::MapReduce(double dummy)
{
  (void)dummy;
  instance_count++;
  mpi_finalize_flag = 1;

  int flag;
  MPI_Initialized(&flag);

  if (!flag) {
    int argc = 0;
    char **argv = NULL;
    MPI_Init(&argc,&argv);
  }

  comm = MPI_COMM_WORLD;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   copy constructor, perform a deep copy
------------------------------------------------------------------------- */

MapReduce::MapReduce(MapReduce &mr)
{
  instance_count++;

  comm = mr.comm;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  if (mr.kv) kv = new KeyValue(*mr.kv);
  if (mr.kmv) kmv = new KeyMultiValue(*mr.kmv);

  mapstyle = mr.mapstyle;
  verbosity = mr.verbosity;
  timer = mr.timer;
}

/* ----------------------------------------------------------------------
   free all memory
   if finalize_flag is set and this is last instance, then finalize MPI
------------------------------------------------------------------------- */

MapReduce::~MapReduce()
{
  delete memory;
  delete error;
  delete kv;
  delete kmv;

  instance_count--;
  if (mpi_finalize_flag && instance_count == 0) MPI_Finalize();
}

/* ----------------------------------------------------------------------
   aggregate a KV across procs to create a new KV
   initially, key copies can exist on many procs
   after aggregation, all copies of key are on same proc
   performed via parallel distributed hashing
   hash = user hash function (NULL if not provided)
   requires irregular all2all communication
------------------------------------------------------------------------- */

int MapReduce::aggregate(int (*hash)(char *, int))
{
  if (kv == NULL) error->all("Cannot aggregate without KeyValue");
  if (timer) start_timer();

  if (nprocs == 1) {
    stats("Aggregate",0,verbosity);
    return kv->nkey;
  }

  KeyValue *kvnew = new KeyValue(comm);
  Irregular *irregular = new Irregular(comm);

  // hash each key to a proc ID
  // either use user-provided hash function or hashlittle()

  int nkey = kv->nkey;
  int *keys = kv->keys;
  int *values = kv->values;
  char *keydata = kv->keydata;

  int *proclist = new int[kv->nkey];

  for (int i = 0; i < nkey; i++) {
    char *key = &keydata[keys[i]];
    int keybytes = keys[i+1] - keys[i];
    if (hash) proclist[i] = hash(key,keybytes) % nprocs;
    else proclist[i] = hashlittle(key,keybytes,nprocs) % nprocs;
  }

  // redistribute key sizes, key data, value sizes, value data
  // convert key/value offsets into sizes

  irregular->pattern(nkey,proclist);

  int *slength = proclist;
  for (int i = 0; i < nkey; i++) slength[i] = keys[i+1] - keys[i];

  int nbytes = irregular->size(sizeof(int));
  kvnew->nkey = kvnew->maxkey = nbytes / sizeof(int);
  kvnew->keys = (int *) memory->smalloc(nbytes,"MR:keys");
  irregular->exchange((char *) slength,(char *) kvnew->keys);

  nbytes = irregular->size(slength,kv->keys,kvnew->keys);
  kvnew->keysize = kvnew->maxkeysize = nbytes;
  kvnew->keydata = (char *) memory->smalloc(nbytes,"MR:keydata");
  irregular->exchange(kv->keydata,kvnew->keydata);

  for (int i = 0; i < nkey; i++) slength[i] = values[i+1] - values[i];

  nbytes = irregular->size(sizeof(int));
  kvnew->values = (int *) memory->smalloc(nbytes,"MR:values");
  irregular->exchange((char *) slength,(char *) kvnew->values);

  nbytes = irregular->size(slength,kv->values,kvnew->values);
  kvnew->valuesize = kvnew->maxvaluesize = nbytes;
  kvnew->valuedata = (char *) memory->smalloc(nbytes,"MR:valuedata");
  irregular->exchange(kv->valuedata,kvnew->valuedata);

  delete [] slength;
  delete irregular;

  // convert key/value sizes back into offsets

  nkey = kvnew->nkey;
  keys = kvnew->keys;
  values = kvnew->values;

  int keysize = 0;
  int valuesize = 0;
  int tmp;

  for (int i = 0; i < nkey; i++) {
    tmp = keys[i];
    keys[i] = keysize;
    keysize += tmp;
    tmp = values[i];
    values[i] = valuesize;
    valuesize += tmp;
  }

  delete kv;
  kv = kvnew;
  kv->complete();

  stats("Aggregate",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   clone KV to KMV so that KMV pairs are one-to-one copies of KV pairs
   each proc clones only its data
   assume each KV key is unique, but is not required
------------------------------------------------------------------------- */

int MapReduce::clone()
{
  if (kv == NULL) error->all("Cannot clone without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->clone(kv);
  delete kv;
  kv = NULL;

  stats("Clone",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   collapse KV into a KMV with a single key/value
   each proc collapses only its data
   new key = provided key name (same on every proc)
   new value = list of old key,value,key,value,etc
------------------------------------------------------------------------- */

int MapReduce::collapse(char *key, int keybytes)
{
  if (kv == NULL) error->all("Cannot collapse without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->collapse(key,keybytes,kv);
  delete kv;
  kv = NULL;

  stats("Collapse",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   collate KV to create a KMV
   aggregate followed by a convert
   hash = user hash function (NULL if not provided)
------------------------------------------------------------------------- */

int MapReduce::collate(int (*hash)(char *, int))
{
  if (kv == NULL) error->all("Cannot collate without KeyValue");
  if (timer) start_timer();

  int verbosity_hold = verbosity;
  int timer_hold = timer;
  verbosity = timer = 0;

  aggregate(hash);
  convert();

  verbosity = verbosity_hold;
  timer = timer_hold;
  stats("Collate",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   compress KV to create a smaller KV
   duplicate keys are replaced with a single key/value
   each proc compresses only its data
   create a temporary KMV
   call appcompress() with each key/multivalue in KMV
   appcompress() returns single key/value to new KV
------------------------------------------------------------------------- */

int MapReduce::compress(void (*appcompress)(char *, int, char *,
              int, int *, KeyValue *, void *),
      void *ptr)
{
  if (kv == NULL) error->all("Cannot compress without KeyValue");
  if (timer) start_timer();

  KeyMultiValue *kmvtmp = new KeyMultiValue(comm);
  kmvtmp->convert(kv);
  kv = new KeyValue(comm);

  int ncompress = kmvtmp->nkey;
  int *keys = kmvtmp->keys;
  int *multivalues = kmvtmp->multivalues;
  int *nvalues = kmvtmp->nvalues;
  int *valuesizes = kmvtmp->valuesizes;
  char *keydata = kmvtmp->keydata;
  char *multivaluedata = kmvtmp->multivaluedata;

  for (int i = 0; i < ncompress; i++)
    appcompress(&keydata[keys[i]],keys[i+1]-keys[i],
    &multivaluedata[multivalues[i]],
    nvalues[i+1]-nvalues[i],&valuesizes[nvalues[i]],
    kv,ptr);

  delete kmvtmp;
  kv->complete();

  stats("Compress",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   convert KV to KMV
   duplicate keys are replaced with a single key/multivalue
   each proc converts only its data
   new key = old unique key
   new multivalue = concatenated list of all values for that key in KV
------------------------------------------------------------------------- */

int MapReduce::convert()
{
  if (kv == NULL) error->all("Cannot convert without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->convert(kv);
  delete kv;
  kv = NULL;

  stats("Convert",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   gather a distributed KV to a new KV on fewer procs
   numprocs = # of procs new KV resides on (0 to numprocs-1)
------------------------------------------------------------------------- */

int MapReduce::gather(int numprocs)
{
  if (kv == NULL) error->all("Cannot gather without KeyValue");
  if (numprocs < 1 || numprocs > nprocs) 
    error->all("Invalid proc count for gather");
  if (timer) start_timer();

  if (nprocs == 1 || numprocs == nprocs) {
    stats("Gather",0,verbosity);
    int nkeyall;
    MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
    return nkeyall;
  }

  // lo procs collect key/value pairs from hi procs
  // lo procs are those with ID < numprocs
  // lo procs recv from set of hi procs with same (ID % numprocs)

  int flag,size;
  MPI_Status status;

  if (me < numprocs) {
    char *buf = NULL;
    int maxsize = 0;

    for (int iproc = me+numprocs; iproc < nprocs; iproc += numprocs) {
      MPI_Send(&flag,0,MPI_INT,iproc,0,comm);
      MPI_Recv(&size,1,MPI_INT,iproc,0,comm,&status);
      if (size > maxsize) {
  delete [] buf;
  buf = new char[size];
      }
      MPI_Recv(buf,size,MPI_BYTE,iproc,0,comm,&status);
      kv->unpack(buf);
    }
    
    delete [] buf;

  } else {
    char *buf;
    size = kv->pack(&buf);

    int iproc = me % numprocs;
    MPI_Recv(&flag,0,MPI_INT,iproc,0,comm,&status);
    MPI_Send(&size,1,MPI_INT,iproc,0,comm);
    MPI_Send(buf,size,MPI_BYTE,iproc,0,comm);

    delete [] buf;
    delete kv;
    kv = new KeyValue(comm);
  }

  kv->complete();
  stats("Gather",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   create a KV via a parallel map operation for nmap tasks
   make one call to appmap() for each task
   mapstyle determines how tasks are partitioned to processors
------------------------------------------------------------------------- */

int MapReduce::map(int nmap, void (*appmap)(int, KeyValue *, void *),
       void *ptr, int addflag)
{
  MPI_Status status;

  if (timer) start_timer();

  delete kmv;
  kmv = NULL;

  if (addflag == 0) {
    delete kv;
    kv = new KeyValue(comm);
  } else if (kv == NULL) 
    kv = new KeyValue(comm);

  // nprocs = 1 = all tasks to single processor
  // mapstyle 0 = chunk of tasks to each proc
  // mapstyle 1 = strided tasks to each proc
  // mapstyle 2 = master/slave assignment of tasks

  if (nprocs == 1) {
    for (int itask = 0; itask < nmap; itask++)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 0) {
    vtkTypeUInt64 nmap64 = nmap;
    int lo = me * (int)(nmap64 / nprocs);
    int hi = (me+1) * (int)(nmap64 / nprocs);
    for (int itask = lo; itask < hi; itask++)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 1) {
    for (int itask = me; itask < nmap; itask += nprocs)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 2) {
    if (me == 0) {
      int doneflag = -1;
      int ndone = 0;
      int itask = 0;
      for (int iproc = 1; iproc < nprocs; iproc++) {
  if (itask < nmap) {
    MPI_Send(&itask,1,MPI_INT,iproc,0,comm);
    itask++;
  } else {
    MPI_Send(&doneflag,1,MPI_INT,iproc,0,comm);
    ndone++;
  }
      }
      while (ndone < nprocs-1) {
  int iproc,tmp;
  MPI_Recv(&tmp,1,MPI_INT,MPI_ANY_SOURCE,0,comm,&status);
  iproc = status.MPI_SOURCE;

  if (itask < nmap) {
    MPI_Send(&itask,1,MPI_INT,iproc,0,comm);
    itask++;
  } else {
    MPI_Send(&doneflag,1,MPI_INT,iproc,0,comm);
    ndone++;
  }
      }

    } else {
      while (1) {
        int itask;
        MPI_Recv(&itask,1,MPI_INT,0,0,comm,&status);
        if (itask < 0) break;
        appmap(itask,kv,ptr);
        MPI_Send(&itask,1,MPI_INT,0,0,comm);
      }
    }

  } else error->all("Invalid mapstyle setting");

  kv->complete();
  stats("Map",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   create a KV via a parallel map operation for list of files in file
   make one call to appmap() for each file in file
   mapstyle determines how tasks are partitioned to processors
------------------------------------------------------------------------- */

int MapReduce::map(char *file, void (*appmap)(int, char *, KeyValue *, void *),
       void *ptr, int addflag)
{
  int n;
  char line[MAXLINE];
  MPI_Status status;

  if (timer) start_timer();

  delete kmv;
  kmv = NULL;

  if (addflag == 0) {
    delete kv;
    kv = new KeyValue(comm);
  } else if (kv == NULL) 
    kv = new KeyValue(comm);

  // open file and extract filenames
  // bcast each filename to all procs
  // trim whitespace from beginning and end of filename

  int nmap = 0;
  int maxfiles = 0;
  char **files = NULL;
  FILE *fp = 0;

  if (me == 0) {<commit_msg>COMP: eleminate warnings from using cast size_t to in.<commit_after>/* ----------------------------------------------------------------------
   MR-MPI = MapReduce-MPI library
   http://www.cs.sandia.gov/~sjplimp/mapreduce.html
   <NAME>, <EMAIL>, Sandia National Laboratories

   Copyright (2009) Sandia Corporation.  Under the terms of Contract
   DE-AC04-94AL85000 with Sandia Corporation, the U.S. Government retains
   certain rights in this software.  This software is distributed under 
   the modified Berkeley Software Distribution (BSD) License.

   See the README file in the top-level MapReduce directory.
------------------------------------------------------------------------- */

#include "mpi.h"
#include "ctype.h"
#include "string.h"
#include "stdio.h"
#include "stdlib.h"
#include "vtkType.h"
#include "sys/types.h"
#include "sys/stat.h"
#include "mapreduce.h"
#include "keyvalue.h"
#include "keymultivalue.h"
#include "irregular.h"
#include "hash.h"
#include "memory.h"
#include "error.h"

using namespace MAPREDUCE_NS;

// allocate space for and initialize static class variables

MapReduce *MapReduce::mrptr;
int MapReduce::instance_count = 0;
int MapReduce::mpi_finalize_flag = 0;

// prototypes for non-class functions

void map_file_standalone(int nmap, KeyValue *ptr, void *data);

extern "C" {
  int compare_keys_standalone(const void *, const void *);
  int compare_values_standalone(const void *, const void *);
  int compare_multivalues_standalone(const void *, const void *);
};

#ifdef MIN
#  undef MIN
#endif
#define MIN(A,B) ((A) < (B)) ? (A) : (B)

#ifdef MAX
#  undef MAX
#endif
#define MAX(A,B) ((A) > (B)) ? (A) : (B)

#define FILECHUNK 1
#define MAXLINE 1024

/* ----------------------------------------------------------------------
   construct using caller's MPI communicator
   perform no MPI_init() and no MPI_Finalize()
------------------------------------------------------------------------- */

MapReduce::MapReduce(MPI_Comm caller)
{
  instance_count++;

  comm = caller;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   construct without MPI communicator, use MPI_COMM_WORLD
   perform MPI_Init() if not already initialized
   perform no MPI_Finalize()
------------------------------------------------------------------------- */

MapReduce::MapReduce()
{
  instance_count++;

  int flag;
  MPI_Initialized(&flag);

  if (!flag) {
    int argc = 0;
    char **argv = NULL;
    MPI_Init(&argc,&argv);
  }

  comm = MPI_COMM_WORLD;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   construct without MPI communicator, use MPI_COMM_WORLD
   perform MPI_Init() if not already initialized
   perform MPI_Finalize() if final instance is destructed
------------------------------------------------------------------------- */

MapReduce::MapReduce(double dummy)
{
  (void)dummy;
  instance_count++;
  mpi_finalize_flag = 1;

  int flag;
  MPI_Initialized(&flag);

  if (!flag) {
    int argc = 0;
    char **argv = NULL;
    MPI_Init(&argc,&argv);
  }

  comm = MPI_COMM_WORLD;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  mapstyle = 0;
  verbosity = 0;
  timer = 0;
}

/* ----------------------------------------------------------------------
   copy constructor, perform a deep copy
------------------------------------------------------------------------- */

MapReduce::MapReduce(MapReduce &mr)
{
  instance_count++;

  comm = mr.comm;
  MPI_Comm_rank(comm,&me);
  MPI_Comm_size(comm,&nprocs);

  memory = new Memory(comm);
  error = new Error(comm);

  kv = NULL;
  kmv = NULL;

  if (mr.kv) kv = new KeyValue(*mr.kv);
  if (mr.kmv) kmv = new KeyMultiValue(*mr.kmv);

  mapstyle = mr.mapstyle;
  verbosity = mr.verbosity;
  timer = mr.timer;
}

/* ----------------------------------------------------------------------
   free all memory
   if finalize_flag is set and this is last instance, then finalize MPI
------------------------------------------------------------------------- */

MapReduce::~MapReduce()
{
  delete memory;
  delete error;
  delete kv;
  delete kmv;

  instance_count--;
  if (mpi_finalize_flag && instance_count == 0) MPI_Finalize();
}

/* ----------------------------------------------------------------------
   aggregate a KV across procs to create a new KV
   initially, key copies can exist on many procs
   after aggregation, all copies of key are on same proc
   performed via parallel distributed hashing
   hash = user hash function (NULL if not provided)
   requires irregular all2all communication
------------------------------------------------------------------------- */

int MapReduce::aggregate(int (*hash)(char *, int))
{
  if (kv == NULL) error->all("Cannot aggregate without KeyValue");
  if (timer) start_timer();

  if (nprocs == 1) {
    stats("Aggregate",0,verbosity);
    return kv->nkey;
  }

  KeyValue *kvnew = new KeyValue(comm);
  Irregular *irregular = new Irregular(comm);

  // hash each key to a proc ID
  // either use user-provided hash function or hashlittle()

  int nkey = kv->nkey;
  int *keys = kv->keys;
  int *values = kv->values;
  char *keydata = kv->keydata;

  int *proclist = new int[kv->nkey];

  for (int i = 0; i < nkey; i++) {
    char *key = &keydata[keys[i]];
    int keybytes = keys[i+1] - keys[i];
    if (hash) proclist[i] = hash(key,keybytes) % nprocs;
    else proclist[i] = hashlittle(key,keybytes,nprocs) % nprocs;
  }

  // redistribute key sizes, key data, value sizes, value data
  // convert key/value offsets into sizes

  irregular->pattern(nkey,proclist);

  int *slength = proclist;
  for (int i = 0; i < nkey; i++) slength[i] = keys[i+1] - keys[i];

  int nbytes = irregular->size(sizeof(int));
  kvnew->nkey = kvnew->maxkey = nbytes / sizeof(int);
  kvnew->keys = (int *) memory->smalloc(nbytes,"MR:keys");
  irregular->exchange((char *) slength,(char *) kvnew->keys);

  nbytes = irregular->size(slength,kv->keys,kvnew->keys);
  kvnew->keysize = kvnew->maxkeysize = nbytes;
  kvnew->keydata = (char *) memory->smalloc(nbytes,"MR:keydata");
  irregular->exchange(kv->keydata,kvnew->keydata);

  for (int i = 0; i < nkey; i++) slength[i] = values[i+1] - values[i];

  nbytes = irregular->size(sizeof(int));
  kvnew->values = (int *) memory->smalloc(nbytes,"MR:values");
  irregular->exchange((char *) slength,(char *) kvnew->values);

  nbytes = irregular->size(slength,kv->values,kvnew->values);
  kvnew->valuesize = kvnew->maxvaluesize = nbytes;
  kvnew->valuedata = (char *) memory->smalloc(nbytes,"MR:valuedata");
  irregular->exchange(kv->valuedata,kvnew->valuedata);

  delete [] slength;
  delete irregular;

  // convert key/value sizes back into offsets

  nkey = kvnew->nkey;
  keys = kvnew->keys;
  values = kvnew->values;

  int keysize = 0;
  int valuesize = 0;
  int tmp;

  for (int i = 0; i < nkey; i++) {
    tmp = keys[i];
    keys[i] = keysize;
    keysize += tmp;
    tmp = values[i];
    values[i] = valuesize;
    valuesize += tmp;
  }

  delete kv;
  kv = kvnew;
  kv->complete();

  stats("Aggregate",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   clone KV to KMV so that KMV pairs are one-to-one copies of KV pairs
   each proc clones only its data
   assume each KV key is unique, but is not required
------------------------------------------------------------------------- */

int MapReduce::clone()
{
  if (kv == NULL) error->all("Cannot clone without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->clone(kv);
  delete kv;
  kv = NULL;

  stats("Clone",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   collapse KV into a KMV with a single key/value
   each proc collapses only its data
   new key = provided key name (same on every proc)
   new value = list of old key,value,key,value,etc
------------------------------------------------------------------------- */

int MapReduce::collapse(char *key, int keybytes)
{
  if (kv == NULL) error->all("Cannot collapse without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->collapse(key,keybytes,kv);
  delete kv;
  kv = NULL;

  stats("Collapse",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   collate KV to create a KMV
   aggregate followed by a convert
   hash = user hash function (NULL if not provided)
------------------------------------------------------------------------- */

int MapReduce::collate(int (*hash)(char *, int))
{
  if (kv == NULL) error->all("Cannot collate without KeyValue");
  if (timer) start_timer();

  int verbosity_hold = verbosity;
  int timer_hold = timer;
  verbosity = timer = 0;

  aggregate(hash);
  convert();

  verbosity = verbosity_hold;
  timer = timer_hold;
  stats("Collate",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   compress KV to create a smaller KV
   duplicate keys are replaced with a single key/value
   each proc compresses only its data
   create a temporary KMV
   call appcompress() with each key/multivalue in KMV
   appcompress() returns single key/value to new KV
------------------------------------------------------------------------- */

int MapReduce::compress(void (*appcompress)(char *, int, char *,
              int, int *, KeyValue *, void *),
      void *ptr)
{
  if (kv == NULL) error->all("Cannot compress without KeyValue");
  if (timer) start_timer();

  KeyMultiValue *kmvtmp = new KeyMultiValue(comm);
  kmvtmp->convert(kv);
  kv = new KeyValue(comm);

  int ncompress = kmvtmp->nkey;
  int *keys = kmvtmp->keys;
  int *multivalues = kmvtmp->multivalues;
  int *nvalues = kmvtmp->nvalues;
  int *valuesizes = kmvtmp->valuesizes;
  char *keydata = kmvtmp->keydata;
  char *multivaluedata = kmvtmp->multivaluedata;

  for (int i = 0; i < ncompress; i++)
    appcompress(&keydata[keys[i]],keys[i+1]-keys[i],
    &multivaluedata[multivalues[i]],
    nvalues[i+1]-nvalues[i],&valuesizes[nvalues[i]],
    kv,ptr);

  delete kmvtmp;
  kv->complete();

  stats("Compress",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   convert KV to KMV
   duplicate keys are replaced with a single key/multivalue
   each proc converts only its data
   new key = old unique key
   new multivalue = concatenated list of all values for that key in KV
------------------------------------------------------------------------- */

int MapReduce::convert()
{
  if (kv == NULL) error->all("Cannot convert without KeyValue");
  if (timer) start_timer();

  kmv = new KeyMultiValue(comm);
  kmv->convert(kv);
  delete kv;
  kv = NULL;

  stats("Convert",1,verbosity);

  int nkeyall;
  MPI_Allreduce(&kmv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   gather a distributed KV to a new KV on fewer procs
   numprocs = # of procs new KV resides on (0 to numprocs-1)
------------------------------------------------------------------------- */

int MapReduce::gather(int numprocs)
{
  if (kv == NULL) error->all("Cannot gather without KeyValue");
  if (numprocs < 1 || numprocs > nprocs) 
    error->all("Invalid proc count for gather");
  if (timer) start_timer();

  if (nprocs == 1 || numprocs == nprocs) {
    stats("Gather",0,verbosity);
    int nkeyall;
    MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
    return nkeyall;
  }

  // lo procs collect key/value pairs from hi procs
  // lo procs are those with ID < numprocs
  // lo procs recv from set of hi procs with same (ID % numprocs)

  int flag,size;
  MPI_Status status;

  if (me < numprocs) {
    char *buf = NULL;
    int maxsize = 0;

    for (int iproc = me+numprocs; iproc < nprocs; iproc += numprocs) {
      MPI_Send(&flag,0,MPI_INT,iproc,0,comm);
      MPI_Recv(&size,1,MPI_INT,iproc,0,comm,&status);
      if (size > maxsize) {
  delete [] buf;
  buf = new char[size];
      }
      MPI_Recv(buf,size,MPI_BYTE,iproc,0,comm,&status);
      kv->unpack(buf);
    }
    
    delete [] buf;

  } else {
    char *buf;
    size = kv->pack(&buf);

    int iproc = me % numprocs;
    MPI_Recv(&flag,0,MPI_INT,iproc,0,comm,&status);
    MPI_Send(&size,1,MPI_INT,iproc,0,comm);
    MPI_Send(buf,size,MPI_BYTE,iproc,0,comm);

    delete [] buf;
    delete kv;
    kv = new KeyValue(comm);
  }

  kv->complete();
  stats("Gather",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   create a KV via a parallel map operation for nmap tasks
   make one call to appmap() for each task
   mapstyle determines how tasks are partitioned to processors
------------------------------------------------------------------------- */

int MapReduce::map(int nmap, void (*appmap)(int, KeyValue *, void *),
       void *ptr, int addflag)
{
  MPI_Status status;

  if (timer) start_timer();

  delete kmv;
  kmv = NULL;

  if (addflag == 0) {
    delete kv;
    kv = new KeyValue(comm);
  } else if (kv == NULL) 
    kv = new KeyValue(comm);

  // nprocs = 1 = all tasks to single processor
  // mapstyle 0 = chunk of tasks to each proc
  // mapstyle 1 = strided tasks to each proc
  // mapstyle 2 = master/slave assignment of tasks

  if (nprocs == 1) {
    for (int itask = 0; itask < nmap; itask++)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 0) {
    vtkTypeUInt64 nmap64 = nmap;
    int lo = me * (int)(nmap64 / nprocs);
    int hi = (me+1) * (int)(nmap64 / nprocs);
    for (int itask = lo; itask < hi; itask++)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 1) {
    for (int itask = me; itask < nmap; itask += nprocs)
      appmap(itask,kv,ptr);

  } else if (mapstyle == 2) {
    if (me == 0) {
      int doneflag = -1;
      int ndone = 0;
      int itask = 0;
      for (int iproc = 1; iproc < nprocs; iproc++) {
  if (itask < nmap) {
    MPI_Send(&itask,1,MPI_INT,iproc,0,comm);
    itask++;
  } else {
    MPI_Send(&doneflag,1,MPI_INT,iproc,0,comm);
    ndone++;
  }
      }
      while (ndone < nprocs-1) {
  int iproc,tmp;
  MPI_Recv(&tmp,1,MPI_INT,MPI_ANY_SOURCE,0,comm,&status);
  iproc = status.MPI_SOURCE;

  if (itask < nmap) {
    MPI_Send(&itask,1,MPI_INT,iproc,0,comm);
    itask++;
  } else {
    MPI_Send(&doneflag,1,MPI_INT,iproc,0,comm);
    ndone++;
  }
      }

    } else {
      while (1) {
        int itask;
        MPI_Recv(&itask,1,MPI_INT,0,0,comm,&status);
        if (itask < 0) break;
        appmap(itask,kv,ptr);
        MPI_Send(&itask,1,MPI_INT,0,0,comm);
      }
    }

  } else error->all("Invalid mapstyle setting");

  kv->complete();
  stats("Map",0,verbosity);

  int nkeyall;
  MPI_Allreduce(&kv->nkey,&nkeyall,1,MPI_INT,MPI_SUM,comm);
  return nkeyall;
}

/* ----------------------------------------------------------------------
   create a KV via a parallel map operation for list of files in file
   make one call to appmap() for each file in file
   mapstyle determines how tasks are partitioned to processors
------------------------------------------------------------------------- */

int MapReduce::map(char *file, void (*appmap)(int, char *, KeyValue *, void *),
       void *ptr, int addflag)
{
  size_t n;
  char line[MAXLINE];
  MPI_Status status;

  if (timer) start_timer();

  delete kmv;
  kmv = NULL;

  if (addflag == 0) {
    delete kv;
    kv = new KeyValue(comm);
  } else if (kv == NULL) 
    kv = new KeyValue(comm);

  // open file and extract filenames
  // bcast each filename to all procs
  // trim whitespace from beginning and end of filename

  int nmap = 0;
  int maxfiles = 0;
  char **files = NULL;
  FILE *fp = 0;

  if (me == 0) {
{'fraction_non_alphanumeric': 0.24962514244587056, 'fraction_numerical': 0.007797037125892161, 'mean_word_length': 2.0768591991142276, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 167, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1665041', 'n_tokens_mistral': 11204, 'n_tokens_neox': 10745, 'n_words': 2850}