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 acqIdentifierToAcqRequestMapIt = impl->m_AcqIdentifierToAcqRequestMap.find(nextAcqId);
+                     auto oldAcqId = it->second.second;
+                     auto acqIdentifierToAcqRequestMapIt = impl->m_AcqIdentifierToAcqRequestMap.find(oldAcqId);
                      if (acqIdentifierToAcqRequestMapIt != impl->m_AcqIdentifierToAcqRequestMap.cend()) {
-                         auto request = acqIdentifierToAcqRequestMapIt->second;
-                         varRequestIdCanceled = request.m_VarRequestId;
+                         auto oldAcqRequest = acqIdentifierToAcqRequestMapIt->second;
+                         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,
-                                               acqRequest.m_CacheRangeRequested, aIdToADPVit->second);
+                             emit dataProvided(varId, rangeRequested, 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: ")
-                                                                          << varRequestIdCanceled;
+                                 qCInfo(LOG_VariableAcquisitionWorker()) << tr("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