mirror of
https://github.com/OPM/ResInsight.git
synced 2026-10-26 15:10:22 -05:00
[ri-cloud-api]: Add grid data from sumo using api (#14653)
Add grid data from Sumo using ri-cloud-api Load Eclipse grid data (roff) from Sumo as RimReservoirGridEnsemble, one case per realization, with async fetch of grid properties per timestep. - New: RiaSumoGrid (transport), RifReaderSumoGridProperty (lazy property reader), RimReservoirGridEnsembleSumo, RimRoffCaseSumo, and a "Create Reservoir Grid Ensemble" command. - Shared grid: IJK dimensions compared before download; one RigMainGrid is reused across the ensemble when they match, each realization still parses its own ACTNUM before releasing geometry. - Rename folders: Summary/Sumo -> Summary/Cloud, RimSumoDataSource -> Cloud/ (PDM keywords unchanged, old project files still load). - UI: cloud data sources also shown in the 3D window's Data Sources dock. - Fix: realizations no longer lost on project reload; realizations missing data are now reported instead of silently skipped.
This commit is contained in:
1 parent
71c5b5a97d
commit
c49f298a7d
70 files changed
+5194
-222
No files matched your search
@@ -4,6 +4,7 @@ set(SOURCE_GROUP_SOURCE_FILES
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaSumoConnector.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaSumoDefines.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaSumoExplore.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaSumoGrid.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaSumoSummary.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaConnectorTools.cpp
|
||||
${CMAKE_CURRENT_LIST_DIR}/RiaOsduConnector.cpp
|
||||
|
||||
@@ -24,6 +24,8 @@
|
||||
|
||||
#include <QCoreApplication>
|
||||
#include <QDir>
|
||||
#include <QElapsedTimer>
|
||||
#include <QEventLoop>
|
||||
#include <QFileInfo>
|
||||
#include <QHostAddress>
|
||||
#include <QNetworkAccessManager>
|
||||
@@ -52,6 +54,10 @@ constexpr int maxConsecutiveFailures = 2;
|
||||
// Per-request timeout for the health check, kept well below the poll interval.
|
||||
constexpr int healthCheckTimeoutMs = 5 * 1000;
|
||||
|
||||
// Interval between the /alive attempts made while waiting for the service to come up. Much shorter than
|
||||
// the periodic health check, as a caller is blocked meanwhile.
|
||||
constexpr int readinessPollIntervalMs = 250;
|
||||
|
||||
// Number of ports scanned, starting at the wanted port, when looking for a free port to bind to.
|
||||
constexpr int portRangeLength = 100;
|
||||
} // namespace
|
||||
@@ -229,6 +235,71 @@ void RiaCloudApiService::start()
|
||||
m_startupTimer.start();
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// The periodic health check first runs a grace period after the launch, which is far too long to make a
|
||||
/// caller wait for. Poll the service directly instead, so a request can be issued as soon as it answers.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
bool RiaCloudApiService::waitUntilResponding( int timeoutMs )
|
||||
{
|
||||
if ( m_isResponding ) return true;
|
||||
|
||||
if ( !isRunning() ) start();
|
||||
|
||||
QElapsedTimer elapsed;
|
||||
elapsed.start();
|
||||
|
||||
while ( elapsed.elapsed() < timeoutMs )
|
||||
{
|
||||
if ( m_port >= 0 )
|
||||
{
|
||||
QNetworkRequest request( QUrl( serverUrl() + "/alive" ) );
|
||||
request.setTransferTimeout( healthCheckTimeoutMs );
|
||||
|
||||
QNetworkReply* reply = m_networkAccessManager->get( request );
|
||||
|
||||
QEventLoop eventLoop;
|
||||
QObject::connect( reply, &QNetworkReply::finished, &eventLoop, &QEventLoop::quit );
|
||||
eventLoop.exec();
|
||||
|
||||
const int statusCode = reply->attribute( QNetworkRequest::HttpStatusCodeAttribute ).toInt();
|
||||
const bool answered = reply->error() == QNetworkReply::NoError && statusCode == 200;
|
||||
reply->deleteLater();
|
||||
|
||||
if ( answered )
|
||||
{
|
||||
m_consecutiveFailures = 0;
|
||||
|
||||
if ( !m_isResponding )
|
||||
{
|
||||
m_isResponding = true;
|
||||
emit statusChanged();
|
||||
}
|
||||
|
||||
// The service is known to be up, so the grace period before the periodic check has served
|
||||
// its purpose. Fall back to the ordinary health check from here.
|
||||
m_startupTimer.stop();
|
||||
if ( !m_healthTimer.isActive() ) m_healthTimer.start();
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// The process may still be booting, or may have exited on a missing dependency.
|
||||
if ( !isRunning() ) break;
|
||||
|
||||
QEventLoop delayLoop;
|
||||
QTimer delayTimer;
|
||||
delayTimer.setSingleShot( true );
|
||||
QObject::connect( &delayTimer, &QTimer::timeout, &delayLoop, &QEventLoop::quit );
|
||||
delayTimer.start( readinessPollIntervalMs );
|
||||
delayLoop.exec();
|
||||
}
|
||||
|
||||
RiaLogging::error( std::format( "Cloud API service: not responding after {} ms, giving up.", elapsed.elapsed() ) );
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
|
||||
@@ -59,6 +59,14 @@ public:
|
||||
// True once the service has answered a health check, i.e. it is actually serving requests.
|
||||
bool isResponding() const;
|
||||
|
||||
// Block until the service answers a health check, starting it when it is not running. Returns
|
||||
// immediately when it is already responding, and false when it does not answer within the timeout.
|
||||
//
|
||||
// Requests must not be issued before this returns true: isRunning() is true as soon as the process is
|
||||
// launched, while uvicorn may still be booting. Call from the thread owning this object, as the health
|
||||
// check uses its network access manager.
|
||||
bool waitUntilResponding( int timeoutMs );
|
||||
|
||||
int port() const;
|
||||
|
||||
// Base URL of the service, including the port actually in use, e.g. "http://127.0.0.1:8001". Empty
|
||||
|
||||
@@ -53,6 +53,7 @@ RiaSumoConnector::RiaSumoConnector( QObject* parent,
|
||||
: RiaCloudConnector( parent, {}, authority, scopes, clientId, port )
|
||||
, m_serverUrlProvider( std::move( serverUrlProvider ) )
|
||||
, m_explore( *this )
|
||||
, m_grid( *this )
|
||||
, m_summary( *this )
|
||||
{
|
||||
// The transfer thread runs the network requests issued by the blocking wrappers, so the calling thread can
|
||||
@@ -65,7 +66,11 @@ RiaSumoConnector::RiaSumoConnector( QObject* parent,
|
||||
QObject::connect( m_transferThread,
|
||||
&QThread::started,
|
||||
m_transferContext,
|
||||
[this]() { m_transferNetworkAccessManager = new QNetworkAccessManager( m_transferContext ); } );
|
||||
[this]()
|
||||
{
|
||||
m_transferNetworkAccessManager = new QNetworkAccessManager( m_transferContext );
|
||||
m_transferBackgroundNetworkAccessManager = new QNetworkAccessManager( m_transferContext );
|
||||
} );
|
||||
|
||||
// The context object lives on the transfer thread, so let that thread delete it when it stops.
|
||||
QObject::connect( m_transferThread, &QThread::finished, m_transferContext, &QObject::deleteLater );
|
||||
@@ -81,6 +86,14 @@ RiaSumoExplore& RiaSumoConnector::explore()
|
||||
return m_explore;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
RiaSumoGrid& RiaSumoConnector::grid()
|
||||
{
|
||||
return m_grid;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -101,6 +114,22 @@ QString RiaSumoConnector::server() const
|
||||
return m_serverUrlProvider();
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// The service is started when the access token is granted, and needs a moment before it answers. Ask for
|
||||
/// the address only once a request is about to be made, so the wait for the service is done here rather
|
||||
/// than producing a request without a host.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoConnector::serviceBaseUrl()
|
||||
{
|
||||
const QString baseUrl = server();
|
||||
if ( baseUrl.isEmpty() )
|
||||
{
|
||||
RiaLogging::error( "The local cloud API service is not available, unable to request data from Sumo." );
|
||||
}
|
||||
|
||||
return baseUrl;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -136,6 +165,20 @@ QNetworkAccessManager* RiaSumoConnector::networkAccessManager()
|
||||
return m_networkAccessManager;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// A separate connection pool for background prefetch, see the header comment. Falls back to the ordinary
|
||||
/// manager off the transfer thread, matching networkAccessManager(), though prefetch always runs there.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QNetworkAccessManager* RiaSumoConnector::backgroundNetworkAccessManager()
|
||||
{
|
||||
if ( m_transferThread && QThread::currentThread() == m_transferThread && m_transferBackgroundNetworkAccessManager )
|
||||
{
|
||||
return m_transferBackgroundNetworkAccessManager;
|
||||
}
|
||||
|
||||
return m_networkAccessManager;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Run work on the transfer thread and return at once. Nothing is waited for here, so the result has to be
|
||||
/// delivered by a callback. The token is requested first, while still on the calling thread, because
|
||||
@@ -198,21 +241,41 @@ void RiaSumoConnector::invokeOnConnectorThread( const std::function<void()>& wor
|
||||
/// A blob is fetched in two steps, first the pre-signed URI and then the data itself. Both are started here
|
||||
/// and neither is waited for: each step continues from the reply of the one before.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
void RiaSumoConnector::downloadBlobAsync( const QString& blobId, const std::function<void( const QByteArray& )>& onFinished )
|
||||
void RiaSumoConnector::downloadBlobAsync( const QString& blobId,
|
||||
const std::function<void( const QByteArray& )>& onFinished,
|
||||
int blobTimeoutMillis,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
const QString url = QString( "%1/blobs/%2/sas_token_and_blob_base_uri" ).arg( server() ).arg( blobId );
|
||||
if ( !onFinished ) return;
|
||||
|
||||
// Every exit below reports through onFinished. Returning without calling it would strand whatever is
|
||||
// waiting for this blob, which can then never be requested again.
|
||||
const QString baseUrl = serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() )
|
||||
{
|
||||
RiaLogging::warning( "No Sumo service address available, unable to download blob." );
|
||||
onFinished( {} );
|
||||
return;
|
||||
}
|
||||
|
||||
const QString url = QString( "%1/blobs/%2/sas_token_and_blob_base_uri" ).arg( baseUrl ).arg( blobId );
|
||||
|
||||
QNetworkRequest networkRequest;
|
||||
networkRequest.setUrl( QUrl( url ) );
|
||||
addStandardHeader( networkRequest, transferToken(), RiaCloudDefines::contentTypeJson() );
|
||||
|
||||
auto accessInfoReply = networkAccessManager()->get( networkRequest );
|
||||
abortIfNotFinishedWithin( accessInfoReply, RiaSumoDefines::requestTimeoutMillis() );
|
||||
auto accessInfoReply = getAndTrackReply( backgroundNetworkAccessManager(), networkRequest, cancelGroup );
|
||||
|
||||
// Deliberately not a short deadline. Several of these run at once, the network manager serves only a few
|
||||
// per host at a time, and the timer runs from when a request is created rather than from when it is
|
||||
// served, so a short value aborts requests that never got a chance. The blocking path puts no deadline on
|
||||
// these at all, see downloadBlobs.
|
||||
abortIfNotFinishedWithin( accessInfoReply, RiaSumoDefines::blobLookupTimeoutMillis() );
|
||||
|
||||
QObject::connect( accessInfoReply,
|
||||
&QNetworkReply::finished,
|
||||
m_transferContext,
|
||||
[this, accessInfoReply, blobId, onFinished]()
|
||||
[this, accessInfoReply, blobId, onFinished, blobTimeoutMillis, cancelGroup]()
|
||||
{
|
||||
const QString sasUri = sasUriFromReply( accessInfoReply, blobId );
|
||||
if ( sasUri.isEmpty() )
|
||||
@@ -230,8 +293,11 @@ void RiaSumoConnector::downloadBlobAsync( const QString& blobId, const std::func
|
||||
// added. Do NOT forward the bearer token to the storage host.
|
||||
blobRequest.setAttribute( QNetworkRequest::RedirectPolicyAttribute, QNetworkRequest::NoLessSafeRedirectPolicy );
|
||||
|
||||
auto blobReply = networkAccessManager()->get( blobRequest );
|
||||
abortIfNotFinishedWithin( blobReply, RiaSumoDefines::requestTimeoutMillis() );
|
||||
auto blobReply = getAndTrackReply( backgroundNetworkAccessManager(), blobRequest, cancelGroup );
|
||||
|
||||
// A deadline on the transfer itself, so it is only applied when the caller asked
|
||||
// for one. A large blob on a slow link is not a failure.
|
||||
if ( blobTimeoutMillis != noTimeout() ) abortIfNotFinishedWithin( blobReply, blobTimeoutMillis );
|
||||
|
||||
QObject::connect( blobReply,
|
||||
&QNetworkReply::finished,
|
||||
@@ -240,6 +306,53 @@ void RiaSumoConnector::downloadBlobAsync( const QString& blobId, const std::func
|
||||
} );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Create a reply and register it for cancelGroup as one atomic step. Held under the same mutex cancelGroup
|
||||
/// locks, so a cancelGroup call from another thread either completes fully before this reply exists (in
|
||||
/// which case it correctly leaves this new, later request alone) or fully after it is tracked (in which case
|
||||
/// it finds and aborts it as usual): there is no gap in between where the reply exists but is not yet
|
||||
/// visible to cancelGroup.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QNetworkReply*
|
||||
RiaSumoConnector::getAndTrackReply( QNetworkAccessManager* networkManager, const QNetworkRequest& networkRequest, const void* groupId )
|
||||
{
|
||||
QMutexLocker locker( &m_activeRepliesMutex );
|
||||
|
||||
QNetworkReply* reply = networkManager->get( networkRequest );
|
||||
if ( groupId && reply ) m_activeReplies.emplace( groupId, reply );
|
||||
|
||||
return reply;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Abort every reply still tracked under groupId. Replies already finished (and thus removed from the
|
||||
/// registry by their QPointer clearing) are not touched, only the ones still in flight.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
void RiaSumoConnector::cancelGroup( const void* groupId )
|
||||
{
|
||||
if ( !groupId ) return;
|
||||
|
||||
std::vector<QPointer<QNetworkReply>> repliesToAbort;
|
||||
{
|
||||
QMutexLocker locker( &m_activeRepliesMutex );
|
||||
auto range = m_activeReplies.equal_range( groupId );
|
||||
for ( auto it = range.first; it != range.second; ++it )
|
||||
{
|
||||
if ( it->second ) repliesToAbort.push_back( it->second );
|
||||
}
|
||||
m_activeReplies.erase( range.first, range.second );
|
||||
}
|
||||
|
||||
for ( auto& reply : repliesToAbort )
|
||||
{
|
||||
if ( !reply ) continue;
|
||||
|
||||
// abort() must run on the thread the reply lives on. QMetaObject::invokeMethod with the default auto
|
||||
// connection marshals the call there when this is a different thread, and calls it directly otherwise.
|
||||
QMetaObject::invokeMethod( reply, &QNetworkReply::abort );
|
||||
}
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Nothing waits on an async reply, so a request that never answers would otherwise keep its data pending
|
||||
/// for good. Aborting makes the reply finish with an error, which the chain reports as a failed transfer.
|
||||
@@ -316,7 +429,7 @@ void RiaSumoConnector::runOnTransferThreadBlocking( const std::function<void()>&
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Wait until every reply has finished, or the timeout expires.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
void RiaSumoConnector::waitForRepliesToFinish( const std::vector<QNetworkReply*>& replies )
|
||||
void RiaSumoConnector::waitForRepliesToFinish( const std::vector<QNetworkReply*>& replies, int perReplyTimeoutMillis )
|
||||
{
|
||||
if ( replies.empty() ) return;
|
||||
|
||||
@@ -348,7 +461,7 @@ void RiaSumoConnector::waitForRepliesToFinish( const std::vector<QNetworkReply*>
|
||||
// given on its own. A batch must not be more likely to time out than the same requests made one
|
||||
// by one, and the server can take a while to answer: a summary vector that has not been
|
||||
// aggregated yet is produced on demand by the first request that asks for it.
|
||||
timer.start( static_cast<int>( replies.size() ) * RiaSumoDefines::requestTimeoutMillis() );
|
||||
timer.start( static_cast<int>( replies.size() ) * perReplyTimeoutMillis );
|
||||
eventLoop.exec();
|
||||
}
|
||||
|
||||
@@ -361,9 +474,9 @@ void RiaSumoConnector::waitForRepliesToFinish( const std::vector<QNetworkReply*>
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Download one blob and return its contents. The download and the wait for it run on the transfer thread.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QByteArray RiaSumoConnector::downloadBlobBlocking( const QString& blobId )
|
||||
QByteArray RiaSumoConnector::downloadBlobBlocking( const QString& blobId, const QString& description )
|
||||
{
|
||||
const auto contentsByBlobId = downloadBlobsBlocking( { blobId } );
|
||||
const auto contentsByBlobId = downloadBlobsBlocking( { blobId }, description );
|
||||
|
||||
if ( auto it = contentsByBlobId.find( blobId ); it != contentsByBlobId.end() ) return it->second;
|
||||
|
||||
@@ -374,10 +487,18 @@ QByteArray RiaSumoConnector::downloadBlobBlocking( const QString& blobId )
|
||||
/// Issue a GET and return the response body, waiting on the transfer thread. Returns an empty array when
|
||||
/// the request fails. This is the primitive the data specific requests are built from.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QByteArray RiaSumoConnector::getBlocking( const QString& url, const QString& progressText )
|
||||
QByteArray RiaSumoConnector::getBlocking( const QString& path, const QString& progressText )
|
||||
{
|
||||
requestTokenBlocking();
|
||||
|
||||
// Resolved here and not at the call site: requesting the token is what starts the service, so the
|
||||
// address only exists from this point on. Done before the work is handed to the transfer thread, as
|
||||
// waiting for the service uses the network access manager of the calling thread.
|
||||
const QString baseUrl = serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return {};
|
||||
|
||||
const QString url = baseUrl + path;
|
||||
|
||||
QByteArray body;
|
||||
|
||||
runOnTransferThreadBlocking(
|
||||
@@ -447,16 +568,21 @@ QByteArray RiaSumoConnector::replyBody( QNetworkReply* reply, const QString& url
|
||||
/// the calling thread dispatches no events while the transfers are in flight. See downloadBlobs for what
|
||||
/// the transfers actually are.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::map<QString, QByteArray> RiaSumoConnector::downloadBlobsBlocking( const std::vector<QString>& blobIds )
|
||||
std::map<QString, QByteArray> RiaSumoConnector::downloadBlobsBlocking( const std::vector<QString>& blobIds, const QString& description )
|
||||
{
|
||||
if ( blobIds.empty() ) return {};
|
||||
|
||||
requestTokenBlocking();
|
||||
|
||||
const QString baseUrl = serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return {};
|
||||
|
||||
std::map<QString, QByteArray> contentsByBlobId;
|
||||
|
||||
runOnTransferThreadBlocking( [&]() { contentsByBlobId = downloadBlobs( blobIds ); },
|
||||
QString( "Downloading %1 file(s) from Sumo" ).arg( blobIds.size() ) );
|
||||
const QString progressText = description.isEmpty() ? QString( "Downloading %1 file(s) from Sumo" ).arg( blobIds.size() )
|
||||
: QString( "Downloading %1 from Sumo" ).arg( description );
|
||||
|
||||
runOnTransferThreadBlocking( [&]() { contentsByBlobId = downloadBlobs( baseUrl, blobIds ); }, progressText );
|
||||
|
||||
return contentsByBlobId;
|
||||
}
|
||||
@@ -469,7 +595,7 @@ std::map<QString, QByteArray> RiaSumoConnector::downloadBlobsBlocking( const std
|
||||
///
|
||||
/// Always called on the transfer thread, where the event loops it waits on dispatch no GUI events.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::map<QString, QByteArray> RiaSumoConnector::downloadBlobs( const std::vector<QString>& blobIds )
|
||||
std::map<QString, QByteArray> RiaSumoConnector::downloadBlobs( const QString& baseUrl, const std::vector<QString>& blobIds )
|
||||
{
|
||||
std::map<QString, QByteArray> contentsByBlobId;
|
||||
if ( blobIds.empty() ) return contentsByBlobId;
|
||||
@@ -478,7 +604,7 @@ std::map<QString, QByteArray> RiaSumoConnector::downloadBlobs( const std::vector
|
||||
std::vector<QNetworkReply*> accessInfoReplies;
|
||||
for ( const auto& blobId : blobIds )
|
||||
{
|
||||
QString url = QString( "%1/blobs/%2/sas_token_and_blob_base_uri" ).arg( server() ).arg( blobId );
|
||||
QString url = QString( "%1/blobs/%2/sas_token_and_blob_base_uri" ).arg( baseUrl ).arg( blobId );
|
||||
|
||||
QNetworkRequest networkRequest;
|
||||
networkRequest.setUrl( QUrl( url ) );
|
||||
@@ -516,7 +642,11 @@ std::map<QString, QByteArray> RiaSumoConnector::downloadBlobs( const std::vector
|
||||
blobIndices.push_back( i );
|
||||
}
|
||||
|
||||
waitForRepliesToFinish( blobReplies );
|
||||
// Phase 2 moves the blob data itself, potentially a large grid or property array, and must not share the
|
||||
// short deadline sized for the metadata lookup above: under load from other concurrent transfers (e.g.
|
||||
// another realization's property prefetch), a real transfer can easily take longer than that. Give it the
|
||||
// same generous deadline the async grid property path uses for its own blob transfer.
|
||||
waitForRepliesToFinish( blobReplies, RiaSumoDefines::gridPropertyTransferTimeoutMillis() );
|
||||
|
||||
for ( size_t i = 0; i < blobReplies.size(); i++ )
|
||||
{
|
||||
|
||||
@@ -21,11 +21,13 @@
|
||||
#include "RiaCloudConnector.h"
|
||||
#include "RiaSumoDefines.h"
|
||||
#include "RiaSumoExplore.h"
|
||||
#include "RiaSumoGrid.h"
|
||||
#include "RiaSumoSummary.h"
|
||||
|
||||
#include <QByteArray>
|
||||
#include <QMutex>
|
||||
#include <QNetworkAccessManager>
|
||||
#include <QPointer>
|
||||
#include <QtNetworkAuth/QOAuth2AuthorizationCodeFlow>
|
||||
|
||||
#include <functional>
|
||||
@@ -59,18 +61,30 @@ public:
|
||||
|
||||
// Download blobs by id and return their contents. Getting a blob takes two round trips, one for the
|
||||
// pre-signed URI and one for the data, and a batch does each of those as one concurrent group.
|
||||
QByteArray downloadBlobBlocking( const QString& blobId );
|
||||
std::map<QString, QByteArray> downloadBlobsBlocking( const std::vector<QString>& blobIds );
|
||||
// The description names what is being fetched in the progress dialog, e.g. "grid DROGON realization 1".
|
||||
// Falls back to a plain file count when it is empty.
|
||||
QByteArray downloadBlobBlocking( const QString& blobId, const QString& description = {} );
|
||||
std::map<QString, QByteArray> downloadBlobsBlocking( const std::vector<QString>& blobIds, const QString& description = {} );
|
||||
|
||||
// What Sumo holds: assets, cases, ensembles and realizations.
|
||||
RiaSumoExplore& explore();
|
||||
|
||||
// The grid data of a case. Owned here so its blob cache lives as long as the connection.
|
||||
RiaSumoGrid& grid();
|
||||
|
||||
// The summary data of a case.
|
||||
RiaSumoSummary& summary();
|
||||
|
||||
// Transport used by the data specific delegates. Every request goes through the transfer thread, so
|
||||
// the calling thread waits without dispatching events.
|
||||
QByteArray getBlocking( const QString& url, const QString& progressText = {} );
|
||||
// Issue a GET against the local cloud API service and return the response body. Takes the path only,
|
||||
// e.g. "/cases/{id}/ensembles": the base URL is prepended here, once the service is known to answer.
|
||||
// Composing the full URL at the call site would capture the address before the service has one.
|
||||
QByteArray getBlocking( const QString& path, const QString& progressText = {} );
|
||||
|
||||
// The base URL of the local cloud API service, empty when it could not be made ready. May block while
|
||||
// the service starts, so call it from the calling thread and not from the transfer thread.
|
||||
QString serviceBaseUrl();
|
||||
|
||||
// The REST API returns a blob id as a plain string, quoted by FastAPI.
|
||||
static QString blobIdFromBody( const QByteArray& body );
|
||||
@@ -90,25 +104,64 @@ public:
|
||||
// async work are delivered through this, so a caller never has its data handed to it on another thread.
|
||||
void invokeOnConnectorThread( const std::function<void()>& work );
|
||||
|
||||
// Download one blob, calling onFinished with its contents. Call on the transfer thread, where onFinished
|
||||
// is called as well. Empty contents mean the transfer failed.
|
||||
void downloadBlobAsync( const QString& blobId, const std::function<void( const QByteArray& )>& onFinished );
|
||||
// Download one blob. Call on the transfer thread, where onFinished is called as well, exactly once on
|
||||
// every path. Empty contents mean the transfer failed.
|
||||
//
|
||||
// blobTimeoutMillis is a deadline on the transfer itself, not an inactivity timeout: pass noTimeout() for
|
||||
// a blob large enough that a slow link is not a failure. The blob id request keeps its own deadline.
|
||||
//
|
||||
// cancelGroup lets the owner of the async request abort it early, see getAndTrackReply/cancelGroup: pass
|
||||
// the requesting object's lifetime token so its transfers are aborted, not just ignored, once it is gone.
|
||||
void downloadBlobAsync( const QString& blobId,
|
||||
const std::function<void( const QByteArray& )>& onFinished,
|
||||
int blobTimeoutMillis = RiaSumoDefines::requestTimeoutMillis(),
|
||||
const void* cancelGroup = nullptr );
|
||||
|
||||
// Issue a GET request and register the reply for cancelGroup in one atomic step, guarded by the same
|
||||
// mutex cancelGroup locks. Creating the reply and tracking it as two separate calls left a gap in which
|
||||
// a cancelGroup call landing on another thread in between would not find the reply yet, and miss it.
|
||||
// groupId identifies the owner requesting the transfer, typically its m_lifetimeToken.get(); a null
|
||||
// groupId tracks nothing. The reply is dropped from the registry automatically once it is deleted.
|
||||
QNetworkReply* getAndTrackReply( QNetworkAccessManager* networkManager, const QNetworkRequest& networkRequest, const void* groupId );
|
||||
|
||||
// Abort every reply still tracked under groupId. Called when the owner of an async request (a reader or
|
||||
// an ensemble) is destroyed, so its in-flight transfers stop consuming connections and logging results
|
||||
// for an object that is no longer there, instead of running until their own timeout.
|
||||
void cancelGroup( const void* groupId );
|
||||
|
||||
// Timeout value meaning "never abort on a deadline".
|
||||
static constexpr int noTimeout() { return -1; }
|
||||
|
||||
// Abort a reply that has not finished in time, so an async chain reports a failure instead of hanging and
|
||||
// leaving whoever waits for the data waiting forever.
|
||||
static void abortIfNotFinishedWithin( QNetworkReply* reply, int timeoutMillis );
|
||||
|
||||
// The network manager belonging to the calling thread: the transfer thread manager when called from
|
||||
// there, otherwise the one owned by RiaCloudConnector on the GUI thread.
|
||||
// there, otherwise the one owned by RiaCloudConnector on the GUI thread. Used for interactive requests:
|
||||
// opening a case, the grid geometry and metadata, and any request the user is waiting on.
|
||||
QNetworkAccessManager* networkAccessManager();
|
||||
|
||||
// A second network manager on the transfer thread, used only by the background prefetch paths
|
||||
// (downloadBlobAsync and the blob id requests feeding it). Qt limits concurrent connections per host
|
||||
// separately for each QNetworkAccessManager, so giving prefetch its own manager keeps a burst of
|
||||
// look-ahead property transfers from starving an interactive request for a different realization or
|
||||
// case, which would otherwise queue behind them and fail on its own short deadline.
|
||||
QNetworkAccessManager* backgroundNetworkAccessManager();
|
||||
|
||||
// The token for a request issued from the transfer thread. token() reads objects owned by another thread.
|
||||
QString transferToken() const;
|
||||
|
||||
static void waitForRepliesToFinish( const std::vector<QNetworkReply*>& replies );
|
||||
// perReplyTimeoutMillis is the deadline given to each individual reply were it waited on alone; the group
|
||||
// as a whole is allowed replies.size() times that, see the implementation. Defaults to requestTimeoutMillis,
|
||||
// right for the small metadata requests (e.g. the SAS token lookup); pass a longer value for a phase that
|
||||
// moves real data, such as the blob transfer itself, so it is not cut off by a deadline sized for a lookup.
|
||||
static void waitForRepliesToFinish( const std::vector<QNetworkReply*>& replies,
|
||||
int perReplyTimeoutMillis = RiaSumoDefines::requestTimeoutMillis() );
|
||||
|
||||
// Issue and collect the two round trips a blob transfer needs. Called on the transfer thread.
|
||||
std::map<QString, QByteArray> downloadBlobs( const std::vector<QString>& blobIds );
|
||||
// Runs on the transfer thread, so the base URL is resolved by the caller and passed in: waiting for the
|
||||
// service must not happen here.
|
||||
std::map<QString, QByteArray> downloadBlobs( const QString& baseUrl, const std::vector<QString>& blobIds );
|
||||
|
||||
public slots:
|
||||
void requestFailed( const QAbstractOAuth::Error error );
|
||||
@@ -127,6 +180,7 @@ private:
|
||||
std::function<QString()> m_serverUrlProvider;
|
||||
|
||||
RiaSumoExplore m_explore;
|
||||
RiaSumoGrid m_grid;
|
||||
RiaSumoSummary m_summary;
|
||||
|
||||
// Transfers run on their own thread so the calling thread can wait without dispatching events. Waiting on
|
||||
@@ -137,7 +191,16 @@ private:
|
||||
QObject* m_transferContext = nullptr; // lives on the transfer thread
|
||||
QNetworkAccessManager* m_transferNetworkAccessManager = nullptr; // created on the transfer thread
|
||||
|
||||
// A second manager on the transfer thread, dedicated to background prefetch transfers, see
|
||||
// backgroundNetworkAccessManager().
|
||||
QNetworkAccessManager* m_transferBackgroundNetworkAccessManager = nullptr; // created on the transfer thread
|
||||
|
||||
// Written on the thread handing work over, read on the transfer thread running it.
|
||||
mutable QMutex m_transferTokenMutex;
|
||||
QString m_transferToken;
|
||||
|
||||
// Replies tracked so cancelGroup can abort them, keyed by the owner's group id. QPointer clears itself
|
||||
// when a reply is deleted, so a finished transfer needs no explicit removal beyond that.
|
||||
QMutex m_activeRepliesMutex;
|
||||
std::multimap<const void*, QPointer<QNetworkReply>> m_activeReplies;
|
||||
};
|
||||
@@ -44,3 +44,27 @@ int RiaSumoDefines::asyncRequestTimeoutMillis()
|
||||
{
|
||||
return 5 * 60 * 1000;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
int RiaSumoDefines::blobLookupTimeoutMillis()
|
||||
{
|
||||
return 2 * 60 * 1000;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
int RiaSumoDefines::gridPropertyTransferTimeoutMillis()
|
||||
{
|
||||
return 5 * 60 * 1000;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
int RiaSumoDefines::serviceReadyTimeoutMillis()
|
||||
{
|
||||
return 20 * 1000;
|
||||
}
|
||||
@@ -36,4 +36,17 @@ int requestTimeoutMillis();
|
||||
// given up on, and generous because a summary vector that has not been aggregated yet is produced on demand
|
||||
// by the request asking for it. Nothing is blocked while it runs, so waiting longer costs nothing.
|
||||
int asyncRequestTimeoutMillis();
|
||||
|
||||
// Deadline for the small requests that resolve which blob to fetch and where it lives. Shorter than the
|
||||
// transfer that follows: they move almost no data, so taking minutes means the answer is not coming.
|
||||
int blobLookupTimeoutMillis();
|
||||
|
||||
// Deadline for one grid property transfer. Long enough that a large blob on a slow link is not cut off,
|
||||
// and short enough that a stalled transfer fails instead of leaving the time step blank for the rest of
|
||||
// the session. A failed step is not retried, so erring on the generous side is the cheaper mistake.
|
||||
int gridPropertyTransferTimeoutMillis();
|
||||
|
||||
// How long to wait for the local cloud API service to answer before giving up on a request. Long enough
|
||||
// to cover a cold uvicorn boot, short enough that a service that will never come up does not hang the UI.
|
||||
int serviceReadyTimeoutMillis();
|
||||
}; // namespace RiaSumoDefines
|
||||
@@ -40,9 +40,7 @@ RiaSumoExplore::RiaSumoExplore( RiaSumoConnector& connector )
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<SumoAsset> RiaSumoExplore::assets()
|
||||
{
|
||||
const QString url = QString( "%1/assets" ).arg( m_connector.server() );
|
||||
|
||||
return parseAssets( m_connector.getBlocking( url, "Loading assets from Sumo" ) );
|
||||
return parseAssets( m_connector.getBlocking( "/assets", "Loading assets from Sumo" ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -50,9 +48,9 @@ std::vector<SumoAsset> RiaSumoExplore::assets()
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<SumoCase> RiaSumoExplore::cases( const QString& assetName )
|
||||
{
|
||||
const QString url = QString( "%1/cases?asset_name=%2" ).arg( m_connector.server() ).arg( QString( QUrl::toPercentEncoding( assetName ) ) );
|
||||
const QString path = QString( "/cases?asset_name=%1" ).arg( QString( QUrl::toPercentEncoding( assetName ) ) );
|
||||
|
||||
return parseCases( m_connector.getBlocking( url, QString( "Loading the cases of %1 from Sumo" ).arg( assetName ) ) );
|
||||
return parseCases( m_connector.getBlocking( path, QString( "Loading the cases of %1 from Sumo" ).arg( assetName ) ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -60,9 +58,9 @@ std::vector<SumoCase> RiaSumoExplore::cases( const QString& assetName )
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<QString> RiaSumoExplore::ensembleNames( const SumoCaseId& caseId )
|
||||
{
|
||||
const QString url = QString( "%1/cases/%2/ensembles" ).arg( m_connector.server() ).arg( caseId.get() );
|
||||
const QString path = QString( "/cases/%1/ensembles" ).arg( caseId.get() );
|
||||
|
||||
return parseEnsembleNames( m_connector.getBlocking( url, "Loading ensembles from Sumo" ) );
|
||||
return parseEnsembleNames( m_connector.getBlocking( path, "Loading ensembles from Sumo" ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -72,10 +70,9 @@ std::vector<QString> RiaSumoExplore::realizationIds( const SumoCaseId& caseId, c
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
|
||||
const QString url =
|
||||
QString( "%1/cases/%2/ensembles/%3/realizations" ).arg( m_connector.server() ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/realizations" ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
|
||||
return parseRealizationIds( m_connector.getBlocking( url, "Loading realizations from Sumo" ) );
|
||||
return parseRealizationIds( m_connector.getBlocking( path, "Loading realizations from Sumo" ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,558 @@
|
||||
/////////////////////////////////////////////////////////////////////////////////
|
||||
//
|
||||
// Copyright (C) 2026- Equinor ASA
|
||||
//
|
||||
// ResInsight is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY
|
||||
// WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
// FITNESS FOR A PARTICULAR PURPOSE.
|
||||
//
|
||||
// See the GNU General Public License at <http://www.gnu.org/licenses/gpl.html>
|
||||
// for more details.
|
||||
//
|
||||
/////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
#include "RiaSumoGrid.h"
|
||||
|
||||
#include "RiaCloudDefines.h"
|
||||
#include "RiaLogging.h"
|
||||
#include "RiaSumoConnector.h"
|
||||
#include "RiaSumoDefines.h"
|
||||
|
||||
#include <QEventLoop>
|
||||
#include <QJsonArray>
|
||||
#include <QJsonDocument>
|
||||
#include <QJsonObject>
|
||||
#include <QNetworkAccessManager>
|
||||
#include <QNetworkReply>
|
||||
#include <QTimer>
|
||||
#include <QUrl>
|
||||
|
||||
#include <algorithm>
|
||||
#include <optional>
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<int> SumoGridInfo::realizationIds() const
|
||||
{
|
||||
std::vector<int> ids;
|
||||
ids.reserve( realizationInfos.size() );
|
||||
|
||||
for ( const auto& realizationInfo : realizationInfos )
|
||||
{
|
||||
ids.push_back( realizationInfo.realization );
|
||||
}
|
||||
|
||||
return ids;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
bool SumoGridInfo::hasIdenticalDimensions( const std::vector<int>& realizations ) const
|
||||
{
|
||||
if ( realizations.empty() ) return false;
|
||||
|
||||
std::optional<SumoGridDimensions> referenceDimensions;
|
||||
|
||||
for ( int realization : realizations )
|
||||
{
|
||||
auto it = std::ranges::find( realizationInfos, realization, &SumoGridRealizationInfo::realization );
|
||||
|
||||
// A realization this grid does not have, or one the server reported no dimensions for, makes the
|
||||
// comparison inconclusive. Report not identical, so a grid is loaded per realization.
|
||||
if ( it == realizationInfos.end() || !it->dimensions.isValid() ) return false;
|
||||
|
||||
if ( !referenceDimensions )
|
||||
{
|
||||
referenceDimensions = it->dimensions;
|
||||
}
|
||||
else if ( it->dimensions != *referenceDimensions )
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
RiaSumoGrid::RiaSumoGrid( RiaSumoConnector& connector )
|
||||
: m_connector( connector )
|
||||
{
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<QString> RiaSumoGrid::gridNames( const SumoCaseId& caseId, const QString& ensembleName )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/grid_names" ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
|
||||
return parseGridNames( m_connector.getBlocking( path ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
SumoGridInfo RiaSumoGrid::gridInfo( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
const QString encodedGridName = QUrl::toPercentEncoding( gridName );
|
||||
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/grid_info/%3" ).arg( caseId.get() ).arg( encodedEnsembleName ).arg( encodedGridName );
|
||||
|
||||
auto gridInfo = parseGridInfo( m_connector.getBlocking( path ) );
|
||||
|
||||
// An empty result also means the request failed: the response body is discarded on any HTTP error.
|
||||
RiaLogging::debug( std::format( "Grid '{}' realization count : {}", gridName.toStdString(), gridInfo.realizationInfos.size() ) );
|
||||
|
||||
return gridInfo;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QByteArray RiaSumoGrid::gridData( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization )
|
||||
{
|
||||
const QString blobId = gridBlobId( caseId, ensembleName, gridName, realization );
|
||||
if ( blobId.isEmpty() ) return {};
|
||||
|
||||
QByteArray contents = m_connector.downloadBlobBlocking( blobId, QString( "grid %1 realization %2" ).arg( gridName ).arg( realization ) );
|
||||
|
||||
// Every realization downloads its own grid, also in shared grid mode, as the roff blob is the only
|
||||
// source of the active cells of a realization. Sharing a grid saves memory, not transfers.
|
||||
RiaLogging::debug( std::format( "Downloaded grid '{}' realization {}: {} bytes", gridName.toStdString(), realization, contents.size() ) );
|
||||
|
||||
return contents;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoGrid::gridBlobId( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
const QString encodedGridName = QUrl::toPercentEncoding( gridName );
|
||||
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/grids/%3/realizations/%4/blob_id" )
|
||||
.arg( caseId.get() )
|
||||
.arg( encodedEnsembleName )
|
||||
.arg( encodedGridName )
|
||||
.arg( realization );
|
||||
|
||||
return blobIdFromBody( m_connector.getBlocking( path ), "grid", gridName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<SumoGridPropertyInfo>
|
||||
RiaSumoGrid::propertyInfo( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
const QString encodedGridName = QUrl::toPercentEncoding( gridName );
|
||||
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/grids/%3/realizations/%4/property_info_list" )
|
||||
.arg( caseId.get() )
|
||||
.arg( encodedEnsembleName )
|
||||
.arg( encodedGridName )
|
||||
.arg( realization );
|
||||
|
||||
return parsePropertyInfo( m_connector.getBlocking( path ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QByteArray RiaSumoGrid::propertyData( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval )
|
||||
{
|
||||
const QString blobId = propertyBlobId( caseId, ensembleName, gridName, realization, propertyName, isoDateOrInterval );
|
||||
if ( blobId.isEmpty() ) return {};
|
||||
|
||||
const QString description = isoDateOrInterval.isEmpty()
|
||||
? QString( "%1 realization %2" ).arg( propertyName ).arg( realization )
|
||||
: QString( "%1 %2 realization %3" ).arg( propertyName, isoDateOrInterval ).arg( realization );
|
||||
|
||||
return m_connector.downloadBlobBlocking( blobId, description );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::map<QString, QByteArray> RiaSumoGrid::propertyDataBatch( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& isoDatesOrIntervals )
|
||||
{
|
||||
// Drop duplicates so a time step is never requested twice in the same batch.
|
||||
std::vector<QString> timestampsToFetch;
|
||||
for ( const auto& isoDateOrInterval : isoDatesOrIntervals )
|
||||
{
|
||||
if ( std::ranges::find( timestampsToFetch, isoDateOrInterval ) != timestampsToFetch.end() ) continue;
|
||||
|
||||
timestampsToFetch.push_back( isoDateOrInterval );
|
||||
}
|
||||
|
||||
if ( timestampsToFetch.empty() ) return {};
|
||||
|
||||
// The requests below are issued directly on the transfer thread, so the token and the address of the
|
||||
// service have to be in place before the work is handed over.
|
||||
m_connector.requestTokenBlocking();
|
||||
|
||||
const QString baseUrl = m_connector.serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return {};
|
||||
|
||||
std::map<QString, QByteArray> contentsByTimestamp;
|
||||
|
||||
m_connector.runOnTransferThreadBlocking(
|
||||
[&]()
|
||||
{
|
||||
contentsByTimestamp = fetchPropertyBatch( baseUrl, caseId, ensembleName, gridName, realization, propertyName, timestampsToFetch );
|
||||
},
|
||||
QString( "Loading %1 time step(s) of %2 from Sumo" ).arg( timestampsToFetch.size() ).arg( propertyName ) );
|
||||
|
||||
return contentsByTimestamp;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// The async twin of propertyDataBatch, built on the same blob id request. Every time step is requested at
|
||||
/// once and delivered on the connector thread as it arrives, so nothing waits for the slowest one.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
void RiaSumoGrid::propertyDataBatchAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& isoDatesOrIntervals,
|
||||
const std::function<void( const QString&, const QByteArray& )>& onTimeStepReady,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
if ( isoDatesOrIntervals.empty() || !onTimeStepReady ) return;
|
||||
|
||||
// Resolved here and not on the transfer thread: requesting the token is what starts the service, and
|
||||
// waiting for it uses the network access manager of the calling thread.
|
||||
const QString baseUrl = m_connector.serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() )
|
||||
{
|
||||
// Report every time step as failed rather than returning silently, so the caller is not left waiting
|
||||
// for replies that will never come.
|
||||
for ( const auto& isoDateOrInterval : isoDatesOrIntervals )
|
||||
{
|
||||
m_connector.invokeOnConnectorThread( [onTimeStepReady, isoDateOrInterval]() { onTimeStepReady( isoDateOrInterval, {} ); } );
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
m_connector.runOnTransferThread(
|
||||
[this, baseUrl, caseId, ensembleName, gridName, realization, propertyName, isoDatesOrIntervals, onTimeStepReady, cancelGroup]()
|
||||
{
|
||||
for ( const auto& isoDateOrInterval : isoDatesOrIntervals )
|
||||
{
|
||||
auto deliver = [this, onTimeStepReady, isoDateOrInterval]( const QByteArray& contents )
|
||||
{
|
||||
m_connector.invokeOnConnectorThread( [onTimeStepReady, isoDateOrInterval, contents]()
|
||||
{ onTimeStepReady( isoDateOrInterval, contents ); } );
|
||||
};
|
||||
|
||||
auto blobIdReply = makePropertyBlobIdRequest( baseUrl,
|
||||
caseId,
|
||||
ensembleName,
|
||||
gridName,
|
||||
realization,
|
||||
propertyName,
|
||||
isoDateOrInterval,
|
||||
m_connector.backgroundNetworkAccessManager(),
|
||||
cancelGroup );
|
||||
if ( !blobIdReply )
|
||||
{
|
||||
deliver( {} );
|
||||
continue;
|
||||
}
|
||||
|
||||
// Shorter than the transfer it precedes: this one moves almost no data, so minutes really
|
||||
// does mean broken.
|
||||
RiaSumoConnector::abortIfNotFinishedWithin( blobIdReply, RiaSumoDefines::blobLookupTimeoutMillis() );
|
||||
|
||||
QObject::connect( blobIdReply,
|
||||
&QNetworkReply::finished,
|
||||
blobIdReply,
|
||||
[this, blobIdReply, propertyName, deliver, cancelGroup]()
|
||||
{
|
||||
const QString blobId = blobIdFromReply( blobIdReply, propertyName );
|
||||
if ( blobId.isEmpty() )
|
||||
{
|
||||
deliver( {} );
|
||||
return;
|
||||
}
|
||||
|
||||
m_connector.downloadBlobAsync( blobId,
|
||||
deliver,
|
||||
RiaSumoDefines::gridPropertyTransferTimeoutMillis(),
|
||||
cancelGroup );
|
||||
} );
|
||||
}
|
||||
} );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Resolve the blob ids of a batch of time steps and download the blobs. Always called on the transfer
|
||||
/// thread, where the event loops it waits on dispatch no GUI events. The contents are returned rather
|
||||
/// than retained, so the caller decides what to keep.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::map<QString, QByteArray> RiaSumoGrid::fetchPropertyBatch( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& timestampsToFetch )
|
||||
{
|
||||
// Phase 1: resolve all blob ids concurrently.
|
||||
std::vector<QNetworkReply*> blobIdReplies;
|
||||
for ( const auto& isoDateOrInterval : timestampsToFetch )
|
||||
{
|
||||
blobIdReplies.push_back( makePropertyBlobIdRequest( baseUrl,
|
||||
caseId,
|
||||
ensembleName,
|
||||
gridName,
|
||||
realization,
|
||||
propertyName,
|
||||
isoDateOrInterval,
|
||||
m_connector.networkAccessManager() ) );
|
||||
}
|
||||
|
||||
RiaSumoConnector::waitForRepliesToFinish( blobIdReplies );
|
||||
|
||||
std::vector<QString> blobIds;
|
||||
for ( auto reply : blobIdReplies )
|
||||
{
|
||||
blobIds.push_back( blobIdFromReply( reply, propertyName ) );
|
||||
}
|
||||
|
||||
// Phase 2: download all resolved blobs as one group.
|
||||
std::vector<QString> blobIdsToDownload;
|
||||
for ( const auto& blobId : blobIds )
|
||||
{
|
||||
if ( !blobId.isEmpty() ) blobIdsToDownload.push_back( blobId );
|
||||
}
|
||||
|
||||
const auto contentsByBlobId = m_connector.downloadBlobs( baseUrl, blobIdsToDownload );
|
||||
|
||||
// Anything missing failed to download, and is left out so the caller can fall back to a single request.
|
||||
std::map<QString, QByteArray> contentsByTimestamp;
|
||||
for ( size_t i = 0; i < blobIds.size(); i++ )
|
||||
{
|
||||
if ( blobIds[i].isEmpty() ) continue;
|
||||
|
||||
if ( auto it = contentsByBlobId.find( blobIds[i] ); it != contentsByBlobId.end() )
|
||||
{
|
||||
contentsByTimestamp[timestampsToFetch[i]] = it->second;
|
||||
}
|
||||
}
|
||||
|
||||
return contentsByTimestamp;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoGrid::propertyBlobIdPath( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
const QString encodedGridName = QUrl::toPercentEncoding( gridName );
|
||||
const QString encodedPropertyName = QUrl::toPercentEncoding( propertyName );
|
||||
|
||||
QString path = QString( "/cases/%1/ensembles/%2/grids/%3/realizations/%4/properties/%5/blob_id" )
|
||||
.arg( caseId.get() )
|
||||
.arg( encodedEnsembleName )
|
||||
.arg( encodedGridName )
|
||||
.arg( realization )
|
||||
.arg( encodedPropertyName );
|
||||
|
||||
// The timestamp/interval is an optional query parameter; omit it for static properties.
|
||||
if ( !isoDateOrInterval.isEmpty() )
|
||||
{
|
||||
path += QString( "?property_iso_date_or_interval=%1" ).arg( QString( QUrl::toPercentEncoding( isoDateOrInterval ) ) );
|
||||
}
|
||||
|
||||
return path;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoGrid::propertyBlobId( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval )
|
||||
{
|
||||
const QString path = propertyBlobIdPath( caseId, ensembleName, gridName, realization, propertyName, isoDateOrInterval );
|
||||
|
||||
return blobIdFromBody( m_connector.getBlocking( path ), "grid property", propertyName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Issue the blob id request for one grid property time step. The reply is returned unfinished, so the
|
||||
/// caller decides how to wait for it: one at a time, or several at once when prefetching. networkManager
|
||||
/// picks the connection pool, see the header comment.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QNetworkReply* RiaSumoGrid::makePropertyBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
const QString url = baseUrl + propertyBlobIdPath( caseId, ensembleName, gridName, realization, propertyName, isoDateOrInterval );
|
||||
|
||||
QNetworkRequest networkRequest;
|
||||
networkRequest.setUrl( QUrl( url ) );
|
||||
m_connector.addStandardHeader( networkRequest, m_connector.transferToken(), RiaCloudDefines::contentTypeJson() );
|
||||
|
||||
return m_connector.getAndTrackReply( networkManager, networkRequest, cancelGroup );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Read the blob id off a finished blob id reply. The reply is consumed and scheduled for deletion.
|
||||
///
|
||||
/// Waiting for one specific reply, rather than for a signal shared by all blob id requests, is what makes
|
||||
/// the mapping correct: a still-pending reply from an earlier property's request can no longer satisfy
|
||||
/// this wait and hand us its blob id, which previously caused e.g. SWAT to be served the SWCR blob.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoGrid::blobIdFromReply( QNetworkReply* reply, const QString& propertyName )
|
||||
{
|
||||
if ( !reply ) return {};
|
||||
|
||||
const bool failed = !reply->isFinished() || reply->error() != QNetworkReply::NoError;
|
||||
QByteArray body = failed ? QByteArray() : reply->readAll();
|
||||
|
||||
if ( failed )
|
||||
{
|
||||
RiaLogging::error( std::format( "Request grid property blob ID failed: '{}'", reply->errorString().toStdString() ) );
|
||||
}
|
||||
|
||||
reply->deleteLater();
|
||||
|
||||
return blobIdFromBody( body, "grid property", propertyName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoGrid::blobIdFromBody( const QByteArray& body, const QString& kind, const QString& name )
|
||||
{
|
||||
const QString blobId = RiaSumoConnector::blobIdFromBody( body );
|
||||
if ( blobId.isEmpty() ) return {};
|
||||
|
||||
RiaLogging::debug( std::format( "Received blob ID for {} '{}': {}", kind.toStdString(), name.toStdString(), blobId.toStdString() ) );
|
||||
|
||||
return blobId;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<QString> RiaSumoGrid::parseGridNames( const QByteArray& body )
|
||||
{
|
||||
std::vector<QString> gridNames;
|
||||
|
||||
QJsonDocument doc = QJsonDocument::fromJson( body );
|
||||
QJsonArray jsonArray = doc.array();
|
||||
|
||||
for ( const QJsonValue& value : jsonArray )
|
||||
{
|
||||
QString gridName = value.toString();
|
||||
|
||||
gridNames.push_back( gridName );
|
||||
}
|
||||
|
||||
RiaLogging::debug( std::format( "Grid names count : {}", gridNames.size() ) );
|
||||
|
||||
return gridNames;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
SumoGridInfo RiaSumoGrid::parseGridInfo( const QByteArray& body )
|
||||
{
|
||||
SumoGridInfo gridInfo;
|
||||
|
||||
QJsonDocument doc = QJsonDocument::fromJson( body );
|
||||
QJsonArray jsonArray = doc.array();
|
||||
|
||||
for ( const QJsonValue& value : jsonArray )
|
||||
{
|
||||
const QJsonObject realizationObj = value.toObject();
|
||||
|
||||
// A missing or null dimensions object leaves the counts at zero, which isValid() rejects, so a
|
||||
// realization Sumo has no dimensions for falls back to a grid of its own.
|
||||
const QJsonObject dimensionsObj = realizationObj["dimensions"].toObject();
|
||||
|
||||
SumoGridRealizationInfo realizationInfo;
|
||||
realizationInfo.realization = realizationObj["realization"].toInt( -1 );
|
||||
realizationInfo.dimensions.iCount = dimensionsObj["iCount"].toInt();
|
||||
realizationInfo.dimensions.jCount = dimensionsObj["jCount"].toInt();
|
||||
realizationInfo.dimensions.kCount = dimensionsObj["kCount"].toInt();
|
||||
|
||||
gridInfo.realizationInfos.push_back( realizationInfo );
|
||||
}
|
||||
|
||||
return gridInfo;
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
std::vector<SumoGridPropertyInfo> RiaSumoGrid::parsePropertyInfo( const QByteArray& body )
|
||||
{
|
||||
std::vector<SumoGridPropertyInfo> propertyInfos;
|
||||
|
||||
QJsonDocument doc = QJsonDocument::fromJson( body );
|
||||
QJsonArray jsonArray = doc.array();
|
||||
|
||||
for ( const QJsonValue& value : jsonArray )
|
||||
{
|
||||
QJsonObject propertyObj = value.toObject();
|
||||
|
||||
SumoGridPropertyInfo propertyInfo;
|
||||
propertyInfo.name = propertyObj["propertyName"].toString();
|
||||
|
||||
// isoDateOrInterval is null for static properties.
|
||||
const auto isoValue = propertyObj["isoDateOrInterval"];
|
||||
if ( !isoValue.isNull() ) propertyInfo.isoDateOrInterval = isoValue.toString();
|
||||
|
||||
propertyInfos.push_back( propertyInfo );
|
||||
}
|
||||
|
||||
RiaLogging::debug( std::format( "Grid property info count : {}", propertyInfos.size() ) );
|
||||
|
||||
return propertyInfos;
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
/////////////////////////////////////////////////////////////////////////////////
|
||||
//
|
||||
// Copyright (C) 2024- Equinor ASA
|
||||
//
|
||||
// ResInsight is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// ResInsight is distributed in the hope that it will be useful, but WITHOUT ANY
|
||||
// WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
// FITNESS FOR A PARTICULAR PURPOSE.
|
||||
//
|
||||
// See the GNU General Public License at <http://www.gnu.org/licenses/gpl.html>
|
||||
// for more details.
|
||||
//
|
||||
/////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
#pragma once
|
||||
|
||||
#include "RiaSumoDefines.h"
|
||||
|
||||
#include <QByteArray>
|
||||
#include <QString>
|
||||
|
||||
#include <functional>
|
||||
#include <map>
|
||||
#include <vector>
|
||||
|
||||
class RiaSumoConnector;
|
||||
class QNetworkReply;
|
||||
class QNetworkAccessManager;
|
||||
|
||||
struct SumoGridDimensions
|
||||
{
|
||||
int iCount = 0;
|
||||
int jCount = 0;
|
||||
int kCount = 0;
|
||||
|
||||
bool isValid() const { return iCount > 0 && jCount > 0 && kCount > 0; }
|
||||
|
||||
bool operator==( const SumoGridDimensions& ) const = default;
|
||||
};
|
||||
|
||||
struct SumoGridRealizationInfo
|
||||
{
|
||||
int realization = -1;
|
||||
SumoGridDimensions dimensions;
|
||||
};
|
||||
|
||||
//==================================================================================================
|
||||
/// The realizations one grid of an ensemble exists for, each with the IJK dimensions Sumo reports. The
|
||||
/// grid is identified by the request, so its name is not repeated here.
|
||||
//==================================================================================================
|
||||
struct SumoGridInfo
|
||||
{
|
||||
std::vector<SumoGridRealizationInfo> realizationInfos;
|
||||
|
||||
// The realizations this grid is available for.
|
||||
std::vector<int> realizationIds() const;
|
||||
|
||||
// True when all the given realizations report the same valid IJK dimensions, which is what makes it
|
||||
// safe to share one grid between them. False when any of them is missing from this grid or reports
|
||||
// invalid dimensions, so the caller falls back to loading a grid per realization.
|
||||
bool hasIdenticalDimensions( const std::vector<int>& realizations ) const;
|
||||
};
|
||||
|
||||
struct SumoGridPropertyInfo
|
||||
{
|
||||
QString name;
|
||||
|
||||
// Empty for a static property. For a dynamic property this is either a single timestamp ("2018-01-01")
|
||||
// or an interval ("2018-01-01/2019-01-01"). ResInsight currently only supports the single-timestamp form.
|
||||
QString isoDateOrInterval;
|
||||
};
|
||||
|
||||
//==================================================================================================
|
||||
/// The grid data of a Sumo case: the grids of an ensemble, the grid geometry itself, and the grid
|
||||
/// properties. Requests are made through RiaSumoConnector, which owns the connection and does the
|
||||
/// transfers, and every call returns its result rather than leaving it in shared state.
|
||||
//==================================================================================================
|
||||
class RiaSumoGrid
|
||||
{
|
||||
public:
|
||||
explicit RiaSumoGrid( RiaSumoConnector& connector );
|
||||
|
||||
// The names of the grids available in an ensemble.
|
||||
std::vector<QString> gridNames( const SumoCaseId& caseId, const QString& ensembleName );
|
||||
|
||||
// The realizations one grid exists for, each with the IJK dimensions Sumo reports for it. Used by the
|
||||
// reservoir grid ensemble, which needs the dimensions to decide whether the realizations can share a grid.
|
||||
SumoGridInfo gridInfo( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName );
|
||||
|
||||
// The grid geometry as a binary roff blob.
|
||||
QByteArray gridData( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization );
|
||||
|
||||
std::vector<SumoGridPropertyInfo>
|
||||
propertyInfo( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization );
|
||||
|
||||
// One time step of one grid property, as a binary roff blob.
|
||||
QByteArray propertyData( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval );
|
||||
|
||||
// Several time steps of one property, downloaded as one concurrent batch. The contents are keyed by
|
||||
// the time step they belong to, and a time step that failed to download is absent from the result.
|
||||
//
|
||||
// Nothing is retained here: the caller decodes the contents and keeps the values, see
|
||||
// RifReaderSumoGridProperty.
|
||||
std::map<QString, QByteArray> propertyDataBatch( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& isoDatesOrIntervals );
|
||||
|
||||
// The async twin of propertyDataBatch. All time steps are requested at once and onTimeStepReady is called
|
||||
// for each as it arrives, on the connector thread, exactly once per requested step. Empty contents mean
|
||||
// that step failed. Transfers get a generous deadline, see
|
||||
// RiaSumoDefines::gridPropertyTransferTimeoutMillis.
|
||||
//
|
||||
// cancelGroup is passed on to RiaSumoConnector::downloadBlobAsync, see its documentation.
|
||||
void propertyDataBatchAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& isoDatesOrIntervals,
|
||||
const std::function<void( const QString&, const QByteArray& )>& onTimeStepReady,
|
||||
const void* cancelGroup = nullptr );
|
||||
|
||||
private:
|
||||
QString gridBlobId( const SumoCaseId& caseId, const QString& ensembleName, const QString& gridName, int realization );
|
||||
|
||||
QString propertyBlobId( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval );
|
||||
|
||||
// Runs on the transfer thread, so the base URL is resolved by the caller and passed in. networkManager
|
||||
// lets the caller choose the connection pool: the shared one for a blocking fetch waited on directly, or
|
||||
// the background one for a prefetch batch, see RiaSumoConnector::backgroundNetworkAccessManager.
|
||||
// cancelGroup is passed on to RiaSumoConnector::getAndTrackReply, see its documentation; defaults to
|
||||
// untracked for the batch path, which has no owner to cancel it early.
|
||||
QNetworkReply* makePropertyBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup = nullptr );
|
||||
|
||||
std::map<QString, QByteArray> fetchPropertyBatch( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const std::vector<QString>& timestampsToFetch );
|
||||
|
||||
static QString propertyBlobIdPath( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& gridName,
|
||||
int realization,
|
||||
const QString& propertyName,
|
||||
const QString& isoDateOrInterval );
|
||||
|
||||
// 'kind' names what the blob holds, e.g. "grid" or "grid property", for the debug log.
|
||||
static QString blobIdFromBody( const QByteArray& body, const QString& kind, const QString& name );
|
||||
static QString blobIdFromReply( QNetworkReply* reply, const QString& propertyName );
|
||||
|
||||
static std::vector<QString> parseGridNames( const QByteArray& body );
|
||||
static SumoGridInfo parseGridInfo( const QByteArray& body );
|
||||
static std::vector<SumoGridPropertyInfo> parsePropertyInfo( const QByteArray& body );
|
||||
|
||||
private:
|
||||
RiaSumoConnector& m_connector;
|
||||
};
|
||||
@@ -46,10 +46,9 @@ std::vector<QString> RiaSumoSummary::vectorNames( const SumoCaseId& caseId, cons
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
|
||||
const QString url =
|
||||
QString( "%1/cases/%2/ensembles/%3/vector_list" ).arg( m_connector.server() ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
const QString path = QString( "/cases/%1/ensembles/%2/vector_list" ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
|
||||
return parseVectorNames( m_connector.getBlocking( url ) );
|
||||
return parseVectorNames( m_connector.getBlocking( path ) );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -90,6 +89,13 @@ std::map<QString, QByteArray>
|
||||
|
||||
if ( namesToFetch.empty() ) return contentsByVectorName;
|
||||
|
||||
// The requests below are issued directly on the transfer thread, so the token and the address of the
|
||||
// service have to be in place before the work is handed over.
|
||||
m_connector.requestTokenBlocking();
|
||||
|
||||
const QString baseUrl = m_connector.serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return contentsByVectorName;
|
||||
|
||||
m_connector.runOnTransferThreadBlocking(
|
||||
[&]()
|
||||
{
|
||||
@@ -97,7 +103,8 @@ std::map<QString, QByteArray>
|
||||
std::vector<QNetworkReply*> blobIdReplies;
|
||||
for ( const auto& vectorName : namesToFetch )
|
||||
{
|
||||
blobIdReplies.push_back( makeVectorBlobIdRequest( caseId, ensembleName, vectorName ) );
|
||||
blobIdReplies.push_back(
|
||||
makeVectorBlobIdRequest( baseUrl, caseId, ensembleName, vectorName, m_connector.networkAccessManager() ) );
|
||||
}
|
||||
|
||||
RiaSumoConnector::waitForRepliesToFinish( blobIdReplies );
|
||||
@@ -115,7 +122,7 @@ std::map<QString, QByteArray>
|
||||
if ( !blobId.isEmpty() ) blobIdsToDownload.push_back( blobId );
|
||||
}
|
||||
|
||||
const auto contentsByBlobId = m_connector.downloadBlobs( blobIdsToDownload );
|
||||
const auto contentsByBlobId = m_connector.downloadBlobs( baseUrl, blobIdsToDownload );
|
||||
|
||||
// Anything missing failed; the caller falls back to fetching it on its own later.
|
||||
for ( size_t i = 0; i < blobIds.size(); i++ )
|
||||
@@ -144,12 +151,16 @@ std::map<QString, QByteArray>
|
||||
void RiaSumoSummary::vectorDataAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const std::vector<QString>& vectorNames,
|
||||
const std::function<void( const QString&, const QByteArray& )>& onVectorReady )
|
||||
const std::function<void( const QString&, const QByteArray& )>& onVectorReady,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
if ( vectorNames.empty() || !onVectorReady ) return;
|
||||
|
||||
const QString baseUrl = m_connector.serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return;
|
||||
|
||||
m_connector.runOnTransferThread(
|
||||
[this, caseId, ensembleName, vectorNames, onVectorReady]()
|
||||
[this, baseUrl, caseId, ensembleName, vectorNames, onVectorReady, cancelGroup]()
|
||||
{
|
||||
for ( const auto& vectorName : vectorNames )
|
||||
{
|
||||
@@ -158,7 +169,8 @@ void RiaSumoSummary::vectorDataAsync( const SumoCaseId&
|
||||
m_connector.invokeOnConnectorThread( [onVectorReady, vectorName, contents]() { onVectorReady( vectorName, contents ); } );
|
||||
};
|
||||
|
||||
auto blobIdReply = makeVectorBlobIdRequest( caseId, ensembleName, vectorName );
|
||||
auto blobIdReply =
|
||||
makeVectorBlobIdRequest( baseUrl, caseId, ensembleName, vectorName, m_connector.backgroundNetworkAccessManager(), cancelGroup );
|
||||
if ( !blobIdReply )
|
||||
{
|
||||
deliver( {} );
|
||||
@@ -172,7 +184,7 @@ void RiaSumoSummary::vectorDataAsync( const SumoCaseId&
|
||||
QObject::connect( blobIdReply,
|
||||
&QNetworkReply::finished,
|
||||
blobIdReply,
|
||||
[this, blobIdReply, vectorName, deliver]()
|
||||
[this, blobIdReply, vectorName, deliver, cancelGroup]()
|
||||
{
|
||||
const QString blobId = blobIdFromReply( blobIdReply, vectorName );
|
||||
if ( blobId.isEmpty() )
|
||||
@@ -181,7 +193,7 @@ void RiaSumoSummary::vectorDataAsync( const SumoCaseId&
|
||||
return;
|
||||
}
|
||||
|
||||
m_connector.downloadBlobAsync( blobId, deliver );
|
||||
m_connector.downloadBlobAsync( blobId, deliver, RiaSumoDefines::requestTimeoutMillis(), cancelGroup );
|
||||
} );
|
||||
}
|
||||
} );
|
||||
@@ -195,7 +207,7 @@ QByteArray RiaSumoSummary::parameterData( const SumoCaseId& caseId, const QStrin
|
||||
const QString blobId = parameterBlobId( caseId, ensembleName );
|
||||
if ( blobId.isEmpty() ) return {};
|
||||
|
||||
return m_connector.downloadBlobBlocking( blobId );
|
||||
return m_connector.downloadBlobBlocking( blobId, "ensemble parameters" );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -203,37 +215,38 @@ QByteArray RiaSumoSummary::parameterData( const SumoCaseId& caseId, const QStrin
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoSummary::vectorBlobId( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName )
|
||||
{
|
||||
const QString url = vectorBlobIdUrl( caseId, ensembleName, vectorName );
|
||||
const QString path = vectorBlobIdPath( caseId, ensembleName, vectorName );
|
||||
|
||||
return logBlobId( RiaSumoConnector::blobIdFromBody( m_connector.getBlocking( url ) ), vectorName );
|
||||
return logBlobId( RiaSumoConnector::blobIdFromBody( m_connector.getBlocking( path ) ), vectorName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoSummary::vectorBlobIdUrl( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName ) const
|
||||
QString RiaSumoSummary::vectorBlobIdPath( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
const QString encodedVectorName = QUrl::toPercentEncoding( vectorName );
|
||||
|
||||
return QString( "%1/cases/%2/ensembles/%3/vectors/%4/blob_id" )
|
||||
.arg( m_connector.server() )
|
||||
.arg( caseId.get() )
|
||||
.arg( encodedEnsembleName )
|
||||
.arg( encodedVectorName );
|
||||
return QString( "/cases/%1/ensembles/%2/vectors/%3/blob_id" ).arg( caseId.get() ).arg( encodedEnsembleName ).arg( encodedVectorName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Issue the blob id request for one vector. The reply is returned unfinished, so the caller decides how
|
||||
/// to wait for it: one at a time, or several at once when batching.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QNetworkReply* RiaSumoSummary::makeVectorBlobIdRequest( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName )
|
||||
QNetworkReply* RiaSumoSummary::makeVectorBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& vectorName,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
QNetworkRequest networkRequest;
|
||||
networkRequest.setUrl( QUrl( vectorBlobIdUrl( caseId, ensembleName, vectorName ) ) );
|
||||
networkRequest.setUrl( QUrl( baseUrl + vectorBlobIdPath( caseId, ensembleName, vectorName ) ) );
|
||||
m_connector.addStandardHeader( networkRequest, m_connector.transferToken(), RiaCloudDefines::contentTypeJson() );
|
||||
|
||||
return m_connector.networkAccessManager()->get( networkRequest );
|
||||
return m_connector.getAndTrackReply( networkManager, networkRequest, cancelGroup );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -273,24 +286,28 @@ QString RiaSumoSummary::logBlobId( const QString& blobId, const QString& vectorN
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
///
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoSummary::parameterBlobIdUrl( const SumoCaseId& caseId, const QString& ensembleName ) const
|
||||
QString RiaSumoSummary::parameterBlobIdPath( const SumoCaseId& caseId, const QString& ensembleName )
|
||||
{
|
||||
const QString encodedEnsembleName = QUrl::toPercentEncoding( ensembleName );
|
||||
|
||||
return QString( "%1/cases/%2/ensembles/%3/parameters/blob_id" ).arg( m_connector.server() ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
return QString( "/cases/%1/ensembles/%2/parameters/blob_id" ).arg( caseId.get() ).arg( encodedEnsembleName );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
/// Issue the blob id request for the ensemble parameters. The reply is returned unfinished, so the caller
|
||||
/// decides how to wait for it.
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QNetworkReply* RiaSumoSummary::makeParameterBlobIdRequest( const SumoCaseId& caseId, const QString& ensembleName )
|
||||
QNetworkReply* RiaSumoSummary::makeParameterBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
QNetworkRequest networkRequest;
|
||||
networkRequest.setUrl( QUrl( parameterBlobIdUrl( caseId, ensembleName ) ) );
|
||||
networkRequest.setUrl( QUrl( baseUrl + parameterBlobIdPath( caseId, ensembleName ) ) );
|
||||
m_connector.addStandardHeader( networkRequest, m_connector.transferToken(), RiaCloudDefines::contentTypeJson() );
|
||||
|
||||
return m_connector.networkAccessManager()->get( networkRequest );
|
||||
return m_connector.getAndTrackReply( networkManager, networkRequest, cancelGroup );
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
@@ -299,17 +316,22 @@ QNetworkReply* RiaSumoSummary::makeParameterBlobIdRequest( const SumoCaseId& cas
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
void RiaSumoSummary::parameterDataAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const std::function<void( const QByteArray& )>& onParametersReady )
|
||||
const std::function<void( const QByteArray& )>& onParametersReady,
|
||||
const void* cancelGroup )
|
||||
{
|
||||
if ( !onParametersReady ) return;
|
||||
|
||||
const QString baseUrl = m_connector.serviceBaseUrl();
|
||||
if ( baseUrl.isEmpty() ) return;
|
||||
|
||||
m_connector.runOnTransferThread(
|
||||
[this, caseId, ensembleName, onParametersReady]()
|
||||
[this, baseUrl, caseId, ensembleName, onParametersReady, cancelGroup]()
|
||||
{
|
||||
auto deliver = [this, onParametersReady]( const QByteArray& contents )
|
||||
{ m_connector.invokeOnConnectorThread( [onParametersReady, contents]() { onParametersReady( contents ); } ); };
|
||||
|
||||
auto blobIdReply = makeParameterBlobIdRequest( caseId, ensembleName );
|
||||
auto blobIdReply =
|
||||
makeParameterBlobIdRequest( baseUrl, caseId, ensembleName, m_connector.backgroundNetworkAccessManager(), cancelGroup );
|
||||
if ( !blobIdReply )
|
||||
{
|
||||
deliver( {} );
|
||||
@@ -321,7 +343,7 @@ void RiaSumoSummary::parameterDataAsync( const SumoCaseId&
|
||||
QObject::connect( blobIdReply,
|
||||
&QNetworkReply::finished,
|
||||
blobIdReply,
|
||||
[this, blobIdReply, deliver]()
|
||||
[this, blobIdReply, deliver, cancelGroup]()
|
||||
{
|
||||
const QString blobId = blobIdFromReply( blobIdReply, "parameters" );
|
||||
if ( blobId.isEmpty() )
|
||||
@@ -330,7 +352,7 @@ void RiaSumoSummary::parameterDataAsync( const SumoCaseId&
|
||||
return;
|
||||
}
|
||||
|
||||
m_connector.downloadBlobAsync( blobId, deliver );
|
||||
m_connector.downloadBlobAsync( blobId, deliver, RiaSumoDefines::requestTimeoutMillis(), cancelGroup );
|
||||
} );
|
||||
} );
|
||||
}
|
||||
@@ -340,7 +362,7 @@ void RiaSumoSummary::parameterDataAsync( const SumoCaseId&
|
||||
//--------------------------------------------------------------------------------------------------
|
||||
QString RiaSumoSummary::parameterBlobId( const SumoCaseId& caseId, const QString& ensembleName )
|
||||
{
|
||||
const QString blobId = RiaSumoConnector::blobIdFromBody( m_connector.getBlocking( parameterBlobIdUrl( caseId, ensembleName ) ) );
|
||||
const QString blobId = RiaSumoConnector::blobIdFromBody( m_connector.getBlocking( parameterBlobIdPath( caseId, ensembleName ) ) );
|
||||
|
||||
if ( !blobId.isEmpty() )
|
||||
{
|
||||
|
||||
@@ -29,6 +29,7 @@
|
||||
|
||||
class RiaSumoConnector;
|
||||
class QNetworkReply;
|
||||
class QNetworkAccessManager;
|
||||
|
||||
//==================================================================================================
|
||||
/// The summary data of a Sumo case: the vectors an ensemble has, their values, and the ensemble
|
||||
@@ -53,10 +54,13 @@ public:
|
||||
// The same again, but without waiting: all vectors are requested at once and onVectorReady is called for
|
||||
// each as it arrives, on the thread the connector lives on. Empty contents mean that vector failed, and
|
||||
// the callback is called exactly once per requested vector.
|
||||
//
|
||||
// cancelGroup is passed on to RiaSumoConnector::downloadBlobAsync, see its documentation.
|
||||
void vectorDataAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const std::vector<QString>& vectorNames,
|
||||
const std::function<void( const QString&, const QByteArray& )>& onVectorReady );
|
||||
const std::function<void( const QString&, const QByteArray& )>& onVectorReady,
|
||||
const void* cancelGroup = nullptr );
|
||||
|
||||
// The ensemble parameters, as a parquet blob.
|
||||
QByteArray parameterData( const SumoCaseId& caseId, const QString& ensembleName );
|
||||
@@ -65,16 +69,32 @@ public:
|
||||
// the first request for them can take a while and is not something to hold the user interface for.
|
||||
void parameterDataAsync( const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const std::function<void( const QByteArray& )>& onParametersReady );
|
||||
const std::function<void( const QByteArray& )>& onParametersReady,
|
||||
const void* cancelGroup = nullptr );
|
||||
|
||||
QString vectorBlobId( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName );
|
||||
QString parameterBlobId( const SumoCaseId& caseId, const QString& ensembleName );
|
||||
|
||||
private:
|
||||
QString vectorBlobIdUrl( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName ) const;
|
||||
QString parameterBlobIdUrl( const SumoCaseId& caseId, const QString& ensembleName ) const;
|
||||
QNetworkReply* makeParameterBlobIdRequest( const SumoCaseId& caseId, const QString& ensembleName );
|
||||
QNetworkReply* makeVectorBlobIdRequest( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName );
|
||||
static QString vectorBlobIdPath( const SumoCaseId& caseId, const QString& ensembleName, const QString& vectorName );
|
||||
static QString parameterBlobIdPath( const SumoCaseId& caseId, const QString& ensembleName );
|
||||
|
||||
// Run on the transfer thread, so the base URL is resolved by the caller and passed in. networkManager
|
||||
// lets the caller choose the connection pool: the shared one for a blocking fetch waited on directly, or
|
||||
// the background one for a prefetch batch, see RiaSumoConnector::backgroundNetworkAccessManager.
|
||||
// cancelGroup is passed on to RiaSumoConnector::getAndTrackReply, see its documentation; defaults to
|
||||
// untracked for the batch path, which has no owner to cancel it early.
|
||||
QNetworkReply* makeParameterBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup = nullptr );
|
||||
QNetworkReply* makeVectorBlobIdRequest( const QString& baseUrl,
|
||||
const SumoCaseId& caseId,
|
||||
const QString& ensembleName,
|
||||
const QString& vectorName,
|
||||
QNetworkAccessManager* networkManager,
|
||||
const void* cancelGroup = nullptr );
|
||||
static QString blobIdFromReply( QNetworkReply* reply, const QString& vectorName );
|
||||
static QString logBlobId( const QString& blobId, const QString& vectorName );
|
||||
|
||||
|
||||
@@ -177,12 +177,13 @@ QString RiaProjectFileTools::transferPathsToGlobalPathList( RimProject* project
|
||||
{
|
||||
if ( gridCase->displayNameType() == RimCaseDisplayNameTools::DisplayName::CUSTOM )
|
||||
{
|
||||
// At this point, after the replace of variables into caf::FilePath objects, the variable name is
|
||||
// stored in the summary case object. Read out the variable name and append "_name" for custom
|
||||
// summary variables.
|
||||
|
||||
// The grid file path was replaced by a path variable above, so reuse that variable as the key for
|
||||
// the case name. A case not written to the project file still holds its raw path, and
|
||||
// RiaVariableMapper cannot read back a key holding '/', '-' or '.', so leave that name alone.
|
||||
QString variableName = gridCase->gridFileName();
|
||||
variableName = variableName.remove( RiaVariableMapper::variableToken() );
|
||||
if ( !variableName.startsWith( RiaVariableMapper::variableToken() ) ) continue;
|
||||
|
||||
variableName = variableName.remove( RiaVariableMapper::variableToken() );
|
||||
|
||||
variableName = RiaVariableMapper::variableToken() + variableName + RiaVariableMapper::postfixName() +
|
||||
RiaVariableMapper::variableToken();
|
||||
|
||||
Reference in new issue
Block a user