Cleanup and removed some unneeded code.

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/CMakeLists.txt

@@ -21,7 +21,7 @@ find_package(Protobuf REQUIRED)
 find_package(pybind11 REQUIRED)
 find_package(Threads REQUIRED)
 find_package(XRootD REQUIRED)
-find_package(httplib REQUIRED)
+#&&& find_package(httplib REQUIRED)
 find_package(aws-c-auth REQUIRED)
 find_package(aws-c-cal REQUIRED)
 find_package(aws-c-common REQUIRED)

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"
@@ -42,20 +41,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
@@ -29,11 +28,15 @@ target_sources(ccontrol PRIVATE
 target_link_libraries(ccontrol PUBLIC
     boost_regex
     cconfig
+    css
     log
     parser
-    replica
+	proto
     sphgeom
-    XrdCl
+)
+
+install(
+	TARGETS ccontrol
 )
 
 install(
@@ -67,3 +70,6 @@ ccontrol_tests(
     testCControl
     testUserQueryType
 )
+
+# set_tests_properties(testCControl PROPERTIES WILL_FAIL 1)
+

src/ccontrol/MergingHandler.cc

@@ -33,7 +33,6 @@
 
 // Third-party headers
 #include "curl/curl.h"
-#include "XrdCl/XrdClFile.hh"
 
 // LSST headers
 #include "lsst/log/Log.h"
@@ -274,18 +273,14 @@ shared_ptr<http::ClientConnPool> const& MergingHandler::_getHttpConnPool() {
     return _httpConnPool;
 }
 
-/*&&&
-MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> merger, std::string const& tableName)
-        : _infileMerger{merger}, _tableNameOld{tableName} {}
-*/
-MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> merger)
-        : _infileMerger{merger} {}
+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__ ); }
+MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_TRACE, __func__); }
 
 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.");
@@ -311,29 +306,17 @@ 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;
 }
 
-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;
-}
 
-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,
@@ -383,10 +366,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

@@ -31,8 +31,6 @@
 // Qserv headers
 #include "qdisp/ResponseHandler.h"
 
-#include "mysql/LocalInfile.h" //&&&
-
 // Forward declarations
 
 namespace lsst::qserv::http {
@@ -45,6 +43,7 @@ class ResponseSummary;
 }  // namespace lsst::qserv::proto
 
 namespace lsst::qserv::qdisp {
+class Executive;
 class JobQuery;
 class UberJob;
 }  // namespace lsst::qserv::qdisp
@@ -64,23 +63,22 @@ namespace lsst::qserv::ccontrol {
 /// Do to the way the code works, MerginHandler is effectively single threaded.
 /// The worker can only send the data for this job back over a single channel
 /// and it can only send one transmit on that channel at a time.
-class MergingHandler : public qdisp::ResponseHandler {  mysql::InstanceMCount icmim{"MergingHandler&&&"};
+class MergingHandler : public qdisp::ResponseHandler {
 public:
     typedef std::shared_ptr<MergingHandler> Ptr;
     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);
-    MergingHandler(std::shared_ptr<rproc::InfileMerger> merger);
+    MergingHandler(std::shared_ptr<rproc::InfileMerger> const& merger,
+                   std::shared_ptr<qdisp::Executive> const& exec);
 
     /// @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;
+    void flushHttpError(int errorCode, std::string const& errorMsg, int errState) override;
 
     /// @see ResponseHandler::flushHttp
     /// @see MerginHandler::_mergeHttp
@@ -96,18 +94,15 @@ class MergingHandler : public qdisp::ResponseHandler {  mysql::InstanceMCount ic
     /// 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;
-    }
+    /// Prepare to scrub the results from jobId-attempt from the result table.
+    void prepScrubResults(int jobId, int attempt) override;
 
 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);
 
     // 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.
@@ -118,12 +113,11 @@ class MergingHandler : public qdisp::ResponseHandler {  mysql::InstanceMCount ic
     static std::mutex _httpConnPoolMutex;
 
     std::shared_ptr<rproc::InfileMerger> _infileMerger;  ///< Merging delegate
-    //&&&std::string _tableNameOld;                              ///< 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/UserQuery.h

@@ -41,8 +41,6 @@
 #include "global/intTypes.h"
 #include "qmeta/types.h"
 
-#include "util/InstanceCount.h" //&&&
-
 // Forward decl
 namespace lsst::qserv::qmeta {
 class MessageStore;
@@ -52,7 +50,7 @@ namespace lsst::qserv::ccontrol {
 
 /// UserQuery : interface for user query data. Not thread-safe, although
 /// its delegates are thread-safe as appropriate.
-class UserQuery { util::InstanceCount ic{"UserQuery&&&"};
+class UserQuery {
 public:
     typedef std::shared_ptr<UserQuery> Ptr;
 

src/ccontrol/UserQuerySelect.cc

@@ -268,7 +268,6 @@ void UserQuerySelect::submit() {
     string dbName("");
     bool dbNameSet = false;
 
-    auto mergingHandler = std::make_shared<MergingHandler>(_infileMerger);
     for (auto i = _qSession->cQueryBegin(), e = _qSession->cQueryEnd(); i != e && !exec->getCancelled();
          ++i) {
         auto& chunkSpec = *i;
@@ -286,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) {
@@ -301,14 +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,
-                mergingHandler, cs, chunkResultName);
+        qdisp::JobDescription::Ptr jobDesc =
+                qdisp::JobDescription::create(_qMetaCzarId, exec->getId(), sequence, ru, cs);
         auto job = exec->add(jobDesc);
         ++sequence;
     }
@@ -479,8 +471,7 @@ 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 = make_shared<ccontrol::MergingHandler>(_infileMerger);
+            auto respHandler = make_shared<ccontrol::MergingHandler>(_infileMerger, exec);
             auto uJob = qdisp::UberJob::create(exec, respHandler, exec->getId(), ujId, _qMetaCzarId,
                                                targetWorker);
             uJob->setWorkerContactInfo(wInfUJ->wInf);
@@ -508,7 +499,8 @@ void UserQuerySelect::buildAndSendUberJobs() {
     }
 
     if (attemptCountIncreased > 0) {
-        LOGS(_log, LOG_LVL_WARN, funcN << " increased attempt count for " << attemptCountIncreased << " Jobs");
+        LOGS(_log, LOG_LVL_WARN,
+             funcN << " increased attempt count for " << attemptCountIncreased << " Jobs");
     }
 
     // Queue unqued UberJobs, these have less than the max number of jobs.

src/ccontrol/UserQuerySelect.h

@@ -47,8 +47,6 @@
 #include "qmeta/types.h"
 #include "qproc/ChunkSpec.h"
 
-#include "util/InstanceCount.h" //&&&
-
 // Forward declarations
 namespace lsst::qserv::qdisp {
 class Executive;
@@ -85,7 +83,7 @@ namespace lsst::qserv::ccontrol {
 class TmpTableName;
 
 /// UserQuerySelect : implementation of the UserQuery for regular SELECT statements.
-class UserQuerySelect : public UserQuery {  util::InstanceCount ic{"UserQuerySelect&&&"};
+class UserQuerySelect : public UserQuery {
 public:
     UserQuerySelect(std::shared_ptr<qproc::QuerySession> const& qs,
                     std::shared_ptr<qmeta::MessageStore> const& messageStore,
@@ -102,7 +100,7 @@ class UserQuerySelect : public UserQuery {  util::InstanceCount ic{"UserQuerySel
     UserQuerySelect(UserQuerySelect const&) = delete;
     UserQuerySelect& operator=(UserQuerySelect const&) = delete;
 
-    //&&& ~UserQuerySelect() override = default;
+    ~UserQuerySelect() override = default;
 
     /**
      *  @param resultLocation:  Result location, if empty use result table with unique