cdo-2.0.0/src/processManager.cc

/*
  This file is part of CDO. CDO is a collection of Operators to manipulate and analyse Climate model Data.

  Author: Oliver Heidmann

*/

#include "processManager.h"
#include "process.h"
#include "cdo_output.h"
#include "cdo_options.h"
#include "mpmo_color.h"
#include "fileStream.h"

#include <stack>
#include <mutex>

static std::mutex processMutex;

static std::string parse_err_msg = "";

void
ProcessManager::run_processes()
{
  for (auto &idProcessPair : m_processes)
    {
      if (idProcessPair.first)
        {
          /*TEMP*/
          if (!Options::silentMode)
            {
              MpMO::Print(Green("%s: ") + "Process started", idProcessPair.second->prompt);
            }
          m_threadIDs.push_back(idProcessPair.second->run());
        }
    }
  m_threadIDs.push_back(pthread_self());
  // MpMO::PrintCerr(Green("%s: ") + "xProcess started", get_process_from_id(0).inq_promt());
  get_process_from_id(0)->m_module.func(get_process_from_id(0).get());
}

void
ProcessManager::kill_processes()
{
  for (auto threadID : m_threadIDs)
    {
      if (threadID != pthread_self())
        {
          pthread_cancel(threadID);
          Debug(PROCESS_MANAGER, "process killed: %d", threadID);
        }
    }
}

int
ProcessManager::get_num_variable_input_operators()
{
  int num = 0;
  for (auto &process : m_processes)
    {
      if (process.second->input_is_variable()) num++;
    }
  return num;
}

void
ProcessManager::validate_processes()
{
  for (auto &process : m_processes)
    {
      process.second->check_stream_cnt();
      auto stat = process.second->m_status;
      if (stat != ProcessStatus::Ok)
        {
          if (stat == ProcessStatus::TooFewStreams || stat == ProcessStatus::TooManyStreams)
            if (get_num_variable_input_operators() > 1)
              {
                cdo_warning("Did you forget to use '[' and/or ']' for multiple variable input operators?");
                cdo_warning("use option --variableInput, for description");
              }
          process.second->handle_process_err();
        }
    }
}

void
ProcessManager::clear_processes()
{
  Debug(PROCESS_MANAGER, "Deleting Processes");
  m_processes.clear();
  m_numProcesses = 0;
  m_numProcessesActive = 0;
}

const std::shared_ptr<Process>
ProcessManager::create_process_from_command(const std::string &command)
{
  Debug(PROCESS_MANAGER, "Creating new process for command: %s with ID: %d ", command, m_numProcesses);
  const auto processID = m_numProcesses++;

  const auto operatorName = extract_operator_name(command);
  const auto arguments = get_operator_argv(command);
  auto success = m_processes.insert(std::make_pair(processID, std::make_shared<Process>(processID, operatorName, arguments)));
  if (!success.second) cdo_abort("Process %d could not be created", processID);

  m_numProcessesActive++;
  Debug(PROCESS_MANAGER, "m_numProcessesActive: %d", m_numProcessesActive);

  if (processID >= MAX_PROCESS) cdo_abort("Limit of %d processes reached!", MAX_PROCESS);

  return success.first->second;
}

int
ProcessManager::get_num_processes(void)
{
  std::unique_lock<std::mutex> locked_mutex(processMutex);
  int pnums = m_processes.size();
  return pnums;
}

int
ProcessManager::get_num_active_processes(void)
{
  std::unique_lock<std::mutex> locked_mutex(processMutex);
  int pnums = m_numProcessesActive;
  return pnums;
}

/**
 * Handles process Creation for cdo command p_argvEntry
 * Adds \p p_parentProcess as parent to the newly created process.
 * Adds newly created process to \p p_parentProcess.
 * Also checks if \p p_parentProcess accepts another processes streams
 * as input and exits with error message if not.
 */
const std::shared_ptr<Process>
ProcessManager::create_child_process_for(const std::shared_ptr<Process> &p_parentProcess, const std::string &argvEntry)
{
  if (p_parentProcess->m_module.restrictions == FilesOnly)
    cdo_abort("operator -%s can not be used with pipes.", p_parentProcess->operatorName);

  const std::shared_ptr<Process> &newProcess = create_process_from_command(argvEntry);

  if (newProcess->m_module.streamOutCnt == 0)
    cdo_abort("operator -%s can not take -%s  with 0 outputs as input", p_parentProcess->operatorName, newProcess->operatorName);

  Debug(PROCESS_MANAGER, "Adding new child %s (id: %d) to %s (idL %d)", newProcess->operatorName, newProcess->get_id(), p_parentProcess->operatorName, p_parentProcess->get_id());

  p_parentProcess->add_child(newProcess);
  newProcess->add_parent(p_parentProcess);

  return newProcess;
}

std::set<std::string>
ProcessManager::handle_first_operator(int p_argcStart, int argc,const std::vector<std::string> &argv, const std::shared_ptr<Process> &p_rootProcess)
{
  std::set<std::string> files;
  for (int i = p_argcStart; i < argc; i++)
    {
      Debug(PROCESS_MANAGER, "Creating new pstream for output file: %s", argv[i]);
      if (argv[i] == "]")
        {
          cdo_abort("missing output file");
        }
      if (argv[i][0] == '-')
        {
          cdo_abort("missing Output file, did you add a '-' to a file?");
        }
      p_rootProcess->add_file_out_stream(argv[i]);
      files.insert(argv[i]);
    }
  return files;
}

void
ProcessManager::check_single_bracket_only(const char *p_argvEntry, char p_bracketType)
{
  if (strlen(p_argvEntry) != 1)
    {
      cdo_abort("Only single %c allowed", p_bracketType);
    }
}

void
ProcessManager::create_processes(int argc,const std::vector<std::string> &argv)
{
  ParseStatus parseStatus = create_processes_from_input(argc, argv);
  if (parseStatus != ParseStatus::Ok)
    {
      // Program Exits here
      handle_parse_error(parseStatus);
    }
  validate_processes();
}

#include <algorithm>
static std::string
get_stack_as_string(std::stack<std::shared_ptr<Process>> p_stack)
{
  std::string stackString = "";
  std::vector<std::string> strVec;
  while (!p_stack.empty())
    {
      strVec.push_back(p_stack.top()->operatorName);
      p_stack.pop();
    }
  std::reverse(strVec.begin(), strVec.end());
  for (auto strEle : strVec)
    {
      stackString += " " + strEle;
    }
  return stackString;
}

/* comment FOR DEVELOPERS ONLY (do not include in docu)
 *
 * This is the so to speak parser for cdo console inputs.
 *
 *  This parser runs over every argv that comes after the cdo options.  The
 *  fist thing that is done is processing the first operator, since this
 *  operator is the only one that has output files we can remove the output
 *  file from out argv by limiting the argc. Obase operators are treated as if
 *  they have a single output since the operator itself takes care of writing
 *  and creating the files it needs for its output. We also create the first
 *  process for the operator and push it on out stack.  Our stack will contain
 *  all operators that do not have all in- or output they need.  After the
 *  first operator is handled the parser will go over each other element in
 *  argv.  Here 4 cases can happen. Only one of the 4 will happen in one
 *  iteration.
 *
 *  If an operator is found we create a new process and add this process as
 *  child to the process on top of the stack. Likewise we add the top process
 *  as parent to the new process. Then the new Process is added to the stack.
 *
 *  Does the argv element represent a file (indicated by the missing '-' in
 *  argv[i][0]) we create a file stream and add it to the process at the top of
 *  the stack.
 *
 *  In case of a '[' or ']' we check if there is only one bracket since we
 *  decided to not allow multiple brackets in the same argv entry.Then we add
 *  ('[') or remove (']') the top of the process stack to a set (named
 *  bracketOperators) which will keep track of which operators used a bracket.
 *  This stack allows to 'mark' an operator so that it is only removed in case
 *  of a ']'.  The ']' indicates that the top process should be removed.and
 *  that it SHOULD have the correct number of inputs.
 *
 *  At the end of each iteration we remove all operators that have all their
 *  inputs AND are not contained in out bracket operators set. So a not closed
 *  bracket will cause a wanted miss function of the parser as the operator
 *  will not be removed and more inputs will be added. This will be found later
 *  by our routine (Process::validate) that checks if every process has the
 *  correct number of inputs and outputs and will throw an error.
 */
std::vector<std::string>
ProcessManager::get_operator_argv(std::string operatorArguments)
{
  std::vector<std::string> argument_vector;
  Debug(PROCESS && strchr(operatorArguments.c_str(), ',') != nullptr, "Setting operator arguments: %s", operatorArguments);

  constexpr char delimiter = ',';

  auto pos = operatorArguments.find(delimiter);
  if (pos != std::string::npos)
    {
      // remove operator name
      operatorArguments.erase(0, pos + 1);

      while ((pos = operatorArguments.find(delimiter)) != std::string::npos)
        {
          argument_vector.push_back(operatorArguments.substr(0, pos));
          Debug("added argument %s", argument_vector.back());
          operatorArguments.erase(0, pos + 1);
        }
      argument_vector.push_back(operatorArguments);
    }
  return argument_vector;
}


ParseStatus
ProcessManager::create_processes_from_input(int argc, const std::vector<std::string> &argv)
{
  Debug(PROCESS_MANAGER, "== Process Creation Start ==");
  Debug(PROCESS_MANAGER, "operators: %s", argv_to_string(argv));

  const std::shared_ptr<Process> &root_process = create_process_from_command(argv[0]);
  int cntOutFiles = (int) root_process->m_module.streamOutCnt;

  std::set<std::string> files;

  unsigned int lastNonOutputIdx = argc - cntOutFiles;
  if (cntOutFiles == -1)
    {
      root_process->set_obase(argv[argc - 1]);
      cntOutFiles = 1;
      lastNonOutputIdx = argc - 1;
    }
  else
    {
      if (lastNonOutputIdx <= 0) return ParseStatus::MissingOutFile;
      files = handle_first_operator(lastNonOutputIdx, argc, argv, root_process);
    }

  std::shared_ptr<Process> currentProcess;
  std::stack<std::shared_ptr<Process>> processStack;
  std::vector<std::shared_ptr<Process>> bracketOperators;
  const char *argvEntry;
  int unclosedBrackets = 0;
  unsigned int idx = 1;

  processStack.push(root_process);
  while (!processStack.empty() && idx < lastNonOutputIdx)
    {
      Debug(PROCESS_MANAGER, "%s", get_stack_as_string(processStack));
      currentProcess = processStack.top();
      Debug(PROCESS_MANAGER, "iteration %d, current argv: %s, currentProcess: %s", idx, argv[idx], currentProcess->m_operatorCommand);

      argvEntry = argv[idx].c_str();
      //------------------------------------------------------
      // case: operator
      if (argvEntry[0] == '-')
        {
          Debug(PROCESS_MANAGER, "Found new Operator: %s", argvEntry);
          currentProcess = create_child_process_for(currentProcess, argvEntry);
          if (currentProcess->m_module.restrictions == OnlyFirst)
            {
              parse_err_msg += "Operator " + std::string(argvEntry) + " can only be used if it is first in the operator chain";
              return ParseStatus::OperatorNotFirst;
            }
          processStack.push(currentProcess);
        }
      // - - - - - - - - - - - - - - - - - - - - - - - - - - -
      // case: bracket start
      else if (argvEntry[0] == '[')
        {
          check_single_bracket_only(argvEntry, '[');
          bracketOperators.push_back(currentProcess);
          unclosedBrackets++;
        }
      // - - - - - - - - - - - - - - - - - - - - - - - - - - -
      // case: bracket end
      else if (argvEntry[0] == ']')
        {
          check_single_bracket_only(argvEntry, ']');
          unclosedBrackets--;
          bracketOperators.pop_back();
          // this check is for double bracktets like -info [ [ file1 file2 file3 ] ]  which can happen when cdo wildcards are used.
          auto it = std::find(bracketOperators.begin(), bracketOperators.end(), processStack.top());
          if (it == bracketOperators.end()) processStack.pop();
        }
      // - - - - - - - - - - - - - - - - - - - - - - - - - - -
      // case: file
      else if (currentProcess->m_module.streamInCnt != 0)
        {
          Debug(PROCESS_MANAGER, "adding in file to %s", currentProcess->operatorName);
          if (files.find(argvEntry) == files.end())
            {
              currentProcess->add_file_in_stream(argvEntry);
            }
          else
            {
              parse_err_msg
                  += std::string(argvEntry) + " is used as an output, files can not be used as in- and output at the same time";
              return ParseStatus::FileIsInAndOutput;
            }
        }
      // -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
      // remove finished
      while (!processStack.empty() && processStack.top()->has_hall_inputs()
             && std::find(bracketOperators.begin(), bracketOperators.end(), processStack.top()) == bracketOperators.end())
        {
          Debug(PROCESS_MANAGER, "Removing %s from stack", processStack.top()->operatorName);
          processStack.pop();
        }
      //------------------------------------------------------
      idx++;
    }
  //---------------------------------------------------------------
  if (unclosedBrackets > 0) return ParseStatus::ClosingBracketMissing;
  if (unclosedBrackets < 0) return ParseStatus::OpenBracketMissing;
  if (idx < lastNonOutputIdx)
    {
      if (argv[idx][0] == ']') return ParseStatus::OpenBracketMissing;
      return ParseStatus::UnprocessedInput;
    }

  Debug(PROCESS_MANAGER, "== Process Creation End ==");

  set_process_num(m_processes.size());
  FileStream::enableTimers(m_processes.size() == 1 && Threading::ompNumThreads == 1);

  return ParseStatus::Ok;
}

void
ProcessManager::handle_parse_error(ParseStatus p_errCode)
{
  switch (p_errCode)
    {
    case ParseStatus::ClosingBracketMissing:
      {
        cdo_abort("Missing ']'.");
        break;
      }
    case ParseStatus::OpenBracketMissing:
      {
        cdo_abort("Missing '['.");
        break;
      }
    case ParseStatus::UnprocessedInput:
      {
        cdo_abort("Unprocessed Input, could not process all Operators/Files");
        break;
      }
    case ParseStatus::MissingOutFile:
      {
        cdo_abort("Missing out file for first operator");
        break;
      }
    case ParseStatus::OperatorNotFirst:
      {
        cdo_abort("%s", parse_err_msg);
        break;
      }
    case ParseStatus::MissingObase:
      {
        break;
      }
    case ParseStatus::FileIsInAndOutput:
      {
        cdo_abort("%s", parse_err_msg);
      }
    case ParseStatus::Ok:
      {
        return;
      }
    }
}

const std::shared_ptr<Process> &
ProcessManager::get_process_from_id(int p_processID)
{
  std::unique_lock<std::mutex> locked_mutex(processMutex);

  const auto process = m_processes.find(p_processID);
  if (process == m_processes.end()) cdo_abort("Process with ID: %d not found", p_processID);

  return process->second;
}