541 lines
18 KiB
C++
Executable File
541 lines
18 KiB
C++
Executable File
/**
|
|
* @RDE_HEADER@
|
|
* Data Plugin for SituationReplayManagement
|
|
* $Id$
|
|
* $HeadURL$
|
|
* $Author$
|
|
* Created by Patric Karwat, SLSE, 2016-07-19
|
|
*/
|
|
|
|
#include "srm_data_plugin.h"
|
|
#include <utility>
|
|
|
|
#include <DDS/Core/core_dtc.h>
|
|
#include <SituationReplayManagement/srm_ftc.h>
|
|
#include <LtsSession/lts_session_data_dds.h>
|
|
#include <DDS/DDSConnector/dds_connector.h>
|
|
#include <DDS/QosProfile/DDS_DefaultQos.h>
|
|
#include <DDS/Core/core_dts_types.h>
|
|
#include <DDS/DTS/generic_dts_impl.h>
|
|
#include <SituationReplayManagement/logger.h>
|
|
#include <LtsUtilsCore/await.h>
|
|
|
|
#include <QCoreApplication>
|
|
#include <QPointer>
|
|
#include <QProcess>
|
|
#include <QTimer>
|
|
#include <QDebug>
|
|
|
|
|
|
namespace SituationReplayManagement
|
|
{
|
|
|
|
class ReplayHandler : public QObject
|
|
{
|
|
Q_OBJECT
|
|
public:
|
|
ReplayHandler()
|
|
: m_logger(getLogger("ReplayHandler"))
|
|
{}
|
|
|
|
~ReplayHandler()
|
|
{
|
|
try
|
|
{
|
|
replayPartitionRemoveRequest(nullptr, 0, 0, 0);
|
|
}
|
|
catch(...)
|
|
{}
|
|
}
|
|
|
|
void addSession(ForeignInterfaceImpl *fi, int partitionId)
|
|
{
|
|
// next 4 lines: comment please
|
|
m_references += fi;
|
|
for (const auto &info : std::as_const(m_replayPartitions))
|
|
{
|
|
fi->write(info);
|
|
}
|
|
|
|
// take ownership of 'own' replays
|
|
for (auto it = m_replayPartitions.begin(); it != m_replayPartitions.end(); ++it)
|
|
{
|
|
ReplayPartitionEx &rp = it.value();
|
|
|
|
if (rp.fi != nullptr)
|
|
continue;
|
|
|
|
if (rp.id() == partitionId)
|
|
rp.fi = fi;
|
|
}
|
|
}
|
|
|
|
void removeSession(ForeignInterfaceImpl *fi, bool wait = true) noexcept(false)
|
|
{
|
|
m_references -= fi;
|
|
for (const auto &info : std::as_const(m_replayPartitions))
|
|
fi->remove(info);
|
|
|
|
replayPartitionRemoveRequest(fi, 0, 0, 0, wait);
|
|
}
|
|
|
|
|
|
int replayPartitionCreateRequest(int domainId, int domainIdReplay, const QUrl &tcpConnectionUrl,
|
|
int exerciseId, int workstationId, int operatorId, Core::ReplayType replayType, bool wait = true) noexcept(false)
|
|
{
|
|
|
|
// clarification need for: domainId, partitionId, replayId, idToUse... : same or different?
|
|
// 'idToUse will become replayPartitionId'
|
|
// assume: replayId == partitionId == replayPartitionId == idToUse
|
|
|
|
DEBUG_LOG(m_logger) << LOGFUNCTION << " START"
|
|
<< ", domainId = " << domainId
|
|
<< ", domainIdReplay = " << domainIdReplay
|
|
<< ", tcpConnectionUrl = " << tcpConnectionUrl.toDisplayString()
|
|
<< ", exerciseId = " << exerciseId
|
|
<< ", workstationId = " << workstationId
|
|
<< ", operatorId = " << operatorId
|
|
<< ", replayType = " << replayType
|
|
<< ", wait = " << wait;
|
|
|
|
// find free replay id
|
|
QSet<int> keys;
|
|
for (auto it = m_replayPartitions.constBegin(); it != m_replayPartitions.constEnd(); ++it)
|
|
keys += it.key();
|
|
|
|
int idToUse = -1;
|
|
for (int i = 1; i < 20; ++i) // why 20??
|
|
{
|
|
if (keys.contains(i))
|
|
continue;
|
|
idToUse = i;
|
|
break;
|
|
}
|
|
|
|
|
|
DEBUG_LOG(m_logger) << LOGFUNCTION << " idToUse = " << idToUse;
|
|
|
|
|
|
// found a valid one
|
|
if (idToUse > 0)
|
|
{
|
|
ReplayPartitionEx repl;
|
|
repl.proc = new QProcess(this);
|
|
QProcess *proc = repl.proc;
|
|
|
|
connect(repl.proc, &QProcess::errorOccurred,
|
|
this, [this, idToUse](const QProcess::ProcessError &error)
|
|
{
|
|
ERROR_LOG(m_logger) << "QProcess::ProcessError occured: "
|
|
<< ", idToUse = " << idToUse;
|
|
qCritical() << "QProcess::ProcessError: " << error;
|
|
} );
|
|
|
|
connect(repl.proc, QOverload<int, QProcess::ExitStatus>::of(&QProcess::finished),
|
|
this, [this, proc, idToUse]() { processFinished(proc, idToUse); } );
|
|
|
|
connect(proc, &QProcess::readyReadStandardError, this, [this, idToUse, proc]()
|
|
{
|
|
DEBUG_LOG(m_logger) << "REPLAY #" << idToUse << " [!]: " << QString::fromUtf8(proc->readAllStandardError());
|
|
});
|
|
connect(proc, &QProcess::readyReadStandardOutput, this, [this, idToUse, proc]()
|
|
{
|
|
DEBUG_LOG(m_logger) << "REPLAY #" << idToUse << " [ ]: " << QString::fromUtf8(proc->readAllStandardOutput());
|
|
});
|
|
|
|
repl.fi = nullptr;
|
|
repl.id(idToUse);
|
|
repl.name((QStringLiteral("srp_%1").arg(idToUse)).toStdString());
|
|
repl.exerciseId(exerciseId);
|
|
repl.workstationId(workstationId);
|
|
repl.operatorId(operatorId);
|
|
repl.replayType(replayType);
|
|
repl.domainId(domainIdReplay);
|
|
repl.proc->setProgram(QCoreApplication::applicationDirPath() + QStringLiteral("/svcSituationReplay"));
|
|
QString input = QString::number(domainId) + QLatin1Char(',') + QString::number(exerciseId);
|
|
if (!tcpConnectionUrl.isEmpty())
|
|
input += QLatin1Char(',') + tcpConnectionUrl.toString();
|
|
|
|
QString output = QString::number(repl.domainId()) + QLatin1Char(',') + QString::number(repl.id());
|
|
|
|
repl.proc->setArguments({QStringLiteral("--input"), input,
|
|
QStringLiteral("--output"), output,
|
|
QStringLiteral("-e"), QString::number(exerciseId)});
|
|
|
|
m_replayPartitions.insert(idToUse, repl);
|
|
|
|
DEBUG_LOG(m_logger) << LOGFUNCTION << " repl.proc->start()...";
|
|
repl.proc->start();
|
|
DEBUG_LOG(m_logger) << LOGFUNCTION << " repl.proc->start()...DONE";
|
|
|
|
if (wait)
|
|
{
|
|
|
|
//#define APPROACH_AWAIT_BY_MIKE
|
|
#ifdef APPROACH_AWAIT_BY_MIKE
|
|
try
|
|
{
|
|
connect(repl.proc, &QProcess::started, this, [this]() {
|
|
NOTICE_LOG(m_logger) << " SRM: Got signal 'QProcess::started'";
|
|
});
|
|
connect(repl.proc, QOverload<int, QProcess::ExitStatus>::of(&QProcess::finished), this, [this](int exitCode, QProcess::ExitStatus exitStatus) {
|
|
NOTICE_LOG(m_logger) << " SRM: Got signal 'QProcess::finished'"
|
|
<< ", exitCode = " << exitCode
|
|
<< ", exitStatus = " << exitStatus;
|
|
});
|
|
|
|
NOTICE_LOG(m_logger) << " LtsUtils::awaitWithCancelationSignal ... ";
|
|
LtsUtils::awaitWithCancelationSignal(repl.proc,
|
|
&QProcess::started,
|
|
repl.proc,
|
|
QOverload<int, QProcess::ExitStatus>::of(&QProcess::finished),
|
|
60 * 1000); // 1 min
|
|
NOTICE_LOG(m_logger) << " LtsUtils::awaitWithCancelationSignal ... DONE";
|
|
|
|
processStarted(proc, idToUse);
|
|
DEBUG_LOG(m_logger) << "Replay started: " << idToUse;
|
|
return idToUse; // <-- return
|
|
}
|
|
catch (const std::exception &ex)
|
|
{
|
|
auto error(repl.proc->errorString());
|
|
throw std::runtime_error(QStringLiteral("Replay failed to start! %1").arg(error.isEmpty() ? QString::fromUtf8(ex.what()) : error).toStdString().c_str());
|
|
}
|
|
catch (...)
|
|
{
|
|
fprintf(stderr, "awaitWithCancelationSignal has thrown an error ...\n");
|
|
} // catch
|
|
|
|
#else // APPROACH_AWAIT_BY_MIKE
|
|
// rebuild by former version... hopefully it works:
|
|
|
|
const bool bIsStartedSuccessfully = repl.proc->waitForStarted(7500); // Qt docu: Returns true if the process was started successfully;
|
|
if (bIsStartedSuccessfully)
|
|
{
|
|
processStarted(proc, idToUse);
|
|
DEBUG_LOG(m_logger) << "Replay started: " << idToUse;
|
|
return idToUse; // <-- return
|
|
}
|
|
else
|
|
{
|
|
// cleanup:
|
|
idToUse = -1;
|
|
wait = false;
|
|
if (repl.proc)
|
|
repl.proc->deleteLater(); // repl.proc is a QPointer
|
|
// repl.proc = nullptr; // repl.proc is a QPointer
|
|
// useless with idToUse == -1 !!??!!
|
|
m_replayPartitions[idToUse].proc = nullptr; // TODO: don't use operator[] !
|
|
}
|
|
#endif // APPROACH_AWAIT_BY_MIKE
|
|
|
|
|
|
}
|
|
else
|
|
{
|
|
connect(repl.proc, &QProcess::started,
|
|
this, [this, proc, idToUse]() { processStarted(proc, idToUse); } );
|
|
}
|
|
} // if (idToUse > 0)
|
|
|
|
// didn't find a valid one!
|
|
|
|
if (wait)
|
|
{
|
|
throw std::runtime_error("Failed to create replay!");
|
|
}
|
|
|
|
return idToUse;
|
|
}
|
|
|
|
void processFinished(QProcess *proc, int replayId)
|
|
{
|
|
auto it = m_replayPartitions.find(replayId);
|
|
if (it == m_replayPartitions.end())
|
|
return;
|
|
|
|
if (proc->exitCode() != 0 || proc->exitStatus() != QProcess::NormalExit)
|
|
{
|
|
WARNING_LOG(m_logger) << "SituationReplay process exited with error: " << proc->exitCode()
|
|
<< ", status: " << proc->exitStatus()
|
|
<< ", stdout: " << proc->readAllStandardOutput()
|
|
<< ", stderr: " << proc->readAllStandardError();
|
|
}
|
|
|
|
const auto &rp = it.value();
|
|
for (auto ptr : std::as_const(m_references))
|
|
ptr->remove(rp);
|
|
|
|
proc->deleteLater();
|
|
m_replayPartitions.erase(it);
|
|
}
|
|
|
|
void processStarted(QProcess *proc, int replayId)
|
|
{
|
|
Q_UNUSED(proc);
|
|
auto it = m_replayPartitions.find(replayId);
|
|
if (it == m_replayPartitions.end())
|
|
{
|
|
WARNING_LOG(m_logger) << "Got a processStarted signal from a process for replay " << replayId
|
|
<< " but found not corresponding entry in m_replayPartitions";
|
|
return;
|
|
}
|
|
|
|
const ReplayPartitionEx &rp = it.value();
|
|
for (ForeignInterfaceImpl* ptr : std::as_const(m_references))
|
|
ptr->write(rp);
|
|
}
|
|
|
|
void replayPartitionRemoveRequest(ForeignInterfaceImpl *fi, int partitionId, int workstationId, int operatorId, bool wait = true) noexcept(false)
|
|
{
|
|
bool bFoundOne = false;
|
|
int idToReturn = partitionId;
|
|
if (idToReturn == 0)
|
|
idToReturn = workstationId;
|
|
if (idToReturn == 0)
|
|
idToReturn = operatorId;
|
|
|
|
auto it = m_replayPartitions.begin();
|
|
while (it != m_replayPartitions.end())
|
|
{
|
|
const ReplayPartitionEx &rp = it.value();
|
|
if ((partitionId != 0 && rp.id() != partitionId) ||
|
|
(workstationId != 0 && rp.workstationId() != workstationId) ||
|
|
(operatorId != 0 && rp.operatorId() != operatorId) ||
|
|
(fi != nullptr && rp.fi != fi))
|
|
{
|
|
++it;
|
|
continue;
|
|
}
|
|
|
|
if (rp.proc)
|
|
{
|
|
rp.proc->disconnect(this);
|
|
if (rp.proc->state() == QProcess::NotRunning)
|
|
{
|
|
delete rp.proc;
|
|
}
|
|
else
|
|
{
|
|
rp.proc->terminate();
|
|
|
|
|
|
QPointer<QProcess> proc = rp.proc;
|
|
auto cb = [this, proc]()
|
|
{
|
|
if (!proc)
|
|
return;
|
|
if (proc->state() != QProcess::NotRunning)
|
|
proc->kill();
|
|
const auto err(proc->readAllStandardError());
|
|
if (!err.isEmpty())
|
|
WARNING_LOG(m_logger) << "stderr: " << err;
|
|
delete proc;
|
|
};
|
|
|
|
|
|
if (wait)
|
|
{
|
|
try
|
|
{
|
|
DEBUG_LOG(m_logger) << "Waiting for replay to process to shutdown... Replay id: " << rp.id();
|
|
LtsUtils::awaitTimeout(10000, rp.proc, QOverload<int, QProcess::ExitStatus>::of(&QProcess::finished), true);
|
|
}
|
|
catch (...)
|
|
{
|
|
WARNING_LOG(m_logger) << "Timeout waiting for the replay process to quit gracefully - killing the process! Replay id: " << rp.id();
|
|
cb();
|
|
}
|
|
}
|
|
else // dont wait
|
|
{
|
|
QTimer::singleShot(10000, rp.proc, cb);
|
|
}
|
|
}
|
|
}
|
|
|
|
for (auto *ptr : std::as_const(m_references))
|
|
ptr->remove(rp);
|
|
|
|
it = m_replayPartitions.erase(it);
|
|
bFoundOne = true;
|
|
}
|
|
|
|
// workstationId / operatorId are wildcard operations which may find nothing which is ok
|
|
if (partitionId == 0)
|
|
bFoundOne = true;
|
|
|
|
if (!bFoundOne && wait)
|
|
{
|
|
throw std::runtime_error("Requested remove replay for a replay that doesn't exist!");
|
|
}
|
|
}
|
|
|
|
|
|
std::shared_ptr<DDSConnector::DdsConnector> getDdsConnector(const std::shared_ptr<DDSConnector::DdsConnector> &sessionConnector)
|
|
{
|
|
if (m_ddsConnector)
|
|
return m_ddsConnector;
|
|
|
|
if (sessionConnector->getTcpConnectionUrl().isEmpty())
|
|
{
|
|
m_ddsConnector = sessionConnector;
|
|
return sessionConnector;
|
|
}
|
|
|
|
int domainId = 1; // ok since we are using ShmUdpLocalhost
|
|
m_ddsConnector.reset(new DDSConnector::DdsConnector(domainId,
|
|
IDL::PartitionType::Exercise,
|
|
sessionConnector->getPartitionId(),
|
|
sessionConnector->getPartitionId(),
|
|
DDS::QosProfile::ConnectionType::ShmUdp));
|
|
m_ddsConnector->initPublisherSubscriber();
|
|
|
|
return m_ddsConnector;
|
|
}
|
|
|
|
std::shared_ptr<Core::Interface::Dtc> dtc(const std::shared_ptr<DDSConnector::DdsConnector> &sessionConnector)
|
|
{
|
|
if (m_dtc)
|
|
return m_dtc;
|
|
auto ddsConnector = getDdsConnector(sessionConnector);
|
|
m_dtc = Core::Interface::Dtc::create(ddsConnector);
|
|
m_dtc->enableReplayPartition();
|
|
return m_dtc;
|
|
}
|
|
|
|
std::shared_ptr<DTS::ReplayPartition> dtsReplay(const std::shared_ptr<DDSConnector::DdsConnector> &sessionConnector)
|
|
{
|
|
if (m_replayPartion)
|
|
return m_replayPartion;
|
|
auto ddsConnector = getDdsConnector(sessionConnector);
|
|
m_replayPartion.reset(ddsConnector->createDTS<DTS::ReplayPartitionImpl>());
|
|
return m_replayPartion;
|
|
}
|
|
|
|
private:
|
|
struct ReplayPartitionEx : public Core::ReplayPartition
|
|
{
|
|
QPointer<QProcess> proc; // = nullptr;
|
|
ForeignInterfaceImpl *fi = nullptr;
|
|
};
|
|
|
|
QMap<int, ReplayPartitionEx> m_replayPartitions; // idToUse - ReplayPartitionEx
|
|
QSet<ForeignInterfaceImpl*> m_references;
|
|
std::shared_ptr<DDSConnector::DdsConnector> m_ddsConnector;
|
|
std::shared_ptr<Core::Interface::Dtc> m_dtc;
|
|
std::shared_ptr<DTS::ReplayPartition> m_replayPartion;
|
|
|
|
NGL::INGLLogger &m_logger;
|
|
};
|
|
|
|
|
|
DataPlugin::DataPlugin()
|
|
: m_replayHandler(std::make_shared<ReplayHandler>())
|
|
{}
|
|
|
|
DataPlugin::~DataPlugin()
|
|
{}
|
|
|
|
std::shared_ptr<DataPluginBase::Interface> DataPlugin::createDataInterface(LtsSession::LtsSessionDataDds *session)
|
|
{
|
|
std::shared_ptr<Core::Interface::Dtc> dtc = m_replayHandler->dtc(session->ddsConnector());
|
|
return std::make_shared<DataInterfaceImpl>(m_replayHandler, session, dtc);
|
|
}
|
|
|
|
DataInterfaceImpl::DataInterfaceImpl(const std::shared_ptr<ReplayHandler> &replayHandler,
|
|
LtsSession::LtsSessionDataDds *session,
|
|
const std::shared_ptr<Core::Interface::Dtc> &dtc)
|
|
: DataInterface()
|
|
, m_dtc(dtc)
|
|
, m_foreignInterface(std::make_shared<ForeignInterfaceImpl>(replayHandler, session))
|
|
{
|
|
}
|
|
|
|
ForeignInterfaceImpl::ForeignInterfaceImpl(const std::shared_ptr<ReplayHandler> &replayHandler,
|
|
LtsSession::LtsSessionDataDds *session)
|
|
: m_replayHandler(replayHandler)
|
|
, m_session(session)
|
|
, m_logger(getLogger("ForeignInterfaceImpl"))
|
|
{
|
|
if (m_session->partitionType() == IDL::PartitionType::Exercise ||
|
|
m_session->partitionType() == IDL::PartitionType::Replay)
|
|
m_replayHandler->addSession(this, session->partitionId());
|
|
}
|
|
|
|
ForeignInterfaceImpl::~ForeignInterfaceImpl()
|
|
{
|
|
m_replayHandler->removeSession(this, false /* dont' wait for replay service to quit */);
|
|
}
|
|
|
|
void ForeignInterfaceImpl::removeReplayPartitionsWorkstation(const int workstationId) noexcept(false)
|
|
{
|
|
if (!m_session)
|
|
{
|
|
throw std::runtime_error("No session available.");
|
|
}
|
|
m_replayHandler->replayPartitionRemoveRequest(nullptr, 0, workstationId, 0);
|
|
}
|
|
|
|
void ForeignInterfaceImpl::removeReplayPartitionsOperator(const int operatorId) noexcept(false)
|
|
{
|
|
if (!m_session)
|
|
{
|
|
throw std::runtime_error("No session available.");
|
|
}
|
|
m_replayHandler->replayPartitionRemoveRequest(nullptr, 0, 0, operatorId);
|
|
}
|
|
|
|
void ForeignInterfaceImpl::replayPartitionRemoveRequest(int partitionId, int workstationId, int operatorId) noexcept(false)
|
|
{
|
|
if (!m_session)
|
|
{
|
|
throw std::runtime_error("No session available.");
|
|
}
|
|
m_replayHandler->replayPartitionRemoveRequest(nullptr, partitionId, workstationId, operatorId);
|
|
}
|
|
|
|
int ForeignInterfaceImpl::replayPartitionCreateRequest(int exerciseId, int workstationId, int operatorId, Core::ReplayType replayType) noexcept(false)
|
|
{
|
|
if (!m_session)
|
|
{
|
|
throw std::runtime_error("No session available.");
|
|
}
|
|
|
|
int domainId = m_session->domainId();
|
|
if (m_session->isReplayPartition())
|
|
{
|
|
/* starting a replay partition from a replay partition would normally inc the
|
|
* domainId to the already inc'd domainId - dec the already inc'd domain id
|
|
* to keep the same domainId for every replay partition */
|
|
if (domainId == 1)
|
|
domainId= 199;
|
|
else
|
|
domainId--;
|
|
}
|
|
int domainIdReplay = (domainId % 200) + 1; // mod aka "if (domainId > 200) domainId = 0;" - domainIds can never be > 199
|
|
|
|
|
|
return m_replayHandler->replayPartitionCreateRequest(domainId, domainIdReplay,
|
|
m_session->tcpConnectionUrl(),
|
|
exerciseId, workstationId, operatorId, replayType);
|
|
}
|
|
|
|
void ForeignInterfaceImpl::write(const Core::ReplayPartition &repl)
|
|
{
|
|
if (!m_session)
|
|
return;
|
|
m_replayHandler->dtsReplay(m_session->ddsConnector())->write(repl);
|
|
}
|
|
|
|
void ForeignInterfaceImpl::remove(const Core::ReplayPartition &repl)
|
|
{
|
|
if (!m_session)
|
|
return;
|
|
m_replayHandler->dtsReplay(m_session->ddsConnector())->remove(repl);
|
|
}
|
|
|
|
} // namespace SituationReplayManagement
|
|
|
|
#include "srm_data_plugin.moc"
|