Algorithm.cpp

        }
      }
    }
  }
}

/** Indicates that this algrithms history should be tracked regardless of if it
 * is a child.
 *  @param parentHist :: the parent algorithm history object the history in.
 */
void Algorithm::trackAlgorithmHistory(
    boost::shared_ptr<AlgorithmHistory> parentHist) {
  enableHistoryRecordingForChild(true);
  m_parentHistory = parentHist;
}

/** Check if we are tracking history for thus algorithm
 *  @return if we are tracking the history of this algorithm
 */
bool Algorithm::trackingHistory() {
  return (!isChild() || m_recordHistoryForChild);
}

/** Populate lists of the input & output workspace properties.
 *  (InOut workspaces go in both lists)
 *  @param inputWorkspaces ::  A reference to a vector for the input workspaces
 *  @param outputWorkspaces :: A reference to a vector for the output workspaces
 */
void Algorithm::findWorkspaceProperties(
    std::vector<Workspace_sptr> &inputWorkspaces,
    std::vector<Workspace_sptr> &outputWorkspaces) const {
  // Loop over properties looking for the workspace properties and putting them
  // in the right list
  const std::vector<Property *> &algProperties = getProperties();
  std::vector<Property *>::const_iterator it;
  for (it = algProperties.begin(); it != algProperties.end(); ++it) {
    const IWorkspaceProperty *wsProp = dynamic_cast<IWorkspaceProperty *>(*it);
    if (wsProp) {
      const Property *wsPropProp = dynamic_cast<Property *>(*it);
      // Check we actually have a workspace, it may have been optional
      Workspace_sptr workspace = wsProp->getWorkspace();
      if (!workspace)
        continue;
      unsigned int direction = wsPropProp->direction();
      if (direction == Direction::Input || direction == Direction::InOut) {
        inputWorkspaces.push_back(workspace);
      }
      if (direction == Direction::Output || direction == Direction::InOut) {
        outputWorkspaces.push_back(workspace);
      }
    }
  }
}

/** Sends out algorithm parameter information to the logger */
void Algorithm::logAlgorithmInfo() const {
  auto &logger = getLogger();

  if (m_isAlgStartupLoggingEnabled) {
    logger.notice() << name() << " started";
    if (this->isChild())
      logger.notice() << " (child)";
    logger.notice() << '\n';
    // Make use of the AlgorithmHistory class, which holds all the info we want
    // here
    AlgorithmHistory algHistory(this);
    size_t maxPropertyLength = 40;
    if (logger.is(Logger::Priority::PRIO_DEBUG)) {
      // include the full property value when logging in debug
      maxPropertyLength = 0;
    }
    algHistory.printSelf(logger.information(), 0, maxPropertyLength);
  }
}

//=============================================================================================
//================================== WorkspaceGroup-related
//===================================
//=============================================================================================

/** Check the input workspace properties for groups.
 *
 * If there are more than one input workspace properties, then:
 *  - All inputs should be groups of the same size
 *    - In this case, algorithms are processed in order
 *  - OR, only one input should be a group, the others being size of 1
 *
 * If the property itself is a WorkspaceProperty<WorkspaceGroup> then
 * this returns false
 *
 * Returns true if processGroups() should be called.
 * It also sets up some other members.
 *
 * Override if it is needed to customize the group checking.
 *
 * @throw std::invalid_argument if the groups sizes are incompatible.
 * @throw std::invalid_argument if a member is not found
 *
 * This method (or an override) must NOT THROW any exception if there are no
 *input workspace groups
 */
bool Algorithm::checkGroups() {
  size_t numGroups = 0;
  bool processGroups = false;

  // Unroll the groups or single inputs into vectors of workspace
  m_groups.clear();
  m_groupWorkspaces.clear();
  for (auto inputWorkspaceProp : m_inputWorkspaceProps) {
    auto prop = dynamic_cast<Property *>(inputWorkspaceProp);
    auto wsGroupProp = dynamic_cast<WorkspaceProperty<WorkspaceGroup> *>(prop);
    std::vector<Workspace_sptr> thisGroup;

    Workspace_sptr ws = inputWorkspaceProp->getWorkspace();
    WorkspaceGroup_sptr wsGroup =
        boost::dynamic_pointer_cast<WorkspaceGroup>(ws);

    // Workspace groups are NOT returned by IWP->getWorkspace() most of the time
    // because WorkspaceProperty is templated by <MatrixWorkspace>
    // and WorkspaceGroup does not subclass <MatrixWorkspace>
    if (!wsGroup && prop && !prop->value().empty()) {
      // So try to use the name in the AnalysisDataService
      try {
        wsGroup = AnalysisDataService::Instance().retrieveWS<WorkspaceGroup>(
            prop->value());
      } catch (Exception::NotFoundError &) { /* Do nothing */
      }
    }

    // Found the group either directly or by name?
    // If the property is of type WorkspaceGroup then don't unroll
    if (wsGroup && !wsGroupProp) {
      numGroups++;
      processGroups = true;
      std::vector<std::string> names = wsGroup->getNames();
      for (auto &name : names) {
        Workspace_sptr memberWS =
            AnalysisDataService::Instance().retrieve(name);
        if (!memberWS)
          throw std::invalid_argument("One of the members of " +
                                      wsGroup->getName() + ", " + name +
                                      " was not found!.");
        thisGroup.push_back(memberWS);
      }
    } else {
      // Single Workspace. Treat it as a "group" with only one member
      if (ws)
        thisGroup.push_back(ws);
    }

    // Add to the list of groups
    m_groups.push_back(thisGroup);
    m_groupWorkspaces.push_back(wsGroup);
  }

  // No groups? Get out.
  if (numGroups == 0)
    return processGroups;

  // ---- Confirm that all the groups are the same size -----
  // Index of the single group
  m_singleGroup = -1;
  // Size of the single or of all the groups
  m_groupSize = 1;
  m_groupsHaveSimilarNames = true;
  for (size_t i = 0; i < m_groups.size(); i++) {
    std::vector<Workspace_sptr> &thisGroup = m_groups[i];
    // We're ok with empty groups if the workspace property is optional
    if (thisGroup.empty() && !m_inputWorkspaceProps[i]->isOptional())
      throw std::invalid_argument("Empty group passed as input");
    if (!thisGroup.empty()) {
      // Record the index of the single group.
      WorkspaceGroup_sptr wsGroup = m_groupWorkspaces[i];
      if (wsGroup && (numGroups == 1))
        m_singleGroup = int(i);

      // For actual groups (>1 members)
      if (thisGroup.size() > 1) {
        // Check for matching group size
        if (m_groupSize > 1)
          if (thisGroup.size() != m_groupSize)
            throw std::invalid_argument(
                "Input WorkspaceGroups are not of the same size.");

        // Are ALL the names similar?
        if (wsGroup)
          m_groupsHaveSimilarNames =
              m_groupsHaveSimilarNames && wsGroup->areNamesSimilar();

        // Save the size for the next group
        m_groupSize = thisGroup.size();
      }
    }
  } // end for each group

  // If you get here, then the groups are compatible
  return processGroups;
}

/**
 * Calls process groups with the required timing checks and algorithm
 * execution finalization steps.
 *
 * @param startTime to record the algorithm execution start
 *
 * @return whether processGroups succeeds.
 */
bool Algorithm::doCallProcessGroups(Mantid::Types::DateAndTime &startTime) {
  // In the base implementation of processGroups, this normally calls
  // this->execute() again on each member of the group. Other algorithms may
  // choose to override that behavior (examples: CompareWorkspaces,
  // CheckWorkspacesMatch, RenameWorkspace)

  startTime = Mantid::Types::DateAndTime::getCurrentTime();
  // Start a timer
  Timer timer;

  bool completed = false;
  try {
    // Call the concrete algorithm's processGroups method
    completed = processGroups();
  } catch (std::exception &ex) {
    // The child algorithm will already have logged the error etc.,
    // but we also need to update flags in the parent algorithm and
    // send an ErrorNotification (because the child isn't registered with the
    // AlgorithmMonitor).
    setExecuted(false);
    m_runningAsync = false;
    m_running = false;
    notificationCenter().postNotification(
        new ErrorNotification(this, ex.what()));
    throw;
  } catch (...) {
    setExecuted(false);
    m_runningAsync = false;
    m_running = false;
    notificationCenter().postNotification(new ErrorNotification(
        this, "UNKNOWN Exception caught from processGroups"));
    throw;
  }

  // Check for a cancellation request in case the concrete algorithm doesn't
  interruption_point();

  if (completed) {
    // in the base processGroups each individual exec stores its outputs
    if (!m_usingBaseProcessGroups && (!isChild() || m_alwaysStoreInADS))
      this->store();

    // Get how long this algorithm took to run
    const float duration = timer.elapsed();
    // Log that execution has completed.
    reportCompleted(duration, true /* this is for group processing*/);
  }

  setExecuted(completed);
  notificationCenter().postNotification(
      new FinishedNotification(this, isExecuted()));

  return completed;
}

//--------------------------------------------------------------------------------------------
/** Process WorkspaceGroup inputs.
 *
 * This should be called after checkGroups(), which sets up required members.
 * It goes through each member of the group(s), creates and sets an algorithm
 * for each and executes them one by one.
 *
 * If there are several group input workspaces, then the member of each group
 * is executed pair-wise.
 *
 * @return true - if all the workspace members are executed.
 */
bool Algorithm::processGroups() {
  m_usingBaseProcessGroups = true;

  std::vector<WorkspaceGroup_sptr> outGroups;

  // ---------- Create all the output workspaces ----------------------------
  for (auto &pureOutputWorkspaceProp : m_pureOutputWorkspaceProps) {
    Property *prop = dynamic_cast<Property *>(pureOutputWorkspaceProp);
    if (prop) {
      auto outWSGrp = boost::make_shared<WorkspaceGroup>();
      outGroups.push_back(outWSGrp);
      // Put the GROUP in the ADS
      AnalysisDataService::Instance().addOrReplace(prop->value(), outWSGrp);
      outWSGrp->observeADSNotifications(false);
    }
  }

  double progress_proportion = 1.0 / static_cast<double>(m_groupSize);
  // Go through each entry in the input group(s)
  for (size_t entry = 0; entry < m_groupSize; entry++) {
    // use create Child Algorithm that look like this one
    Algorithm_sptr alg_sptr = this->createChildAlgorithm(
        this->name(), progress_proportion * static_cast<double>(entry),
        progress_proportion * (1 + static_cast<double>(entry)),
        this->isLogging(), this->version());
    // Don't make the new algorithm a child so that it's workspaces are stored
    // correctly
    alg_sptr->setChild(false);

    alg_sptr->setRethrows(true);

    IAlgorithm *alg = alg_sptr.get();
    // Set all non-workspace properties
    this->copyNonWorkspaceProperties(alg, int(entry) + 1);

    std::string outputBaseName;

    // ---------- Set all the input workspaces ----------------------------
    for (size_t iwp = 0; iwp < m_groups.size(); iwp++) {
      std::vector<Workspace_sptr> &thisGroup = m_groups[iwp];
      if (!thisGroup.empty()) {
        // By default (for a single group) point to the first/only workspace
        Workspace_sptr ws = thisGroup[0];

        if ((m_singleGroup == int(iwp)) || m_singleGroup < 0) {
          // Either: this is the single group
          // OR: all inputs are groups
          // ... so get then entry^th workspace in this group
          ws = thisGroup[entry];
        }
        // Append the names together
        if (!outputBaseName.empty())
          outputBaseName += "_";
        outputBaseName += ws->getName();

        // Set the property using the name of that workspace
        if (Property *prop =
                dynamic_cast<Property *>(m_inputWorkspaceProps[iwp])) {
          alg->setPropertyValue(prop->name(), ws->getName());
        } else {
          throw std::logic_error("Found a Workspace property which doesn't "
                                 "inherit from Property.");
        }
      } // not an empty (i.e. optional) input
    }   // for each InputWorkspace property

    std::vector<std::string> outputWSNames(m_pureOutputWorkspaceProps.size());
    // ---------- Set all the output workspaces ----------------------------
    for (size_t owp = 0; owp < m_pureOutputWorkspaceProps.size(); owp++) {
      if (Property *prop =
              dynamic_cast<Property *>(m_pureOutputWorkspaceProps[owp])) {
        // Default name = "in1_in2_out"
        const std::string inName = prop->value();
        std::string outName;
        if (m_groupsHaveSimilarNames) {
          outName.append(inName).append("_").append(
              Strings::toString(entry + 1));
        } else {
          outName.append(outputBaseName).append("_").append(inName);
        }

        auto inputProp = std::find_if(m_inputWorkspaceProps.begin(),
                                      m_inputWorkspaceProps.end(),
                                      WorkspacePropertyValueIs(inName));

        // Overwrite workspaces in any input property if they have the same
        // name as an output (i.e. copy name button in algorithm dialog used)
        // (only need to do this for a single input, multiple will be handled
        // by ADS)
        if (inputProp != m_inputWorkspaceProps.end()) {
          const auto &inputGroup =
              m_groups[inputProp - m_inputWorkspaceProps.begin()];
          if (!inputGroup.empty())
            outName = inputGroup[entry]->getName();
        }
        // Except if all inputs had similar names, then the name is "out_1"

        // Set in the output
        alg->setPropertyValue(prop->name(), outName);

        outputWSNames[owp] = outName;
      } else {
        throw std::logic_error(
            "Found a Workspace property which doesn't inherit from Property.");
      }
    } // for each OutputWorkspace property

    // ------------ Execute the algo --------------
    try {
      alg->execute();
    } catch (std::exception &e) {
      std::ostringstream msg;
      msg << "Execution of " << this->name() << " for group entry "
          << (entry + 1) << " failed: ";
      msg << e.what(); // Add original message
      throw std::runtime_error(msg.str());
    }

    // ------------ Fill in the output workspace group ------------------
    // this has to be done after execute() because a workspace must exist
    // when it is added to a group
    for (size_t owp = 0; owp < m_pureOutputWorkspaceProps.size(); owp++) {
      // And add it to the output group
      outGroups[owp]->add(outputWSNames[owp]);
    }

  } // for each entry in each group

  // restore group notifications
  for (auto &outGroup : outGroups) {
    outGroup->observeADSNotifications(true);
  }

  return true;
}

//--------------------------------------------------------------------------------------------
/** Copy all the non-workspace properties from this to alg
 *
 * @param alg :: other IAlgorithm
 * @param periodNum :: number of the "period" = the entry in the group + 1
 */
void Algorithm::copyNonWorkspaceProperties(IAlgorithm *alg, int periodNum) {
  if (!alg)
    throw std::runtime_error("Algorithm not created!");
  std::vector<Property *> props = this->getProperties();
  for (auto prop : props) {
    if (prop) {
      IWorkspaceProperty *wsProp = dynamic_cast<IWorkspaceProperty *>(prop);
      // Copy the property using the string
      if (!wsProp)
        this->setOtherProperties(alg, prop->name(), prop->value(), periodNum);
    }
  }
}

//--------------------------------------------------------------------------------------------
/** Virtual method to set the non workspace properties for this algorithm.
 * To be overridden by specific algorithms when needed.
 *
 *  @param alg :: pointer to the algorithm
 *  @param propertyName :: name of the property
 *  @param propertyValue :: value  of the property
 *  @param periodNum :: period number
 */
void Algorithm::setOtherProperties(IAlgorithm *alg,
                                   const std::string &propertyName,
                                   const std::string &propertyValue,
                                   int periodNum) {
  (void)periodNum; // Avoid compiler warning
  if (alg)
    alg->setPropertyValue(propertyName, propertyValue);
}

//--------------------------------------------------------------------------------------------
/** To query the property is a workspace property
 *  @param prop :: pointer to input properties
 *  @returns true if this is a workspace property
 */
bool Algorithm::isWorkspaceProperty(const Kernel::Property *const prop) const {
  if (!prop) {
    return false;
  }
  const IWorkspaceProperty *const wsProp =
      dynamic_cast<const IWorkspaceProperty *>(prop);
  return (wsProp != nullptr);
}

//=============================================================================================
//================================== Asynchronous Execution
//===================================
//=============================================================================================
namespace {
/**
 * A object to set the flag marking asynchronous running correctly
 */
struct AsyncFlagHolder {
  /** Constructor
   * @param A :: reference to the running flag
   */
  explicit AsyncFlagHolder(bool &running_flag) : m_running_flag(running_flag) {
    m_running_flag = true;
  }
  /// Destructor
  ~AsyncFlagHolder() { m_running_flag = false; }

private:
  /// Default constructor
  AsyncFlagHolder();
  /// Running flag
  bool &m_running_flag;
};
} // namespace

//--------------------------------------------------------------------------------------------
/**
 * Asynchronous execution
 */
Poco::ActiveResult<bool> Algorithm::executeAsync() {
  m_executeAsync = new Poco::ActiveMethod<bool, Poco::Void, Algorithm>(
      this, &Algorithm::executeAsyncImpl);
  return (*m_executeAsync)(Poco::Void());
}

/**Callback when an algorithm is executed asynchronously
 * @param i :: Unused argument
 * @return true if executed successfully.
 */
bool Algorithm::executeAsyncImpl(const Poco::Void &) {
  AsyncFlagHolder running(m_runningAsync);
  return this->execute();
}

/**
 * @return A reference to the Poco::NotificationCenter object that dispatches
 * notifications
 */
Poco::NotificationCenter &Algorithm::notificationCenter() const {
  if (!m_notificationCenter)
    m_notificationCenter = new Poco::NotificationCenter;
  return *m_notificationCenter;
}

/** Handles and rescales child algorithm progress notifications.
 *  @param pNf :: The progress notification from the child algorithm.
 */
void Algorithm::handleChildProgressNotification(
    const Poco::AutoPtr<ProgressNotification> &pNf) {
  double p = m_startChildProgress +
             (m_endChildProgress - m_startChildProgress) * pNf->progress;

  progress(p, pNf->message);
}

/**
 * @return A Poco:NObserver object that is responsible for reporting progress
 */
const Poco::AbstractObserver &Algorithm::progressObserver() const {
  if (!m_progressObserver)
    m_progressObserver = new Poco::NObserver<Algorithm, ProgressNotification>(
        *const_cast<Algorithm *>(this),
        &Algorithm::handleChildProgressNotification);

  return *m_progressObserver;
}

//--------------------------------------------------------------------------------------------
/**
 * Cancel an algorithm
 */
void Algorithm::cancel() {
  // set myself to be cancelled
  m_cancel = true;

  // Loop over the output workspaces and try to cancel them
  for (auto &weakPtr : m_ChildAlgorithms) {
    if (IAlgorithm_sptr sharedPtr = weakPtr.lock()) {
      sharedPtr->cancel();
    }
  }
}

/// Returns the cancellation state
bool Algorithm::getCancel() const { return m_cancel; }

/// Returns a reference to the logger.
Kernel::Logger &Algorithm::getLogger() const { return g_log; }
/// Logging can be disabled by passing a value of false
void Algorithm::setLogging(const bool value) { g_log.setEnabled(value); }
/// returns the status of logging, True = enabled
bool Algorithm::isLogging() const { return g_log.getEnabled(); }

/* Sets the logging priority offset. Values are subtracted from the log level.
 *
 * Example value=1 will turn warning into notice
 * Example value=-1 will turn notice into warning
 */
void Algorithm::setLoggingOffset(const int value) {
  if (m_communicator->rank() == 0)
    g_log.setLevelOffset(value);
  else {
    int offset{1};
    ConfigService::Instance().getValue("mpi.loggingOffset", offset);
    g_log.setLevelOffset(value + offset);
  }
}

/// returns the logging priority offset
int Algorithm::getLoggingOffset() const { return g_log.getLevelOffset(); }

//--------------------------------------------------------------------------------------------
/** This is called during long-running operations,
 * and check if the algorithm has requested that it be cancelled.
 */
void Algorithm::interruption_point() {
  // only throw exceptions if the code is not multi threaded otherwise you
  // contravene the OpenMP standard
  // that defines that all loops must complete, and no exception can leave an
  // OpenMP section
  // openmp cancel handling is performed using the ??, ?? and ?? macros in each
  // algrothim
  IF_NOT_PARALLEL
  if (m_cancel)
    throw CancelException();
}

/**
Report that the algorithm has completed.
@param duration : Algorithm duration
@param groupProcessing : We have been processing via processGroups if true.
*/
void Algorithm::reportCompleted(const double &duration,
                                const bool groupProcessing) {
  std::string optionalMessage;
  if (groupProcessing) {
    optionalMessage = ". Processed as a workspace group";
  }

  if (!m_isChildAlgorithm || m_alwaysStoreInADS) {
    if (m_isAlgStartupLoggingEnabled) {

      std::stringstream msg;
      msg << name() << " successful, Duration ";
      double seconds = duration;
      if (seconds > 60.) {
        int minutes = static_cast<int>(seconds / 60.);
        msg << minutes << " minutes ";
        seconds = seconds - static_cast<double>(minutes) * 60.;
      }
      msg << std::fixed << std::setprecision(2) << seconds << " seconds"
          << optionalMessage;
      getLogger().notice(msg.str());
    }
  }

  else {
    getLogger().debug() << name() << " finished with isChild = " << isChild()
                        << '\n';
  }
  m_running = false;
}

/** Registers the usage of the algorithm with the UsageService
 */
void Algorithm::registerFeatureUsage() const {
  if (UsageService::Instance().isEnabled()) {
    std::ostringstream oss;
    oss << this->name() << ".v" << this->version();
    UsageService::Instance().registerFeatureUsage("Algorithm", oss.str(),
                                                  isChild());
  }
}

/** Enable or disable Logging of start and end messages
@param enabled : true to enable logging, false to disable
*/
void Algorithm::setAlgStartupLogging(const bool enabled) {
  m_isAlgStartupLoggingEnabled = enabled;
}

/** return the state of logging of start and end messages
@returns : true to logging is enabled
*/
bool Algorithm::getAlgStartupLogging() const {
  return m_isAlgStartupLoggingEnabled;
}

bool Algorithm::isCompoundProperty(const std::string &name) const {
  return std::find(m_reservedList.cbegin(), m_reservedList.cend(), name) !=
         m_reservedList.cend();
}

/// Runs the algorithm with the specified execution mode.
void Algorithm::exec(Parallel::ExecutionMode executionMode) {
  switch (executionMode) {
  case Parallel::ExecutionMode::Serial:
  case Parallel::ExecutionMode::Identical:
    return exec();
  case Parallel::ExecutionMode::Distributed:
    return execDistributed();
  case Parallel::ExecutionMode::MasterOnly:
    return execMasterOnly();
  default:
    throw(std::runtime_error("Algorithm " + name() +
                             " does not support execution mode " +
                             Parallel::toString(executionMode)));
  }
}

/** Runs the algorithm in `distributed` execution mode.
 *
 * The default implementation runs the normal exec() method on all ranks.
 * Classes inheriting from Algorithm can re-implement this if they support
 * execution with multiple MPI ranks and require a special implementation for
 * distributed execution. */
void Algorithm::execDistributed() { exec(); }

/** Runs the algorithm in `master-only` execution mode.
 *
 * The default implementation runs the normal exec() method on rank 0 and
 * execNonMaster() on all other ranks. Classes inheriting from Algorithm can
 * re-implement this if they support execution with multiple MPI ranks and
 * require a special implementation for master-only execution. */
void Algorithm::execMasterOnly() {
  if (communicator().rank() == 0)
    exec();
  else
    execNonMaster();
}

/** By default execMasterOnly() runs this in `master-only` execution mode on all
 * but rank 0.
 *
 * The default implementation creates dummy workspaces for all pure output
 * workspaces. Classes inheriting from Algorithm can re-implement this if they
 * support execution with multiple MPI ranks and require a special behavior on
 * non-master ranks in master-only execution. */
void Algorithm::execNonMaster() {
  // If there is no output we can simply do nothing.
  if (m_pureOutputWorkspaceProps.empty())
    return;
  // Does Algorithm have exactly one input and one output workspace property?
  if (m_inputWorkspaceProps.size() == 1 &&
      m_pureOutputWorkspaceProps.size() == 1) {
    // Does the input workspace property point to an actual workspace?
    if (const auto &ws = m_inputWorkspaceProps.front()->getWorkspace()) {
      if (ws->storageMode() == Parallel::StorageMode::MasterOnly) {
        const auto &wsProp = m_pureOutputWorkspaceProps.front();
        // This is the reverse cast of what is done in
        // cacheWorkspaceProperties(), so it should never fail.
        const Property &prop = dynamic_cast<Property &>(*wsProp);
        auto clone = ws->cloneEmpty();
        // Currently we have not implemented a proper cloneEmpty() for all
        // workspace types, in particular the abundance of Workspace2D subtypes,
        // so we do a safety check here.
        if (ws->storageMode() != clone->storageMode())
          throw std::runtime_error(clone->id() +
                                   "::cloneEmpty() did not return a workspace "
                                   "with the correct storage mode. Make sure "
                                   "cloneEmpty() sets the storage mode.");
        setProperty(prop.name(), std::move(clone));
        return;
      }
    }
  }
  throw std::runtime_error(
      "Attempt to run algorithm with " +
      Parallel::toString(Parallel::ExecutionMode::MasterOnly) +
      ": Execution in this mode not implemented.");
}

/** Get a (valid) execution mode for this algorithm.
 *
 * "Valid" implies that this function does check whether or not the Algorithm
 * actually supports the mode. If it cannot return a valid mode it throws an
 * error. As a consequence, the return value of this function can be used
 * without further sanitization of the return value. */
Parallel::ExecutionMode Algorithm::getExecutionMode() const {
  if (communicator().size() == 1)
    return Parallel::ExecutionMode::Serial;

  const auto storageModes = getInputWorkspaceStorageModes();
  const auto executionMode = getParallelExecutionMode(storageModes);
  if (executionMode == Parallel::ExecutionMode::Invalid) {
    std::string error("Algorithm does not support execution with input "
                      "workspaces of the following storage types: " +
                      Parallel::toString(storageModes) + ".");
    getLogger().error() << error << "\n";
    throw(std::runtime_error(error));
  }
  if (executionMode == Parallel::ExecutionMode::Serial) {
    std::string error(Parallel::toString(executionMode) +
                      " is not a valid *parallel* execution mode.");
    getLogger().error() << error << "\n";
    throw(std::runtime_error(error));
  }
  return executionMode;
}

/** Get map of storage modes of all input workspaces.
 *
 * The key to the name is the property name of the respective workspace. */
std::map<std::string, Parallel::StorageMode>
Algorithm::getInputWorkspaceStorageModes() const {
  std::map<std::string, Parallel::StorageMode> map;
  for (const auto &wsProp : m_inputWorkspaceProps) {
    // This is the reverse cast of what is done in cacheWorkspaceProperties(),
    // so it should never fail.
    const Property &prop = dynamic_cast<Property &>(*wsProp);
    // Check if we actually have that input workspace
    if (wsProp->getWorkspace())
      map.emplace(prop.name(), wsProp->getWorkspace()->storageMode());
  }
  return map;
}

/** Get correct execution mode based on input storage modes for an MPI run.
 *
 * The default implementation returns ExecutionMode::Invalid. Classes inheriting
 * from Algorithm can re-implement this if they support execution with multiple
 * MPI ranks. May not return ExecutionMode::Serial, because that is not a
 * "parallel" execution mode. */
Parallel::ExecutionMode Algorithm::getParallelExecutionMode(
    const std::map<std::string, Parallel::StorageMode> &storageModes) const {
  UNUSED_ARG(storageModes)
  // By default no parallel execution is possible.
  return Parallel::ExecutionMode::Invalid;
}

/// Sets up skipping workspace validation on non-master ranks for
/// StorageMode::MasterOnly.
void Algorithm::setupSkipValidationMasterOnly() {
  // If workspaces have StorageMode::MasterOnly, validation on non-master ranks
  // would usually fail. Therefore, WorkspaceProperty needs to skip validation.
  // Thus, we must notify it whether or not it is on the master rank or not.
  if (communicator().rank() != 0)
    for (auto *prop : getProperties())
      if (auto *wsProp = dynamic_cast<IWorkspaceProperty *>(prop))
        wsProp->setIsMasterRank(false);
}

/// Returns a const reference to the (MPI) communicator of the algorithm.
const Parallel::Communicator &Algorithm::communicator() const {
  return *m_communicator;
}

/// Sets the (MPI) communicator of the algorithm.
void Algorithm::setCommunicator(const Parallel::Communicator &communicator) {
  m_communicator = Kernel::make_unique<Parallel::Communicator>(communicator);
}

//---------------------------------------------------------------------------
// Algorithm's inner classes
//---------------------------------------------------------------------------

Algorithm::AlgorithmNotification::AlgorithmNotification(
    const Algorithm *const alg)
    : Poco::Notification(), m_algorithm(alg) {}

const IAlgorithm *Algorithm::AlgorithmNotification::algorithm() const {
  return m_algorithm;
}

Algorithm::StartedNotification::StartedNotification(const Algorithm *const alg)
    : AlgorithmNotification(alg) {}
std::string Algorithm::StartedNotification::name() const {
  return "StartedNotification";
} ///< class name

Algorithm::FinishedNotification::FinishedNotification(
    const Algorithm *const alg, bool res)
    : AlgorithmNotification(alg), success(res) {}
std::string Algorithm::FinishedNotification::name() const {
  return "FinishedNotification";
}

Algorithm::ProgressNotification::ProgressNotification(
    const Algorithm *const alg, double p, const std::string &msg,
    double estimatedTime, int progressPrecision)
    : AlgorithmNotification(alg), progress(p), message(msg),
      estimatedTime(estimatedTime), progressPrecision(progressPrecision) {}

std::string Algorithm::ProgressNotification::name() const {
  return "ProgressNotification";
}

Algorithm::ErrorNotification::ErrorNotification(const Algorithm *const alg,
                                                const std::string &str)
    : AlgorithmNotification(alg), what(str) {}

std::string Algorithm::ErrorNotification::name() const {
  return "ErrorNotification";
}

const char *Algorithm::CancelException::what() const noexcept {
  return "Algorithm terminated";
}

} // namespace API

//---------------------------------------------------------------------------
// Specialized templated PropertyManager getValue definitions for Algorithm
// types
//---------------------------------------------------------------------------
namespace Kernel {
/**
 * Get the value of a given property as the declared concrete type
 * @param name :: The name of the property
 * @returns A pointer to an algorithm
 */
template <>
MANTID_API_DLL API::IAlgorithm_sptr
IPropertyManager::getValue<API::IAlgorithm_sptr>(
    const std::string &name) const {
  PropertyWithValue<API::IAlgorithm_sptr> *prop =
      dynamic_cast<PropertyWithValue<API::IAlgorithm_sptr> *>(
          getPointerToProperty(name));
  if (prop) {
    return *prop;
  } else {
    std::string message = "Attempt to assign property " + name +
                          " to incorrect type. Expected shared_ptr<IAlgorithm>";
    throw std::runtime_error(message);
  }
}

/**
 * Get the value of a given property as the declared concrete type (const
 * version)
 * @param name :: The name of the property
 * @returns A pointer to an algorithm
 */
template <>
MANTID_API_DLL API::IAlgorithm_const_sptr
IPropertyManager::getValue<API::IAlgorithm_const_sptr>(
    const std::string &name) const {
  PropertyWithValue<API::IAlgorithm_sptr> *prop =
      dynamic_cast<PropertyWithValue<API::IAlgorithm_sptr> *>(
          getPointerToProperty(name));
  if (prop) {
    return prop->operator()();
  } else {
    std::string message =
        "Attempt to assign property " + name +
        " to incorrect type. Expected const shared_ptr<IAlgorithm>";
    throw std::runtime_error(message);
  }
}
} // namespace Kernel

} // namespace Mantid