Improve import from rdb files

Author: uglide

src/app/models/treeoperations.cpp

@@ -114,7 +114,8 @@ void TreeOperations::requestBulkOperation(
           .arg(ns.getFullPath().size() > 0 ? conf().namespaceSeparator() : "");
   QRegExp filter(pattern, Qt::CaseSensitive, QRegExp::Wildcard);
 
-  emit m_events->requestBulkOperation(m_connection, ns.getDbIndex(), op, filter,
+  // NOTE(u_glide): Use "clean" connection wihout logger here for better performance
+  emit m_events->requestBulkOperation(m_connection->clone(), ns.getDbIndex(), op, filter,
                                       callback);
 }
 
@@ -271,7 +272,7 @@ void TreeOperations::copyKeys(ConnectionsTree::AbstractNamespaceItem& ns) {
 
 void TreeOperations::importKeysFromRdb(ConnectionsTree::DatabaseItem& db) {
   emit m_events->requestBulkOperation(
-      m_connection, db.getDbIndex(),
+      m_connection->clone(), db.getDbIndex(),
       BulkOperations::Manager::Operation::IMPORT_RDB_KEYS, QRegExp(".*"),
       [&db](QRegExp, int, const QStringList&) { db.reload(); });
 }

src/modules/bulk-operations/operations/abstractoperation.cpp

@@ -8,7 +8,8 @@ BulkOperations::AbstractOperation::AbstractOperation(
       m_keyPattern(keyPattern),
       m_currentState(State::READY),
       m_progress(0),
-      m_callback(callback) {}
+      m_callback(callback),
+      m_lastProgressNotification(0) {}
 
 void BulkOperations::AbstractOperation::getAffectedKeys(
     std::function<void(QVariant, QString)> callback) {
@@ -82,5 +83,17 @@ int BulkOperations::AbstractOperation::currentProgress() const {
 }
 
 void BulkOperations::AbstractOperation::setMetadata(const QVariantMap& meta) {
-  m_metadata = meta;
+    m_metadata = meta;
+}
+
+void BulkOperations::AbstractOperation::incrementProgress()
+{
+    QMutexLocker l(&m_processedKeysMutex);
+    m_progress++;
+
+    if (QDateTime::currentMSecsSinceEpoch() - m_lastProgressNotification >= 1000) {
+        qDebug() << "Notify UI about progress";
+        emit progress(m_progress);
+        m_lastProgressNotification = QDateTime::currentMSecsSinceEpoch();
+    }
 }

src/modules/bulk-operations/operations/abstractoperation.h

@@ -59,6 +59,8 @@ class AbstractOperation : public QObject {
       QSharedPointer<RedisClient::Connection> targetConnection,
       int targetDbIndex) = 0;
 
+  void incrementProgress();
+
  protected:
   QSharedPointer<RedisClient::Connection> m_connection;
   int m_dbIndex;
@@ -72,5 +74,6 @@ class AbstractOperation : public QObject {
   QStringList m_errors;
   QMutex m_errorsMutex;
   QMutex m_processedKeysMutex;
+  qint64 m_lastProgressNotification;
 };
 }  // namespace BulkOperations

src/modules/bulk-operations/operations/rdbimport.cpp

@@ -1,6 +1,7 @@
 #include "rdbimport.h"
 #include <qpython.h>
 #include <QFileInfo>
+#include <QtConcurrent>
 
 BulkOperations::RDBImportOperation::RDBImportOperation(
     QSharedPointer<RedisClient::Connection> connection, int dbIndex,
@@ -63,36 +64,42 @@ void BulkOperations::RDBImportOperation::performOperation(
     return returnResults();
   }
 
+  auto processCommands = [this, returnResults](const QVariantList& commands) {    
+    for (QVariant cmd : commands) {
+      auto rawCmd = convertToByteArray(cmd);
+
+      if (rawCmd.at(0).toLower() == QByteArray("select")) {
+          continue;
+      }
+
+      auto future = m_connection->cmd(
+          rawCmd, this, -1,
+          [this](const RedisClient::Response&) { incrementProgress(); },
+          [this, rawCmd](const QString& err) {
+            QMutexLocker l(&m_errorsMutex);
+            m_errors.append(
+                QCoreApplication::translate("RDM", "Cannot execute command ") +
+                QString("%1: %2")
+                    .arg(QString::fromUtf8(rawCmd.join(QByteArray(" "))))
+                    .arg(err));
+          });
+
+      m_combinator->combine(future);
+    }
+    m_combinator->subscribe(returnResults, returnResults);
+  };
+
   m_python->call_native(
       "rdb.rdb_export_as_commands",
       QVariantList{m_metadata["path"].toString(), m_metadata["db"].toInt(),
                    m_keyPattern.pattern()},
-      [returnResults, this](QVariant v) {
+      [processCommands, this](QVariant v) {
         QVariantList commands = v.toList();
 
-        for (QVariant cmd : commands) {
-          auto rawCmd = convertToByteArray(cmd);
-
-          auto future = m_connection->cmd(
-              rawCmd, this, m_dbIndex,
-              [this](const RedisClient::Response&) {
-                QMutexLocker l(&m_processedKeysMutex);
-                m_progress++;
-                emit progress(m_progress);
-              },
-              [this, rawCmd](const QString& err) {
-                QMutexLocker l(&m_errorsMutex);
-                m_errors.append(
-                    QCoreApplication::translate("RDM",
-                                                "Cannot execute command ") +
-                    QString("%1: %2")
-                        .arg(QString::fromUtf8(rawCmd.join(QByteArray(" "))))
-                        .arg(err));
-              });
-
-          m_combinator->combine(future);
-        }
-
-        m_combinator->subscribe(returnResults, returnResults);
+        m_connection->cmd(
+                {"select", QByteArray::number(m_dbIndex)}, this, -1,
+                [processCommands, commands](const RedisClient::Response&) {
+            QtConcurrent::run(processCommands, commands);
+        }, [](const QString&) {});
       });
 }