Merge pull request #8048 from nextcloud/bugfix/slowBulkUpload

Bugfix/slow bulk upload

Author: mgallien

src/common/checksums.cpp

@@ -201,7 +201,7 @@ QByteArray ComputeChecksum::checksumType() const
 
 void ComputeChecksum::start(const QString &filePath)
 {
-    qCInfo(lcChecksums) << "Computing" << checksumType() << "checksum of" << filePath << "in a thread";
+    qCDebug(lcChecksums) << "Computing" << checksumType() << "checksum of" << filePath << "in a thread";
     startImpl(filePath);
 }
 

src/common/syncjournaldb.cpp

@@ -1711,13 +1711,11 @@ static void toDownloadInfo(SqlQuery &query, SyncJournalDb::DownloadInfo *res)
     res->_valid = ok;
 }
 
-static bool deleteBatch(SqlQuery &query, const QStringList &entries, const QString &name)
+static bool deleteBatch(SqlQuery &query, const QStringList &entries)
 {
     if (entries.isEmpty())
         return true;
 
-    qCDebug(lcDb) << "Removing stale" << name << "entries:" << entries.join(QStringLiteral(", "));
-    // FIXME: Was ported from execBatch, check if correct!
     for (const auto &entry : entries) {
         query.reset_and_clear_bindings();
         query.bindValue(1, entry);
@@ -1831,7 +1829,7 @@ QVector<SyncJournalDb::DownloadInfo> SyncJournalDb::getAndDeleteStaleDownloadInf
             qCDebug(lcDb) << "database error:" << query->error();
             return empty_result;
         }
-        if (!deleteBatch(*query, superfluousPaths, QStringLiteral("downloadinfo"))) {
+        if (!deleteBatch(*query, superfluousPaths)) {
             return empty_result;
         }
     }
@@ -1965,7 +1963,7 @@ QVector<uint> SyncJournalDb::deleteStaleUploadInfos(const QSet<QString> &keep)
     }
 
     const auto deleteUploadInfoQuery = _queryManager.get(PreparedSqlQueryManager::DeleteUploadInfoQuery);
-    deleteBatch(*deleteUploadInfoQuery, superfluousPaths, QStringLiteral("uploadinfo"));
+    deleteBatch(*deleteUploadInfoQuery, superfluousPaths);
     return ids;
 }
 
@@ -2033,7 +2031,7 @@ bool SyncJournalDb::deleteStaleErrorBlacklistEntries(const QSet<QString> &keep)
 
     SqlQuery delQuery(_db);
     delQuery.prepare("DELETE FROM blacklist WHERE path = ?");
-    return deleteBatch(delQuery, superfluousPaths, QStringLiteral("blacklist"));
+    return deleteBatch(delQuery, superfluousPaths);
 }
 
 void SyncJournalDb::deleteStaleFlagsEntries()

src/gui/folderwatcher.cpp

@@ -223,8 +223,6 @@ void FolderWatcher::changeDetected(const QStringList &paths)
             _lockedFiles.insert(checkResult.path);
         }
 
-        qCDebug(lcFolderWatcher) << "Locked files:" << _lockedFiles.values();
-
         // ------- handle ignores:
         if (pathIsIgnored(path)) {
             continue;
@@ -233,9 +231,6 @@ void FolderWatcher::changeDetected(const QStringList &paths)
         changedPaths.insert(path);
     }
 
-    qCDebug(lcFolderWatcher) << "Unlocked files:" << _unlockedFiles.values();
-    qCDebug(lcFolderWatcher) << "Locked files:" << _lockedFiles;
-
     if (!_lockedFiles.isEmpty() || !_unlockedFiles.isEmpty()) {
         if (_lockChangeDebouncingTimer.isActive()) {
             _lockChangeDebouncingTimer.stop();

src/libsync/bulkpropagatorjob.cpp

@@ -58,7 +58,6 @@ QByteArray getHeaderFromJsonReply(const QJsonObject &reply, const QByteArray &he
     return reply.value(headerName).toString().toLatin1();
 }
 
-constexpr auto batchSize = 100;
 constexpr auto parallelJobsMaximumCount = 1;
 
 }
@@ -70,10 +69,10 @@ Q_LOGGING_CATEGORY(lcBulkPropagatorJob, "nextcloud.sync.propagator.bulkupload",
 BulkPropagatorJob::BulkPropagatorJob(OwncloudPropagator *propagator, const std::deque<SyncFileItemPtr> &items)
     : PropagatorJob(propagator)
     , _items(items)
-    , _currentBatchSize(batchSize)
+    , _currentBatchSize(_items.size())
 {
-    _filesToUpload.reserve(batchSize);
-    _pendingChecksumFiles.reserve(batchSize);
+    _filesToUpload.reserve(_items.size());
+    _pendingChecksumFiles.reserve(_items.size());
 }
 
 bool BulkPropagatorJob::scheduleSelfOrChild()
@@ -84,11 +83,15 @@ bool BulkPropagatorJob::scheduleSelfOrChild()
 
     _state = Running;
 
-    for(auto i = 0; i < _currentBatchSize && !_items.empty(); ++i) {
+    qCDebug(lcBulkPropagatorJob()) << "max chunk size" << PropagatorJob::propagator()->syncOptions().maxChunkSize();
+
+    for(auto batchDataSize = 0; batchDataSize <= PropagatorJob::propagator()->syncOptions().maxChunkSize() && !_items.empty(); ) {
         const auto currentItem = _items.front();
         _items.pop_front();
         _pendingChecksumFiles.insert(currentItem->_file);
 
+        batchDataSize += currentItem->_size;
+
         QMetaObject::invokeMethod(this, [this, currentItem] {
             UploadFileInfo fileToUpload;
             fileToUpload._file = currentItem->_file;
@@ -117,7 +120,7 @@ bool BulkPropagatorJob::handleBatchSize()
     }
 
     // change batch size before trying it again
-    const auto halfBatchSize = batchSize / 2;
+    const auto halfBatchSize = static_cast<int>(_items.size() / 2);
 
     // we already tried to upload with half of the batch size
     if(_currentBatchSize == halfBatchSize) {
@@ -221,7 +224,6 @@ void BulkPropagatorJob::doStartUpload(SyncFileItemPtr item,
                 remotePath, fileToUpload._path,
                 fileToUpload._size, currentHeaders};
 
-    qCInfo(lcBulkPropagatorJob) << remotePath << "transmission checksum" << transmissionChecksumHeader << fileToUpload._path;
     _filesToUpload.push_back(std::move(newUploadFile));
     _pendingChecksumFiles.remove(item->_file);
 
@@ -583,7 +585,7 @@ void BulkPropagatorJob::finalizeOneFile(const BulkUploadItem &oneFile)
 
 void BulkPropagatorJob::finalize(const QJsonObject &fullReply)
 {
-    qCDebug(lcBulkPropagatorJob) << "Received a full reply" << fullReply;
+    qCDebug(lcBulkPropagatorJob) << "Received a full reply" << QJsonDocument::fromVariant(fullReply).toJson();
 
     for(auto singleFileIt = std::begin(_filesToUpload); singleFileIt != std::end(_filesToUpload); ) {
         const auto &singleFile = *singleFileIt;

src/libsync/configfile.cpp

@@ -272,13 +272,13 @@ qint64 ConfigFile::chunkSize() const
 qint64 ConfigFile::maxChunkSize() const
 {
     QSettings settings(configFile(), QSettings::IniFormat);
-    return settings.value(QLatin1String(maxChunkSizeC), 5LL * 1000LL * 1000LL * 1000LL).toLongLong(); // default to 5000 MB
+    return settings.value(QLatin1String(maxChunkSizeC), 100LL * 1024LL * 1024LL).toLongLong(); // default to 100 MiB
 }
 
 qint64 ConfigFile::minChunkSize() const
 {
     QSettings settings(configFile(), QSettings::IniFormat);
-    return settings.value(QLatin1String(minChunkSizeC), 5LL * 1000LL * 1000LL).toLongLong(); // default to 5 MB
+    return settings.value(QLatin1String(minChunkSizeC), 5LL * 1024LL * 1024LL).toLongLong(); // default to 5 MiB
 }
 
 chrono::milliseconds ConfigFile::targetChunkUploadDuration() const

src/libsync/owncloudpropagator.cpp

@@ -1267,7 +1267,9 @@ bool PropagatorCompositeJob::scheduleSelfOrChild()
         _tasksToDo.remove(0);
         PropagatorJob *job = propagator()->createJob(nextTask);
         if (!job) {
-            qCWarning(lcDirectory) << "Useless task found for file" << nextTask->destination() << "instruction" << nextTask->_instruction;
+            if (!propagator()->isDelayedUploadItem(nextTask)) {
+                qCWarning(lcDirectory) << "Useless task found for file" << nextTask->destination() << "instruction" << nextTask->_instruction;
+            }
             continue;
         }
         appendJob(job);

src/libsync/propagateupload.cpp

@@ -64,18 +64,27 @@ void PUTFileJob::start()
 
     req.setPriority(QNetworkRequest::LowPriority); // Long uploads must not block non-propagation jobs.
 
+    auto requestID = QByteArray{};
+
     if (_url.isValid()) {
-        sendRequest("PUT", _url, req, _device);
+        const auto reply = sendRequest("PUT", _url, req, _device);
+        requestID = reply->request().rawHeader("X-Request-ID");
     } else {
-        sendRequest("PUT", makeDavUrl(path()), req, _device);
+        const auto reply = sendRequest("PUT", makeDavUrl(path()), req, _device);
+        requestID = reply->request().rawHeader("X-Request-ID");
     }
 
     if (reply()->error() != QNetworkReply::NoError) {
         qCWarning(lcPutJob) << " Network error: " << reply()->errorString();
     }
 
+    connect(reply(), &QNetworkReply::uploadProgress, this, [requestID] (qint64 bytesSent, qint64 bytesTotal) {
+        qCDebug(lcPutJob()) << requestID << "upload progress" << bytesSent << bytesTotal;
+    });
+
     connect(reply(), &QNetworkReply::uploadProgress, this, &PUTFileJob::uploadProgress);
     connect(this, &AbstractNetworkJob::networkActivity, account().data(), &Account::propagatorNetworkActivity);
+
     _requestTimer.start();
     AbstractNetworkJob::start();
 }

src/libsync/putmultifilejob.cpp

@@ -32,8 +32,6 @@ PutMultiFileJob::PutMultiFileJob(AccountPtr account,
 
     for(const auto &singleDevice : _devices) {
         singleDevice._device->setParent(this);
-        connect(this, &PutMultiFileJob::uploadProgress,
-                singleDevice._device.get(), &UploadDevice::slotJobUploadProgress);
     }
 }
 
@@ -56,7 +54,12 @@ void PutMultiFileJob::start()
         if (oneDevice._device->size() == 0) {
             onePart.setBody({});
         } else {
-            onePart.setBodyDevice(oneDevice._device.get());
+            const auto allData = oneDevice._device->readAll();
+            onePart.setBody(allData);
+        }
+
+        if (oneDevice._device->isOpen()) {
+            oneDevice._device->close();
         }
 
         for (auto it = oneDevice._headers.begin(); it != oneDevice._headers.end(); ++it) {
@@ -68,13 +71,17 @@ void PutMultiFileJob::start()
         _body.append(onePart);
     }
 
-    sendRequest("POST", _url, req, &_body);
+    const auto newReply = sendRequest("POST", _url, req, &_body);
+    const auto &requestID = newReply->request().rawHeader("X-Request-ID");
 
     if (reply()->error() != QNetworkReply::NoError) {
         qCWarning(lcPutMultiFileJob) << " Network error: " << reply()->errorString();
     }
 
     connect(reply(), &QNetworkReply::uploadProgress, this, &PutMultiFileJob::uploadProgress);
+    connect(reply(), &QNetworkReply::uploadProgress, this, [requestID] (qint64 bytesSent, qint64 bytesTotal) {
+        qCDebug(lcPutMultiFileJob()) << requestID << "upload progress" << bytesSent << bytesTotal;
+    });
     connect(this, &AbstractNetworkJob::networkActivity, account().data(), &Account::propagatorNetworkActivity);
     _requestTimer.start();
     AbstractNetworkJob::start();
@@ -90,15 +97,12 @@ bool PutMultiFileJob::finished()
     for(const auto &oneDevice : _devices) {
         Q_ASSERT(oneDevice._device);
 
-        if (!oneDevice._device->errorString().isEmpty()) {
-            qCWarning(lcPutMultiFileJob) << "oneDevice has error:" << oneDevice._device->errorString();
-        }
-
         if (oneDevice._device->isOpen()) {
+            if (!oneDevice._device->errorString().isEmpty()) {
+                qCWarning(lcPutMultiFileJob) << "oneDevice has error:" << oneDevice._device->errorString();
+            }
+
             oneDevice._device->close();
-        } else {
-            qCWarning(lcPutMultiFileJob) << "Did not close device" << oneDevice._device.get()
-                                         << "as it was not open";
         }
     }
 

src/libsync/syncengine.cpp

@@ -228,7 +228,6 @@ void SyncEngine::deleteStaleDownloadInfos(const SyncFileItemVector &syncItems)
         _journal->getAndDeleteStaleDownloadInfos(download_file_paths);
     for (const SyncJournalDb::DownloadInfo &deleted_info : deleted_infos) {
         const QString tmppath = _propagator->fullLocalPath(deleted_info._tmpfile);
-        qCInfo(lcEngine) << "Deleting stale temporary file: " << tmppath;
         FileSystem::remove(tmppath);
     }
 }

test/testsyncengine.cpp

@@ -1208,7 +1208,7 @@ private slots:
 
         QVERIFY(fakeFolder.syncOnce());
         QCOMPARE(nPUT, 0);
-        QCOMPARE(nPOST, 2);
+        QCOMPARE(nPOST, 1);
         nPUT = 0;
         nPOST = 0;
 
@@ -1219,7 +1219,7 @@ private slots:
 
         QVERIFY(!fakeFolder.syncOnce());
         QCOMPARE(nPUT, 120);
-        QCOMPARE(nPOST, 2);
+        QCOMPARE(nPOST, 1);
         nPUT = 0;
         nPOST = 0;