Rebase fixes.

Author: jgates108

src/ccontrol/MergingHandler.cc

@@ -39,6 +39,7 @@
 #include "lsst/log/Log.h"
 
 // Qserv headers
+#include "cconfig/CzarConfig.h"
 #include "ccontrol/msgCode.h"
 #include "global/clock_defs.h"
 #include "global/debugUtil.h"
@@ -49,7 +50,6 @@
 #include "qdisp/CzarStats.h"
 #include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
-#include "qdisp/QueryRequest.h"
 #include "qdisp/UberJob.h"
 #include "rproc/InfileMerger.h"
 #include "util/Bug.h"
@@ -84,7 +84,6 @@ lsst::qserv::TimeCountTracker<double>::CALLBACKFUNC const reportFileRecvRate =
             }
         };
 
-
 string readHttpFileAndMerge(lsst::qserv::qdisp::UberJob::Ptr const& uberJob, string const& httpUrl,
                             size_t fileSize, function<void(char const*, uint32_t)> const& messageIsReady,
                             shared_ptr<http::ClientConnPool> const& httpConnPool) {
@@ -180,52 +179,6 @@ MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> const& merge
 
 MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_TRACE, __func__); }
 
-
-bool MergingHandler::flush(proto::ResponseSummary const& resp) {
-    _wName = resp.wname();
-
-    // This is needed to ensure the job query would be staying alive for the duration
-    // of the operation to prevent inconsistency within the application.
-    auto const jobQuery = getJobQuery().lock();
-    if (jobQuery == nullptr) {
-        LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, jobQuery was NULL");
-        return false;
-    }
-    auto const jobQuery = std::dynamic_pointer_cast<qdisp::JobQuery>(jobBase);
-
-    LOGS(_log, LOG_LVL_TRACE,
-         "MergingHandler::" << __func__ << " jobid=" << resp.jobid() << " transmitsize="
-                            << resp.transmitsize() << " rowcount=" << resp.rowcount() << " rowSize="
-                            << " attemptcount=" << resp.attemptcount() << " errorcode=" << resp.errorcode()
-                            << " errormsg=" << resp.errormsg());
-
-    if (resp.errorcode() != 0 || !resp.errormsg().empty()) {
-        _error = util::Error(resp.errorcode(), resp.errormsg(), util::ErrorCode::MYSQLEXEC);
-        _setError(ccontrol::MSG_RESULT_ERROR, _error.getMsg());
-        LOGS(_log, LOG_LVL_ERROR,
-             "MergingHandler::" << __func__ << " error from worker:" << resp.wname() << " error: " << _error);
-        // This way we can track if the worker has reported this error. The current implementation
-        // requires the large result size to be reported as an error via the InfileMerger regardless
-        // of an origin of the error (Czar or the worker). Note that large results can be produced
-        // by the Czar itself, e.g., when the aggregate result of multiple worker queries is too large
-        // or by the worker when the result set of a single query is too large.
-        // The error will be reported to the Czar as a part of the response summary.
-        if (resp.errorcode() == util::ErrorCode::WORKER_RESULT_TOO_LARGE) {
-            _infileMerger->setResultSizeLimitExceeded();
-        }
-        return false;
-    }
-
-    bool const success = _merge(resp, jobQuery);
-
-    if (success) {
-        _infileMerger->mergeCompleteFor(resp.jobid());
-        qdisp::CzarStats::get()->addTotalRowsRecv(resp.rowcount());
-        qdisp::CzarStats::get()->addTotalBytesRecv(resp.transmitsize());
-    }
-    return success;
-}
-
 void MergingHandler::errorFlush(std::string const& msg, int code) {
     _setError(code, msg, util::ErrorCode::RESULT_IMPORT);
     // Might want more info from result service.

src/ccontrol/MergingHandler.h

@@ -91,12 +91,13 @@ class MergingHandler : public qdisp::ResponseHandler {
     /// Set error code and string.
     void _setError(int code, std::string const& msg, int errorState);
 
-    /// Check if the query is no longer active.
-    /// This is used to prevent the query from being processed after it has been cancelled
-    /// or finished for any reason.
-    /// @param jobQuery the query to check
-    /// @return true if the query is no longer active
-    bool _queryIsNoLongerActive(std::shared_ptr<qdisp::JobQuery> const& jobQuery) const;
+    // 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()
+    // because the instantiation depends on the availability of the Czar configuration.
+    static std::shared_ptr<http::ClientConnPool> const& _getHttpConnPool();
+    static std::shared_ptr<http::ClientConnPool> _httpConnPool;
+    static std::mutex _httpConnPoolMutex;
 
     std::shared_ptr<rproc::InfileMerger> _infileMerger;  ///< Merging delegate
     std::atomic<bool> _errorSet{false};                  ///< Set to true when an error is set.

src/ccontrol/UserQueryFactory.cc

@@ -59,6 +59,7 @@
 #include "qmeta/QMetaMysql.h"
 #include "qmeta/QMetaSelect.h"
 #include "qmeta/QProgress.h"
+#include "qmeta/QProgressHistory.h"
 #include "qproc/DatabaseModels.h"
 #include "qproc/QuerySession.h"
 #include "qproc/SecondaryIndex.h"
@@ -134,7 +135,7 @@ std::shared_ptr<UserQuery> _makeUserQueryProcessList(query::SelectStmt::Ptr& stm
     LOGS(_log, LOG_LVL_DEBUG, "SELECT query is a PROCESSLIST");
     try {
         return std::make_shared<UserQueryProcessList>(stmt, sharedResources->qMetaSelect,
-                                                      sharedResources->qMetaCzarId, userQueryId, resultDb);
+                                                      sharedResources->czarId, userQueryId, resultDb);
     } catch (std::exception const& exc) {
         return std::make_shared<UserQueryInvalid>(exc.what());
     }
@@ -215,6 +216,7 @@ std::shared_ptr<UserQuerySharedResources> makeUserQuerySharedResources(
             std::make_shared<qmeta::QMetaMysql>(czarConfig->getMySqlQmetaConfig(),
                                                 czarConfig->getMaxMsgSourceStore()),
             std::make_shared<qmeta::QProgress>(czarConfig->getMySqlQStatusDataConfig()),
+            qmeta::QProgressHistory::create(czarConfig->getMySqlQStatusDataConfig()),
             std::make_shared<qmeta::QMetaSelect>(czarConfig->getMySqlQmetaConfig()), dbModels, czarName,
             czarConfig->getInteractiveChunkLimit());
 }
@@ -352,9 +354,10 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
         std::shared_ptr<qdisp::Executive> executive;
         std::shared_ptr<rproc::InfileMergerConfig> infileMergerConfig;
         if (sessionValid) {
-            executive =
-                    qdisp::Executive::create(_qmetaSecondsBetweenUpdates, messageStore, qdispPool,
-                                             _userQuerySharedResources->queryStatsData, qs, _asioIoService);
+            executive = qdisp::Executive::create(_qmetaSecondsBetweenUpdates, messageStore, qdispPool,
+                                                 _userQuerySharedResources->queryProgress,
+                                                 _userQuerySharedResources->queryProgressHistory, qs,
+                                                 _asioIoService);
             infileMergerConfig =
                     std::make_shared<rproc::InfileMergerConfig>(_userQuerySharedResources->mysqlResultConfig);
             infileMergerConfig->debugNoMerge = _debugNoMerge;
@@ -369,7 +372,8 @@ UserQuery::Ptr UserQueryFactory::newUserQuery(std::string const& aQuery, std::st
                 qs, messageStore, executive, _userQuerySharedResources->databaseModels, infileMergerConfig,
                 _userQuerySharedResources->secondaryIndex, _userQuerySharedResources->queryMetadata,
                 _userQuerySharedResources->queryProgress, _userQuerySharedResources->czarId, errorExtra,
-                async, resultDb);
+                async, resultDb, uberJobMaxChunks);
+
         if (sessionValid) {
             uq->qMetaRegister(resultLocation, msgTableName);
             uq->setupMerger();

src/ccontrol/UserQueryProcessList.cc

@@ -67,7 +67,7 @@ UserQueryProcessList::UserQueryProcessList(std::shared_ptr<query::SelectStmt> co
                                            qmeta::CzarId czarId, std::string const& userQueryId,
                                            std::string const& resultDb)
         : _qMetaSelect(qMetaSelect),
-          _qMetaCzarId(qMetaCzarId),
+          _czarId(czarId),
           _messageStore(std::make_shared<qmeta::MessageStore>()),
           _resultTableName(::g_nextResultTableId(userQueryId)),
           _resultDb(resultDb) {
@@ -94,7 +94,7 @@ UserQueryProcessList::UserQueryProcessList(bool full, std::shared_ptr<qmeta::QMe
                                            qmeta::CzarId czarId, std::string const& userQueryId,
                                            std::string const& resultDb)
         : _qMetaSelect(qMetaSelect),
-          _qMetaCzarId(qMetaCzarId),
+          _czarId(czarId),
           _messageStore(std::make_shared<qmeta::MessageStore>()),
           _resultTableName(::g_nextResultTableId(userQueryId)),
           _resultDb(resultDb) {

src/ccontrol/UserQueryResources.cc

@@ -36,6 +36,7 @@ UserQuerySharedResources::UserQuerySharedResources(
         std::shared_ptr<qproc::SecondaryIndex> const& secondaryIndex_,
         std::shared_ptr<qmeta::QMeta> const& queryMetadata_,
         std::shared_ptr<qmeta::QProgress> const& queryProgress_,
+        std::shared_ptr<qmeta::QProgressHistory> const& queryProgressHistory_,
         std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect_,
         std::shared_ptr<qproc::DatabaseModels> const& dbModels_, std::string const& czarName,
         int interactiveChunkLimit_)
@@ -44,6 +45,7 @@ UserQuerySharedResources::UserQuerySharedResources(
           secondaryIndex(secondaryIndex_),
           queryMetadata(queryMetadata_),
           queryProgress(queryProgress_),
+          queryProgressHistory(queryProgressHistory_),
           qMetaSelect(qMetaSelect_),
           databaseModels(dbModels_),
           interactiveChunkLimit(interactiveChunkLimit_) {

src/ccontrol/UserQueryResources.h

@@ -46,6 +46,7 @@ namespace lsst::qserv::qmeta {
 class QMeta;
 class QMetaSelect;
 class QProgress;
+class QProgressHistory;
 }  // namespace lsst::qserv::qmeta
 
 namespace lsst::qserv::qproc {
@@ -66,6 +67,7 @@ class UserQuerySharedResources {
                              std::shared_ptr<qproc::SecondaryIndex> const& secondaryIndex_,
                              std::shared_ptr<qmeta::QMeta> const& queryMetadata_,
                              std::shared_ptr<qmeta::QProgress> const& queryProgress_,
+                             std::shared_ptr<qmeta::QProgressHistory> const& queryProgressHistory_,
                              std::shared_ptr<qmeta::QMetaSelect> const& qMetaSelect_,
                              std::shared_ptr<qproc::DatabaseModels> const& databaseModels_,
                              std::string const& czarName, int interactiveChunkLimit_);
@@ -78,6 +80,7 @@ class UserQuerySharedResources {
     std::shared_ptr<qproc::SecondaryIndex> secondaryIndex;
     std::shared_ptr<qmeta::QMeta> queryMetadata;
     std::shared_ptr<qmeta::QProgress> queryProgress;
+    std::shared_ptr<qmeta::QProgressHistory> queryProgressHistory;
     std::shared_ptr<qmeta::QMetaSelect> qMetaSelect;
     std::shared_ptr<qproc::DatabaseModels> databaseModels;
     qmeta::CzarId czarId;  ///< Czar ID in QMeta database

src/ccontrol/UserQueryResultDelete.cc

@@ -39,7 +39,7 @@
 #include "cconfig/CzarConfig.h"
 #include "qmeta/Exceptions.h"
 #include "qmeta/QMeta.h"
-#include "qdisp/MessageStore.h"
+#include "qmeta/MessageStore.h"
 #include "sql/SqlConnection.h"
 #include "sql/SqlConnectionFactory.h"
 #include "sql/SqlErrorObject.h"
@@ -57,7 +57,7 @@ namespace lsst::qserv::ccontrol {
 
 UserQueryResultDelete::UserQueryResultDelete(shared_ptr<UserQueryResources> const& queryResources,
                                              string const& value)
-        : _value(value), _queryResources(queryResources), _messageStore(make_shared<qdisp::MessageStore>()) {}
+        : _value(value), _queryResources(queryResources), _messageStore(make_shared<qmeta::MessageStore>()) {}
 
 void UserQueryResultDelete::submit() {
     LOGS(_log, LOG_LVL_DEBUG, "UserQueryResultDelete::submit: " << _value);

src/ccontrol/UserQuerySelect.cc

@@ -241,7 +241,7 @@ void UserQuerySelect::submit() {
     LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect beginning submission");
     assert(_infileMerger);
 
-    _ttn = std::make_shared<TmpTableName>(_qMetaQueryId, _qSession->getOriginal());
+    _ttn = std::make_shared<TmpTableName>(_queryId, _qSession->getOriginal());
     std::vector<int> chunks;
     std::mutex chunksMtx;
     JobId sequence = 0;
@@ -269,10 +269,7 @@ void UserQuerySelect::submit() {
          ++i) {
         auto& chunkSpec = *i;
 
-        std::function<void(util::CmdData*)> funcBuildJob = [this, sequence,  // sequence must be a copy
-                                                            &chunkSpec, &queryTemplates, &ttn,
-                                                            &taskMsgFactory](util::CmdData*) {
-            QSERV_LOGCONTEXT_QUERY(_queryId);
+        QSERV_LOGCONTEXT_QUERY(_queryId);
 
         // TODO:UJ The template(s) is generated here and later it is compared to other
         //         templates. It would be better to create the list of query templates here
@@ -299,7 +296,7 @@ void UserQuerySelect::submit() {
         ResourceUnit ru;
         ru.setAsDbChunk(cs->db, cs->chunkId);
         qdisp::JobDescription::Ptr jobDesc =
-                qdisp::JobDescription::create(_qMetaCzarId, exec->getId(), sequence, ru, cs);
+                qdisp::JobDescription::create(_czarId, exec->getId(), sequence, ru, cs);
         auto job = exec->add(jobDesc);
         ++sequence;
     }
@@ -312,17 +309,11 @@ void UserQuerySelect::submit() {
     LOGS(_log, LOG_LVL_DEBUG, "total jobs in query=" << sequence);
     // TODO:UJ Waiting for all jobs to start may not be needed anymore?
     exec->waitForAllJobsToStart();
-
-    // we only care about per-chunk info for ASYNC queries
-    if (_async) {
-        std::lock_guard<std::mutex> lock(chunksMtx);
-        _qMetaAddChunks(chunks);
-    }
 }
 
 void UserQuerySelect::buildAndSendUberJobs() {
     // TODO:UJ Is special handling needed for the dummy chunk, 1234567890 ?
-    string const funcN("UserQuerySelect::" + string(__func__) + " QID=" + to_string(_qMetaQueryId));
+    string const funcN("UserQuerySelect::" + string(__func__) + " QID=" + to_string(_queryId));
     LOGS(_log, LOG_LVL_DEBUG, funcN << " start " << _uberJobMaxChunks);
 
     // Ensure `_monitor()` doesn't do anything until everything is ready.
@@ -473,8 +464,7 @@ void UserQuerySelect::buildAndSendUberJobs() {
             string uberResultName = _ttn->make(ujId);
             auto respHandler =
                     ccontrol::MergingHandler::Ptr(new ccontrol::MergingHandler(_infileMerger, exec));
-            auto uJob = qdisp::UberJob::create(exec, respHandler, exec->getId(), ujId, _qMetaCzarId,
-                                               targetWorker);
+            auto uJob = qdisp::UberJob::create(exec, respHandler, exec->getId(), ujId, _czarId, targetWorker);
             uJob->setWorkerContactInfo(wInfUJ->wInf);
             wInfUJ->uberJobPtr = uJob;
         };
@@ -574,7 +564,6 @@ QueryState UserQuerySelect::join() {
         auto const status = resultSizeLimitExceeded ? qmeta::QInfo::FAILED_LR : qmeta::QInfo::FAILED;
         _qMetaUpdateStatus(status, collectedRows, collectedBytes, finalRows);
         LOGS(_log, LOG_LVL_ERROR, "Joined everything (failure!) QID=" << getQueryId());
-        operation = proto::QueryManagement::CANCEL;
         state = ERROR;
     }
     auto const czarConfig = cconfig::CzarConfig::instance();
@@ -807,7 +796,7 @@ void UserQuerySelect::qMetaRegister(std::string const& resultLocation, std::stri
 
     auto exec = _executive;
     if (exec != nullptr) {
-        exec->setQueryId(_qMetaQueryId);
+        exec->setQueryId(_queryId);
     } else {
         LOGS(_log, LOG_LVL_WARN, "No Executive, assuming invalid query");
     }

src/ccontrol/UserQuerySelectCountStar.cc

@@ -62,7 +62,7 @@ UserQuerySelectCountStar::UserQuerySelectCountStar(std::string query,
         : _qMetaSelect(qMetaSelect),
           _queryMetadata(queryMetadata),
           _messageStore(std::make_shared<qmeta::MessageStore>()),
-          _resultTableName(::g_nextResultTableId(userQueryId)),
+          //&&&_resultTableName(::g_nextResultTableId(userQueryId)),
           _userQueryId(userQueryId),
           _rowsTable(rowsTable),
           _resultDb(resultDb),

src/ccontrol/UserQuerySelectCountStar.h

@@ -119,7 +119,6 @@ class UserQuerySelectCountStar : public UserQuery {
     std::shared_ptr<qmeta::MessageStore> _messageStore;
     std::string _resultTable;
     std::string _resultLoc;  ///< Result location
-    std::string _resultTableName;
     std::string _userQueryId;
     std::string _rowsTable;
     std::string _resultDb;