<commit_before>/*
 * reduce.cpp
 *
 *  Created on: 2 Jan 2016
 *      Author: nick
 */

#include <vector>
#include <iterator>
#include <string.h>
#include <stdlib.h>
#include "messaging.h"
#include "reduce.h"
#include "ndm.h"
#include "data_types.h"
#include "pthread.h"
#include "misc.h"
#include "groups.h"

static int REDUCE_ACTION_ID = 2;

static std::vector<ReductionState*> reduction_state;
static pthread_mutex_t reduction_state_lock;
static Messaging messaging;
static ThreadPool threadPool;

static void reduction_callback_at_root(void*, NDM_Metadata);
static std::vector<ReductionState*>::iterator findReductionState(const char*);
static void sendToSpecificProcess(Messaging*, void*, int, int, int, int, NDM_Group, const char*);
static void applyOperation(ReductionState*, void*, int, int);
static void applyActualOperation(int*, int*, NDM_Op);
static void applyActualOperation(float*, float*, NDM_Op);
static void applyActualOperation(double*, double*, NDM_Op);

void initialise_ndmReduce(Messaging messaging_arg, ThreadPool threadPool_arg) {
  messaging = messaging_arg;
  threadPool = threadPool_arg;
  pthread_mutex_init(&reduction_state_lock, NULL);
}

void collective_ndmReduce(Messaging* messaging, ThreadPool* threadPool, void* data, int type, int size, int totalSize,
                          int contributionsPerElement, int startPoint, NDM_Op operation, void (*callback)(void*, NDM_Metadata),
                          int root, int my_rank, NDM_Group comm_group, const char* unique_id) {
  pthread_mutex_lock(&reduction_state_lock);
  std::vector<ReductionState*>::iterator it = findReductionState(unique_id);
  ReductionState* state;
  if (it == reduction_state.end()) {
    state = new ReductionState(type, totalSize, contributionsPerElement, callback, operation, root, comm_group, unique_id);
    reduction_state.insert(reduction_state.begin(), state);
  } else {
    state = *it;
  }
  applyOperation(state, data, size, startPoint);
  if (state->isDataSettingComplete()) {
    if (getNumberDistinctProcesses(comm_group) == 1) {
      it = findReductionState(unique_id);
      reduction_state.erase(it);
      pthread_mutex_unlock(&reduction_state_lock);
      state->getCallback()(state->getData(), messaging->generateMetaData(state->getType(), state->getSize(), -1, state->getRoot(),
                                                                         state->getCommGroup(), state->getUniqueId().c_str()));
      delete (state);
    } else {
      pthread_mutex_unlock(&reduction_state_lock);
      state->incrementContributedProcesses();
      if ((my_rank >= 0 && my_rank == root) || (my_rank == NDM_ANY_MYRANK && isRankLocalToGroup(comm_group, root))) {
        int i;
        for (i = 1; i < getNumberDistinctProcesses(comm_group); i++) {
          messaging->registerCommand(unique_id, NDM_ANY_MYRANK, root, comm_group, REDUCE_ACTION_ID, false, reduction_callback_at_root);
        }
      } else {
        sendToSpecificProcess(messaging, state->getData(), type, totalSize, my_rank, root, comm_group, unique_id);
      }
    }
  } else {
    pthread_mutex_unlock(&reduction_state_lock);
  }
}

static std::vector<ReductionState*>::iterator findReductionState(const char* uniqueId) {
  std::string searchSalt = std::string(uniqueId);
  size_t wildCardLocB = searchSalt.find('*');
  std::vector<ReductionState*>::iterator it;
  ReductionState* specificState = NULL;
  for (it = reduction_state.begin(); it != reduction_state.end(); it++) {
    size_t wildCardLocA = (*it)->getUniqueId().find('*');
    if (wildCardLocA != std::string::npos || wildCardLocB != std::string::npos) {
      if (wildCardLocA == std::string::npos) {
        if ((*it)->getUniqueId().substr(0, wildCardLocB).compare(searchSalt.substr(0, wildCardLocB)) == 0) return it;
      } else if (wildCardLocB == std::string::npos) {
        if ((*it)->getUniqueId().substr(0, wildCardLocA).compare(searchSalt.substr(0, wildCardLocA)) == 0) return it;
      } else {
        if ((*it)->getUniqueId().substr(0, wildCardLocA).compare(searchSalt.substr(0, wildCardLocB)) == 0) return it;
      }
    } else {
      if ((*it)->getUniqueId().compare(searchSalt) == 0) return it;
    }
  }
  return reduction_state.end();
}

static void sendToSpecificProcess(Messaging* messaging, void* data, int type, int size, int source, int target, NDM_Group comm_group,
                                  const char* unique_id) {
  int dataSize;
  char* buffer = messaging->packageMessage(data, type, size, source, target, comm_group, REDUCE_ACTION_ID, unique_id, &dataSize);
  messaging->sendMessage(buffer, dataSize, target, comm_group);
}

static void reduction_callback_at_root(void* buffer, NDM_Metadata metaData) {
  pthread_mutex_lock(&reduction_state_lock);
  std::vector<ReductionState*>::iterator it = findReductionState(metaData.unique_id);
  if (it == reduction_state.end()) raiseError("Reduction state not found");
  ReductionState* specificState = *it;
  specificState->lock();
  specificState->incrementContributedProcesses();
  bool isComplete = specificState->hasCompletedAllProcessesReduction();
  if (isComplete) reduction_state.erase(it);
  pthread_mutex_unlock(&reduction_state_lock);
  int i;
  if (specificState->getSize() > 0 && specificState->getType() != NDM_NOTYPE && metaData.data_type != NDM_NOTYPE &&
      metaData.number_elements > 0) {
    applyOperation(specificState, buffer, specificState->getSize(), 0);
  }
  specificState->unlock();<commit_msg>Reduce uses a map for lookup<commit_after>/*
 * reduce.cpp
 *
 *  Created on: 2 Jan 2016
 *      Author: nick
 */

#include <map>
#include <vector>
#include <iterator>
#include <string.h>
#include <stdlib.h>
#include "messaging.h"
#include "reduce.h"
#include "ndm.h"
#include "data_types.h"
#include "pthread.h"
#include "misc.h"
#include "groups.h"

static int REDUCE_ACTION_ID = 2;

struct ReductionStateComparitor {
  bool operator()(std::string a, std::string b) {
    size_t wildCardLocA = a.find('*');
    size_t wildCardLocB = b.find('*');
    if (wildCardLocA != std::string::npos || wildCardLocB != std::string::npos) {
      if (wildCardLocA == std::string::npos) {
        return a.substr(0, wildCardLocB) < b.substr(0, wildCardLocB);
      } else if (wildCardLocB == std::string::npos) {
        return a.substr(0, wildCardLocA) < b.substr(0, wildCardLocA);
      } else {
        return a.substr(0, wildCardLocA) < b.substr(0, wildCardLocB);
      }
    } else {
      return a < b;
    }
  }
};

static std::map<std::string, ReductionState*, ReductionStateComparitor> reduction_state;
static pthread_mutex_t reduction_state_lock;
static Messaging messaging;
static ThreadPool threadPool;

static void reduction_callback_at_root(void*, NDM_Metadata);
static void sendToSpecificProcess(Messaging*, void*, int, int, int, int, NDM_Group, const char*);
static void applyOperation(ReductionState*, void*, int, int);
static void applyActualOperation(int*, int*, NDM_Op);
static void applyActualOperation(float*, float*, NDM_Op);
static void applyActualOperation(double*, double*, NDM_Op);

void initialise_ndmReduce(Messaging messaging_arg, ThreadPool threadPool_arg) {
  messaging = messaging_arg;
  threadPool = threadPool_arg;
  pthread_mutex_init(&reduction_state_lock, NULL);
}

void collective_ndmReduce(Messaging* messaging, ThreadPool* threadPool, void* data, int type, int size, int totalSize,
                          int contributionsPerElement, int startPoint, NDM_Op operation, void (*callback)(void*, NDM_Metadata),
                          int root, int my_rank, NDM_Group comm_group, const char* unique_id) {
  pthread_mutex_lock(&reduction_state_lock);
  std::map<std::string, ReductionState*, ReductionStateComparitor>::iterator it = reduction_state.find(unique_id);
  ReductionState* state;
  if (it == reduction_state.end()) {
    state = new ReductionState(type, totalSize, contributionsPerElement, callback, operation, root, comm_group, unique_id);
    it = reduction_state.insert(std::pair<std::string, ReductionState*>(state->getUniqueId(), state)).first;
  } else {
    state = it->second;
  }
  applyOperation(state, data, size, startPoint);
  if (state->isDataSettingComplete()) {
    if (getNumberDistinctProcesses(comm_group) == 1) {
      reduction_state.erase(it);
      pthread_mutex_unlock(&reduction_state_lock);
      state->getCallback()(state->getData(), messaging->generateMetaData(state->getType(), state->getSize(), -1, state->getRoot(),
                                                                         state->getCommGroup(), state->getUniqueId().c_str()));
      delete (state);
    } else {
      pthread_mutex_unlock(&reduction_state_lock);
      state->incrementContributedProcesses();
      if ((my_rank >= 0 && my_rank == root) || (my_rank == NDM_ANY_MYRANK && isRankLocalToGroup(comm_group, root))) {
        int i;
        for (i = 1; i < getNumberDistinctProcesses(comm_group); i++) {
          messaging->registerCommand(unique_id, NDM_ANY_MYRANK, root, comm_group, REDUCE_ACTION_ID, false, reduction_callback_at_root);
        }
      } else {
        sendToSpecificProcess(messaging, state->getData(), type, totalSize, my_rank, root, comm_group, unique_id);
      }
    }
  } else {
    pthread_mutex_unlock(&reduction_state_lock);
  }
}

static void sendToSpecificProcess(Messaging* messaging, void* data, int type, int size, int source, int target, NDM_Group comm_group,
                                  const char* unique_id) {
  int dataSize;
  char* buffer = messaging->packageMessage(data, type, size, source, target, comm_group, REDUCE_ACTION_ID, unique_id, &dataSize);
  messaging->sendMessage(buffer, dataSize, target, comm_group);
}

static void reduction_callback_at_root(void* buffer, NDM_Metadata metaData) {
  pthread_mutex_lock(&reduction_state_lock);
  std::map<std::string, ReductionState*, ReductionStateComparitor>::iterator it = reduction_state.find(metaData.unique_id);
  if (it == reduction_state.end()) raiseError("Reduction state not found");
  ReductionState* specificState = it->second;
  specificState->lock();
  specificState->incrementContributedProcesses();
  bool isComplete = specificState->hasCompletedAllProcessesReduction();
  if (isComplete) reduction_state.erase(it);
  pthread_mutex_unlock(&reduction_state_lock);
  int i;
  if (specificState->getSize() > 0 && specificState->getType() != NDM_NOTYPE && metaData.data_type != NDM_NOTYPE &&
      metaData.number_elements > 0) {
    applyOperation(specificState, buffer, specificState->getSize(), 0);
  }
  specificState->unlock();
{'fraction_non_alphanumeric': 0.12960682141165325, 'fraction_numerical': 0.003979156797726196, 'mean_word_length': 4.512271540469974, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 78, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4175963', 'n_tokens_mistral': 3356, 'n_tokens_neox': 3256, 'n_words': 708}