[stable-3.16] Bugfix/slow bulk upload

src/common/checksums.cpp

@@ -15,7 +15,7 @@
  * License along with this library; if not, write to the Free Software
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 #include "config.h"
 #include "filesystembase.h"
 #include "common/checksums.h"
 #include "checksumcalculator.h"
@@ -201,7 +201,7 @@
 
 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

@@ -16,7 +16,7 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
 #include <QCryptographicHash>
 #include <QFile>
 #include <QLoggingCategory>
 #include <QStringList>
@@ -1711,13 +1711,11 @@
     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 @@
             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 @@
     }
 
     const auto deleteUploadInfoQuery = _queryManager.get(PreparedSqlQueryManager::DeleteUploadInfoQuery);
-    deleteBatch(*deleteUploadInfoQuery, superfluousPaths, QStringLiteral("uploadinfo"));
+    deleteBatch(*deleteUploadInfoQuery, superfluousPaths);
     return ids;
 }
 
@@ -2033,7 +2031,7 @@
 
     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

@@ -1,3 +1,3 @@
 /*
  * Copyright 2021 (c) Matthieu Gallien <matthieu.gallien@nextcloud.com>
  *
@@ -58,7 +58,6 @@
     return reply.value(headerName).toString().toLatin1();
 }
 
-constexpr auto batchSize = 100;
 constexpr auto parallelJobsMaximumCount = 1;
 
 }
@@ -70,9 +69,10 @@
 BulkPropagatorJob::BulkPropagatorJob(OwncloudPropagator *propagator, const std::deque<SyncFileItemPtr> &items)
     : PropagatorJob(propagator)
     , _items(items)
+    , _currentBatchSize(_items.size())
 {
-    _filesToUpload.reserve(batchSize);
-    _pendingChecksumFiles.reserve(batchSize);
+    _filesToUpload.reserve(_items.size());
+    _pendingChecksumFiles.reserve(_items.size());
 }
 
 bool BulkPropagatorJob::scheduleSelfOrChild()
@@ -83,11 +83,15 @@
 
     _state = Running;
 
-    for(auto i = 0; i < batchSize && !_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;
@@ -107,6 +111,29 @@
     return _items.empty() && _filesToUpload.empty();
 }
 
+bool BulkPropagatorJob::handleBatchSize()
+{
+    // no error, no batch size to change
+    if (_finalStatus == SyncFileItem::Success || _finalStatus == SyncFileItem::NoStatus) {
+        qCDebug(lcBulkPropagatorJob) << "No error, no need to change the bulk upload batch size!";
+        return true;
+    }
+
+    // change batch size before trying it again
+    const auto halfBatchSize = static_cast<int>(_items.size() / 2);
+
+    // we already tried to upload with half of the batch size
+    if(_currentBatchSize == halfBatchSize) {
+        qCDebug(lcBulkPropagatorJob) << "There was another error, stop syncing now!";
+        return false;
+    }
+
+    // try to upload with half of the batch size
+    _currentBatchSize = halfBatchSize;
+    qCDebug(lcBulkPropagatorJob) << "There was an error, sync again with bulk upload batch size cut to half!";
+    return true;
+}
+
 PropagatorJob::JobParallelism BulkPropagatorJob::parallelism() const
 {
     return PropagatorJob::JobParallelism::FullParallelism;
@@ -197,7 +224,6 @@
                 remotePath, fileToUpload._path,
                 fileToUpload._size, currentHeaders};
 
-    qCInfo(lcBulkPropagatorJob) << remotePath << "transmission checksum" << transmissionChecksumHeader << fileToUpload._path;
     _filesToUpload.push_back(std::move(newUploadFile));
     _pendingChecksumFiles.remove(item->_file);
 
@@ -265,13 +291,16 @@
             // just wait for the other job to finish.
             return;
         }
-
-        qCInfo(lcBulkPropagatorJob) << "final status" << _finalStatus;
-        emit finished(_finalStatus);
-        propagator()->scheduleNextJob();
     } else {
-        scheduleSelfOrChild();
+        if (handleBatchSize()) {
+            scheduleSelfOrChild();
+            return;
+        }
     }
+
+    qCInfo(lcBulkPropagatorJob) << "final status" << _finalStatus;
+    emit finished(_finalStatus);
+    propagator()->scheduleNextJob();
 }
 
 void BulkPropagatorJob::slotComputeTransmissionChecksum(SyncFileItemPtr item,
@@ -418,6 +447,9 @@
 
     singleFile._item->_status = SyncFileItem::Success;
 
+    // upload succeeded, so remove from black list
+    propagator()->removeFromBulkUploadBlackList(singleFile._item->_file);
+
     // Check the file again post upload.
     // Two cases must be considered separately: If the upload is finished,
     // the file is on the server and has a changed ETag. In that case,
@@ -553,7 +585,7 @@
 
 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/bulkpropagatorjob.h

@@ -162,6 +162,8 @@ private slots:
 
     void checkPropagationIsDone();
 
+    bool handleBatchSize();
+
     std::deque<SyncFileItemPtr> _items;
 
     QVector<AbstractNetworkJob *> _jobs; /// network jobs that are currently in transit
@@ -173,6 +175,7 @@ private slots:
     qint64 _sentTotal = 0;
 
     SyncFileItem::Status _finalStatus = SyncFileItem::Status::NoStatus;
+    int _currentBatchSize = 0;
 };
 
 }

src/libsync/configfile.cpp

@@ -12,7 +12,7 @@
  * for more details.
  */
 
 #include "config.h"
 
 #include "configfile.h"
 #include "theme.h"
@@ -270,13 +270,13 @@
 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

@@ -427,8 +427,6 @@ std::unique_ptr<PropagateUploadFileCommon> OwncloudPropagator::createUploadJob(S
 
     job->setDeleteExisting(deleteExisting);
 
-    removeFromBulkUploadBlackList(item->_file);
-
     return job;
 }
 
@@ -1269,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);
@@ -1338,8 +1338,9 @@ void PropagatorCompositeJob::finalize()
 {
     // The propagator will do parallel scheduling and this could be posted
     // multiple times on the event loop, ignore the duplicate calls.
-    if (_state == Finished)
+    if (_state == Finished) {
         return;
+    }
 
     _state = Finished;
     emit finished(_hasError == SyncFileItem::NoStatus ? SyncFileItem::Success : _hasError);

src/libsync/propagateupload.cpp

@@ -1,3 +1,3 @@
 /*
  * Copyright (C) by Olivier Goffart <ogoffart@owncloud.com>
  *
@@ -12,7 +12,7 @@
  * for more details.
  */
 
 #include "config.h"
 #include "propagateupload.h"
 #include "propagateuploadencrypted.h"
 #include "owncloudpropagator_p.h"
@@ -64,18 +64,27 @@
 
     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

@@ -1,3 +1,3 @@
 /*
  * Copyright 2021 (c) Matthieu Gallien <matthieu.gallien@nextcloud.com>
  *
@@ -32,8 +32,6 @@
 
     for(const auto &singleDevice : _devices) {
         singleDevice._device->setParent(this);
-        connect(this, &PutMultiFileJob::uploadProgress,
-                singleDevice._device.get(), &UploadDevice::slotJobUploadProgress);
     }
 }
 
@@ -56,7 +54,12 @@
         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 @@
         _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 @@
     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);
     }
 }