Line data    Source code

       1            0 : // Distributed under the MIT License.
       2              : // See LICENSE.txt for details.
       3              : 
       4              : #pragma once
       5              : 
       6              : #include <cstddef>
       7              : #include <mutex>
       8              : 
       9              : #include "DataStructures/DataBox/DataBox.hpp"
      10              : #include "DataStructures/TaggedTuple.hpp"
      11              : #include "Domain/Structure/ElementId.hpp"
      12              : #include "Parallel/GlobalCache.hpp"
      13              : #include "Parallel/Invoke.hpp"
      14              : #include "Parallel/Local.hpp"
      15              : #include "Parallel/NodeLock.hpp"
      16              : #include "Parallel/Phase.hpp"
      17              : #include "Utilities/ErrorHandling/Error.hpp"
      18              : #include "Utilities/Gsl.hpp"
      19              : 
      20              : namespace Parallel::Actions {
      21              : /// \brief Receive data for a specific element on the nodegroup.
      22              : ///
      23              : /// If `StartPhase` is `true` then `start_phase(phase)` is called on the
      24              : /// `element_to_execute_on`, otherwise `perform_algorithm()` is called on the
      25              : /// `element_to_execute_on`.
      26              : template <bool StartPhase = false>
      27            1 : struct ReceiveDataForElement {
      28              :   /// \brief Entry method called when receiving data from another node.
      29              :   template <typename ParallelComponent, typename DbTagsList,
      30              :             typename Metavariables, typename ArrayIndex, typename ReceiveData,
      31              :             typename ReceiveTag, size_t Dim, typename DistributedObject>
      32            1 :   static void apply(db::DataBox<DbTagsList>& box,
      33              :                     Parallel::GlobalCache<Metavariables>& cache,
      34              :                     const ArrayIndex& /*array_index*/,
      35              :                     const gsl::not_null<Parallel::NodeLock*> /*node_lock*/,
      36              :                     const DistributedObject* /*distributed_object*/,
      37              :                     const ReceiveTag& /*meta*/,
      38              :                     const ElementId<Dim>& element_to_execute_on,
      39              :                     typename ReceiveTag::temporal_id instance,
      40              :                     ReceiveData receive_data) {
      41              :     [[maybe_unused]] const size_t my_node = Parallel::my_node<size_t>(cache);
      42              :     auto& element_collection = db::get_mutable_reference<
      43              :         typename ParallelComponent::element_collection_tag>(
      44              :         make_not_null(&box));
      45              : 
      46              :     ASSERT(
      47              :         element_collection.count(element_to_execute_on) == 1,
      48              :         "ElementId " << element_to_execute_on << " is not on node " << my_node);
      49              :     const bool run_algorithm = ReceiveTag::insert_into_inbox(
      50              :         make_not_null(&tuples::get<ReceiveTag>(
      51              :             element_collection.at(element_to_execute_on).inboxes())),
      52              :         instance, std::move(receive_data));
      53              :     if (run_algorithm) {
      54              :       apply_impl<ParallelComponent>(cache, element_to_execute_on,
      55              :                                     make_not_null(&element_collection));
      56              :     }
      57              :   }
      58              : 
      59              :   /// \brief Entry method call when receiving from same node.
      60              :   template <typename ParallelComponent, typename DbTagsList,
      61              :             typename Metavariables, typename ArrayIndex, size_t Dim,
      62              :             typename DistributedObject>
      63            1 :   static void apply(db::DataBox<DbTagsList>& box,
      64              :                     Parallel::GlobalCache<Metavariables>& cache,
      65              :                     const ArrayIndex& /*array_index*/,
      66              :                     const gsl::not_null<Parallel::NodeLock*> /*node_lock*/,
      67              :                     const DistributedObject* /*distributed_object*/,
      68              :                     const ElementId<Dim>& element_to_execute_on) {
      69              :     auto& element_collection = db::get_mutable_reference<
      70              :         typename ParallelComponent::element_collection_tag>(
      71              :         make_not_null(&box));
      72              :     apply_impl<ParallelComponent>(cache, element_to_execute_on,
      73              :                                   make_not_null(&element_collection));
      74              :   }
      75              : 
      76              :  private:
      77              :   template <typename ParallelComponent, typename Metavariables,
      78              :             typename ElementCollection, size_t Dim>
      79            0 :   static void apply_impl(
      80              :       Parallel::GlobalCache<Metavariables>& cache,
      81              :       const ElementId<Dim>& element_to_execute_on,
      82              :       const gsl::not_null<ElementCollection*> element_collection) {
      83              :     if constexpr (StartPhase) {
      84              :       const Phase current_phase =
      85              :           Parallel::local_branch(
      86              :               Parallel::get_parallel_component<ParallelComponent>(cache))
      87              :               ->phase();
      88              :       ASSERT(element_collection->count(element_to_execute_on) == 1,
      89              :              "ElementId " << element_to_execute_on << " is not on node "
      90              :                           << Parallel::my_node<size_t>(cache););
      91              :       auto& element = element_collection->at(element_to_execute_on);
      92              :       const std::lock_guard element_lock(element.element_lock());
      93              :       // We always force the phase to start because if a previous phase
      94              :       // terminated properly, then this does nothing. But if a phase didn't
      95              :       // terminate properly, then it likely went into deadlock analysis or
      96              :       // post-failure cleanup, in which case we want to run code after a
      97              :       // failure, so we must force the phase to start.
      98              :       element.start_phase(current_phase, true);
      99              :     } else {
     100              :       auto& element = element_collection->at(element_to_execute_on);
     101              :       std::unique_lock element_lock(element.element_lock(), std::defer_lock);
     102              :       if (element_lock.try_lock()) {
     103              :         element.perform_algorithm();
     104              :       } else {
     105              :         const size_t my_node = Parallel::my_node<size_t>(cache);
     106              :         auto& my_proxy =
     107              :             Parallel::get_parallel_component<ParallelComponent>(cache);
     108              :         Parallel::threaded_action<Parallel::Actions::ReceiveDataForElement<>>(
     109              :             my_proxy[my_node], element_to_execute_on);
     110              :       }
     111              :     }
     112              :   }
     113              : };
     114              : }  // namespace Parallel::Actions