SimilarityScheduler.h

Go to the documentation of this file.

//
// Created by Nico on 13/05/2019.
//
 
#ifndef JODA_SIMILARITYSCHEDULER_H
#define JODA_SIMILARITYSCHEDULER_H
 
#include <joda/config/config.h>
#include <joda/container/ContainerFlags.h>
#include <joda/container/JSONContainer.h>
#include <joda/storage/JSONStorage.h>
#include <rapidjson/istreamwrapper.h>
#include "measures/IJSONSimilarityMeasure.h"
 
template <typename SIM>
class SimilarityScheduler {
 public:
  typedef size_t ContainerIdentifier;
 
  explicit SimilarityScheduler(JsonContainerQueue::queue_t *queue,
                               size_t contSize = 0);
 
  ContainerIdentifier getContainerForDoc(std::string &raw);
 
  ContainerIdentifier getContainerForDoc(const RJDocument &doc);
 
  ContainerIdentifier getContainerForDoc(rapidjson::IStreamWrapper &stream);
 
  std::unique_ptr<RJDocument> getNewDoc(ContainerIdentifier id);
 
  void scheduleDocument(ContainerIdentifier id,
                        std::unique_ptr<RJDocument> &&doc,
                        std::unique_ptr<IOrigin> &&origin, size_t size);
  virtual ~SimilarityScheduler() = default;
 
  void finalize();
 
 private:
  typedef std::pair<std::unique_ptr<JSONContainer>,
                    typename similarityRepresentation<SIM>::Representation>
      sCont;
  std::vector<sCont> container;
  JsonContainerQueue::queue_t *queue;
  size_t contSize;
  std::unique_ptr<JSONContainer> createContainer(size_t contSize) const;
};
 
template <typename SIM>
std::unique_ptr<JSONContainer> SimilarityScheduler<SIM>::createContainer(
    size_t contSize) const {
  if (contSize == 0)
    return std::make_unique<JSONContainer>();  //& Create new one
  else
    return std::make_unique<JSONContainer>(contSize);  //& Create new one
}
 
template <typename SIM>
std::unique_ptr<RJDocument> SimilarityScheduler<SIM>::getNewDoc(
    ContainerIdentifier id) {
  return std::make_unique<RJDocument>(container[id].first->getAlloc());
}
 
template <typename SIM>
void SimilarityScheduler<SIM>::scheduleDocument(
    ContainerIdentifier id, std::unique_ptr<RJDocument> &&doc,
    std::unique_ptr<IOrigin> &&origin, size_t size) {
  container[id].first->insertDoc(std::move(doc), std::move(origin));
  if (!(container[id].first->hasSpace(0) || container[id].first->size() == 0)) {
    container[id].first->finalize();
    DCHECK(container[id].first != nullptr);
    queue->send(std::move(container[id].first));  // Enqueue
    DCHECK(container[id].first == nullptr);
    container[id].first = createContainer(contSize);
    DCHECK(container[id].first != nullptr);
  }
}
 
template <typename SIM>
SimilarityScheduler<SIM>::SimilarityScheduler(
    JsonContainerQueue::queue_t *queue, size_t contSize)
    : queue(queue), contSize(contSize) {}
 
template <typename SIM>
typename SimilarityScheduler<SIM>::ContainerIdentifier
SimilarityScheduler<SIM>::getContainerForDoc(std::string &raw) {
  SIM measure;
  similarityRepresentation<SIM> simRep;
  DCHECK(simRep.is_implemented)
      << "Function called with wrong Similarity measure";
  auto docRep = simRep.getRepresentation(raw);
 
  if (!container.empty()) {
    double maxSim = 0;
    int maxSimi = -1;
    for (size_t i = 0; i < container.size(); ++i) {
      double sim = measure.measure(docRep, container[i].second);
      if (sim > maxSim) {
        maxSim = sim;
        maxSimi = i;
      }
    }
    if (maxSim >= config::sim_min_similarity) {
      return maxSimi;
    }
  }
  container.emplace_back(createContainer(contSize), std::move(docRep));
  return container.size() - 1;
}
 
template <typename SIM>
typename SimilarityScheduler<SIM>::ContainerIdentifier
SimilarityScheduler<SIM>::getContainerForDoc(const RJDocument &doc) {
  SIM measure;
  similarityRepresentation<SIM> simRep;
  DCHECK(simRep.is_implemented)
      << "Function called with wrong Similarity measure";
  auto docRep = simRep.getRepresentation(doc);
 
  if (!container.empty()) {
    double maxSim = 0;
    int maxSimi = -1;
    for (size_t i = 0; i < container.size(); ++i) {
      double sim = measure.measure(docRep, container[i].second);
      if (sim > maxSim) {
        maxSim = sim;
        maxSimi = i;
      }
    }
    if (maxSim >= config::sim_min_similarity) {
      return maxSimi;
    }
  }
  container.emplace_back(createContainer(contSize), std::move(docRep));
  return container.size() - 1;
}
 
template <typename SIM>
typename SimilarityScheduler<SIM>::ContainerIdentifier
SimilarityScheduler<SIM>::getContainerForDoc(
    rapidjson::IStreamWrapper &stream) {
  SIM measure;
  similarityRepresentation<SIM> simRep;
  DCHECK(simRep.is_implemented)
      << "Function called with wrong Similarity measure";
  auto docRep = simRep.getRepresentation(stream);
 
  if (!container.empty()) {
    double maxSim = 0;
    int maxSimi = -1;
    for (auto i = 0; i < container.size(); ++i) {
      double sim = measure.measure(docRep, container[i].second);
      if (sim > maxSim) {
        maxSim = sim;
        maxSimi = i;
      }
    }
    if (maxSim >= config::sim_min_similarity) {
      return maxSimi;
    }
  }
  container.emplace_back(createContainer(contSize), std::move(docRep));
  return container.size() - 1;
}
 
template <typename SIM>
void SimilarityScheduler<SIM>::finalize() {
  // TODO Merge small container
 
  for (auto &&currentSimContainer : container) {
    auto &currentContainer = currentSimContainer.first;
    if (currentContainer->size() > 0) {
      DCHECK(currentContainer != nullptr);
      currentContainer->finalize();
      if (!config::storeJson) currentContainer->removeDocuments();
      queue->send(std::move(currentContainer));  // Enqueue
      DCHECK(currentContainer == nullptr);
    }
  }
  queue->producerFinished();
}
 
#endif  // JODA_SIMILARITYSCHEDULER_H