Merge pull request #911 from lsst/tickets/DM-49661

Tickets/dm 49661

Author: jgates108

admin/tools/docker/base/Dockerfile

@@ -37,7 +37,6 @@ RUN dnf install -y 'dnf-command(config-manager)' \
         glib2-devel \
         glibc-langpack-en \
         java-devel \
-        jemalloc \
         libcurl-devel \
         libevent-devel \
         libtool \
@@ -53,6 +52,7 @@ RUN dnf install -y 'dnf-command(config-manager)' \
         protobuf-devel \
         python3.12 \
         python3.12-devel \
+        jemalloc \
         tree \
         vim \
         zip \
@@ -73,6 +73,17 @@ RUN dnf update -y \
 RUN curl -s "https://cmake.org/files/v3.31/cmake-3.31.5-linux-x86_64.tar.gz" \
     | tar --strip-components=1 -xz -C /usr/local
 
+RUN cd /tmp \
+    && git clone https://github.com/jemalloc/jemalloc \
+    && cd jemalloc \
+    && git checkout tags/5.3.0 \
+    && ./autogen.sh \
+    && ./configure --enable-prof \
+    && make -j8 \
+    && make install \
+    && cd /tmp \
+    && rm -rf jemalloc
+
 RUN cd /tmp \
     && git clone https://github.com/apache/logging-log4cxx \
     && cd logging-log4cxx \
@@ -127,6 +138,9 @@ RUN cd /tmp \
     && git clone https://github.com/xrootd/xrootd.git \
     && cd xrootd \
     && git checkout tags/v5.6.2 \
+    && git config --global user.email "[email protected]" \
+    && git config --global user.name "Qserv" \
+    && git cherry-pick d85915a3927261e49859c3e13075bce1dfefcbe4 \
     && mkdir build \
     && cd build \
     && cmake -DENABLE_PYTHON=off .. \
@@ -220,7 +234,6 @@ RUN dnf install -y 'dnf-command(config-manager)' \
         boost-thread \
         glib2 \
         glibc-langpack-en \
-        jemalloc \
         libevent \
         libuuid \
         lua5.1 \
@@ -262,6 +275,10 @@ RUN mkdir -p /qserv/data && \
 RUN alternatives --install /usr/bin/python3 python3 /usr/bin/python3.12 1
 ENV PYTHONPATH "${PYTHONPATH}:/usr/local/python"
 
+COPY --from=lite-build /usr/local/bin/jemalloc* /usr/local/bin/
+COPY --from=lite-build /usr/local/lib/pkgconfig/jemalloc.pc /usr/local/lib/pkgconfig/
+COPY --from=lite-build /usr/local/lib/libjemalloc* /usr/local/lib
+
 COPY --from=lite-build /usr/local/lib64/liblog4cxx.so /usr/local/lib64/
 COPY --from=lite-build /usr/local/lib/libantlr4-runtime.so /usr/local/lib/
 COPY --from=lite-build /usr/local/bin/mysql-proxy /usr/local/bin/

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)
@@ -94,58 +94,3 @@ add_subdirectory(xrdsvc)
 
 #-----------------------------------------------------------------------------
 
-add_library(qserv_common SHARED)
-
-target_link_libraries(qserv_common PUBLIC
-    global
-    memman
-    proto
-    mysql
-    sql
-    util
-    protojson
-)
-
-install(
-    TARGETS qserv_common
-)
-
-#-----------------------------------------------------------------------------
-
-add_library(xrdsvc SHARED)
-
-target_link_libraries(xrdsvc PUBLIC
-    wbase
-    wcontrol
-    wconfig
-    wdb
-    wpublish
-    wsched
-    qserv_xrdsvc
-    qserv_common
-)
-
-install(
-    TARGETS xrdsvc
-)
-
-#-----------------------------------------------------------------------------
-
-add_library(qserv_czar SHARED)
-
-target_link_libraries(qserv_czar PUBLIC
-    ccontrol
-    czar
-    parser
-    qana
-    query
-    qdisp
-    qproc
-    rproc
-    qserv_css
-    qserv_meta
-)
-
-install(
-    TARGETS qserv_czar
-)

src/cconfig/CMakeLists.txt

@@ -1,14 +1,14 @@
-add_library(cconfig OBJECT)
+add_library(cconfig SHARED)
 
 target_sources(cconfig PRIVATE
     CzarConfig.cc
 )
 
-target_include_directories(cconfig PRIVATE
-    ${XROOTD_INCLUDE_DIRS}
-)
 
 target_link_libraries(cconfig PUBLIC
     log
-    XrdSsiLib
+)
+
+install(
+	TARGETS cconfig
 )

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

@@ -1,9 +1,8 @@
-add_library(ccontrol OBJECT)
+add_library(ccontrol SHARED)
 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
 )
 
 FUNCTION(ccontrol_tests)
@@ -47,8 +50,8 @@ FUNCTION(ccontrol_tests)
             qana
             qdisp
             qproc
-            qserv_css
-            qserv_meta
+            css
+            qmeta
             query
             rproc
             Boost::unit_test_framework
@@ -63,3 +66,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"
@@ -49,6 +48,7 @@
 #include "proto/ProtoHeaderWrap.h"
 #include "proto/worker.pb.h"
 #include "qdisp/CzarStats.h"
+#include "qdisp/Executive.h"
 #include "qdisp/JobQuery.h"
 #include "qdisp/UberJob.h"
 #include "rproc/InfileMerger.h"
@@ -273,13 +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}, _tableName{tableName} {}
+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__ << " " << _tableName); }
+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.");
@@ -293,7 +294,7 @@ void MergingHandler::prepScrubResults(int jobId, int attemptCount) {
 }
 
 std::ostream& MergingHandler::print(std::ostream& os) const {
-    return os << "MergingRequester(" << _tableName << ", flushed=" << (_flushed ? "true)" : "false)");
+    return os << "MergingRequester(flushed=" << (_flushed ? "true)" : "false)");
 }
 
 bool MergingHandler::_mergeHttp(shared_ptr<qdisp::UberJob> const& uberJob,
@@ -305,15 +306,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,
@@ -363,10 +365,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

@@ -43,6 +43,7 @@ class ResponseSummary;
 }  // namespace lsst::qserv::proto
 
 namespace lsst::qserv::qdisp {
+class Executive;
 class JobQuery;
 class UberJob;
 }  // namespace lsst::qserv::qdisp
@@ -68,29 +69,23 @@ 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);
+    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;
 
     /// Signal an unrecoverable error condition. No further calls are expected.
     void errorFlush(std::string const& msg, int code) override;
 
     /// 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;
 
@@ -99,7 +94,7 @@ class MergingHandler : public qdisp::ResponseHandler {
     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.
@@ -110,12 +105,11 @@ class MergingHandler : public qdisp::ResponseHandler {
     static std::mutex _httpConnPoolMutex;
 
     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