OilfieldToolsNet/ResInsight
4
0
Fork 0
mirror of https://github.com/OPM/ResInsight.git synced 2025-02-25 18:55:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

Use block transfer for data transfer of cell properties

Browse Source
This commit is contained in:
Magne Sjaastad 2014-04-15 13:40:53 +02:00
parent 8ea418272d
commit b4821ed7a3
9 changed files with 155 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
ApplicationCode/SocketInterface/RiaPropertyDataCommands.cpp
View File

@ -42,6 +42,7 @@
#include <QTcpSocket>
#include "RiaApplication.h"
#include "RiaPreferences.h"
#include "RiaSocketDataTransfer.h"
//--------------------------------------------------------------------------------------------------
@ -145,6 +146,9 @@ public:
socketStream << timestepByteCount ;
// Then write the data.
size_t valueCount = RiaSocketDataTransfer::doubleValueCountInBlock();
std::vector<double> values(valueCount);
size_t valueIndex = 0;
size_t globalCellCount = activeInfo->globalCellCount();
for (size_t tIdx = 0; tIdx < requestedTimesteps.size(); ++tIdx)
@ -156,37 +160,34 @@ public:
{
if (resultIdx < scalarResultFrames->at(requestedTimesteps[tIdx]).size())
{
socketStream << scalarResultFrames->at(requestedTimesteps[tIdx])[resultIdx];
values[valueIndex] = scalarResultFrames->at(requestedTimesteps[tIdx])[resultIdx];
}
else
{
socketStream << HUGE_VAL;
values[valueIndex] = HUGE_VAL;
}
valueIndex++;
if (valueIndex >= valueCount)
{
if (!RiaSocketTools::writeBlockData(server, server->currentClient(), (const char *)values.data(), valueIndex * sizeof(double)))
{
return false;
}
valueIndex = 0;
}
}
}
}
#if 0
// This aproach is faster but does not handle coarsening
size_t timestepResultCount = scalarResultFrames->front().size();
quint64 timestepByteCount = (quint64)(timestepResultCount*sizeof(double));
socketStream << timestepByteCount ;
// Then write the data.
for (size_t tIdx = 0; tIdx < requestedTimesteps.size(); ++tIdx)
// Write remaining data
if (!RiaSocketTools::writeBlockData(server, server->currentClient(), (const char *)values.data(), valueIndex * sizeof(double)))
{
#if 1 // Write data as raw bytes, fast but does not handle byteswapping
server->currentClient()->write((const char *)scalarResultFrames->at(requestedTimesteps[tIdx]).data(), timestepByteCount); // Raw print of data. Fast but no platform conversion
#else // Write data using QDataStream, does byteswapping for us. Must use QDataStream on client as well
for (size_t cIdx = 0; cIdx < scalarResultFrames->at(requestedTimesteps[tIdx]).size(); ++cIdx)
{
socketStream << scalarResultFrames->at(tIdx)[cIdx];
}
#endif
return false;
}
#endif
}
}
return true;
}
};
@ -314,7 +315,9 @@ public:
continue;
}
std::vector<double> values(rigGrid->cellCount());
size_t valueCount = RiaSocketDataTransfer::doubleValueCountInBlock();
std::vector<double> values(valueCount);
size_t valueIndex = 0;
for (size_t cellIdx = 0; cellIdx < rigGrid->cellCount(); cellIdx++)
{
double cellValue = cellCenterDataAccessObject->cellScalar(cellIdx);
@ -322,10 +325,24 @@ public:
{
cellValue = 0.0;
}
values[cellIdx] = cellValue;
values[valueIndex++] = cellValue;
if (valueIndex >= valueCount)
{
if (!RiaSocketTools::writeBlockData(server, server->currentClient(), (const char *)values.data(), valueIndex * sizeof(double)))
{
return false;
}
valueIndex = 0;
}
}
RiaSocketTools::writeBlockData(server, server->currentClient(), (const char *)values.data(), values.size() * sizeof(double));
// Write remaining data
if (!RiaSocketTools::writeBlockData(server, server->currentClient(), (const char *)values.data(), valueIndex * sizeof(double)))
{
return false;
}
}
return true;
@ -552,16 +569,18 @@ public:
internalMatrixData = m_scalarResultsToAdd->at(m_requestedTimesteps[m_currentTimeStepNumberToRead]).data();
}
#if 1 // Use raw data transfer. Faster.
bytesRead = currentClient->read((char*)(internalMatrixData), m_bytesPerTimeStepToRead);
#else
for (size_t cIdx = 0; cIdx < cellCountFromOctave; ++cIdx)
QStringList errorMessages;
if (!RiaSocketDataTransfer::readBlockDataFromSocket(currentClient, (char*)(internalMatrixData), m_bytesPerTimeStepToRead, errorMessages))
{
socketStream >> internalMatrixData[cIdx];
for (int i = 0; i < errorMessages.size(); i++)
{
server->errorMessageDialog()->showMessage(errorMessages[i]);
}
if (socketStream.status() == QDataStream::Ok) bytesRead += sizeof(double);
currentClient->abort();
return true;
}
#endif
// Map data from active to result index based container ( Coarsening is active)
if (isCoarseningActive)
{
@ -576,12 +595,6 @@ public:
}
}
if ((int)m_bytesPerTimeStepToRead != bytesRead)
{
server->errorMessageDialog()->showMessage(RiaSocketServer::tr("ResInsight SocketServer: \n") +
RiaSocketServer::tr("Could not read binary double data properly from socket"));
}
++m_currentTimeStepNumberToRead;
}
@ -896,8 +909,17 @@ public:
std::vector<double> doubleValues(cellCountFromOctave);
qint64 bytesRead = currentClient->read((char*)(doubleValues.data()), m_bytesPerTimeStepToRead);
size_t doubleValueIndex = 0;
QStringList errorMessages;
if (!RiaSocketDataTransfer::readBlockDataFromSocket(currentClient, (char*)(doubleValues.data()), m_bytesPerTimeStepToRead, errorMessages))
{
for (int i = 0; i < errorMessages.size(); i++)
{
server->errorMessageDialog()->showMessage(errorMessages[i]);
}
currentClient->abort();
return true;
}
cvf::ref<cvf::StructGridScalarDataAccess> cellCenterDataAccessObject =
m_currentReservoir->reservoirData()->dataAccessObject(grid, m_porosityModelEnum, m_requestedTimesteps[m_currentTimeStepNumberToRead], m_currentScalarIndex);

13
ApplicationCode/SocketInterface/RiaSocketDataTransfer.cpp
View File

@ -18,6 +18,7 @@
#include "RiaSocketDataTransfer.h"
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
@ -55,15 +56,12 @@ bool RiaSocketDataTransfer::writeBlockDataToSocket(QTcpSocket* socket, const cha
bool RiaSocketDataTransfer::readBlockDataFromSocket(QTcpSocket* socket, char* data, quint64 bytesToRead, QStringList& errorMessages)
{
quint64 bytesRead = 0;
int blockCount = 0;
quint64 maxBlockSize = doubleValueCountInBlock() * sizeof(double);
while (bytesRead < bytesToRead)
{
if (socket->bytesAvailable())
{
quint64 byteCountToRead = qMin(bytesToRead - bytesRead, maxBlockSize);
quint64 byteCountToRead = bytesToRead - bytesRead;
qint64 actuallyBytesRead = socket->read(data + bytesRead, byteCountToRead);
if (actuallyBytesRead < 0)
@ -75,7 +73,6 @@ bool RiaSocketDataTransfer::readBlockDataFromSocket(QTcpSocket* socket, char* da
}
bytesRead += actuallyBytesRead;
blockCount++;
}
else
{
@ -87,6 +84,12 @@ bool RiaSocketDataTransfer::readBlockDataFromSocket(QTcpSocket* socket, char* da
return false;
}
}
// Allow Octave process to end a long running Octave function
#ifdef octave_oct_h
OCTAVE_QUIT;
#endif
}
return true;

8
ApplicationCode/SocketInterface/RiaSocketTools.cpp
View File

@ -122,10 +122,10 @@ bool RiaSocketTools::writeBlockData(RiaSocketServer* server, QTcpSocket* socket,
server->errorMessageDialog()->showMessage(errorMessages[i]);
}
double totalTimeMS = timer.time() * 1000.0;
QString resultInfo = QString("Total time '%1 ms'").arg(totalTimeMS);
server->errorMessageDialog()->showMessage(resultInfo);
// double totalTimeMS = timer.time() * 1000.0;
// QString resultInfo = QString("Total time '%1 ms'").arg(totalTimeMS);
//
// server->errorMessageDialog()->showMessage(resultInfo);
}
return writeSucceded;

36
OctavePlugin/CMakeLists.txt
View File

@ -6,26 +6,26 @@
set(CPP_SOURCES
riGetActiveCellProperty.cpp
riSetActiveCellProperty.cpp
riGetActiveCellInfo.cpp
riGetMainGridDimensions.cpp
riGetCurrentCase.cpp
riGetCaseGroups.cpp
riGetSelectedCases.cpp
riGetCases.cpp
riGetTimeStepDates.cpp
riGetTimeStepDays.cpp
riGetGridDimensions.cpp
riGetCoarseningInfo.cpp
riGetCellCenters.cpp
riGetActiveCellCenters.cpp
riGetCellCorners.cpp
riGetActiveCellCorners.cpp
# riGetActiveCellInfo.cpp
# riGetMainGridDimensions.cpp
# riGetCurrentCase.cpp
# riGetCaseGroups.cpp
# riGetSelectedCases.cpp
# riGetCases.cpp
# riGetTimeStepDates.cpp
# riGetTimeStepDays.cpp
# riGetGridDimensions.cpp
# riGetCoarseningInfo.cpp
# riGetCellCenters.cpp
# riGetActiveCellCenters.cpp
# riGetCellCorners.cpp
# riGetActiveCellCorners.cpp
riGetGridProperty.cpp
riSetGridProperty.cpp
riGetPropertyNames.cpp
riGetWellNames.cpp
riGetWellStatus.cpp
riGetWellCells.cpp
# riGetPropertyNames.cpp
# riGetWellNames.cpp
# riGetWellStatus.cpp
# riGetWellCells.cpp
)
if (${CMAKE_SYSTEM_NAME} MATCHES "Linux")

45
OctavePlugin/riGetActiveCellProperty.cpp
View File

@ -1,7 +1,12 @@
#include <QtNetwork>
#include <QStringList>
#include <octave/oct.h>
#include "riSettings.h"
#include "RiaSocketDataTransfer.cpp" // NB! Include cpp-file to avoid linking of additional file in oct-compile configuration
void getActiveCellProperty(Matrix& propertyFrames, const QString &serverName, quint16 serverPort,
const qint64& caseId, QString propertyName, const int32NDArray& requestedTimeSteps, QString porosityModel)
{
@ -63,44 +68,18 @@ void getActiveCellProperty(Matrix& propertyFrames, const QString &serverName, qu
return;
}
// Wait for available data for each timestep, then read data for each timestep
quint64 totalByteCount = byteCount * timestepCount;
for (size_t tIdx = 0; tIdx < timestepCount; ++tIdx)
double* internalMatrixData = propertyFrames.fortran_vec();
QStringList errorMessages;
if (!RiaSocketDataTransfer::readBlockDataFromSocket(&socket, (char*)(internalMatrixData), totalByteCount, errorMessages))
{
while (socket.bytesAvailable() < (qint64)byteCount)
for (int i = 0; i < errorMessages.size(); i++)
{
if (!socket.waitForReadyRead(riOctavePlugin::longTimeOutMilliSecs))
{
error((("Waiting for timestep data number: ") + QString::number(tIdx)+ ": " + socket.errorString()).toLatin1().data());
octave_stdout << "Active cells: " << activeCellCount << ", Timesteps: " << timestepCount << std::endl;
return ;
}
OCTAVE_QUIT;
error(errorMessages[i].toLatin1().data());
}
qint64 bytesRead = 0;
double * internalMatrixData = propertyFrames.fortran_vec();
#if 0
// Raw data transfer. Faster. Not possible when dealing with coarsening
// bytesRead = socket.read((char*)(internalMatrixData + tIdx * activeCellCount), byteCount);
#else
// Compatible transfer. Now the only one working
for (size_t cIdx = 0; cIdx < activeCellCount; ++cIdx)
{
socketStream >> internalMatrixData[tIdx * activeCellCount + cIdx];
if (socketStream.status() == QDataStream::Ok) bytesRead += sizeof(double);
}
#endif
if ((qint64)byteCount != bytesRead)
{
error("Could not read binary double data properly from socket");
octave_stdout << "Active cells: " << activeCellCount << ", Timesteps: " << timestepCount << std::endl;
}
OCTAVE_QUIT;
return;
}
QString tmp = QString("riGetActiveCellProperty : Read %1").arg(propertyName);

6
OctavePlugin/riGetGridProperty.cpp
View File

@ -4,7 +4,7 @@
#include <octave/oct.h>
#include "riSettings.h"
#include "riSocketTools.h"
#include "RiaSocketDataTransfer.cpp" // NB! Include cpp-file to avoid linking of additional file in oct-compile configuration
void getGridProperty(NDArray& propertyFrames, const QString &serverName, quint16 serverPort,
@ -83,14 +83,14 @@ void getGridProperty(NDArray& propertyFrames, const QString &serverName, quint16
double* internalMatrixData = propertyFrames.fortran_vec();
QStringList errorMessages;
if (!readBlockData(socket, (char*)(internalMatrixData), totalByteCount, errorMessages))
if (!RiaSocketDataTransfer::readBlockDataFromSocket(&socket, (char*)(internalMatrixData), totalByteCount, errorMessages))
{
for (int i = 0; i < errorMessages.size(); i++)
{
error(errorMessages[i].toLatin1().data());
}
OCTAVE_QUIT;
return;
}
QString tmp = QString("riGetGridProperty : Read %1").arg(propertyName);

34
OctavePlugin/riSetActiveCellProperty.cpp
View File

@ -1,6 +1,10 @@
#include <QtNetwork>
#include <QStringList>
#include <octave/oct.h>
#include "riSettings.h"
#include "RiaSocketDataTransfer.cpp" // NB! Include cpp-file to avoid linking of additional file in oct-compile configuration
void setEclipseProperty(const Matrix& propertyFrames, const QString &hostName, quint16 port,
@ -47,27 +51,29 @@ void setEclipseProperty(const Matrix& propertyFrames, const QString &hostName, q
socketStream << (qint64)timeStepByteCount;
const double* internalData = propertyFrames.fortran_vec();
qint64 dataWritten = socket.write((const char *)internalData, timeStepByteCount*timeStepCount);
if (dataWritten == timeStepByteCount*timeStepCount)
QStringList errorMessages;
if (!RiaSocketDataTransfer::writeBlockDataToSocket(&socket, (const char *)internalData, timeStepByteCount*timeStepCount, errorMessages))
{
QString tmp = QString("riSetActiveCellProperty : Wrote %1").arg(propertyName);
for (int i = 0; i < errorMessages.size(); i++)
{
octave_stdout << errorMessages[i].toStdString();
}
if (caseId == -1)
{
tmp += QString(" to current case.");
}
else
{
tmp += QString(" to case with Id = %1.").arg(caseId);
}
octave_stdout << tmp.toStdString() << " Active Cells : " << cellCount << " Time steps : " << timeStepCount << std::endl;
return;
}
QString tmp = QString("riSetActiveCellProperty : Wrote %1").arg(propertyName);
if (caseId == -1)
{
tmp += QString(" to current case.");
}
else
{
error("riSetActiveCellProperty : Was not able to write the proper amount of data to ResInsight:");
octave_stdout << " Active Cells : " << cellCount << "Time steps : " << timeStepCount << " Data Written: " << dataWritten << " Should have written: " << timeStepCount * cellCount * sizeof(double) << std::endl;
tmp += QString(" to case with Id = %1.").arg(caseId);
}
octave_stdout << tmp.toStdString() << " Active Cells : " << cellCount << " Time steps : " << timeStepCount << std::endl;
while(socket.bytesToWrite() && socket.state() == QAbstractSocket::ConnectedState)
{

52
OctavePlugin/riSetGridProperty.cpp
View File

@ -1,6 +1,10 @@
#include <QtNetwork>
#include <QStringList>
#include <octave/oct.h>
#include "riSettings.h"
#include "RiaSocketDataTransfer.cpp" // NB! Include cpp-file to avoid linking of additional file in oct-compile configuration
void setEclipseProperty(const NDArray& propertyFrames, const QString &hostName, quint16 port,
@ -64,41 +68,41 @@ void setEclipseProperty(const NDArray& propertyFrames, const QString &hostName,
socketStream << (qint64)singleTimeStepByteCount;
const double* internalData = propertyFrames.fortran_vec();
qint64 dataWritten = 0;
for (size_t tsIdx = 0; tsIdx < timeStepCount; ++tsIdx)
QStringList errorMessages;
if (!RiaSocketDataTransfer::writeBlockDataToSocket(&socket, (const char *)internalData, timeStepCount*singleTimeStepByteCount, errorMessages))
{
dataWritten += socket.write(((const char *)internalData) + tsIdx*singleTimeStepByteCount, singleTimeStepByteCount);
for (int i = 0; i < errorMessages.size(); i++)
{
octave_stdout << errorMessages[i].toStdString();
}
size_t cellCount = cellCountI * cellCountJ * cellCountK;
error("riSetGridProperty : Was not able to write the proper amount of data to ResInsight:");
octave_stdout << " Cell count : " << cellCount << "Time steps : " << timeStepCount << std::endl;
return;
}
if (dataWritten == singleTimeStepByteCount*timeStepCount)
QString tmp = QString("riSetGridProperty : Wrote %1").arg(propertyName);
if (caseId == -1)
{
QString tmp = QString("riSetGridProperty : Wrote %1").arg(propertyName);
if (caseId == -1)
{
tmp += QString(" to current case,");
}
else
{
tmp += QString(" to case with Id = %1,").arg(caseId);
}
tmp += QString(" grid index: %1, ").arg(gridIndex);
octave_stdout << tmp.toStdString() << " Time steps : " << timeStepCount << std::endl;
tmp += QString(" to current case,");
}
else
{
size_t cellCount = cellCountI * cellCountJ * cellCountK;
error("riSetGridProperty : Was not able to write the proper amount of data to ResInsight:");
octave_stdout << " Cell count : " << cellCount << "Time steps : " << timeStepCount << " Data Written: " << dataWritten << " Should have written: " << timeStepCount * cellCount * sizeof(double) << std::endl;
tmp += QString(" to case with Id = %1,").arg(caseId);
}
tmp += QString(" grid index: %1, ").arg(gridIndex);
octave_stdout << tmp.toStdString() << " Time steps : " << timeStepCount << std::endl;
while(socket.bytesToWrite() && socket.state() == QAbstractSocket::ConnectedState)
{
octave_stdout << "Bytes to write: " << socket.bytesToWrite() << std::endl << std::flush;
socket.waitForBytesWritten(2000);
// octave_stdout << "Bytes to write: " << socket.bytesToWrite() << std::endl << std::flush;
socket.waitForBytesWritten(riOctavePlugin::longTimeOutMilliSecs);
OCTAVE_QUIT;
}

4
OctavePlugin/riSocketTools.h
View File

@ -2,7 +2,7 @@
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
bool readBlockData(QTcpSocket& socket, char* data, quint64 bytesToRead, QStringList& errorMessages)
bool readBlockData_to_be_deleted(QTcpSocket& socket, char* data, quint64 bytesToRead, QStringList& errorMessages)
{
quint64 bytesRead = 0;
int blockCount = 0;
@ -51,7 +51,7 @@ bool readBlockData(QTcpSocket& socket, char* data, quint64 bytesToRead, QStringL
//--------------------------------------------------------------------------------------------------
///
//--------------------------------------------------------------------------------------------------
bool writeBlockData(QTcpSocket& socket, const char* data, quint64 bytesToWrite, QStringList& errorMessages)
bool writeBlockData_to_be_deleted(QTcpSocket& socket, const char* data, quint64 bytesToWrite, QStringList& errorMessages)
{
quint64 bytesWritten = 0;
int blockCount = 0;