tickets/DM-43715

admin/local/docker/compose/docker-compose.yml

@@ -457,6 +457,7 @@ services:
           --registry-host=repl-registry
           --controller-auto-register-workers=1
           --qserv-sync-force
+          --qserv-chunk-map-update
           --debug
         expose:
             - "25081"

src/CMakeLists.txt

@@ -68,6 +68,7 @@ add_subdirectory(mysql)
 add_subdirectory(parser)
 add_subdirectory(partition)
 add_subdirectory(proto)
+add_subdirectory(protojson)
 add_subdirectory(proxy)
 add_subdirectory(qana)
 add_subdirectory(qdisp)
@@ -89,7 +90,6 @@ add_subdirectory(wpublish)
 add_subdirectory(wsched)
 add_subdirectory(www)
 add_subdirectory(xrdlog)
-add_subdirectory(xrdreq)
 add_subdirectory(xrdsvc)
 
 #-----------------------------------------------------------------------------
@@ -103,6 +103,7 @@ target_link_libraries(qserv_common PUBLIC
     mysql
     sql
     util
+    protojson
 )
 
 install(
@@ -143,7 +144,6 @@ target_link_libraries(qserv_czar PUBLIC
     rproc
     qserv_css
     qserv_meta
-    xrdreq
 )
 
 install(

src/admin/python/lsst/qserv/admin/itest.py

@@ -978,6 +978,7 @@ def compareQueryResults(run_cases: List[str], outputs_dir: str) -> List[ITestCas
         if not os.path.exists(os.path.join(outputs_dir, case)):
             _log.warn("There are no query results to compare for %s", case)
             continue
+
         comparisons = (
             (query_mode_mysql, query_mode_qserv_attached),
             (query_mode_mysql, query_mode_qserv_detached),

src/admin/python/lsst/qserv/admin/qservCli/launch.py

@@ -262,7 +262,9 @@ def cmake(
         build_image,
         "cmake",
         "..",
+        "-DCMAKE_BUILD_TYPE=Debug"
     ]
+    # "-DCMAKE_BUILD_TYPE=Debug"
     if dry:
         print(" ".join(args))
         return

src/admin/templates/http/etc/qserv-czar.cnf.jinja

@@ -109,8 +109,6 @@ largestPriority = 3
 vectRunSizes = 50:50:50:50
 # Minimum number of threads running for each queue. No spaces. Values separated by ':'
 vectMinRunningSizes = 0:1:3:3
-# Maximum number of QueryRequests allowed to be running at one time.
-qReqPseudoFifoMaxRunning = 299
 
 [replication]
 

src/admin/templates/proxy/etc/qserv-czar.cnf.jinja

@@ -98,21 +98,19 @@ notifyWorkersOnCzarRestart = 1
 #[debug]
 #chunkLimit = -1
 
-# Please see qdisp/QdispPool.h QdispPool::QdispPool for more information
+# Please see util/QdispPool.h QdispPool::QdispPool for more information
 [qdisppool]
 #size of the pool
-poolSize = 50
+poolSize = 1000
 # Low numbers are higher priority. Largest priority 3 creates 4 priority queues 0, 1, 2, 3
 #     Must be greater than 0.
 largestPriority = 3
 # Maximum number of threads running for each queue. No spaces. Values separated by ':'
 #   Using largestPriority = 2  and   vectRunsizes = 3:5:8
 #    queue 0 would have runSize 3, queue 1 would have runSize 5, and queue 2 would have runSize 8.
-vectRunSizes = 50:50:50:50
+vectRunSizes = 800:800:500:500
 # Minimum number of threads running for each queue. No spaces. Values separated by ':'
-vectMinRunningSizes = 0:1:3:3
-# Maximum number of QueryRequests allowed to be running at one time.
-qReqPseudoFifoMaxRunning = 299
+vectMinRunningSizes = 0:3:3:3
 
 [replication]
 

src/cconfig/CzarConfig.cc

@@ -62,18 +62,17 @@ namespace lsst::qserv::cconfig {
 
 std::mutex CzarConfig::_mtxOnInstance;
 
-std::shared_ptr<CzarConfig> CzarConfig::_instance;
+CzarConfig::Ptr CzarConfig::_instance;
 
-std::shared_ptr<CzarConfig> CzarConfig::create(std::string const& configFileName,
-                                               std::string const& czarName) {
+CzarConfig::Ptr CzarConfig::create(std::string const& configFileName, std::string const& czarName) {
     std::lock_guard<std::mutex> const lock(_mtxOnInstance);
     if (_instance == nullptr) {
         _instance = std::shared_ptr<CzarConfig>(new CzarConfig(util::ConfigStore(configFileName), czarName));
     }
     return _instance;
 }
 
-std::shared_ptr<CzarConfig> CzarConfig::instance() {
+CzarConfig::Ptr CzarConfig::instance() {
     std::lock_guard<std::mutex> const lock(_mtxOnInstance);
     if (_instance == nullptr) {
         throw std::logic_error("CzarConfig::" + std::string(__func__) + ": instance has not been created.");

src/cconfig/CzarConfig.h

@@ -53,6 +53,7 @@ namespace lsst::qserv::cconfig {
  */
 class CzarConfig {
 public:
+    using Ptr = std::shared_ptr<CzarConfig>;
     /**
      * Create an instance of CzarConfig and load parameters from the specifid file.
      * @note One has to call this method at least once before trying to obtain
@@ -63,15 +64,15 @@ class CzarConfig {
      * @param czarName - the unique name of Czar.
      * @return the shared pointer to the configuration object
      */
-    static std::shared_ptr<CzarConfig> create(std::string const& configFileName, std::string const& czarName);
+    static Ptr create(std::string const& configFileName, std::string const& czarName);
 
     /**
      * Get a pointer to an instance that was created by the last call to
      * the method 'create'.
      * @return the shared pointer to the configuration object
      * @throws std::logic_error when attempting to call the bethod before creating an instance.
      */
-    static std::shared_ptr<CzarConfig> instance();
+    static Ptr instance();
 
     CzarConfig() = delete;
     CzarConfig(CzarConfig const&) = delete;
@@ -108,36 +109,8 @@ class CzarConfig {
      */
     int getInteractiveChunkLimit() const { return _interactiveChunkLimit->getVal(); }
 
-    /* Get hostname and port for xrootd manager
-     *
-     * "localhost:1094" is the most reasonable default, even though it is
-     * the wrong choice for all but small developer installations
-     *
-     * @return a string containing "<hostname>:<port>"
-     */
-    std::string const& getXrootdFrontendUrl() const { return _xrootdFrontendUrl->getVal(); }
-
-    /* Get the maximum number of threads for xrootd to use.
-     *
-     * @return the maximum number of threads for xrootd to use.
-     */
-    int getXrootdCBThreadsMax() const { return _xrootdCBThreadsMax->getVal(); }
-
-    /* Get the initial number of threads for xrootd to create and maintain.
-     *
-     * @return the initial number of threads for xrootd to use.
-     */
-    int getXrootdCBThreadsInit() const { return _xrootdCBThreadsInit->getVal(); }
-
     bool getQueryDistributionTestVer() const { return _queryDistributionTestVer->getVal(); }
 
-    /*
-     * @return A value of the "spread" parameter. This may improve a performance
-     * of xrootd for catalogs with the large number of chunks. The default value
-     * of this parameter in xrootd is 4.
-     */
-    int getXrootdSpread() const { return _xrootdSpread->getVal(); }
-
     /* Get minimum number of seconds between QMeta chunk completion updates.
      *
      * @return seconds between QMeta chunk completion updates.
@@ -198,6 +171,31 @@ class CzarConfig {
     /// the OOM situation.
     unsigned int czarStatsRetainPeriodSec() const { return _czarStatsRetainPeriodSec->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(); }
+
+    /// A worker is considered DEAD if it hasn't been heard from in more than
+    /// _activeWorkerTimeoutDeadSecs.
+    int getActiveWorkerTimeoutDeadSecs() const { return _activeWorkerTimeoutDeadSecs->getVal(); }
+
+    /// Max lifetime of a message to be sent to an active worker. If the czar has been
+    /// trying to send a message to a worker and has failed for this many seconds,
+    /// it gives up at this point, removing elements of the message to save memory.
+    int getActiveWorkerMaxLifetimeSecs() const { return _activeWorkerMaxLifetimeSecs->getVal(); }
+
+    /// The maximum number of chunks (basically Jobs) allowed in a single UberJob.
+    int getUberJobMaxChunks() const { return _uberJobMaxChunks->getVal(); }
+
+    /// Return the maximum number of http connections to use for czar commands.
+    int getCommandMaxHttpConnections() const { return _commandMaxHttpConnections->getVal(); }
+
+    /// Return the sleep time (in milliseconds) between messages sent to active workers.
+    int getMonitorSleepTimeMilliSec() const { return _monitorSleepTimeMilliSec->getVal(); }
+
+    /// Return true if family map chunk distribution should depend on chunk size.
+    bool getFamilyMapUsingChunkSize() const { return _familyMapUsingChunkSize->getVal(); }
+
     // Parameters of the Czar management service
 
     std::string const& replicationInstanceId() const { return _replicationInstanceId->getVal(); }
@@ -293,7 +291,7 @@ class CzarConfig {
     CVTIntPtr _resultMaxConnections =
             util::ConfigValTInt::create(_configValMap, "resultdb", "maxconnections", notReq, 40);
     CVTIntPtr _resultMaxHttpConnections =
-            util::ConfigValTInt::create(_configValMap, "resultdb", "maxhttpconnections", notReq, 8192);
+            util::ConfigValTInt::create(_configValMap, "resultdb", "maxhttpconnections", notReq, 2000);
     CVTIntPtr _oldestResultKeptDays =
             util::ConfigValTInt::create(_configValMap, "resultdb", "oldestResultKeptDays", notReq, 30);
 
@@ -332,8 +330,6 @@ class CzarConfig {
     CVTStrPtr _qstatusDb =
             util::ConfigValTStr::create(_configValMap, "qstatus", "db", notReq, "qservStatusData");
 
-    CVTStrPtr _xrootdFrontendUrl =
-            util::ConfigValTStr::create(_configValMap, "frontend", "xrootd", notReq, "localhost:1094");
     CVTStrPtr _emptyChunkPath =
             util::ConfigValTStr::create(_configValMap, "partitioner", "emptyChunkPath", notReq, ".");
     CVTIntPtr _maxMsgSourceStore =
@@ -344,19 +340,18 @@ class CzarConfig {
     CVTIntPtr _qdispMaxPriority =
             util::ConfigValTInt::create(_configValMap, "qdisppool", "largestPriority", notReq, 2);
     CVTStrPtr _qdispVectRunSizes =
-            util::ConfigValTStr::create(_configValMap, "qdisppool", "vectRunSizes", notReq, "50:50:50:50");
+            util::ConfigValTStr::create(_configValMap, "qdisppool", "vectRunSizes", notReq, "800:800:500:50");
     CVTStrPtr _qdispVectMinRunningSizes =
-            util::ConfigValTStr::create(_configValMap, "qdisppool", "vectMinRunningSizes", notReq, "0:1:3:3");
+            util::ConfigValTStr::create(_configValMap, "qdisppool", "vectMinRunningSizes", notReq, "0:3:3:3");
+
+    // UberJobs
+    CVTIntPtr _uberJobMaxChunks =
+            util::ConfigValTInt::create(_configValMap, "uberjob", "maxChunks", notReq, 10000);
 
-    CVTIntPtr _xrootdSpread = util::ConfigValTInt::create(_configValMap, "tuning", "xrootdSpread", notReq, 4);
     CVTIntPtr _qMetaSecsBetweenChunkCompletionUpdates = util::ConfigValTInt::create(
             _configValMap, "tuning", "qMetaSecsBetweenChunkCompletionUpdates", notReq, 60);
     CVTIntPtr _interactiveChunkLimit =
             util::ConfigValTInt::create(_configValMap, "tuning", "interactiveChunkLimit", notReq, 10);
-    CVTIntPtr _xrootdCBThreadsMax =
-            util::ConfigValTInt::create(_configValMap, "tuning", "xrootdCBThreadsMax", notReq, 500);
-    CVTIntPtr _xrootdCBThreadsInit =
-            util::ConfigValTInt::create(_configValMap, "tuning", "xrootdCBThreadsInit", notReq, 50);
     CVTIntPtr _queryDistributionTestVer =
             util::ConfigValTInt::create(_configValMap, "tuning", "queryDistributionTestVer", notReq, 0);
     CVTBoolPtr _notifyWorkersOnQueryFinish =
@@ -385,6 +380,24 @@ class CzarConfig {
             util::ConfigValTInt::create(_configValMap, "replication", "http_port", notReq, 0);
     CVTUIntPtr _replicationNumHttpThreads =
             util::ConfigValTUInt::create(_configValMap, "replication", "num_http_threads", notReq, 2);
+
+    // Active Worker
+    CVTIntPtr _activeWorkerTimeoutAliveSecs =  // 5min
+            util::ConfigValTInt::create(_configValMap, "activeworker", "timeoutAliveSecs", notReq, 60 * 5);
+    CVTIntPtr _activeWorkerTimeoutDeadSecs =  // 10min
+            util::ConfigValTInt::create(_configValMap, "activeworker", "timeoutDeadSecs", notReq, 60 * 10);
+    CVTIntPtr _activeWorkerMaxLifetimeSecs =  // 1hr
+            util::ConfigValTInt::create(_configValMap, "activeworker", "maxLifetimeSecs", notReq, 60 * 60);
+    CVTIntPtr _monitorSleepTimeMilliSec = util::ConfigValTInt::create(
+            _configValMap, "activeworker", "monitorSleepTimeMilliSec", notReq, 15'000);
+
+    // FamilyMap
+    CVTBoolPtr _familyMapUsingChunkSize =
+            util::ConfigValTBool::create(_configValMap, "familymap", "usingChunkSize", notReq, 0);
+
+    /// This may impact `_resultMaxHttpConnections` as too many connections may cause kernel memory issues.
+    CVTIntPtr _commandMaxHttpConnections =
+            util::ConfigValTInt::create(_configValMap, "uberjob", "commandMaxHttpConnections", notReq, 2000);
 };
 
 }  // namespace lsst::qserv::cconfig

src/ccontrol/CMakeLists.txt

@@ -33,7 +33,6 @@ target_link_libraries(ccontrol PUBLIC
     parser
     replica
     sphgeom
-    xrdreq
     XrdCl
 )
 
@@ -52,7 +51,6 @@ FUNCTION(ccontrol_tests)
             qserv_meta
             query
             rproc
-            xrdreq
             Boost::unit_test_framework
             Threads::Threads
         )