Improved Sumo summary data fetching (#14551)

Sumo data is loaded so the application stays responsive while it waits.

Run Sumo transfers on a dedicated thread
Split RiaSumoConnector into transport and data delegates
Load Sumo summary vectors concurrently and without waiting
Show that Sumo data is being loaded
This commit is contained in:
Jørgen Herje authored and GitHub committed 2026-08-20 11:48:25 +02:00
1 parent 9b268e6d98
commit 0d9d4fed19
30 files changed
+2124 -920

No files matched your search

@@ -167,14 +167,21 @@ void RimCloudDataSourceCollection::fieldChangedByUi( const caf::PdmFieldHandle*
if ( changedField == &m_sumoFieldName )
{
// What was picked below belonged to the asset that was just left, both the selection and the options
// it was chosen from. Forget the cached answers as well, or the case list would keep offering the
// cases of the previous asset. The editors are refreshed by the caller, which updates them as soon
// as this returns, so asking for that here would only fetch everything a second time.
m_sumoCaseId = "";
m_sumoEnsembleNames.v().clear();
m_sumoEnsembleNames.setValue( {} );
m_sumoConnector->requestCasesForFieldBlocking( m_sumoFieldName );
clearCachedCases();
clearCachedEnsembleNames();
}
else if ( changedField == &m_sumoCaseId )
{
m_sumoEnsembleNames.v().clear();
m_sumoEnsembleNames.setValue( {} );
clearCachedEnsembleNames();
}
if ( changedField == &m_addEnsembles )
{
@@ -214,12 +221,7 @@ QList<caf::PdmOptionItemInfo> RimCloudDataSourceCollection::calculateValueOption
QList<caf::PdmOptionItemInfo> options;
if ( fieldNeedingOptions == &m_sumoFieldName )
{
if ( m_sumoConnector->assets().empty() )
{
m_sumoConnector->requestAssetsBlocking();
}
for ( const auto& asset : m_sumoConnector->assets() )
for ( const auto& asset : cachedAssets() )
{
if ( m_sumoFieldName().isEmpty() )
{
@@ -231,24 +233,14 @@ QList<caf::PdmOptionItemInfo> RimCloudDataSourceCollection::calculateValueOption
}
else if ( fieldNeedingOptions == &m_sumoCaseId && !m_sumoFieldName().isEmpty() )
{
if ( m_sumoConnector->cases().empty() )
{
m_sumoConnector->requestCasesForFieldBlocking( m_sumoFieldName );
}
for ( const auto& sumoCase : m_sumoConnector->cases() )
for ( const auto& sumoCase : cachedCases( m_sumoFieldName ) )
{
options.push_back( { sumoCase.name, sumoCase.caseId.get() } );
}
}
else if ( fieldNeedingOptions == &m_sumoEnsembleNames && !m_sumoCaseId().isEmpty() )
{
if ( m_sumoConnector->ensembleNamesForCase( SumoCaseId( m_sumoCaseId ) ).empty() )
{
m_sumoConnector->requestEnsembleByCasesIdBlocking( SumoCaseId( m_sumoCaseId ) );
}
for ( const auto& name : m_sumoConnector->ensembleNamesForCase( SumoCaseId( m_sumoCaseId ) ) )
for ( const auto& name : cachedEnsembleNames( SumoCaseId( m_sumoCaseId ) ) )
{
options.push_back( { name, name } );
}
@@ -257,6 +249,66 @@ QList<caf::PdmOptionItemInfo> RimCloudDataSourceCollection::calculateValueOption
return options;
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
const std::vector<SumoAsset>& RimCloudDataSourceCollection::cachedAssets()
{
if ( m_assets.empty() && m_sumoConnector )
{
m_assets = m_sumoConnector->explore().assets();
}
return m_assets;
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
const std::vector<SumoCase>& RimCloudDataSourceCollection::cachedCases( const QString& assetName )
{
if ( m_casesAssetName != assetName && m_sumoConnector )
{
m_cases = m_sumoConnector->explore().cases( assetName );
m_casesAssetName = assetName;
}
return m_cases;
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
const std::vector<QString>& RimCloudDataSourceCollection::cachedEnsembleNames( const SumoCaseId& caseId )
{
if ( m_ensembleNamesCaseId != caseId.get() && m_sumoConnector )
{
m_ensembleNames = m_sumoConnector->explore().ensembleNames( caseId );
m_ensembleNamesCaseId = caseId.get();
}
return m_ensembleNames;
}
//--------------------------------------------------------------------------------------------------
/// Forget the cases Sumo answered with, so the next request for them asks again. Called when the asset they
/// belong to is left behind.
//--------------------------------------------------------------------------------------------------
void RimCloudDataSourceCollection::clearCachedCases()
{
m_casesAssetName.clear();
m_cases.clear();
}
//--------------------------------------------------------------------------------------------------
/// The same for the ensemble names, which belong to a case.
//--------------------------------------------------------------------------------------------------
void RimCloudDataSourceCollection::clearCachedEnsembleNames()
{
m_ensembleNamesCaseId.clear();
m_ensembleNames.clear();
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
@@ -402,7 +454,7 @@ std::vector<RimSumoDataSource*> RimCloudDataSourceCollection::addDataSources()
}
QString caseName;
for ( const auto& sumoCase : m_sumoConnector->cases() )
for ( const auto& sumoCase : cachedCases( m_sumoFieldName ) )
{
if ( sumoCase.caseId == sumoCaseId )
{
@@ -411,11 +463,8 @@ std::vector<RimSumoDataSource*> RimCloudDataSourceCollection::addDataSources()
}
}
m_sumoConnector->requestRealizationIdsForEnsembleBlocking( sumoCaseId, ensembleName );
m_sumoConnector->requestVectorNamesForEnsembleBlocking( sumoCaseId, ensembleName );
auto availableRealizationIds = m_sumoConnector->realizationIds();
auto vectorNames = m_sumoConnector->vectorNames();
const auto availableRealizationIds = m_sumoConnector->explore().realizationIds( sumoCaseId, ensembleName );
const auto vectorNames = m_sumoConnector->summary().vectorNames( sumoCaseId, ensembleName );
auto dataSource = new RimSumoDataSource();
dataSource->setCaseId( sumoCaseId );
@@ -58,6 +58,16 @@ private:
static bool isCloudApiServerAvailable();
// The option lists are rebuilt every time the property editor refreshes, so what Sumo answered is kept
// here and only asked for again when the selection it belongs to changes.
const std::vector<SumoAsset>& cachedAssets();
const std::vector<SumoCase>& cachedCases( const QString& assetName );
const std::vector<QString>& cachedEnsembleNames( const SumoCaseId& caseId );
void clearCachedCases();
void clearCachedEnsembleNames();
private:
caf::PdmField<bool> m_authenticate;
caf::PdmField<QString> m_sumoFieldName;
@@ -73,4 +83,12 @@ private:
caf::PdmField<bool> m_restartServer;
QPointer<RiaSumoConnector> m_sumoConnector;
std::vector<SumoAsset> m_assets;
QString m_casesAssetName;
std::vector<SumoCase> m_cases;
QString m_ensembleNamesCaseId;
std::vector<QString> m_ensembleNames;
};
@@ -857,6 +857,10 @@ void RimMultiPlot::updatePlots()
{
if ( m_showWindow )
{
// The plots are loaded one after the other, so a source loading data remotely would make its requests
// one plot at a time. Give it the chance to load everything this window needs in one go first.
prefetchPlotData();
for ( RimPlot* plot : plots() )
{
plot->loadDataAndUpdate();
@@ -142,6 +142,11 @@ protected:
virtual void onPlotAdditionOrRemoval();
// Called before the plots are loaded, giving a window the chance to load what all of its plots need at
// once instead of leaving each plot to request its own. Data is still loaded on demand, so this is an
// optimization only: not overriding it changes nothing but speed.
virtual void prefetchPlotData() {}
bool isMouseCursorInsidePlot();
private:
@@ -83,6 +83,15 @@ public:
virtual std::set<RifEclipseSummaryAddress> ensembleSummaryAddresses() const;
virtual std::set<time_t> ensembleTimeSteps() const;
// Hint that these addresses are about to be read, given before the curves pull their values. Sources that
// fetch data remotely can use it to load everything in one go instead of one blocking request per address.
// Data is still loaded on demand, so this is an optimization only: not calling it changes nothing but speed.
virtual void prefetchSummaryData( const std::vector<RifEclipseSummaryAddress>& resultAddresses ) {}
// Whether any of these addresses is being loaded right now, for sources that load without waiting. A plot
// asks about the addresses of its own curves, so it can say it is still waiting for data.
virtual bool isSummaryDataPending( const std::vector<RifEclipseSummaryAddress>& resultAddresses ) const { return false; }
void setEnsembleId( int ensembleId );
int ensembleId() const;
bool hasEnsembleParameters() const;
@@ -1588,6 +1588,32 @@ void RimSummaryMultiPlot::onPlotAdditionOrRemoval()
RimMultiPlot::onPlotAdditionOrRemoval();
}
//--------------------------------------------------------------------------------------------------
/// Gather what every plot in the window is about to read and hand it to each ensemble as one set. A plot
/// prefetches for itself as well, but the plots are loaded one at a time, so doing it here is what lets a
/// remote source load the whole window in one request group rather than one per plot.
//--------------------------------------------------------------------------------------------------
void RimSummaryMultiPlot::prefetchPlotData()
{
std::map<RimSummaryEnsemble*, std::vector<RifEclipseSummaryAddress>> addressesByEnsemble;
for ( RimSummaryPlot* plot : summaryPlots() )
{
if ( !plot ) continue;
for ( const auto& [ensemble, addresses] : plot->summaryAddressesByEnsemble() )
{
auto& allAddresses = addressesByEnsemble[ensemble];
allAddresses.insert( allAddresses.end(), addresses.begin(), addresses.end() );
}
}
for ( const auto& [ensemble, addresses] : addressesByEnsemble )
{
ensemble->prefetchSummaryData( addresses );
}
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
@@ -123,6 +123,8 @@ protected:
void onPlotAdditionOrRemoval() override;
void prefetchPlotData() override;
private:
void defineUiOrdering( QString uiConfigName, caf::PdmUiOrdering& uiOrdering ) override;
void defineEditorAttribute( const caf::PdmFieldHandle* field, QString uiConfigName, caf::PdmUiEditorAttribute* attribute ) override;
@@ -63,6 +63,8 @@
#include "SummaryPlotCommands/RicSummaryPlotEditorUi.h"
#include "Tools/RimPlotAxisTools.h"
#include "RiuAbstractOverlayContentFrame.h"
#include "RiuDraggableOverlayFrame.h"
#include "RiuPlotAxis.h"
#include "RiuPlotMainWindow.h"
#include "RiuPlotMainWindowTools.h"
@@ -1805,6 +1807,109 @@ void RimSummaryPlot::defineUiTreeOrdering( caf::PdmUiTreeOrdering& uiTreeOrderin
uiTreeOrdering.skipRemainingChildren( true );
}
//--------------------------------------------------------------------------------------------------
/// The addresses this plot's curves are about to read, grouped by the ensemble owning them. Used to tell a
/// source what is coming before any of it is read, see prefetchSummaryData().
//--------------------------------------------------------------------------------------------------
std::map<RimSummaryEnsemble*, std::vector<RifEclipseSummaryAddress>> RimSummaryPlot::summaryAddressesByEnsemble() const
{
std::map<RimSummaryEnsemble*, std::vector<RifEclipseSummaryAddress>> addressesByEnsemble;
auto addAddress = []( auto& addressesByEnsemble, RimSummaryEnsemble* ensemble, const RifEclipseSummaryAddress& address )
{
if ( !ensemble || !address.isValid() ) return;
addressesByEnsemble[ensemble].push_back( address );
};
if ( m_summaryCurveCollection )
{
for ( RimSummaryCurve* curve : m_summaryCurveCollection->curves() )
{
if ( !curve ) continue;
if ( auto summaryCase = curve->summaryCaseY() )
{
addAddress( addressesByEnsemble, summaryCase->firstAncestorOrThisOfType<RimSummaryEnsemble>(), curve->summaryAddressY() );
}
if ( auto summaryCase = curve->summaryCaseX() )
{
addAddress( addressesByEnsemble, summaryCase->firstAncestorOrThisOfType<RimSummaryEnsemble>(), curve->summaryAddressX() );
}
}
}
for ( RimEnsembleCurveSet* curveSet : m_ensembleCurveSetCollection->curveSets() )
{
if ( !curveSet ) continue;
addAddress( addressesByEnsemble, curveSet->summaryEnsemble(), curveSet->summaryAddressY() );
}
return addressesByEnsemble;
}
//--------------------------------------------------------------------------------------------------
/// Tell each ensemble in the plot which addresses its curves are about to read, before any of them read
/// one. Curves pull their values one at a time, so a source loading data remotely would otherwise make one
/// blocking request per curve; given the whole set up front it can load them together.
///
/// This is a hint only. Every curve still loads its own data, and an ensemble that does not need the hint
/// ignores it. A plot in a plot window is usually covered by the window prefetching for all its plots at
/// once, in which case there is nothing left for this to do.
//--------------------------------------------------------------------------------------------------
void RimSummaryPlot::prefetchSummaryData()
{
for ( const auto& [ensemble, addresses] : summaryAddressesByEnsemble() )
{
ensemble->prefetchSummaryData( addresses );
}
}
//--------------------------------------------------------------------------------------------------
/// Say that this plot is waiting for data. A source can load without waiting, and then the curves are drawn
/// with what has arrived so far, leaving a plot looking finished when it is not. Called after every load, so
/// the frame appears when the data is asked for and goes away when the last of it has arrived.
//--------------------------------------------------------------------------------------------------
void RimSummaryPlot::updateLoadingOverlayFrame()
{
if ( !plotWidget() ) return;
bool isWaitingForData = false;
for ( const auto& [ensemble, addresses] : summaryAddressesByEnsemble() )
{
if ( ensemble->isSummaryDataPending( addresses ) )
{
isWaitingForData = true;
break;
}
}
if ( !isWaitingForData )
{
if ( m_loadingOverlayFrame )
{
plotWidget()->removeOverlayFrame( m_loadingOverlayFrame );
delete m_loadingOverlayFrame;
m_loadingOverlayFrame = nullptr;
}
return;
}
if ( !m_loadingOverlayFrame )
{
m_loadingOverlayFrame = new RiuDraggableOverlayFrame( plotWidget()->getParentForOverlay(), plotWidget()->overlayMargins() );
m_loadingOverlayFrame->setAnchorCorner( RiuDraggableOverlayFrame::AnchorCorner::TopLeft );
auto* spinnerFrame = new RiuSpinnerOverlayContentFrame( m_loadingOverlayFrame );
m_loadingOverlayFrame->setContentFrame( spinnerFrame );
spinnerFrame->setText( "Loading" );
}
plotWidget()->addOverlayFrame( m_loadingOverlayFrame );
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
@@ -1815,6 +1920,8 @@ void RimSummaryPlot::onLoadDataAndUpdate()
auto plotWindow = firstAncestorOrThisOfType<RimMultiPlot>();
if ( plotWindow == nullptr ) updateDockWindowVisibility();
prefetchSummaryData();
if ( m_summaryCurveCollection )
{
m_summaryCurveCollection->loadDataAndUpdate( false );
@@ -1861,6 +1968,8 @@ void RimSummaryPlot::onLoadDataAndUpdate()
updateAxes();
updateStackedCurveData();
updateLoadingOverlayFrame();
}
//--------------------------------------------------------------------------------------------------
@@ -56,6 +56,7 @@ class RimSummaryTimeAxisProperties;
class RimPlotAxisPropertiesInterface;
class RimPlotAxisProperties;
class RiuSummaryQwtPlot;
class RiuDraggableOverlayFrame;
class RimSummaryNameHelper;
class RimSummaryPlotNameHelper;
class RimPlotTemplateFileItem;
@@ -150,6 +151,8 @@ public:
void deleteAllSummaryCurves();
RimSummaryCurveCollection* summaryCurveCollection() const;
std::map<RimSummaryEnsemble*, std::vector<RifEclipseSummaryAddress>> summaryAddressesByEnsemble() const;
void updatePlotTitle();
const RimSummaryNameHelper* activePlotTitleHelperAllCurves() const;
@@ -250,6 +253,9 @@ protected:
void defineUiOrdering( QString uiConfigName, caf::PdmUiOrdering& uiOrdering ) override;
void onLoadDataAndUpdate() override;
void prefetchSummaryData();
void updateLoadingOverlayFrame();
bool handleGlobalKeyEvent( QKeyEvent* keyEvent ) override;
private slots:
@@ -341,8 +347,9 @@ private:
caf::PdmChildArrayField<RimPlotAxisPropertiesInterface*> m_axisPropertiesArray;
QPointer<RiuSummaryPlot> m_summaryPlot;
std::unique_ptr<QwtPlotTextLabel> m_plotInfoLabel;
QPointer<RiuSummaryPlot> m_summaryPlot;
QPointer<RiuDraggableOverlayFrame> m_loadingOverlayFrame;
std::unique_ptr<QwtPlotTextLabel> m_plotInfoLabel;
std::unique_ptr<RimSummaryPlotNameHelper> m_nameHelperAllCurves;
caf::PdmChildField<RimSummaryPlotSourceStepping*> m_sourceStepping;
@@ -30,19 +30,61 @@
#include "RifEclipseSummaryAddress.h"
#include "Cloud/RimCloudDataSourceCollection.h"
#include "RimProject.h"
#include "RimSummaryCaseMainCollection.h"
#include "RimSummaryCaseSumo.h"
#include "RimSummaryCurve.h"
#include "RimSummaryMultiPlot.h"
#include "RimSummaryPlot.h"
#include "RimSumoDataSource.h"
#include <arrow/type_fwd.h>
#include "RiuPlotCurve.h"
#include <arrow/type_fwd.h>
#include <arrow/util/key_value_metadata.h>
#include <algorithm>
#include <map>
#include <memory>
#include <optional>
CAF_PDM_SOURCE_INIT( RimSummaryEnsembleSumo, "RimSummaryEnsembleSumo" );
namespace
{
//--------------------------------------------------------------------------------------------------
/// Whether any curve of the plot holds samples. A plot with none has nothing to fit its axes to, which is the
/// state a plot is left in when it is created for data that has not arrived yet.
//--------------------------------------------------------------------------------------------------
bool hasCurveSamples( const RimSummaryPlot* summaryPlot )
{
for ( const RimSummaryCurve* curve : summaryPlot->summaryAndEnsembleCurves() )
{
if ( curve && curve->plotCurve() && curve->plotCurve()->numSamples() > 0 ) return true;
}
return false;
}
//--------------------------------------------------------------------------------------------------
/// Whether any other plot of the same window already shows something. The plots of a window share their time
/// axis, so a window that shows data has a range that fitting one plot would move for all of them.
//--------------------------------------------------------------------------------------------------
bool windowShowsData( RimSummaryPlot* summaryPlot )
{
auto multiPlot = summaryPlot->firstAncestorOrThisOfType<RimSummaryMultiPlot>();
if ( !multiPlot ) return false;
for ( const RimSummaryPlot* plot : multiPlot->summaryPlots() )
{
if ( plot == summaryPlot ) continue;
if ( hasCurveSamples( plot ) ) return true;
}
return false;
}
//--------------------------------------------------------------------------------------------------
/// Read an integer column of any width as int64. The bit width of a column is decided by the producer
/// of the parquet file, and can not be assumed to be a specific type.
@@ -75,12 +117,36 @@ std::optional<std::vector<double>> readFloatingPointColumn( const std::shared_pt
return {};
}
//--------------------------------------------------------------------------------------------------
/// The unit of a summary vector, as written by the producer of the parquet file into the metadata of the
/// column holding the values. Empty when the file carries no unit for the column.
//--------------------------------------------------------------------------------------------------
std::string unitFromTableColumn( const std::shared_ptr<arrow::Table>& table, const std::string& columnName )
{
if ( !table || !table->schema() ) return {};
auto field = table->schema()->GetFieldByName( columnName );
if ( !field || !field->HasMetadata() ) return {};
auto metadata = field->metadata();
for ( int64_t i = 0; i < metadata->size(); i++ )
{
if ( QString::fromStdString( metadata->key( i ) ).compare( "unit", Qt::CaseInsensitive ) == 0 )
{
return metadata->value( i );
}
}
return {};
}
} // namespace
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
RimSummaryEnsembleSumo::RimSummaryEnsembleSumo()
: summaryDataLoaded( this )
{
CAF_PDM_InitObject( "Sumo Ensemble", ":/SummaryCase.svg", "", "The Base Class for all Summary Cases" );
@@ -93,6 +159,8 @@ RimSummaryEnsembleSumo::RimSummaryEnsembleSumo()
setAsEnsemble( true );
m_sumoConnector = RiaApplication::instance()->makeSumoConnector();
m_lifetimeToken = std::make_shared<bool>( true );
}
//--------------------------------------------------------------------------------------------------
@@ -108,8 +176,19 @@ void RimSummaryEnsembleSumo::setSumoDataSource( RimSumoDataSource* sumoDataSourc
//--------------------------------------------------------------------------------------------------
std::string RimSummaryEnsembleSumo::unitName( const RifEclipseSummaryAddress& resultAddress )
{
// TODO: Not implemented yet. Need to get the unit name from the Sumo data source
return {};
if ( !m_sumoDataSource() ) return {};
// Only the table already fetched for this vector is looked at. Asking for a unit must not put a request on
// its way: it is read while the plot axes are built, and the curves that trigger the load are drawn again
// once the data arrives, which is when the unit becomes available too.
const auto columnName = resultAddress.toEclipseTextAddress();
const auto parquetKey =
ParquetKey{ m_sumoDataSource()->caseId(), m_sumoDataSource()->ensembleName(), QString::fromStdString( columnName ), false };
auto it = m_parquetTable.find( parquetKey );
if ( it == m_parquetTable.end() ) return {};
return unitFromTableColumn( it->second, columnName );
}
//--------------------------------------------------------------------------------------------------
@@ -180,53 +259,229 @@ void RimSummaryEnsembleSumo::updateName( const std::set<QString>& existingEnsemb
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::loadSummaryData( const RifEclipseSummaryAddress& resultAddress )
{
if ( resultAddress.isStatistics() ) return;
loadSummaryData( std::vector<RifEclipseSummaryAddress>{ resultAddress } );
}
// An address without a vector name has no blob to fetch. The special time address used as curve
// x-axis is one such address. Requesting it would produce a URL with an empty vector segment, which
// the service answers with 404, surfacing as a misleading "parquet file size is 0 bytes" error.
if ( resultAddress.vectorName().empty() ) return;
if ( !m_sumoDataSource() ) return;
auto resultText = QString::fromStdString( resultAddress.toEclipseTextAddress() );
auto sumoCaseId = m_sumoDataSource->caseId();
auto sumoEnsembleName = m_sumoDataSource->ensembleName();
auto key = ParquetKey{ sumoCaseId, sumoEnsembleName, resultText, false };
if ( m_parquetTable.find( key ) == m_parquetTable.end() )
{
auto contents = loadParquetData( key );
RiaLogging::debug( std::format( "Load Summary Data. Contents size: {}", contents.size() ) );
std::shared_ptr<arrow::Table> table = readParquetTable( contents, QString::fromStdString( resultAddress.uiText() ) );
m_parquetTable[key] = table;
distributeDataToRealizations( resultAddress, table );
}
auto parametersKey = ParquetKey{ sumoCaseId, sumoEnsembleName, "", true };
if ( m_parquetTable.find( parametersKey ) == m_parquetTable.end() )
{
auto contents = m_sumoConnector->requestParametersParquetDataBlocking( sumoCaseId, sumoEnsembleName );
RiaLogging::debug( std::format( "Load ensemble parameter sensitivities. Contents size: {}", contents.size() ) );
std::shared_ptr<arrow::Table> table = readParquetTable( contents, QString( "%1 parameter sensitivities" ).arg( sumoEnsembleName ) );
m_parquetTable[parametersKey] = table;
distributeParametersDataToRealizations( table );
}
//--------------------------------------------------------------------------------------------------
/// Load several vectors at once. Each vector is one parquet blob covering every realization, so the
/// addresses that are not cached yet are fetched as one concurrent batch rather than one after another.
/// That matters because a vector the service has not aggregated yet is produced on demand by the request
/// asking for it, so fetching serially costs the sum of those aggregations.
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::loadSummaryData( const std::vector<RifEclipseSummaryAddress>& resultAddresses )
{
// Nothing is fetched while the caller waits. A curve asking for values it does not have yet gets none,
// the request is put on its way, and the curve is drawn again once it arrives. Waiting here instead
// stopped the application for as long as the service took, which for a vector it has not aggregated yet
// is a good while.
prefetchSummaryData( resultAddresses );
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
QByteArray RimSummaryEnsembleSumo::loadParquetData( const ParquetKey& parquetKey )
void RimSummaryEnsembleSumo::loadEnsembleParameters()
{
if ( !m_sumoConnector ) return {};
if ( !m_sumoDataSource() || !m_sumoConnector ) return;
return m_sumoConnector->requestParquetDataBlocking( SumoCaseId( parquetKey.caseId ), parquetKey.ensembleId, parquetKey.vectorName );
auto sumoCaseId = m_sumoDataSource->caseId();
auto sumoEnsembleName = m_sumoDataSource->ensembleName();
auto parametersKey = ParquetKey{ sumoCaseId, sumoEnsembleName, "", true };
if ( m_parquetTable.find( parametersKey ) != m_parquetTable.end() ) return;
if ( m_pendingVectors.find( parametersKey ) != m_pendingVectors.end() ) return;
// Asked for without waiting, like the vectors. The service aggregates the parameters on demand too, so
// the first request for them can take a while, and it used to be made from inside the read of a curve
// value: dropping a vector into a plot stopped the application until the parameters had been fetched.
m_pendingVectors[parametersKey] = RifEclipseSummaryAddress();
std::weak_ptr<bool> isAlive = m_lifetimeToken;
m_sumoConnector->summary().parameterDataAsync( sumoCaseId,
sumoEnsembleName,
[this, isAlive, parametersKey]( const QByteArray& contents )
{
if ( isAlive.expired() ) return;
onParameterDataReceived( parametersKey, contents );
} );
}
//--------------------------------------------------------------------------------------------------
/// The ensemble parameters have arrived. Called on the thread owning the user interface.
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::onParameterDataReceived( const ParquetKey& parquetKey, const QByteArray& contents )
{
auto it = m_pendingVectors.find( parquetKey );
// No longer wanted: the data source changed, or the cache was cleared, while this was on its way.
if ( it == m_pendingVectors.end() ) return;
m_pendingVectors.erase( it );
RiaLogging::debug( std::format( "Load ensemble parameter sensitivities. Contents size: {}", contents.size() ) );
std::shared_ptr<arrow::Table> table = readParquetTable( contents, QString( "%1 parameter sensitivities" ).arg( parquetKey.ensembleId ) );
m_parquetTable[parquetKey] = table;
distributeParametersDataToRealizations( table );
updatePlotsUsingThisEnsemble();
}
//--------------------------------------------------------------------------------------------------
/// Ask for everything the plots are about to read, and return without waiting for any of it. Each vector is
/// taken in as it arrives, so a plot shows the vectors that are ready while the rest are still on their way
/// rather than staying blank until the slowest one is done.
///
/// A vector still on its way is reported as having no data, and the curves using it are drawn empty. They are
/// redrawn when it arrives, see onVectorDataReceived.
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::prefetchSummaryData( const std::vector<RifEclipseSummaryAddress>& resultAddresses )
{
if ( !m_sumoDataSource() || !m_sumoConnector ) return;
auto sumoCaseId = m_sumoDataSource->caseId();
auto sumoEnsembleName = m_sumoDataSource->ensembleName();
std::vector<QString> vectorNamesToFetch;
for ( const auto& resultAddress : resultAddresses )
{
if ( resultAddress.isStatistics() ) continue;
if ( resultAddress.vectorName().empty() ) continue;
auto resultText = QString::fromStdString( resultAddress.toEclipseTextAddress() );
auto key = ParquetKey{ sumoCaseId, sumoEnsembleName, resultText, false };
if ( m_parquetTable.find( key ) != m_parquetTable.end() ) continue;
if ( m_pendingVectors.find( key ) != m_pendingVectors.end() ) continue;
m_pendingVectors[key] = resultAddress;
vectorNamesToFetch.push_back( resultText );
}
// The parameters belong to the ensemble rather than to any one vector, and are wanted as soon as
// anything of it is read. Asked for here so they travel alongside the vectors.
loadEnsembleParameters();
if ( vectorNamesToFetch.empty() ) return;
std::weak_ptr<bool> isAlive = m_lifetimeToken;
m_sumoConnector->summary().vectorDataAsync( sumoCaseId,
sumoEnsembleName,
vectorNamesToFetch,
[this, isAlive, sumoCaseId, sumoEnsembleName]( const QString& vectorName,
const QByteArray& contents )
{
// The request outlived the ensemble that asked for it.
if ( isAlive.expired() ) return;
onVectorDataReceived( ParquetKey{ sumoCaseId, sumoEnsembleName, vectorName, false },
contents );
} );
}
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
bool RimSummaryEnsembleSumo::isSummaryDataPending( const std::vector<RifEclipseSummaryAddress>& resultAddresses ) const
{
if ( m_pendingVectors.empty() || !m_sumoDataSource() ) return false;
auto sumoCaseId = m_sumoDataSource()->caseId();
auto sumoEnsembleName = m_sumoDataSource()->ensembleName();
for ( const auto& resultAddress : resultAddresses )
{
auto resultText = QString::fromStdString( resultAddress.toEclipseTextAddress() );
if ( m_pendingVectors.contains( ParquetKey{ sumoCaseId, sumoEnsembleName, resultText, false } ) ) return true;
}
return false;
}
//--------------------------------------------------------------------------------------------------
/// One requested vector has arrived. Called on the thread owning the user interface, once per vector.
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::onVectorDataReceived( const ParquetKey& parquetKey, const QByteArray& contents )
{
auto it = m_pendingVectors.find( parquetKey );
// No longer wanted: the data source changed, or the cache was cleared, while this was on its way.
if ( it == m_pendingVectors.end() ) return;
const auto resultAddress = it->second;
m_pendingVectors.erase( it );
RiaLogging::debug( std::format( "Load Summary Data. Contents size: {}", contents.size() ) );
// Empty contents mean the request failed. The empty result is stored like any other, so a failure is not
// retried on every redraw, matching what a failed blocking load does.
std::shared_ptr<arrow::Table> table = readParquetTable( contents, QString::fromStdString( resultAddress.uiText() ) );
m_parquetTable[parquetKey] = table;
distributeDataToRealizations( resultAddress, table );
loadEnsembleParameters();
updatePlotsUsingThisEnsemble();
}
//--------------------------------------------------------------------------------------------------
/// Redraw with what has arrived so far. The curves read their values again, those still waiting for data come
/// back empty, and the replot itself is coalesced by the redraw scheduler.
///
/// A plot is created and fitted to its data in one go. Fetching from Sumo does not wait, so a plot created for
/// data still on its way was fitted while it had none, and kept that range. Such a plot is fitted here when its
/// first data arrives, which is what creation would have done had the data been there.
///
/// Only a plot that is still empty is fitted. A plot that already shows something has a range the user may have
/// chosen, and loading another vector into it must not throw that away.
//--------------------------------------------------------------------------------------------------
void RimSummaryEnsembleSumo::updatePlotsUsingThisEnsemble()
{
// Loading a plot can bring in the next vector, which asks for this update again. Finish the pass that is
// running and repeat it afterwards, rather than reloading plots from inside their own load.
if ( m_isUpdatingPlots )
{
m_hasMissedPlotUpdate = true;
return;
}
m_isUpdatingPlots = true;
do
{
m_hasMissedPlotUpdate = false;
for ( RimSummaryPlot* summaryPlot : RimProject::current()->descendantsOfType<RimSummaryPlot>() )
{
if ( !summaryPlot->summaryAddressesByEnsemble().contains( this ) ) continue;
// Fit only a plot that has nothing to show and sits in a window where nothing else does either.
// Plots of a window share their time axis, so fitting one moves them all, and a window already
// showing data has a range the user may have chosen and must keep. A plot added to such a window
// takes the shared range instead, which is what happens when its data comes from the cache.
const bool needsInitialFit = !hasCurveSamples( summaryPlot ) && !windowShowsData( summaryPlot );
summaryPlot->loadDataAndUpdate();
// zoomAll turns the automatic range back on, the one the fit onto an empty plot switched off.
if ( needsInitialFit ) summaryPlot->zoomAll();
summaryPlot->scheduleReplotIfVisible();
}
} while ( m_hasMissedPlotUpdate );
m_isUpdatingPlots = false;
// Sent after the guard is released, so a listener reloading something is not treated as a nested update.
// The plots of the project are already done above; this reaches the views held outside it.
summaryDataLoaded.send();
}
//--------------------------------------------------------------------------------------------------
@@ -354,6 +609,8 @@ void RimSummaryEnsembleSumo::distributeDataToRealizations( const RifEclipseSumma
RiaLogging::warning( "Failed to find values column" );
return;
}
RiaLogging::debug( std::format( "Unit of '{}': '{}'", columnName, unitFromTableColumn( table, columnName ) ) );
}
// find unique realizations
@@ -654,6 +911,10 @@ void RimSummaryEnsembleSumo::clearCachedData()
{
m_resultAddresses.clear();
m_parquetTable.clear();
// Anything still on its way belongs to the data just thrown away. Forgetting it here makes those replies
// drop their contents on arrival, and lets the vectors be requested again if they are still wanted.
m_pendingVectors.clear();
}
//--------------------------------------------------------------------------------------------------
@@ -57,6 +57,10 @@ class RimSummaryEnsembleSumo : public RimSummaryEnsemble
{
CAF_PDM_HEADER_INIT;
public:
// Sent when data asked for earlier has arrived, for views the reload below cannot reach.
caf::Signal<> summaryDataLoaded;
public:
RimSummaryEnsembleSumo();
@@ -65,12 +69,15 @@ public:
void onRealizationSelectionChanged();
void loadSummaryData( const RifEclipseSummaryAddress& resultAddress );
void loadSummaryData( const std::vector<RifEclipseSummaryAddress>& resultAddresses );
std::string unitName( const RifEclipseSummaryAddress& resultAddress );
RiaDefines::EclipseUnitSystem unitSystem() const;
std::set<RifEclipseSummaryAddress> allResultAddresses() const;
std::pair<std::string, std::string> nameKeys() const override;
void updateName( const std::set<QString>& existingEnsembleNames ) override;
void prefetchSummaryData( const std::vector<RifEclipseSummaryAddress>& resultAddresses ) override;
bool isSummaryDataPending( const std::vector<RifEclipseSummaryAddress>& resultAddresses ) const override;
protected:
void onLoadDataAndUpdate() override;
@@ -83,13 +90,16 @@ private:
void updateResultAddresses();
void clearCachedData();
QByteArray loadParquetData( const ParquetKey& parquetKey );
void distributeDataToRealizations( const RifEclipseSummaryAddress& resultAddress, std::shared_ptr<arrow::Table> table );
void buildMetaData();
void distributeParametersDataToRealizations( std::shared_ptr<arrow::Table> table );
void redistributeCachedDataToRealizations();
void loadEnsembleParameters();
void onVectorDataReceived( const ParquetKey& parquetKey, const QByteArray& contents );
void onParameterDataReceived( const ParquetKey& parquetKey, const QByteArray& contents );
void updatePlotsUsingThisEnsemble();
static std::shared_ptr<arrow::Table> readParquetTable( const QByteArray& contents, const QString& messageTag );
@@ -100,4 +110,15 @@ private:
std::set<RifEclipseSummaryAddress> m_resultAddresses;
std::map<ParquetKey, std::shared_ptr<arrow::Table>> m_parquetTable;
// The vectors requested but not yet arrived, and the address each belongs to. A vector in here is not
// requested again, and is reported as having no data yet rather than being waited for.
std::map<ParquetKey, RifEclipseSummaryAddress> m_pendingVectors;
// Held by the callbacks of requests still on their way. They check it before touching this object, so a
// reply arriving after the ensemble is gone is dropped instead of writing into freed memory.
std::shared_ptr<bool> m_lifetimeToken;
bool m_isUpdatingPlots = false;
bool m_hasMissedPlotUpdate = false;
};