Improved implementatin of the chunk map building algorithm

Author: iagaponenko

src/replica/contr/ReplicationTask.cc

@@ -93,45 +93,71 @@ ReplicationTask::ReplicationTask(Controller::Ptr const& controller,
           _disableQservSync(disableQservSync),
           _forceQservSync(forceQservSync),
           _qservChunkMapUpdate(qservChunkMapUpdate),
-          _purge(purge) {}
-
-void ReplicationTask::_updateChunkMap() {
-    // Open MySQL connection using the RAII-style handler that would automatically
-    // abort the transaction should any problem occured when loading data into the table.
-    ConnectionHandler h;
-    try {
-        h.conn = Connection::open(Configuration::qservCzarDbParams("qservMeta"));
-    } catch (exception const& ex) {
-        error("failed to connect to the czar's database server, ex: " + string(ex.what()));
-        return;
-    }
-    QueryGenerator const g(h.conn);
+          _purge(purge),
+          _chunkMap(make_shared<ChunkMap>()) {}
 
+bool ReplicationTask::_getChunkMap() {
     // Get info on known chunk replicas from the persistent store of the Replication system
-    // and package those into ready-to-ingest data.
+    // and package those into the new chunk disposition map.
     bool const allDatabases = true;
     string const emptyDatabaseFilter;
     bool const isPublished = true;
     bool const includeFileInfo = true;  // need this to access tables sizes
-    vector<string> rows;
+    shared_ptr<ChunkMap> newChunkMap = make_shared<ChunkMap>();
     for (auto const& workerName : serviceProvider()->config()->workers()) {
         vector<ReplicaInfo> replicas;
         serviceProvider()->databaseServices()->findWorkerReplicas(replicas, workerName, emptyDatabaseFilter,
                                                                   allDatabases, isPublished, includeFileInfo);
         for (auto const& replica : replicas) {
+            // Incomplete replicas should not be used by Czar for query processing.
+            if (replica.status() != ReplicaInfo::Status::COMPLETE) continue;
             for (auto const& fileInfo : replica.fileInfo()) {
                 if (fileInfo.isData() && !fileInfo.isOverlap()) {
-                    rows.push_back(g.packVals(workerName, replica.database(), fileInfo.baseTable(),
-                                              replica.chunk(), fileInfo.size));
+                    (*newChunkMap)[workerName][replica.database()][fileInfo.baseTable()][replica.chunk()] =
+                            fileInfo.size;
                 }
             }
         }
     }
-    if (rows.empty()) {
-        warn("no replicas found in the persistent state of the Replication system");
+
+    // Update the current map if the new one is different from the current one.
+    if (*_chunkMap != *newChunkMap) {
+        _chunkMap = newChunkMap;
+        return true;
+    }
+    return false;
+}
+
+void ReplicationTask::_updateChunkMap() {
+    if (!_getChunkMap() || _chunkMap->empty()) {
+        // No changes in the chunk map, or the map is still empty so there's
+        // nothing to do.
         return;
     }
 
+    // Open MySQL connection using the RAII-style handler that would automatically
+    // abort the transaction should any problem occured when loading data into the table.
+    ConnectionHandler h;
+    try {
+        h.conn = Connection::open(Configuration::qservCzarDbParams("qservMeta"));
+    } catch (exception const& ex) {
+        error("failed to connect to the czar's database server, ex: " + string(ex.what()));
+        return;
+    }
+    QueryGenerator const g(h.conn);
+
+    // Pack the map into ready-to-ingest data.
+    vector<string> rows;
+    for (auto const& [workerName, databases] : *_chunkMap) {
+        for (auto const& [databaseName, tables] : databases) {
+            for (auto const& [tableName, chunks] : tables) {
+                for (auto const [chunkId, size] : chunks) {
+                    rows.push_back(g.packVals(workerName, databaseName, tableName, chunkId, size));
+                }
+            }
+        }
+    }
+
     // Get the limit for the length of the bulk insert queries. The limit is needed
     // to run the query generation.
     size_t maxQueryLength = 0;
@@ -163,6 +189,7 @@ void ReplicationTask::_updateChunkMap() {
         error("failed to update chunk map in the Czar database, ex: " + string(ex.what()));
         return;
     }
+    info("chunk map has been updated in the Czar database");
 }
 
 }  // namespace lsst::qserv::replica

src/replica/contr/ReplicationTask.h

@@ -21,6 +21,11 @@
 #ifndef LSST_QSERV_REPLICATIONTASK_H
 #define LSST_QSERV_REPLICATIONTASK_H
 
+// System headers
+#include <cstdint>
+#include <map>
+#include <string>
+
 // Qserv headers
 #include "replica/contr/Task.h"
 
@@ -56,7 +61,8 @@ class ReplicationTask : public Task {
      *   up on the Qserv synchronization requests.
      * @param disableQservSync Disable replica synchronization at Qserv workers if 'true'.
      * @param forceQservSync Force chunk removal at worker resource collections if 'true'.
-     * @param qservChunkMapUpdate Update the chunk disposition map in Qserv's QMeta database if 'true'.
+     * @param qservChunkMapUpdate Enable updating the chunk disposition map in Qserv's QMeta database if
+     * 'true'.
      * @param replicationIntervalSec The number of seconds to wait in the end of each
      *   iteration loop before to begin the new one.
      * @param purge Purge excess replicas if 'true'.
@@ -77,17 +83,36 @@ class ReplicationTask : public Task {
                     unsigned int qservSyncTimeoutSec, bool disableQservSync, bool forceQservSync,
                     bool qservChunkMapUpdate, unsigned int replicationIntervalSec, bool purge);
 
+    /// Get info on known chunk replicas from the persistent store of the Replication system
+    /// and package those into the new chunk disposition map. Update the current map if the new one is
+    /// different from the current one.
+    /// @return 'true' if the map has been updated, 'false' otherwise.
+    bool _getChunkMap();
+
+    /// Update the chunk disposition map in QMeta when changes in the map are detected.
     void _updateChunkMap();
 
     /// The maximum number of seconds to be waited before giving up
     /// on the Qserv synchronization requests.
     unsigned int const _qservSyncTimeoutSec;
 
-    bool const _disableQservSync;  ///< Disable replica synchroization at Qserv workers if 'true'.
-    bool const _forceQservSync;    ///< Force removal at worker resource collections if 'true'.
-    bool const
-            _qservChunkMapUpdate;  /// Update the chunk disposition map in Qserv's QMeta database if 'true'.
-    bool const _purge;             ///< Purge excess replicas if 'true'.
+    bool const _disableQservSync;     ///< Disable replica synchroization at Qserv workers if 'true'.
+    bool const _forceQservSync;       ///< Force removal at worker resource collections if 'true'.
+    bool const _qservChunkMapUpdate;  ///< Enable updating the chunk disposition map in Qserv's QMeta
+                                      /// database if 'true'.
+    bool const _purge;                ///< Purge excess replicas if 'true'.
+
+    /// [worker] -> [database] -> [baseTable] -> [chunk] -> size
+    ///
+    ///   The map represents the information on the replica disposition across Qserv workers.
+    ///   The information is obtained from the persistent state of the Replication system on each
+    ///   run of the task. The maps gets updated only if the new map is different from the current one.
+    ///
+    using ChunkMap =
+            std::map<std::string,
+                     std::map<std::string, std::map<std::string, std::map<unsigned int, std::uint64_t>>>>;
+
+    std::shared_ptr<ChunkMap> _chunkMap;  ///< The current chunk disposition map
 };
 
 }  // namespace lsst::qserv::replica