<commit_before>#ifndef __TASK_DISTRIBUTION__TASK_MANAGER_IMPL_HPP__
#define __TASK_DISTRIBUTION__TASK_MANAGER_IMPL_HPP__

#include "clean_tuple.hpp"
#include "is_tuple_convertible.hpp"
#include "repeated_tuple.hpp"

#include "task_manager.hpp"

#include "computing_unit.hpp"
#include "task.hpp"

#include <boost/serialization/set.hpp>
#include <functional>
#include <type_traits>

namespace TaskDistribution {
  template <class Unit, class... Args>
  Task<typename CompileUtils::function_traits<Unit>::return_type>
  TaskManager::new_task(Unit const& computing_unit, Args const&... args) {
    if (id() != 0)
      return new_invalid_task(computing_unit);

    typedef typename CompileUtils::clean_tuple<Args...>::type
      given_args_tuple_type;
    typedef typename CompileUtils::repeated_tuple<sizeof...(Args), Key>::type
      args_tasks_tuple_type;
    typedef typename CompileUtils::clean_tuple_from_tuple<
      typename CompileUtils::function_traits<Unit>::arg_tuple_type>::type
      unit_args_tuple_type;

    // Checks if the arguments are valid
    static_assert(
        std::tuple_size<given_args_tuple_type>::value ==
        std::tuple_size<unit_args_tuple_type>::value,
        "Invalid number of arguments."
    );

    static_assert(
        CompileUtils::is_tuple_convertible<
          given_args_tuple_type,
          unit_args_tuple_type
        >::value,
        "Can't convert from arguments provided to expected."
    );

    // Make tuples of normal arguments and task arguments
    unit_args_tuple_type args_tuple(
        make_args_tuple<unit_args_tuple_type>(args...));
    args_tasks_tuple_type args_tasks_tuple(
        make_args_tasks_tuple<args_tasks_tuple_type>(args...));

    // Gets keys
    Key computing_unit_key = get_key(computing_unit,
        Key::ComputingUnit);
    Key computing_unit_id_key = get_key(computing_unit.get_id(),
        Key::ComputingUnitId);
    Key arguments_key = get_key(args_tuple,
        Key::Arguments);
    Key arguments_tasks_key = get_key(args_tasks_tuple,
        Key::ArgumentsTasks);

    // Builds entry
    TaskEntry task_entry;
    task_entry.computing_unit_key = computing_unit_key;
    task_entry.computing_unit_id_key = computing_unit_id_key;
    task_entry.arguments_key = arguments_key;
    task_entry.arguments_tasks_key = arguments_tasks_key;
    task_entry.run_locally = computing_unit.run_locally();

    // Stores task entry and update its internal data
    Key task_key = get_key(task_entry, Key::Task);
    archive_.load(task_key, task_entry);
    task_entry.task_key = task_key;

    // Creates children and parents if they don't exist
    create_family_lists(task_entry);

    // Do dependency analysis
    KeyList dependencies({get_task_key(args)...});

    // Add dependencies
    KeySet parents;
    archive_.load(task_entry.parents_key, parents);

    for (auto& parent_key: dependencies)
      if (parent_key.is_valid())
        add_dependency(task_entry, parent_key, parents);

    archive_.insert(task_entry.parents_key, parents);

    // Check if task can be run now
    if (task_entry.active_parents == 0 && !task_entry.result_key.is_valid()) {
      bool found = false;
      for (auto& key : ready_) {
        if (key == task_key) {
          found = true;
          break;
        }
      }

      // If task doesn't exist already, tell everyone we have a new task
      if (!found)
        ready_.push_back(task_key);
    }

    archive_.insert(task_key, task_entry);

    task_creation_handler_(computing_unit.get_id(), task_key);

    return Task<
      typename CompileUtils::function_traits<Unit>::return_type>(task_key,
          this);
  }

  template <class T>
  Task<T> TaskManager::new_identity_task(T const& arg) {
    Key result_key = get_key(arg, Key::Result);
    TaskEntry task_entry;
    task_entry.result_key = result_key;
    task_entry.run_locally = false;

    Key task_key = get_key(task_entry, Key::Task);
    archive_.load(task_key, task_entry);
    task_entry.task_key = task_key;
    archive_.insert(task_key, task_entry);

    return Task<T>(task_key, this);
  }

  template <class T>
  Task<T> TaskManager::new_identity_task(Task<T> const& arg) {
    return arg;
  }

  template <class Unit>
  Task<typename CompileUtils::function_traits<Unit>::return_type><commit_msg>Fixed bug that caused some identity tasks to get new keys.<commit_after>#ifndef __TASK_DISTRIBUTION__TASK_MANAGER_IMPL_HPP__
#define __TASK_DISTRIBUTION__TASK_MANAGER_IMPL_HPP__

#include "clean_tuple.hpp"
#include "is_tuple_convertible.hpp"
#include "repeated_tuple.hpp"

#include "task_manager.hpp"

#include "computing_unit.hpp"
#include "task.hpp"

#include <boost/serialization/set.hpp>
#include <functional>
#include <type_traits>

namespace TaskDistribution {
  template <class Unit, class... Args>
  Task<typename CompileUtils::function_traits<Unit>::return_type>
  TaskManager::new_task(Unit const& computing_unit, Args const&... args) {
    if (id() != 0)
      return new_invalid_task(computing_unit);

    typedef typename CompileUtils::clean_tuple<Args...>::type
      given_args_tuple_type;
    typedef typename CompileUtils::repeated_tuple<sizeof...(Args), Key>::type
      args_tasks_tuple_type;
    typedef typename CompileUtils::clean_tuple_from_tuple<
      typename CompileUtils::function_traits<Unit>::arg_tuple_type>::type
      unit_args_tuple_type;

    // Checks if the arguments are valid
    static_assert(
        std::tuple_size<given_args_tuple_type>::value ==
        std::tuple_size<unit_args_tuple_type>::value,
        "Invalid number of arguments."
    );

    static_assert(
        CompileUtils::is_tuple_convertible<
          given_args_tuple_type,
          unit_args_tuple_type
        >::value,
        "Can't convert from arguments provided to expected."
    );

    // Make tuples of normal arguments and task arguments
    unit_args_tuple_type args_tuple(
        make_args_tuple<unit_args_tuple_type>(args...));
    args_tasks_tuple_type args_tasks_tuple(
        make_args_tasks_tuple<args_tasks_tuple_type>(args...));

    // Gets keys
    Key computing_unit_key = get_key(computing_unit,
        Key::ComputingUnit);
    Key computing_unit_id_key = get_key(computing_unit.get_id(),
        Key::ComputingUnitId);
    Key arguments_key = get_key(args_tuple,
        Key::Arguments);
    Key arguments_tasks_key = get_key(args_tasks_tuple,
        Key::ArgumentsTasks);

    // Builds entry
    TaskEntry task_entry;
    task_entry.computing_unit_key = computing_unit_key;
    task_entry.computing_unit_id_key = computing_unit_id_key;
    task_entry.arguments_key = arguments_key;
    task_entry.arguments_tasks_key = arguments_tasks_key;
    task_entry.run_locally = computing_unit.run_locally();

    // Stores task entry and update its internal data
    Key task_key = get_key(task_entry, Key::Task);
    archive_.load(task_key, task_entry);
    task_entry.task_key = task_key;

    // Creates children and parents if they don't exist
    create_family_lists(task_entry);

    // Do dependency analysis
    KeyList dependencies({get_task_key(args)...});

    // Add dependencies
    KeySet parents;
    archive_.load(task_entry.parents_key, parents);

    for (auto& parent_key: dependencies)
      if (parent_key.is_valid())
        add_dependency(task_entry, parent_key, parents);

    archive_.insert(task_entry.parents_key, parents);

    // Check if task can be run now
    if (task_entry.active_parents == 0 && !task_entry.result_key.is_valid()) {
      bool found = false;
      for (auto& key : ready_) {
        if (key == task_key) {
          found = true;
          break;
        }
      }

      // If task doesn't exist already, tell everyone we have a new task
      if (!found)
        ready_.push_back(task_key);
    }

    archive_.insert(task_key, task_entry);

    task_creation_handler_(computing_unit.get_id(), task_key);

    return Task<
      typename CompileUtils::function_traits<Unit>::return_type>(task_key,
          this);
  }

  template <class T>
  Task<T> TaskManager::new_identity_task(T const& arg) {
    if (id() != 0)
      return Task<T>(Key(), this);

    Key result_key = get_key(arg, Key::Result);
    TaskEntry task_entry;
    task_entry.result_key = result_key;
    task_entry.run_locally = false;

    Key task_key = get_key(task_entry, Key::Task);
    archive_.load(task_key, task_entry);
    task_entry.task_key = task_key;
    archive_.insert(task_key, task_entry);

    return Task<T>(task_key, this);
  }

  template <class T>
  Task<T> TaskManager::new_identity_task(Task<T> const& arg) {
    return arg;
  }

  template <class Unit>
  Task<typename CompileUtils::function_traits<Unit>::return_type>
{'fraction_non_alphanumeric': 0.10147092622385659, 'fraction_numerical': 0.0005745805561939783, 'mean_word_length': 3.729891304347826, 'pattern_counts': {'":': 0, '<': 56, '<?xml version=': 0, '>': 56, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4527532', 'n_tokens_mistral': 2828, 'n_tokens_neox': 2733, 'n_words': 549}