Generate RFT Data Only for Owned Wells

Otherwise, we risk parallel run race conditions in
EclWriter<>::writeOutput() due to multiple ranks constructing
data::Connection objects for shut/inactive distributed wells.

opm/simulators/flow/GenericOutputBlackoilModule.cpp

@@ -157,7 +157,7 @@ GenericOutputBlackoilModule(const EclipseState& eclState,
     , flowsC_(schedule, summaryConfig)
     , rftC_(eclState_, schedule_,
             [this](const std::string& wname)
-            { return !isDefunctParallelWell(wname); })
+            { return this->isOwnedByCurrentRank(wname); })
     , rst_conv_(std::move(globalCell), comm)
     , local_data_valid_(false)
 {

opm/simulators/flow/GenericOutputBlackoilModule.hpp

@@ -330,7 +330,8 @@ protected:
     void setupExtraBlockData(const std::size_t        reportStepNum,
                              std::function<bool(int)> isCartIdxOnThisRank);
 
-    virtual bool isDefunctParallelWell(std::string wname) const = 0;
+    virtual bool isDefunctParallelWell(const std::string& wname) const = 0;
+    virtual bool isOwnedByCurrentRank(const std::string& wname) const = 0;
 
     const EclipseState& eclState_;
     const Schedule& schedule_;

opm/simulators/flow/OutputBlackoilModule.hpp

@@ -597,7 +597,7 @@ private:
         Problem, std::void_t<decltype(std::declval<Problem>().geoMechModel())>
     > : public std::true_type {};
 
-    bool isDefunctParallelWell(std::string wname) const override
+    bool isDefunctParallelWell(const std::string& wname) const override
     {
         if (simulator_.gridView().comm().size() == 1)
             return false;
@@ -607,6 +607,11 @@ private:
         return candidate == parallelWells.end() || *candidate != value;
     }
 
+    bool isOwnedByCurrentRank(const std::string& wname) const override
+    {
+        return this->simulator_.problem().wellModel().isOwner(wname);
+    }
+
     void updateFluidInPlace_(const ElementContext& elemCtx, const unsigned dofIdx)
     {
         const auto& intQuants = elemCtx.intensiveQuantities(dofIdx, /*timeIdx=*/0);

opm/simulators/flow/OutputCompositionalModule.hpp

@@ -354,7 +354,7 @@ public:
     }
 
 private:
-    bool isDefunctParallelWell(std::string wname) const override
+    bool isDefunctParallelWell(const std::string& wname) const override
     {
         if (simulator_.gridView().comm().size() == 1)
             return false;
@@ -364,6 +364,14 @@ private:
         return candidate == parallelWells.end() || *candidate != value;
     }
 
+    bool isOwnedByCurrentRank(const std::string& wname) const override
+    {
+        // Note: This statement is not correct for distributed wells and
+        // will need additional logic once those are supported for
+        // compositional flows.
+        return ! this->isDefunctParallelWell(wname);
+    }
+
     void createLocalRegion_(std::vector<int>& region)
     {
         std::size_t elemIdx = 0;

opm/simulators/flow/RFTContainer.cpp

@@ -85,19 +85,21 @@ addToWells(data::Wells& wellDatas,
         gatherAndUpdateMap(gasConnectionSaturations_, comm);
     }
 
-    for (const auto& well: schedule_.getWells(reportStepNum)) {
-
-        // don't bother with wells not on this process
-        if (!wellQuery_(well.name())) {
+    for (const auto& wname : this->schedule_.wellNames(reportStepNum)) {
+        // Don't bother with wells not on this process.
+        if (!wellQuery_(wname)) {
             continue;
         }
 
-        //add data infrastructure for shut wells
-        if (!wellDatas.count(well.name())) {
-            auto& wellData = wellDatas[well.name()];
-            wellData.connections.reserve(well.getConnections().size());
-            std::transform(well.getConnections().begin(),
-                           well.getConnections().end(),
+        // Add data infrastructure for shut wells.
+        if (!wellDatas.count(wname)) {
+            const auto& conns = this->schedule_[reportStepNum]
+                .wells(wname).getConnections();
+
+            auto& wellData = wellDatas[wname];
+            wellData.connections.reserve(conns.size());
+
+            std::transform(conns.begin(), conns.end(),
                            std::back_inserter(wellData.connections),
                            [](const auto& connection)
                            {
@@ -107,7 +109,7 @@ addToWells(data::Wells& wellDatas,
                            });
         }
 
-        data::Well& wellData = wellDatas.at(well.name());
+        data::Well& wellData = wellDatas.at(wname);
 
         auto cond_assign = [](double& dest, unsigned idx, const auto& map)
         {

opm/simulators/wells/BlackoilWellModelGeneric.hpp

@@ -36,12 +36,14 @@
 #include <opm/simulators/wells/ConnectionIndexMap.hpp>
 #include <opm/simulators/wells/ParallelPAvgDynamicSourceData.hpp>
 #include <opm/simulators/wells/ParallelWBPCalculation.hpp>
+#include <opm/simulators/wells/ParallelWellInfo.hpp>
 #include <opm/simulators/wells/PerforationData.hpp>
 #include <opm/simulators/wells/WellFilterCake.hpp>
 #include <opm/simulators/wells/WellProdIndexCalculator.hpp>
 #include <opm/simulators/wells/WellTracerRate.hpp>
 #include <opm/simulators/wells/WGState.hpp>
 
+#include <algorithm>
 #include <cstddef>
 #include <functional>
 #include <map>
@@ -61,7 +63,6 @@ namespace Opm {
     template<class Scalar> class GasLiftWellState;
     class Group;
     class GuideRateConfig;
-    template<class Scalar> class ParallelWellInfo;
     class RestartValue;
     class Schedule;
     struct SimulatorUpdate;
@@ -268,6 +269,17 @@ public:
     parallelWellInfo(const std::size_t idx) const
     { return local_parallel_well_info_[idx].get(); }
 
+    bool isOwner(const std::string& wname) const
+    {
+        auto pwInfoPos = std::find_if(this->parallel_well_info_.begin(),
+                                      this->parallel_well_info_.end(),
+                                      [&wname](const auto& pwInfo)
+                                      { return pwInfo.name() == wname; });
+
+        return (pwInfoPos != this->parallel_well_info_.end())
+            && pwInfoPos->isOwner();
+    }
+
     const ConnectionIndexMap& connectionIndexMap(const std::size_t idx)
     { return conn_idx_map_[idx]; }