GIT_REPOSITORIES/LPP/Users/mperrinel/SciQLop-fork Commit - r818:68959f479714

Implementation of varRequestId cancel for the worker

perrinel -

r818:68959f479714

parent child

Context file:

r818:68959f479714

core/src/Variable/VariableAcquisitionWorker.cpp +100 -13

                 /// Remove the current request and execute the next one if exist
                 void updateToNextRequest(QUuid vIdentifier);
+                /// Remove and/or abort all AcqRequest in link with varRequestId
+                void cancelVarRequest(QUuid varRequestId);
+                void removeAcqRequest(QUuid acqRequestId);
                 QMutex m_WorkingMutex;
                 QReadWriteLock m_Lock;
                 // Request creation
                 auto acqRequest = AcquisitionRequest{};
-                qCInfo(LOG_VariableAcquisitionWorker()) << tr("TpushVariableRequest ") << vIdentifier;
+                qCInfo(LOG_VariableAcquisitionWorker()) << tr("TpushVariableRequest ") << vIdentifier
+                                                        << varRequestId;
                 acqRequest.m_VarRequestId = varRequestId;
                 acqRequest.m_vIdentifier = vIdentifier;
                 acqRequest.m_DataProviderParameters = parameters;
                 auto it = impl->m_VIdentifierToCurrrentAcqIdNextIdPairMap.find(vIdentifier);
                 if (it != impl->m_VIdentifierToCurrrentAcqIdNextIdPairMap.cend()) {
                     // A current request already exists, we can replace the next one
-                    auto nextAcqId = it->second.second;
+                    auto oldAcqId = it->second.second;
-                    auto acqIdentifierToAcqRequestMapIt = impl->m_AcqIdentifierToAcqRequestMap.find(nextAcqId);
+                    auto acqIdentifierToAcqRequestMapIt = impl->m_AcqIdentifierToAcqRequestMap.find(oldAcqId);
                     if (acqIdentifierToAcqRequestMapIt != impl->m_AcqIdentifierToAcqRequestMap.cend()) {
-                        auto request = acqIdentifierToAcqRequestMapIt->second;
+                        auto oldAcqRequest = acqIdentifierToAcqRequestMapIt->second;
-                        varRequestIdCanceled = request.m_VarRequestId;
+                        varRequestIdCanceled = oldAcqRequest.m_VarRequestId;
                     }
+                    // remove old acqIdentifier from the worker
+                    impl->cancelVarRequest(oldAcqId);
+                    //        impl->m_AcqIdentifierToAcqRequestMap.erase(oldAcqId);
                     it->second.second = acqRequest.m_AcqIdentifier;
                     impl->unlock();
                 }
                 else {
                         impl->unlock();
                         // Remove the current request from the worker
-                        impl->lockWrite();
                         impl->updateToNextRequest(vIdentifier);
-                        impl->unlock();
                         // notify the request aborting to the provider
                         request.m_Provider->requestDataAborting(currentAcqId);
                     // if the counter is 0, we can return data then run the next request if it exists and
                     // removed the finished request
                     if (acqRequest.m_Size == acqRequest.m_Progression) {
+                        auto varId = acqRequest.m_vIdentifier;
+                        auto rangeRequested = acqRequest.m_RangeRequested;
+                        auto cacheRangeRequested = acqRequest.m_CacheRangeRequested;
                         // Return the data
                         aIdToADPVit = impl->m_AcqIdentifierToAcqDataPacketVectorMap.find(acqIdentifier);
                         if (aIdToADPVit != impl->m_AcqIdentifierToAcqDataPacketVectorMap.cend()) {
-                            emit dataProvided(acqRequest.m_vIdentifier, acqRequest.m_RangeRequested,
+                            emit dataProvided(varId, rangeRequested, cacheRangeRequested, aIdToADPVit->second);
-                                              acqRequest.m_CacheRangeRequested, aIdToADPVit->second);
                         }
+                        impl->unlock();
                         // Update to the next request
                         impl->updateToNextRequest(acqRequest.m_vIdentifier);
                     }
+                    else {
+                        impl->unlock();
+                    }
                 }
                 else {
+                    impl->unlock();
                     qCWarning(LOG_VariableAcquisitionWorker())
                         << tr("Impossible to retrieve AcquisitionRequest for the incoming data.");
                 }
-                impl->unlock();
             }
             void VariableAcquisitionWorker::onExecuteRequest(QUuid acqIdentifier)
             void VariableAcquisitionWorker::VariableAcquisitionWorkerPrivate::updateToNextRequest(
                 QUuid vIdentifier)
             {
+                lockRead();
                 auto it = m_VIdentifierToCurrrentAcqIdNextIdPairMap.find(vIdentifier);
                 if (it != m_VIdentifierToCurrrentAcqIdNextIdPairMap.cend()) {
                     if (it->second.second.isNull()) {
+                        unlock();
                         // There is no next request, we can remove the variable request
                         removeVariableRequest(vIdentifier);
                     }
                     else {
                         auto acqIdentifierToRemove = it->second.first;
                         // Move the next request to the current request
-                        it->second.first = it->second.second;
+                        auto nextRequestId = it->second.second;
+                        it->second.first = nextRequestId;
                         it->second.second = QUuid();
+                        unlock();
                         // Remove AcquisitionRequest and results;
+                        lockWrite();
                         m_AcqIdentifierToAcqRequestMap.erase(acqIdentifierToRemove);
                         m_AcqIdentifierToAcqDataPacketVectorMap.erase(acqIdentifierToRemove);
+                        unlock();
                         // Execute the current request
                         QMetaObject::invokeMethod(q, "onExecuteRequest", Qt::QueuedConnection,
-                                                  Q_ARG(QUuid, it->second.first));
+                                                  Q_ARG(QUuid, nextRequestId));
                     }
                 }
                 else {
+                    unlock();
                     qCCritical(LOG_VariableAcquisitionWorker())
                         << tr("Impossible to execute the acquisition on an unfound variable ");
                 }
             }
+            void VariableAcquisitionWorker::VariableAcquisitionWorkerPrivate::cancelVarRequest(
+                QUuid varRequestId)
+            {
+                lockRead();
+                // get all AcqIdentifier in link with varRequestId
+                QVector<QUuid> acqIdsToRm;
+                auto cend = m_AcqIdentifierToAcqRequestMap.cend();
+                for (auto it = m_AcqIdentifierToAcqRequestMap.cbegin(); it != cend; ++it) {
+                    if (it->second.m_VarRequestId == varRequestId) {
+                        acqIdsToRm << it->first;
+                    }
+                }
+                unlock();
+                // run aborting or removing of acqIdsToRm
+                for (auto acqId : acqIdsToRm) {
+                    removeAcqRequest(acqId);
+                }
+            }
+            void VariableAcquisitionWorker::VariableAcquisitionWorkerPrivate::removeAcqRequest(
+                QUuid acqRequestId)
+            {
+                QUuid vIdentifier;
+                std::shared_ptr<IDataProvider> provider;
+                lockRead();
+                auto acqIt = m_AcqIdentifierToAcqRequestMap.find(acqRequestId);
+                if (acqIt != m_AcqIdentifierToAcqRequestMap.cend()) {
+                    vIdentifier = acqIt->second.m_vIdentifier;
+                    provider = acqIt->second.m_Provider;
+                    auto it = m_VIdentifierToCurrrentAcqIdNextIdPairMap.find(vIdentifier);
+                    if (it != m_VIdentifierToCurrrentAcqIdNextIdPairMap.cend()) {
+                        if (it->second.first == acqRequestId) {
+                            // acqRequest is currently running -> let's aborting it
+                            unlock();
+                            // Remove the current request from the worker
+                            updateToNextRequest(vIdentifier);
+                            // notify the request aborting to the provider
+                            provider->requestDataAborting(acqRequestId);
+                        }
+                        else if (it->second.second == acqRequestId) {
+                            it->second.second = QUuid();
+                            unlock();
+                        }
+                        else {
+                            unlock();
+                        }
+                    }
+                    else {
+                        unlock();
+                    }
+                }
+                else {
+                    unlock();
+                }
+                lockWrite();
+                m_AcqIdentifierToAcqDataPacketVectorMap.erase(acqRequestId);
+                m_AcqIdentifierToAcqRequestMap.erase(acqRequestId);
+                unlock();
+            }

core/src/Variable/VariableController.cpp +7 -3

                                 varProvider);
                             if (!varRequestIdCanceled.isNull()) {
-                                qCDebug(LOG_VariableAcquisitionWorker()) << tr("vsarRequestIdCanceled: ")
+                                qCInfo(LOG_VariableAcquisitionWorker()) << tr("varRequestIdCanceled: ")
-                                                                         << varRequestIdCanceled;
+                                                                        << varRequestIdCanceled;
                                 cancelVariableRequest(varRequestIdCanceled);
                             }
                         }
                     varRequestIdQueue.pop_front();
                     if (varRequestIdQueue.empty()) {
-                        qCDebug(LOG_VariableController())
+                        qCCritical(LOG_VariableController())
                             << tr("TORM Erase REQUEST because it has been accepted") << varId;
                         m_VarIdToVarRequestIdQueueMap.erase(varId);
                     }
                         std::remove(varRequestIdQueue.begin(), varRequestIdQueue.end(), varRequestId),
                         varRequestIdQueue.end());
                     if (varRequestIdQueue.empty()) {
+                        qCCritical(LOG_VariableController())
+                            << tr("VariableControllerPrivate::cancelVariableRequest")
+                            << varIdToVarRequestIdQueueMapIt->first;
                         varIdToVarRequestIdQueueMapIt
                             = m_VarIdToVarRequestIdQueueMap.erase(varIdToVarRequestIdQueueMapIt);
                     }

plugins/mockplugin/src/CosinusProvider.cpp +1 -1

                 for (const auto &dateTime : qAsConst(times)) {
                     if (m_VariableToEnableProvider[acqIdentifier]) {
                         auto scalarSeries = this->retrieveData(acqIdentifier, dateTime, parameters.m_Data);
-                        qCDebug(LOG_CosinusProvider()) << "TORM: CosinusProvider::dataProvided";
+                        qCCritical(LOG_CosinusProvider()) << "TORM: CosinusProvider::dataProvided";
                         emit dataProvided(acqIdentifier, scalarSeries, dateTime);
                     }
                 }

General Comments 3

perrinel

|

Auto status change to "Under Review"

perrinel

|

Status change > Approved

Write
Preview

You need to be logged in to leave comments. Login now

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages