<commit_before>/**
 * This file is part of the "libfnord" project
 *   Copyright (c) 2015 <NAME>
 *
 * FnordMetric is free software: you can redistribute it and/or modify it under
 * the terms of the GNU General Public License v3.0. You should have received a
 * copy of the GNU General Public License along with this program. If not, see
 * <http://www.gnu.org/licenses/>.
 */
#include <unistd.h>
#include <fnord-base/io/file.h>
#include <fnord-base/io/fileutil.h>
#include <fnord-base/io/mmappedfile.h>
#include <fnord-base/logging.h>
#include <fnord-base/io/fileutil.h>
#include <fnord-base/util/binarymessagereader.h>
#include <fnord-base/util/binarymessagewriter.h>
#include <fnord-dproc/LocalScheduler.h>

namespace fnord {
namespace dproc {

LocalScheduler::LocalScheduler(
    const String& tempdir /* = "/tmp" */,
    size_t max_threads /* = 8 */,
    size_t max_requests /* = 32 */) :
    tempdir_(tempdir),
    tpool_(max_threads),
    req_tpool_(max_requests) {}

void LocalScheduler::start() {
  req_tpool_.start();
  tpool_.start();
}

void LocalScheduler::stop() {
  req_tpool_.stop();
  tpool_.stop();
}

RefPtr<TaskResultFuture> LocalScheduler::run(
    RefPtr<Application> app,
    const TaskSpec& task) {
  RefPtr<TaskResultFuture> result(new TaskResultFuture());

  try {
    auto instance = mkRef(
        new LocalTaskRef(
            app,
            task.task_name(),
            Buffer(task.params().data(), task.params().size())));

    req_tpool_.run([this, app, result, instance] () {
      try {
        LocalTaskPipeline pipeline;
        pipeline.tasks.push_back(instance);
        runPipeline(app.get(), &pipeline, result);

        if (instance->failed) {
          RAISE(kRuntimeError, "task failed");
        }

        result->returnResult(instance->task);
      } catch (const StandardException& e) {
        fnord::logError("dproc.scheduler", e, "task failed");
        result->returnError(e);
      }
    });
  } catch (const StandardException& e) {
    fnord::logError("dproc.scheduler", e, "task failed");
    result->returnError(e);
  }

  return result;
}

void LocalScheduler::runPipeline(
    Application* app,
    LocalTaskPipeline* pipeline,
    RefPtr<TaskResultFuture> result) {
  fnord::logInfo(
      "fnord.dproc",
      "Starting local pipeline id=$0 tasks=$1",
      (void*) pipeline,
      pipeline->tasks.size());

  std::unique_lock<std::mutex> lk(pipeline->mutex);
  result->updateStatus([&pipeline] (TaskStatus* status) {
    status->num_subtasks_total = pipeline->tasks.size();
  });

  while (pipeline->tasks.size() > 0) {
    bool waiting = true;
    size_t num_waiting = 0;
    size_t num_running = 0;
    size_t num_completed = 0;

    for (auto& taskref : pipeline->tasks) {

      if (taskref->finished) {
        ++num_completed;
        continue;
      }

      if (taskref->running) {
        ++num_running;
        continue;
      }

      if (!taskref->expanded) {
        taskref->expanded = true;

        auto cache_key = taskref->task->cacheKey();
        if (!cache_key.isEmpty()) {
          taskref->cache_filename = FileUtil::joinPaths(
              tempdir_,
              StringUtil::format("cache_$0", cache_key.get()));
        }

        auto cached =
            !cache_key.isEmpty() &&
            FileUtil::exists(taskref->cache_filename);

        if (cached) {
          auto cache_file = File::openFile(
              taskref->cache_filename,
              File::O_READ);

          Buffer cache_hdr(sizeof(uint64_t));
          cache_file.read(&cache_hdr);
          util::BinaryMessageReader reader(cache_hdr.data(), cache_hdr.size());
          auto cache_version = *reader.readUInt64();
          auto cache_size = cache_file.size() - cache_hdr.size();

          auto ckey = taskref->task->cacheKeySHA1();
          fnord::logDebug(
              "fnord.dproc",
              "Reading RDD from cache: $0, key=$1, version=$2",
              taskref->debug_name,
              ckey.isEmpty() ? "<nil>" : ckey.get(),
              cache_version);

          taskref->task->decode(
                new io::MmappedFile(
                    std::move(cache_file),
                    cache_hdr.size(),
                    cache_size));

          result->updateStatus([&pipeline] (TaskStatus* status) {
            ++status->num_subtasks_completed;
          });

          taskref->finished = true;
          waiting = false;
          break;
        }

        auto parent_task = taskref;
        size_t numdeps = 0;
        for (const auto& dep : taskref->task->dependencies()) {
          RefPtr<LocalTaskRef> depref(new LocalTaskRef(app, dep.task_name, dep.params));
          parent_task->dependencies.emplace_back(depref);
          pipeline->tasks.emplace_back(depref);
          ++numdeps;
        }

        if (numdeps > 0) {<commit_msg>use SHA1 hash key<commit_after>/**
 * This file is part of the "libfnord" project
 *   Copyright (c) 2015 <NAME>
 *
 * FnordMetric is free software: you can redistribute it and/or modify it under
 * the terms of the GNU General Public License v3.0. You should have received a
 * copy of the GNU General Public License along with this program. If not, see
 * <http://www.gnu.org/licenses/>.
 */
#include <unistd.h>
#include <fnord-base/io/file.h>
#include <fnord-base/io/fileutil.h>
#include <fnord-base/io/mmappedfile.h>
#include <fnord-base/logging.h>
#include <fnord-base/io/fileutil.h>
#include <fnord-base/util/binarymessagereader.h>
#include <fnord-base/util/binarymessagewriter.h>
#include <fnord-dproc/LocalScheduler.h>

namespace fnord {
namespace dproc {

LocalScheduler::LocalScheduler(
    const String& tempdir /* = "/tmp" */,
    size_t max_threads /* = 8 */,
    size_t max_requests /* = 32 */) :
    tempdir_(tempdir),
    tpool_(max_threads),
    req_tpool_(max_requests) {}

void LocalScheduler::start() {
  req_tpool_.start();
  tpool_.start();
}

void LocalScheduler::stop() {
  req_tpool_.stop();
  tpool_.stop();
}

RefPtr<TaskResultFuture> LocalScheduler::run(
    RefPtr<Application> app,
    const TaskSpec& task) {
  RefPtr<TaskResultFuture> result(new TaskResultFuture());

  try {
    auto instance = mkRef(
        new LocalTaskRef(
            app,
            task.task_name(),
            Buffer(task.params().data(), task.params().size())));

    req_tpool_.run([this, app, result, instance] () {
      try {
        LocalTaskPipeline pipeline;
        pipeline.tasks.push_back(instance);
        runPipeline(app.get(), &pipeline, result);

        if (instance->failed) {
          RAISE(kRuntimeError, "task failed");
        }

        result->returnResult(instance->task);
      } catch (const StandardException& e) {
        fnord::logError("dproc.scheduler", e, "task failed");
        result->returnError(e);
      }
    });
  } catch (const StandardException& e) {
    fnord::logError("dproc.scheduler", e, "task failed");
    result->returnError(e);
  }

  return result;
}

void LocalScheduler::runPipeline(
    Application* app,
    LocalTaskPipeline* pipeline,
    RefPtr<TaskResultFuture> result) {
  fnord::logInfo(
      "fnord.dproc",
      "Starting local pipeline id=$0 tasks=$1",
      (void*) pipeline,
      pipeline->tasks.size());

  std::unique_lock<std::mutex> lk(pipeline->mutex);
  result->updateStatus([&pipeline] (TaskStatus* status) {
    status->num_subtasks_total = pipeline->tasks.size();
  });

  while (pipeline->tasks.size() > 0) {
    bool waiting = true;
    size_t num_waiting = 0;
    size_t num_running = 0;
    size_t num_completed = 0;

    for (auto& taskref : pipeline->tasks) {

      if (taskref->finished) {
        ++num_completed;
        continue;
      }

      if (taskref->running) {
        ++num_running;
        continue;
      }

      if (!taskref->expanded) {
        taskref->expanded = true;

        auto cache_key = taskref->task->cacheKeySHA1();
        if (!cache_key.isEmpty()) {
          taskref->cache_filename = FileUtil::joinPaths(
              tempdir_,
              StringUtil::format("cache_$0", cache_key.get()));
        }

        auto cached =
            !cache_key.isEmpty() &&
            FileUtil::exists(taskref->cache_filename);

        if (cached) {
          auto cache_file = File::openFile(
              taskref->cache_filename,
              File::O_READ);

          Buffer cache_hdr(sizeof(uint64_t));
          cache_file.read(&cache_hdr);
          util::BinaryMessageReader reader(cache_hdr.data(), cache_hdr.size());
          auto cache_version = *reader.readUInt64();
          auto cache_size = cache_file.size() - cache_hdr.size();

          fnord::logDebug(
              "fnord.dproc",
              "Reading RDD from cache: $0, key=$1, version=$2",
              taskref->debug_name,
              cache_key.isEmpty() ? "<nil>" : cache_key.get(),
              cache_version);

          taskref->task->decode(
                new io::MmappedFile(
                    std::move(cache_file),
                    cache_hdr.size(),
                    cache_size));

          result->updateStatus([&pipeline] (TaskStatus* status) {
            ++status->num_subtasks_completed;
          });

          taskref->finished = true;
          waiting = false;
          break;
        }

        auto parent_task = taskref;
        size_t numdeps = 0;
        for (const auto& dep : taskref->task->dependencies()) {
          RefPtr<LocalTaskRef> depref(new LocalTaskRef(app, dep.task_name, dep.params));
          parent_task->dependencies.emplace_back(depref);
          pipeline->tasks.emplace_back(depref);
          ++numdeps;
        }

        if (numdeps > 0) {
{'fraction_non_alphanumeric': 0.1365003630328804, 'fraction_numerical': 0.005497355046157038, 'mean_word_length': 2.4509663564781676, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 107, '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': '1658187', 'n_tokens_mistral': 3029, 'n_tokens_neox': 2942, 'n_words': 598}