Merge pull request #976 from lsst/tickets/DM-52880

Tickets/dm 52880

Author: jgates108

src/CMakeLists.txt

@@ -87,3 +87,4 @@ add_subdirectory(wpublish)
 add_subdirectory(wsched)
 add_subdirectory(www)
 
+

src/cconfig/CzarConfig.cc

@@ -143,7 +143,7 @@ http::AuthContext CzarConfig::httpAuthContext() const {
                              _replicationAdminAuthKey->getVal());
 }
 
-void CzarConfig::setId(qmeta::CzarId id) {
+void CzarConfig::setId(CzarId id) {
     _czarId = id;
     // Update the relevant section of the JSON-ified configuration.
     _jsonConfig["actual"]["identity"]["id"] = std::to_string(_czarId);

src/cconfig/CzarConfig.h

@@ -34,8 +34,8 @@
 #include <nlohmann/json.hpp>
 
 // Qserv headers
+#include "global/intTypes.h"
 #include "mysql/MySqlConfig.h"
-#include "qmeta/types.h"
 #include "util/ConfigStore.h"
 #include "util/ConfigValMap.h"
 
@@ -177,6 +177,9 @@ class CzarConfig {
     /// the method then the monitoring will be disabled.
     unsigned int czarStatsUpdateIvalSec() const { return _czarStatsUpdateIvalSec->getVal(); }
 
+    /// Maximum number of attempts to run a given job before aborting the entire user query.
+    unsigned int jobMaxAttempts() const { return _jobMaxAttempts->getVal(); }
+
     /// A worker is considered fully ALIVE if the last update from the worker has been
     /// heard in less than _activeWorkerTimeoutAliveSecs seconds.
     int getActiveWorkerTimeoutAliveSecs() const { return _activeWorkerTimeoutAliveSecs->getVal(); }
@@ -230,14 +233,14 @@ class CzarConfig {
     std::string const& name() const { return _czarName; }
 
     /// @return The unique identifier of Czar.
-    qmeta::CzarId id() const { return _czarId; }
+    CzarId id() const { return _czarId; }
 
     /// Set a unique identifier of Czar.
     /// @note In the current implementation of Qserv a value of the identifier is not
     /// available at a time when the configuration is initialized. The identifier is generated
     /// when registering Czar by name in a special table of teh Qserv database.
     /// This logic should be fixed in some future version of Qserv.
-    void setId(qmeta::CzarId id);
+    void setId(CzarId id);
 
     /// @return The interval in seconds for cleaning up the in-progress queries in QMeta.
     unsigned int getInProgressCleanupIvalSec() const { return _inProgressCleanupIvalSec->getVal(); }
@@ -266,8 +269,7 @@ class CzarConfig {
     /// The unique identifier of the Czar instance, the real vale cannot be
     /// acquired until later. Using a crazy initial value in hopes of highlighting
     /// issues.
-    /// TODO: Is this really the right place for this? (previously undefined)
-    qmeta::CzarId _czarId = std::numeric_limits<qmeta::CzarId>::max();
+    CzarId _czarId = std::numeric_limits<CzarId>::max();
 
     nlohmann::json _jsonConfig;  ///< JSON-ified configuration
 
@@ -370,7 +372,6 @@ class CzarConfig {
     // UberJobs
     CVTIntPtr _uberJobMaxChunks =
             util::ConfigValTInt::create(_configValMap, "uberjob", "maxChunks", notReq, 10000);
-
     CVTIntPtr _qMetaSecsBetweenChunkCompletionUpdates = util::ConfigValTInt::create(
             _configValMap, "tuning", "qMetaSecsBetweenChunkCompletionUpdates", notReq, 60);
     CVTIntPtr _interactiveChunkLimit =
@@ -383,6 +384,8 @@ class CzarConfig {
             util::ConfigValTBool::create(_configValMap, "tuning", "notifyWorkersOnCzarRestart", notReq, 1);
     CVTIntPtr _czarStatsUpdateIvalSec =
             util::ConfigValTInt::create(_configValMap, "tuning", "czarStatsUpdateIvalSec", notReq, 1);
+    CVTUIntPtr _jobMaxAttempts =
+            util::ConfigValTUInt::create(_configValMap, "tuning", "jobMaxAttempts", notReq, 5);
 
     // Replicator
     CVTStrPtr _replicationInstanceId =

src/ccontrol/MergingHandler.cc

@@ -234,6 +234,17 @@ qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uber
         }
     };
     csvMemDisk->transferDataFromWorker(transferFunc);
+    if (csvMemDisk->isCancelled()) {
+        // Since csvMemDisk was cancelled, avoid merging to avoid risks of contamination.
+        LOGS(_log, LOG_LVL_DEBUG, __func__ << " csvMemDisk cancelled");
+        return qdisp::MergeEndStatus(false);
+    }
+
+    bool mergeOk = _startMerge();
+    if (!mergeOk) {
+        LOGS(_log, LOG_LVL_DEBUG, __func__ << " merge cancelled");
+        return qdisp::MergeEndStatus(false);
+    }
 
     // Attempt the actual merge.
     bool fileMergeSuccess = _infileMerger->mergeHttp(uberJob, fileSize, csvMemDisk);
@@ -264,11 +275,31 @@ qdisp::MergeEndStatus MergingHandler::_mergeHttp(qdisp::UberJob::Ptr const& uber
     return mergeEStatus;
 }
 
-void MergingHandler::cancelFileMerge() {
-    auto csvStrm = _csvMemDisk.lock();
-    if (csvStrm != nullptr) {
-        csvStrm->cancel();
+bool MergingHandler::cancelFileMerge() {
+    lock_guard mergeStateLock(_mergeStateMtx);
+    if (_mergeState == PREMERGE || _mergeState == CANCELLED) {
+        _mergeState = CANCELLED;
+        auto csvStrm = _csvMemDisk.lock();
+        if (csvStrm != nullptr) {
+            csvStrm->cancel();
+        }
+        // Merging to the result table hasn't been started, so
+        // this can be cancelled.
+        return true;
+    }
+    // Cancelling at this point would probably corrupt the result table.
+    return false;
+}
+
+bool MergingHandler::_startMerge() {
+    lock_guard mergeStateLock(_mergeStateMtx);
+    if (_mergeState == PREMERGE) {
+        _mergeState = MERGING;
+        // Merging hasn't been cancelled, so it's ok to start.
+        return true;
     }
+    // Merge was cancelled.
+    return false;
 }
 
 void MergingHandler::_setError(int code, std::string const& msg, int errorState) {

src/ccontrol/MergingHandler.h

@@ -60,6 +60,9 @@ namespace lsst::qserv::ccontrol {
 class MergingHandler : public qdisp::ResponseHandler {
 public:
     typedef std::shared_ptr<MergingHandler> Ptr;
+
+    enum MergeState { PREMERGE, MERGING, CANCELLED };
+
     virtual ~MergingHandler();
 
     /// @param merger downstream merge acceptor
@@ -78,7 +81,8 @@ class MergingHandler : public qdisp::ResponseHandler {
     void errorFlush(std::string const& msg, int code) override;
 
     /// Stop an ongoing file merge, if possible.
-    void cancelFileMerge() override;
+    /// @return true if the merge was cancelled.
+    bool cancelFileMerge() override;
 
     /// Print a string representation of the receiver to an ostream
     std::ostream& print(std::ostream& os) const override;
@@ -91,6 +95,11 @@ class MergingHandler : public qdisp::ResponseHandler {
     /// Set error code and string.
     void _setError(int code, std::string const& msg, int errorState);
 
+    /// Return true if merging should be started and set _mergeState to MERGING.
+    /// This should only be called once after the file has been collected and
+    /// before merging with the result table starts.
+    bool _startMerge();
+
     // All instances of the HTTP client class are members of the same pool. This allows
     // connection reuse and a significant reduction of the kernel memory pressure.
     // Note that the pool gets instantiated at the very first call to method _getHttpConnPool()
@@ -106,6 +115,11 @@ class MergingHandler : public qdisp::ResponseHandler {
 
     std::weak_ptr<qdisp::Executive> _executive;    ///< Weak pointer to the executive for errors.
     std::weak_ptr<mysql::CsvMemDisk> _csvMemDisk;  ///< Weak pointer to cancel infile merge.
+
+    /// Indicates merge state of the result table relating to the UberJob associated with
+    /// instance of MergingHandler.
+    MergeState _mergeState = PREMERGE;
+    std::mutex _mergeStateMtx;  ///< Protectes _mergeState
 };
 
 }  // namespace lsst::qserv::ccontrol

src/ccontrol/UserQuery.h

@@ -39,7 +39,6 @@
 // Qserv headers
 #include "ccontrol/QueryState.h"
 #include "global/intTypes.h"
-#include "qmeta/types.h"
 
 // Forward decl
 namespace lsst::qserv::qmeta {

src/ccontrol/UserQueryAsyncResult.cc

@@ -47,7 +47,7 @@ LOG_LOGGER _log = LOG_GET("lsst.qserv.ccontrol.UserQueryAsyncResult");
 namespace lsst::qserv::ccontrol {
 
 // Constructors
-UserQueryAsyncResult::UserQueryAsyncResult(QueryId queryId, qmeta::CzarId czarId,
+UserQueryAsyncResult::UserQueryAsyncResult(QueryId queryId, CzarId czarId,
                                            std::shared_ptr<qmeta::QMeta> const& qMeta)
         : UserQuery(),
           _queryId(queryId),

src/ccontrol/UserQueryAsyncResult.h

@@ -29,7 +29,6 @@
 // Qserv headers
 #include "ccontrol/UserQuery.h"
 #include "qmeta/QInfo.h"
-#include "qmeta/types.h"
 
 namespace lsst::qserv::qmeta {
 class MessageStore;
@@ -58,7 +57,7 @@ class UserQueryAsyncResult : public UserQuery {
      *  @param czarId ID for current czar
      *  @param qMeta QMeta instance
      */
-    UserQueryAsyncResult(QueryId queryId, qmeta::CzarId czarId, std::shared_ptr<qmeta::QMeta> const& qMeta);
+    UserQueryAsyncResult(QueryId queryId, CzarId czarId, std::shared_ptr<qmeta::QMeta> const& qMeta);
 
     // Destructor
     ~UserQueryAsyncResult();
@@ -103,7 +102,7 @@ class UserQueryAsyncResult : public UserQuery {
 protected:
 private:
     QueryId _queryId;
-    qmeta::CzarId _czarId;
+    CzarId _czarId;
     std::shared_ptr<qmeta::QMeta> _qMeta;
     qmeta::QInfo _qInfo;
     std::shared_ptr<qmeta::MessageStore> _messageStore;

src/ccontrol/UserQueryFactory.cc

@@ -234,7 +234,7 @@ UserQueryFactory::UserQueryFactory(qproc::DatabaseModels::Ptr const& dbModels, s
     _userQuerySharedResources->queryMetadata->cleanupQueriesAtStart(_userQuerySharedResources->czarId);
 
     // Add logging context with czar ID
-    qmeta::CzarId czarId = _userQuerySharedResources->czarId;
+    auto const czarId = _userQuerySharedResources->czarId;
     LOG_MDC_INIT([czarId]() { LOG_MDC("CZID", std::to_string(czarId)); });
 
     // BOOST ASIO service is started to process asynchronous timer requests

src/ccontrol/UserQueryInvalid.h

@@ -33,7 +33,6 @@
 // Qserv headers
 #include "ccontrol/UserQuery.h"
 #include "qmeta/MessageStore.h"
-#include "qmeta/types.h"
 
 // Forward decl