Added family map option to not use chunk size for distribution.

Author: jgates108

admin/tools/docker/base/Dockerfile

@@ -52,6 +52,7 @@ RUN dnf install -y 'dnf-command(config-manager)' \
         protobuf-devel \
         python3.12 \
         python3.12-devel \
+        jemalloc \
         tree \
         vim \
         zip \

src/cconfig/CMakeLists.txt

@@ -4,13 +4,13 @@ target_sources(cconfig PRIVATE
     CzarConfig.cc
 )
 
-target_include_directories(cconfig PRIVATE
-    ${XROOTD_INCLUDE_DIRS}
-)
 
 target_link_libraries(cconfig PUBLIC
     log
-    XrdSsiLib
+)
+
+install(
+	TARGETS cconfig
 )
 
 install(

src/cconfig/CzarConfig.cc

@@ -28,7 +28,6 @@
 #include <stdexcept>
 
 // Third party headers
-#include "XrdSsi/XrdSsiLogger.hh"
 
 // LSST headers
 #include "lsst/log/Log.h"
@@ -43,20 +42,6 @@ namespace {
 
 LOG_LOGGER _log = LOG_GET("lsst.qserv.cconfig.CzarConfig");
 
-void QservLogger(struct timeval const& mtime, unsigned long tID, const char* msg, int mlen) {
-    static log4cxx::spi::LocationInfo xrdLoc(
-            "client", log4cxx::spi::LocationInfo::calcShortFileName("client"), "<xrdssi>", 0);
-    static LOG_LOGGER myLog = LOG_GET("lsst.qserv.xrdssi.msgs");
-
-    if (myLog.isInfoEnabled()) {
-        while (mlen && msg[mlen - 1] == '\n') --mlen;  // strip all trailing newlines
-        std::string theMsg(msg, mlen);
-        lsst::log::Log::MDC("LWP", std::to_string(tID));
-        myLog.logMsg(log4cxx::Level::getInfo(), xrdLoc, theMsg);
-    }
-}
-
-bool dummy = XrdSsiLogger::SetMCB(QservLogger, XrdSsiLogger::mcbClient);
 }  // namespace
 
 namespace lsst::qserv::cconfig {

src/ccontrol/CMakeLists.txt

@@ -3,7 +3,6 @@ add_dependencies(ccontrol proto)
 
 target_include_directories(ccontrol PRIVATE
     ${ANTLR4_INCLUDE_DIR}
-    ${XROOTD_INCLUDE_DIRS}
 )
 
 target_sources(ccontrol PRIVATE
@@ -27,11 +26,15 @@ target_sources(ccontrol PRIVATE
 target_link_libraries(ccontrol PUBLIC
     boost_regex
     cconfig
+    css
     log
     parser
-    replica
+	proto
     sphgeom
-    XrdCl
+)
+
+install(
+	TARGETS ccontrol
 )
 
 install(
@@ -65,3 +68,6 @@ ccontrol_tests(
     testCControl
     testUserQueryType
 )
+
+# set_tests_properties(testCControl PROPERTIES WILL_FAIL 1)
+

src/ccontrol/MergingHandler.cc

@@ -34,7 +34,6 @@
 
 // Third-party headers
 #include "curl/curl.h"
-#include "XrdCl/XrdClFile.hh"
 
 // LSST headers
 #include "lsst/log/Log.h"
@@ -127,7 +126,7 @@ std::tuple<bool, bool> readHttpFileAndMergeHttp(
     uint32_t msgSizeBytes = 0;
     bool success = true;
     bool mergeHappened = false;
-    int headerCount = 0;
+    uint64_t headerCount = 0;
     uint64_t totalBytesRead = 0;
     try {
         auto exec = uberJob->getExecutive();
@@ -211,11 +210,9 @@ std::tuple<bool, bool> readHttpFileAndMergeHttp(
                         }
 
                         // Parse and evaluate the message.
-                        //&&&mergeHappened = messageIsReady(msgBuf.get(), msgSizeBytes, last);
                         mergeHappened = true;
                         bool messageReadyResult = messageIsReady(msgBuf.get(), msgSizeBytes, last);
                         totalBytesRead += msgSizeBytes;
-                        //&&&if (!mergeHappened) {
                         if (!messageReadyResult) {
                             success = false;
                             throw runtime_error("message processing failed at offset " +
@@ -265,10 +262,23 @@ std::tuple<bool, bool> readHttpFileAndMergeHttp(
 
 namespace lsst::qserv::ccontrol {
 
-MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> merger, std::string const& tableName)
-        : _infileMerger{merger}, _tableName{tableName} {}
+shared_ptr<http::ClientConnPool> MergingHandler::_httpConnPool;
+mutex MergingHandler::_httpConnPoolMutex;
 
-MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_TRACE, __func__ << " " << _tableName); }
+shared_ptr<http::ClientConnPool> const& MergingHandler::_getHttpConnPool() {
+    lock_guard<mutex> const lock(_httpConnPoolMutex);
+    if (nullptr == _httpConnPool) {
+        _httpConnPool = make_shared<http::ClientConnPool>(
+                cconfig::CzarConfig::instance()->getResultMaxHttpConnections());
+    }
+    return _httpConnPool;
+}
+
+MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> const& merger,
+                               std::shared_ptr<qdisp::Executive> const& exec)
+        : _infileMerger(merger), _executive(exec) {}
+
+MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_TRACE, __func__); }
 
 
 bool MergingHandler::flush(proto::ResponseSummary const& resp) {
@@ -317,35 +327,14 @@ bool MergingHandler::flush(proto::ResponseSummary const& resp) {
 }
 
 void MergingHandler::errorFlush(std::string const& msg, int code) {
-    _setError(code, msg);
+    _setError(code, msg, util::ErrorCode::RESULT_IMPORT);
     // Might want more info from result service.
     // Do something about the error. FIXME.
     LOGS(_log, LOG_LVL_ERROR, "Error receiving result.");
 }
 
-// Note that generally we always have an _infileMerger object except during
-// a unit test. I suppose we could try to figure out how to create one.
-//
-void MergingHandler::prepScrubResults(int jobId, int attemptCount) {
-    if (_infileMerger) _infileMerger->prepScrub(jobId, attemptCount);
-}
-
 std::ostream& MergingHandler::print(std::ostream& os) const {
-    return os << "MergingRequester(" << _tableName << ", flushed=" << (_flushed ? "true)" : "false)");
-}
-
-bool MergingHandler::_mergeHttp(shared_ptr<qdisp::UberJob> const& uberJob,
-                                proto::ResponseData const& responseData) {
-    if (_flushed) {
-        throw util::Bug(ERR_LOC, "already flushed");
-    }
-    bool const success = _infileMerger->mergeHttp(uberJob, responseData);
-    if (!success) {
-        LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
-        util::Error const& err = _infileMerger->getError();
-        _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
-    }
-    return success;
+    return os << "MergingRequester(flushed=" << (_flushed ? "true)" : "false)");
 }
 
 bool MergingHandler::_mergeHttp(shared_ptr<qdisp::UberJob> const& uberJob,
@@ -357,15 +346,16 @@ bool MergingHandler::_mergeHttp(shared_ptr<qdisp::UberJob> const& uberJob,
     if (!success) {
         LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
         util::Error const& err = _infileMerger->getError();
-        _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
+        _setError(ccontrol::MSG_RESULT_ERROR, err.getMsg(), util::ErrorCode::RESULT_IMPORT);
     }
     return success;
 }
 
-void MergingHandler::_setError(int code, std::string const& msg) {
+void MergingHandler::_setError(int code, std::string const& msg, int errorState) {
     LOGS(_log, LOG_LVL_DEBUG, "_setError: code: " << code << ", message: " << msg);
-    std::lock_guard<std::mutex> lock(_errorMutex);
-    _error = Error(code, msg);
+    auto exec = _executive.lock();
+    if (exec == nullptr) return;
+    exec->addMultiError(code, msg, errorState);
 }
 
 tuple<bool, bool> MergingHandler::flushHttp(string const& fileUrl, uint64_t expectedRows,
@@ -415,10 +405,9 @@ tuple<bool, bool> MergingHandler::flushHttp(string const& fileUrl, uint64_t expe
     return {success, shouldCancel};
 }
 
-void MergingHandler::flushHttpError(int errorCode, std::string const& errorMsg, int status) {
+void MergingHandler::flushHttpError(int errorCode, std::string const& errorMsg, int errState) {
     if (!_errorSet.exchange(true)) {
-        _error = util::Error(errorCode, errorMsg, util::ErrorCode::MYSQLEXEC);
-        _setError(ccontrol::MSG_RESULT_ERROR, _error.getMsg());
+        _setError(errorCode, errorMsg, errState);
     }
 }
 

src/ccontrol/MergingHandler.h

@@ -39,6 +39,7 @@ class ResponseSummary;
 }  // namespace lsst::qserv::proto
 
 namespace lsst::qserv::qdisp {
+class Executive;
 class JobQuery;
 class UberJob;
 }  // namespace lsst::qserv::qdisp
@@ -64,20 +65,8 @@ class MergingHandler : public qdisp::ResponseHandler {
     virtual ~MergingHandler();
 
     /// @param merger downstream merge acceptor
-    /// @param tableName target table for incoming data
-    MergingHandler(std::shared_ptr<rproc::InfileMerger> merger, std::string const& tableName);
-
-    /// Process the response and read the result file if no error was reported by a worker.
-    /// @return true if successful (no error)
-    bool flush(proto::ResponseSummary const& resp) override;
-
-    /// @see ResponseHandler::flushHttp
-    /// @see MerginHandler::_mergeHttp
-    std::tuple<bool, bool> flushHttp(std::string const& fileUrl, uint64_t expectedRows,
-                                     uint64_t& resultRows) override;
-
-    /// @see ResponseHandler::flushHttpError
-    void flushHttpError(int errorCode, std::string const& errorMsg, int status) override;
+    MergingHandler(std::shared_ptr<rproc::InfileMerger> const& merger,
+                   std::shared_ptr<qdisp::Executive> const& exec);
 
     /// @see ResponseHandler::flushHttp
     /// @see MerginHandler::_mergeHttp
@@ -93,18 +82,12 @@ class MergingHandler : public qdisp::ResponseHandler {
     /// Print a string representation of the receiver to an ostream
     std::ostream& print(std::ostream& os) const override;
 
-    /// @return an error code and description
-    Error getError() const override {
-        std::lock_guard<std::mutex> lock(_errorMutex);
-        return _error;
-    }
-
 private:
     /// Call InfileMerger to do the work of merging this data to the result.
     bool _mergeHttp(std::shared_ptr<qdisp::UberJob> const& uberJob, proto::ResponseData const& responseData);
 
     /// Set error code and string.
-    void _setError(int code, std::string const& msg);
+    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
@@ -114,12 +97,11 @@ class MergingHandler : public qdisp::ResponseHandler {
     bool _queryIsNoLongerActive(std::shared_ptr<qdisp::JobQuery> const& jobQuery) const;
 
     std::shared_ptr<rproc::InfileMerger> _infileMerger;  ///< Merging delegate
-    std::string _tableName;                              ///< Target table name
-    Error _error;                                        ///< Error description
     std::atomic<bool> _errorSet{false};                  ///< Set to true when an error is set.
-    mutable std::mutex _errorMutex;                      ///< Protect readers from partial updates
     bool _flushed{false};                                ///< flushed to InfileMerger?
     std::string _wName{"~"};                             ///< worker name
+
+    std::weak_ptr<qdisp::Executive> _executive;  ///< Weak pointer to the executive for errors.
 };
 
 }  // namespace lsst::qserv::ccontrol

src/ccontrol/UserQuerySelect.cc

@@ -86,7 +86,8 @@
 #include "global/LogContext.h"
 #include "proto/worker.pb.h"
 #include "qdisp/Executive.h"
-#include "qdisp/MessageStore.h"
+#include "qdisp/JobQuery.h"
+#include "qmeta/MessageStore.h"
 #include "qmeta/QMeta.h"
 #include "qmeta/Exceptions.h"
 #include "qmeta/QMeta.h"
@@ -284,7 +285,6 @@ void UserQuerySelect::submit() {
             cs = _qSession->buildChunkQuerySpec(queryTemplates, chunkSpec, fillInChunkIdTag);
             chunks.push_back(cs->chunkId);
         }
-        std::string chunkResultName = _ttn->make(cs->chunkId);
 
         // This should only need to be set once as all jobs should have the same database name.
         if (cs->db != dbName) {
@@ -299,9 +299,8 @@ void UserQuerySelect::submit() {
 
         ResourceUnit ru;
         ru.setAsDbChunk(cs->db, cs->chunkId);
-        qdisp::JobDescription::Ptr jobDesc = qdisp::JobDescription::create(
-                _qMetaCzarId, exec->getId(), sequence, ru,
-                std::make_shared<MergingHandler>(_infileMerger, chunkResultName), cs, chunkResultName);
+        qdisp::JobDescription::Ptr jobDesc =
+                qdisp::JobDescription::create(_qMetaCzarId, exec->getId(), sequence, ru, cs);
         auto job = exec->add(jobDesc);
         ++sequence;
     }
@@ -339,6 +338,14 @@ void UserQuerySelect::buildAndSendUberJobs() {
         return;
     }
 
+    if (exec->getCancelled() || exec->getSuperfluous()) {
+        LOGS(_log, LOG_LVL_INFO, funcN << " executive cancelled.");
+    }
+
+    if (exec->getSuperfluous()) {
+        LOGS(_log, LOG_LVL_INFO, funcN << " executive superfluous, result already found.");
+    }
+
     // Only one thread should be generating UberJobs for this user query at any given time.
     lock_guard fcLock(_buildUberJobMtx);
     LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect::" << __func__ << " totalJobs=" << exec->getTotalJobs());
@@ -394,13 +401,15 @@ void UserQuerySelect::buildAndSendUberJobs() {
     map<string, WInfoAndUJPtr::Ptr> workerJobMap;
     vector<qdisp::Executive::ChunkIdType> missingChunks;
 
+    int attemptCountIncreased = 0;
     // unassignedChunksInQuery needs to be in numerical order so that UberJobs contain chunk numbers in
     // numerical order. The workers run shared scans in numerical order of chunkId numbers.
     // Numerical order keeps the number of partially complete UberJobs running on a worker to a minimum,
     // and should minimize the time for the first UberJob on the worker to complete.
     for (auto const& [chunkId, jqPtr] : unassignedChunksInQuery) {
         bool const increaseAttemptCount = true;
         jqPtr->getDescription()->incrAttemptCount(exec, increaseAttemptCount);
+        attemptCountIncreased++;
 
         // If too many workers are down, there will be a chunk that cannot be found.
         // Just continuing should leave jobs `unassigned` with their attempt count
@@ -462,7 +471,8 @@ void UserQuerySelect::buildAndSendUberJobs() {
         if (wInfUJ->uberJobPtr == nullptr) {
             auto ujId = _uberJobIdSeq++;  // keep ujId consistent
             string uberResultName = _ttn->make(ujId);
-            auto respHandler = make_shared<ccontrol::MergingHandler>(_infileMerger, uberResultName);
+            auto respHandler =
+                    ccontrol::MergingHandler::Ptr(new ccontrol::MergingHandler(_infileMerger, exec));
             auto uJob = qdisp::UberJob::create(exec, respHandler, exec->getId(), ujId, _qMetaCzarId,
                                                targetWorker);
             uJob->setWorkerContactInfo(wInfUJ->wInf);
@@ -489,6 +499,11 @@ void UserQuerySelect::buildAndSendUberJobs() {
         LOGS(_log, LOG_LVL_ERROR, errStr);
     }
 
+    if (attemptCountIncreased > 0) {
+        LOGS(_log, LOG_LVL_WARN,
+             funcN << " increased attempt count for " << attemptCountIncreased << " Jobs");
+    }
+
     // Queue unqued UberJobs, these have less than the max number of jobs.
     for (auto const& [wIdKey, winfUjPtr] : workerJobMap) {
         if (winfUjPtr != nullptr) {

src/ccontrol/UserQuerySelect.h

@@ -95,6 +95,8 @@ class UserQuerySelect : public UserQuery {
     UserQuerySelect(UserQuerySelect const&) = delete;
     UserQuerySelect& operator=(UserQuerySelect const&) = delete;
 
+    ~UserQuerySelect() override = default;
+
     /**
      *  @param resultLocation:  Result location, if empty use result table with unique
      *                          name generated from query ID.