diff options
author | Markus Goetz <markus@woboq.com> | 2016-11-15 17:56:52 +0300 |
---|---|---|
committer | GitHub <noreply@github.com> | 2016-11-15 17:56:52 +0300 |
commit | 52552a420449c4882a00f044f9c75449fee60743 (patch) | |
tree | a9de7d921a84a9871681a10bd39279e49938f068 | |
parent | 2f3db04e87a4e3edff389389d46d1c7a5a3e0d03 (diff) | |
parent | 92027e869230d82de1bbbd0d486ebf14bfc471f7 (diff) |
Merge pull request #5102 from owncloud/chunking-ng
Chunking ng
-rw-r--r-- | src/libsync/CMakeLists.txt | 2 | ||||
-rw-r--r-- | src/libsync/account.cpp | 12 | ||||
-rw-r--r-- | src/libsync/account.h | 5 | ||||
-rw-r--r-- | src/libsync/capabilities.cpp | 6 | ||||
-rw-r--r-- | src/libsync/capabilities.h | 1 | ||||
-rw-r--r-- | src/libsync/connectionvalidator.cpp | 20 | ||||
-rw-r--r-- | src/libsync/connectionvalidator.h | 14 | ||||
-rw-r--r-- | src/libsync/networkjobs.cpp | 20 | ||||
-rw-r--r-- | src/libsync/networkjobs.h | 6 | ||||
-rw-r--r-- | src/libsync/owncloudpropagator.cpp | 9 | ||||
-rw-r--r-- | src/libsync/owncloudpropagator.h | 3 | ||||
-rw-r--r-- | src/libsync/propagateremotedelete.cpp | 5 | ||||
-rw-r--r-- | src/libsync/propagateremotedelete.h | 2 | ||||
-rw-r--r-- | src/libsync/propagateremotemove.cpp | 10 | ||||
-rw-r--r-- | src/libsync/propagateremotemove.h | 4 | ||||
-rw-r--r-- | src/libsync/propagateupload.cpp | 462 | ||||
-rw-r--r-- | src/libsync/propagateupload.h | 166 | ||||
-rw-r--r-- | src/libsync/propagateuploadng.cpp | 488 | ||||
-rw-r--r-- | src/libsync/propagateuploadv1.cpp | 375 | ||||
-rw-r--r-- | test/CMakeLists.txt | 1 | ||||
-rw-r--r-- | test/syncenginetestutils.h | 168 | ||||
-rw-r--r-- | test/testchunkingng.cpp | 78 |
22 files changed, 1383 insertions, 474 deletions
diff --git a/src/libsync/CMakeLists.txt b/src/libsync/CMakeLists.txt index abfb2cfd9..92b69c152 100644 --- a/src/libsync/CMakeLists.txt +++ b/src/libsync/CMakeLists.txt @@ -50,6 +50,8 @@ set(libsync_SRCS propagatorjobs.cpp propagatedownload.cpp propagateupload.cpp + propagateuploadv1.cpp + propagateuploadng.cpp propagateremotedelete.cpp propagateremotemove.cpp propagateremotemkdir.cpp diff --git a/src/libsync/account.cpp b/src/libsync/account.cpp index 5c6450673..3dfa7bcb8 100644 --- a/src/libsync/account.cpp +++ b/src/libsync/account.cpp @@ -75,11 +75,19 @@ AccountPtr Account::sharedFromThis() return _sharedThis.toStrongRef(); } +QString Account::user() const +{ + return _user.isEmpty() ? _credentials->user() : _user; +} + +void Account::setUser(const QString &user) +{ + _user = user; +} QString Account::displayName() const { - auto user = _credentials->user(); - QString dn = QString("%1@%2").arg(user, _url.host()); + QString dn = QString("%1@%2").arg(user(), _url.host()); int port = url().port(); if (port > 0 && port != 80 && port != 443) { dn.append(QLatin1Char(':')); diff --git a/src/libsync/account.h b/src/libsync/account.h index 4ff73748e..36e606ebf 100644 --- a/src/libsync/account.h +++ b/src/libsync/account.h @@ -68,6 +68,10 @@ public: AccountPtr sharedFromThis(); + /// The user that can be used in dav url + QString user() const; + void setUser(const QString &user); + /// The name of the account as shown in the toolbar QString displayName() const; @@ -198,6 +202,7 @@ private: QWeakPointer<Account> _sharedThis; QString _id; + QString _user; QMap<QString, QVariant> _settingsMap; QUrl _url; QList<QSslCertificate> _approvedCerts; diff --git a/src/libsync/capabilities.cpp b/src/libsync/capabilities.cpp index ca72a70bb..6bf352f9a 100644 --- a/src/libsync/capabilities.cpp +++ b/src/libsync/capabilities.cpp @@ -107,4 +107,10 @@ QByteArray Capabilities::uploadChecksumType() const return QByteArray(); } +bool Capabilities::chunkingNg() const +{ + return _capabilities["dav"].toMap()["chunking"].toByteArray() >= "1.0"; +} + + } diff --git a/src/libsync/capabilities.h b/src/libsync/capabilities.h index 04eef6e39..5f1936aba 100644 --- a/src/libsync/capabilities.h +++ b/src/libsync/capabilities.h @@ -39,6 +39,7 @@ public: bool sharePublicLinkEnforceExpireDate() const; int sharePublicLinkExpireDateDays() const; bool shareResharing() const; + bool chunkingNg() const; /// returns true if the capabilities report notifications bool notificationsAvailable() const; diff --git a/src/libsync/connectionvalidator.cpp b/src/libsync/connectionvalidator.cpp index a1eead282..d9f0c3e16 100644 --- a/src/libsync/connectionvalidator.cpp +++ b/src/libsync/connectionvalidator.cpp @@ -229,10 +229,26 @@ void ConnectionValidator::slotCapabilitiesRecieved(const QVariantMap &json) auto caps = json.value("ocs").toMap().value("data").toMap().value("capabilities"); qDebug() << "Server capabilities" << caps; _account->setCapabilities(caps.toMap()); - reportResult(Connected); - return; + fetchUser(); } +void ConnectionValidator::fetchUser() +{ + + JsonApiJob *job = new JsonApiJob(_account, QLatin1String("ocs/v1.php/cloud/user"), this); + job->setTimeout(timeoutToUseMsec); + QObject::connect(job, SIGNAL(jsonReceived(QVariantMap, int)), this, SLOT(slotUserFetched(QVariantMap))); + job->start(); +} + +void ConnectionValidator::slotUserFetched(const QVariantMap &json) +{ + QString user = json.value("ocs").toMap().value("data").toMap().value("id").toString(); + if (!user.isEmpty()) { + _account->setUser(user); + } + reportResult(Connected); +} void ConnectionValidator::reportResult(Status status) { diff --git a/src/libsync/connectionvalidator.h b/src/libsync/connectionvalidator.h index e1264628a..b97d665ec 100644 --- a/src/libsync/connectionvalidator.h +++ b/src/libsync/connectionvalidator.h @@ -27,7 +27,7 @@ namespace OCC { * This is a job-like class to check that the server is up and that we are connected. * There are two entry points: checkServerAndAuth and checkAuthentication * checkAuthentication is the quick version that only does the propfind - * while checkServerAndAuth is doing the 3 calls. + * while checkServerAndAuth is doing the 4 calls. * * We cannot use the capabilites call to test the login and the password because of * https://github.com/owncloud/core/issues/12930 @@ -60,7 +60,15 @@ namespace OCC { +-> checkServerCapabilities (cloud/capabilities) JsonApiJob | - +-> slotCapabilitiesRecieved --> X + +-> slotCapabilitiesRecieved -+ + | + +-----------------------------------+ + | + +-> fetchUser + PropfindJob + | + +-> slotUserFetched --> X + \endcode */ class OWNCLOUDSYNC_EXPORT ConnectionValidator : public QObject @@ -109,10 +117,12 @@ protected slots: void slotAuthSuccess(); void slotCapabilitiesRecieved(const QVariantMap&); + void slotUserFetched(const QVariantMap &); private: void reportResult(Status status); void checkServerCapabilities(); + void fetchUser(); QStringList _errors; AccountPtr _account; diff --git a/src/libsync/networkjobs.cpp b/src/libsync/networkjobs.cpp index e6d5a0e06..10fd3d545 100644 --- a/src/libsync/networkjobs.cpp +++ b/src/libsync/networkjobs.cpp @@ -106,14 +106,24 @@ MkColJob::MkColJob(AccountPtr account, const QString &path, QObject *parent) { } +MkColJob::MkColJob(AccountPtr account, const QUrl &url, + const QMap<QByteArray, QByteArray> &extraHeaders, QObject *parent) + : AbstractNetworkJob(account, QString(), parent), _url(url), _extraHeaders(extraHeaders) +{ +} + void MkColJob::start() { // add 'Content-Length: 0' header (see https://github.com/owncloud/client/issues/3256) QNetworkRequest req; req.setRawHeader("Content-Length", "0"); + for(auto it = _extraHeaders.constBegin(); it != _extraHeaders.constEnd(); ++it) { + req.setRawHeader(it.key(), it.value()); + } // assumes ownership - QNetworkReply *reply = davRequest("MKCOL", path(), req); + QNetworkReply *reply = _url.isValid() ? davRequest("MKCOL", _url, req) + : davRequest("MKCOL", path(), req); setReply(reply); setupConnections(reply); AbstractNetworkJob::start(); @@ -264,6 +274,11 @@ LsColJob::LsColJob(AccountPtr account, const QString &path, QObject *parent) { } +LsColJob::LsColJob(AccountPtr account, const QUrl &url, QObject *parent) + : AbstractNetworkJob(account, QString(), parent), _url(url) +{ +} + void LsColJob::setProperties(QList<QByteArray> properties) { _properties = properties; @@ -307,7 +322,8 @@ void LsColJob::start() QBuffer *buf = new QBuffer(this); buf->setData(xml); buf->open(QIODevice::ReadOnly); - QNetworkReply *reply = davRequest("PROPFIND", path(), req, buf); + QNetworkReply *reply = _url.isValid() ? davRequest("PROPFIND", _url, req, buf) + : davRequest("PROPFIND", path(), req, buf); buf->setParent(reply); setReply(reply); setupConnections(reply); diff --git a/src/libsync/networkjobs.h b/src/libsync/networkjobs.h index c2cc978c4..f7a37d950 100644 --- a/src/libsync/networkjobs.h +++ b/src/libsync/networkjobs.h @@ -62,6 +62,7 @@ class OWNCLOUDSYNC_EXPORT LsColJob : public AbstractNetworkJob { Q_OBJECT public: explicit LsColJob(AccountPtr account, const QString &path, QObject *parent = 0); + explicit LsColJob(AccountPtr account, const QUrl &url, QObject *parent = 0); void start() Q_DECL_OVERRIDE; QHash<QString, qint64> _sizes; @@ -87,6 +88,7 @@ private slots: private: QList<QByteArray> _properties; + QUrl _url; // Used instead of path() if the url is specified in the constructor }; /** @@ -170,8 +172,12 @@ private: */ class OWNCLOUDSYNC_EXPORT MkColJob : public AbstractNetworkJob { Q_OBJECT + QUrl _url; // Only used if the constructor taking a url is taken. + QMap<QByteArray, QByteArray> _extraHeaders; public: explicit MkColJob(AccountPtr account, const QString &path, QObject *parent = 0); + explicit MkColJob(AccountPtr account, const QUrl &url, + const QMap<QByteArray, QByteArray> &extraHeaders, QObject *parent = 0); void start() Q_DECL_OVERRIDE; signals: diff --git a/src/libsync/owncloudpropagator.cpp b/src/libsync/owncloudpropagator.cpp index 575d72abc..d8ad23ea9 100644 --- a/src/libsync/owncloudpropagator.cpp +++ b/src/libsync/owncloudpropagator.cpp @@ -270,7 +270,14 @@ PropagateItemJob* OwncloudPropagator::createJob(const SyncFileItemPtr &item) { job->setDeleteExistingFolder(deleteExisting); return job; } else { - auto job = new PropagateUploadFile(this, item); + PropagateUploadFileCommon *job = 0; + static const auto chunkng = qgetenv("OWNCLOUD_CHUNKING_NG"); + if (item->_size > chunkSize() + && (account()->capabilities().chunkingNg() || chunkng == "1") && chunkng != "0") { + job = new PropagateUploadFileNG(this, item); + } else { + job = new PropagateUploadFileV1(this, item); + } job->setDeleteExisting(deleteExisting); return job; } diff --git a/src/libsync/owncloudpropagator.h b/src/libsync/owncloudpropagator.h index b9862eaaf..3c6930662 100644 --- a/src/libsync/owncloudpropagator.h +++ b/src/libsync/owncloudpropagator.h @@ -380,10 +380,11 @@ private: #if QT_VERSION < QT_VERSION_CHECK(5, 0, 0) // access to signals which are protected in Qt4 friend class PropagateDownloadFile; - friend class PropagateUploadFile; friend class PropagateLocalMkdir; friend class PropagateLocalRename; friend class PropagateRemoteMove; + friend class PropagateUploadFileV1; + friend class PropagateUploadFileNG; #endif }; diff --git a/src/libsync/propagateremotedelete.cpp b/src/libsync/propagateremotedelete.cpp index 30b06fe41..9be19c309 100644 --- a/src/libsync/propagateremotedelete.cpp +++ b/src/libsync/propagateremotedelete.cpp @@ -22,11 +22,14 @@ DeleteJob::DeleteJob(AccountPtr account, const QString& path, QObject* parent) : AbstractNetworkJob(account, path, parent) { } +DeleteJob::DeleteJob(AccountPtr account, const QUrl& url, QObject* parent) + : AbstractNetworkJob(account, QString(), parent), _url(url) +{ } void DeleteJob::start() { QNetworkRequest req; - setReply(davRequest("DELETE", path(), req)); + setReply(_url.isValid() ? davRequest("DELETE", _url, req) : davRequest("DELETE", path(), req)); setupConnections(reply()); if( reply()->error() != QNetworkReply::NoError ) { diff --git a/src/libsync/propagateremotedelete.h b/src/libsync/propagateremotedelete.h index 0473042c8..44f4ebd3b 100644 --- a/src/libsync/propagateremotedelete.h +++ b/src/libsync/propagateremotedelete.h @@ -24,8 +24,10 @@ namespace OCC { */ class DeleteJob : public AbstractNetworkJob { Q_OBJECT + QUrl _url; // Only used if the constructor taking a url is taken. public: explicit DeleteJob(AccountPtr account, const QString& path, QObject* parent = 0); + explicit DeleteJob(AccountPtr account, const QUrl& url, QObject* parent = 0); void start() Q_DECL_OVERRIDE; bool finished() Q_DECL_OVERRIDE; diff --git a/src/libsync/propagateremotemove.cpp b/src/libsync/propagateremotemove.cpp index bc0b4efa1..bcceec709 100644 --- a/src/libsync/propagateremotemove.cpp +++ b/src/libsync/propagateremotemove.cpp @@ -28,12 +28,20 @@ MoveJob::MoveJob(AccountPtr account, const QString& path, : AbstractNetworkJob(account, path, parent), _destination(destination) { } +MoveJob::MoveJob(AccountPtr account, const QUrl& url, const QString &destination, + QMap<QByteArray, QByteArray> extraHeaders, QObject* parent) + : AbstractNetworkJob(account, QString(), parent), _destination(destination), _url(url) + , _extraHeaders(extraHeaders) +{ } void MoveJob::start() { QNetworkRequest req; req.setRawHeader("Destination", QUrl::toPercentEncoding(_destination, "/")); - setReply(davRequest("MOVE", path(), req)); + for(auto it = _extraHeaders.constBegin(); it != _extraHeaders.constEnd(); ++it) { + req.setRawHeader(it.key(), it.value()); + } + setReply(_url.isValid() ? davRequest("MOVE", _url, req) : davRequest("MOVE", path(), req)); setupConnections(reply()); if( reply()->error() != QNetworkReply::NoError ) { diff --git a/src/libsync/propagateremotemove.h b/src/libsync/propagateremotemove.h index 829e7a1fb..8d213e430 100644 --- a/src/libsync/propagateremotemove.h +++ b/src/libsync/propagateremotemove.h @@ -25,8 +25,12 @@ namespace OCC { class MoveJob : public AbstractNetworkJob { Q_OBJECT const QString _destination; + const QUrl _url; // Only used (instead of path) when the constructor taking an URL is used + QMap<QByteArray, QByteArray> _extraHeaders; public: explicit MoveJob(AccountPtr account, const QString& path, const QString &destination, QObject* parent = 0); + explicit MoveJob(AccountPtr account, const QUrl& url, const QString &destination, + QMap<QByteArray, QByteArray> _extraHeaders, QObject* parent = 0); void start() Q_DECL_OVERRIDE; bool finished() Q_DECL_OVERRIDE; diff --git a/src/libsync/propagateupload.cpp b/src/libsync/propagateupload.cpp index 2783ac9a5..388752ad9 100644 --- a/src/libsync/propagateupload.cpp +++ b/src/libsync/propagateupload.cpp @@ -72,7 +72,8 @@ void PUTFileJob::start() { req.setRawHeader(it.key(), it.value()); } - setReply(davRequest("PUT", path(), req, _device.data())); + setReply(_url.isValid() ? davRequest("PUT", _url, req, _device.data()) + : davRequest("PUT", path(), req, _device.data())); setupConnections(reply()); if( reply()->error() != QNetworkReply::NoError ) { @@ -184,7 +185,13 @@ bool PollJob::finished() return true; } -void PropagateUploadFile::start() +void PropagateUploadFileCommon::setDeleteExisting(bool enabled) +{ + _deleteExisting = enabled; +} + + +void PropagateUploadFileCommon::start() { if (_propagator->_abortRequested.fetchAndAddRelaxed(0)) { return; @@ -205,7 +212,7 @@ void PropagateUploadFile::start() job->start(); } -void PropagateUploadFile::slotComputeContentChecksum() +void PropagateUploadFileCommon::slotComputeContentChecksum() { if (_propagator->_abortRequested.fetchAndAddRelaxed(0)) { return; @@ -239,12 +246,7 @@ void PropagateUploadFile::slotComputeContentChecksum() computeChecksum->start(filePath); } -void PropagateUploadFile::setDeleteExisting(bool enabled) -{ - _deleteExisting = enabled; -} - -void PropagateUploadFile::slotComputeTransmissionChecksum(const QByteArray& contentChecksumType, const QByteArray& contentChecksum) +void PropagateUploadFileCommon::slotComputeTransmissionChecksum(const QByteArray& contentChecksumType, const QByteArray& contentChecksum) { _item->_contentChecksum = contentChecksum; _item->_contentChecksumType = contentChecksumType; @@ -276,7 +278,7 @@ void PropagateUploadFile::slotComputeTransmissionChecksum(const QByteArray& cont computeChecksum->start(filePath); } -void PropagateUploadFile::slotStartUpload(const QByteArray& transmissionChecksumType, const QByteArray& transmissionChecksum) +void PropagateUploadFileCommon::slotStartUpload(const QByteArray& transmissionChecksumType, const QByteArray& transmissionChecksum) { // Remove ourselfs from the list of active job, before any posible call to done() // When we start chunks, we will add it again, once for every chunks. @@ -322,23 +324,7 @@ void PropagateUploadFile::slotStartUpload(const QByteArray& transmissionChecksum return; } - _chunkCount = std::ceil(fileSize/double(chunkSize())); - _startChunk = 0; - _transferId = qrand() ^ _item->_modtime ^ (_item->_size << 16); - - const SyncJournalDb::UploadInfo progressInfo = _propagator->_journal->getUploadInfo(_item->_file); - - if (progressInfo._valid && Utility::qDateTimeToTime_t(progressInfo._modtime) == _item->_modtime ) { - _startChunk = progressInfo._chunk; - _transferId = progressInfo._transferid; - qDebug() << Q_FUNC_INFO << _item->_file << ": Resuming from chunk " << _startChunk; - } - - _currentChunk = 0; - _duration.start(); - - emit progress(*_item, 0); - this->startNextChunk(); + doStartUpload(); } UploadDevice::UploadDevice(BandwidthManager *bwm) @@ -476,371 +462,7 @@ void UploadDevice::setChoked(bool b) { } } -void PropagateUploadFile::startNextChunk() -{ - if (_propagator->_abortRequested.fetchAndAddRelaxed(0)) - return; - - if (! _jobs.isEmpty() && _currentChunk + _startChunk >= _chunkCount - 1) { - // Don't do parallel upload of chunk if this might be the last chunk because the server cannot handle that - // https://github.com/owncloud/core/issues/11106 - // We return now and when the _jobs are finished we will proceed with the last chunk - // NOTE: Some other parts of the code such as slotUploadProgress also assume that the last chunk - // is sent last. - return; - } - quint64 fileSize = _item->_size; - QMap<QByteArray, QByteArray> headers; - headers["OC-Total-Length"] = QByteArray::number(fileSize); - headers["OC-Async"] = "1"; - headers["OC-Chunk-Size"]= QByteArray::number(quint64(chunkSize())); - headers["Content-Type"] = "application/octet-stream"; - headers["X-OC-Mtime"] = QByteArray::number(qint64(_item->_modtime)); - - if(_item->_file.contains(".sys.admin#recall#")) { - // This is a file recall triggered by the admin. Note: the - // recall list file created by the admin and downloaded by the - // client (.sys.admin#recall#) also falls into this category - // (albeit users are not supposed to mess up with it) - - // We use a special tag header so that the server may decide to store this file away in some admin stage area - // And not directly in the user's area (which would trigger redownloads etc). - headers["OC-Tag"] = ".sys.admin#recall#"; - } - - if (!_item->_etag.isEmpty() && _item->_etag != "empty_etag" - && _item->_instruction != CSYNC_INSTRUCTION_NEW // On new files never send a If-Match - && _item->_instruction != CSYNC_INSTRUCTION_TYPE_CHANGE - && !_deleteExisting - ) { - // We add quotes because the owncloud server always adds quotes around the etag, and - // csync_owncloud.c's owncloud_file_id always strips the quotes. - headers["If-Match"] = '"' + _item->_etag + '"'; - } - - QString path = _item->_file; - - UploadDevice *device = new UploadDevice(&_propagator->_bandwidthManager); - qint64 chunkStart = 0; - qint64 currentChunkSize = fileSize; - bool isFinalChunk = false; - if (_chunkCount > 1) { - int sendingChunk = (_currentChunk + _startChunk) % _chunkCount; - // XOR with chunk size to make sure everything goes well if chunk size changes between runs - uint transid = _transferId ^ chunkSize(); - qDebug() << "Upload chunk" << sendingChunk << "of" << _chunkCount << "transferid(remote)=" << transid; - path += QString("-chunking-%1-%2-%3").arg(transid).arg(_chunkCount).arg(sendingChunk); - - headers["OC-Chunked"] = "1"; - - chunkStart = chunkSize() * quint64(sendingChunk); - currentChunkSize = chunkSize(); - if (sendingChunk == _chunkCount - 1) { // last chunk - currentChunkSize = (fileSize % chunkSize()); - if( currentChunkSize == 0 ) { // if the last chunk pretends to be 0, its actually the full chunk size. - currentChunkSize = chunkSize(); - } - isFinalChunk = true; - } - } else { - // if there's only one chunk, it's the final one - isFinalChunk = true; - } - - if (isFinalChunk && !_transmissionChecksumType.isEmpty()) { - headers[checkSumHeaderC] = makeChecksumHeader( - _transmissionChecksumType, _transmissionChecksum); - } - - const QString fileName = _propagator->getFilePath(_item->_file); - if (! device->prepareAndOpen(fileName, chunkStart, currentChunkSize)) { - qDebug() << "ERR: Could not prepare upload device: " << device->errorString(); - - // If the file is currently locked, we want to retry the sync - // when it becomes available again. - if (FileSystem::isFileLocked(fileName)) { - emit _propagator->seenLockedFile(fileName); - } - - // Soft error because this is likely caused by the user modifying his files while syncing - abortWithError( SyncFileItem::SoftError, device->errorString() ); - delete device; - return; - } - - // job takes ownership of device via a QScopedPointer. Job deletes itself when finishing - PUTFileJob* job = new PUTFileJob(_propagator->account(), _propagator->_remoteFolder + path, device, headers, _currentChunk); - _jobs.append(job); - connect(job, SIGNAL(finishedSignal()), this, SLOT(slotPutFinished())); - connect(job, SIGNAL(uploadProgress(qint64,qint64)), this, SLOT(slotUploadProgress(qint64,qint64))); - connect(job, SIGNAL(uploadProgress(qint64,qint64)), device, SLOT(slotJobUploadProgress(qint64,qint64))); - connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); - job->start(); - _propagator->_activeJobList.append(this); - _currentChunk++; - - bool parallelChunkUpload = true; - QByteArray env = qgetenv("OWNCLOUD_PARALLEL_CHUNK"); - if (!env.isEmpty()) { - parallelChunkUpload = env != "false" && env != "0"; - } else { - int versionNum = _propagator->account()->serverVersionInt(); - if (versionNum < 0x080003) { - // Disable parallel chunk upload severs older than 8.0.3 to avoid too many - // internal sever errors (#2743, #2938) - parallelChunkUpload = false; - } - } - - if (_currentChunk + _startChunk >= _chunkCount - 1) { - // Don't do parallel upload of chunk if this might be the last chunk because the server cannot handle that - // https://github.com/owncloud/core/issues/11106 - parallelChunkUpload = false; - } - - if (parallelChunkUpload && (_propagator->_activeJobList.count() < _propagator->maximumActiveJob()) - && _currentChunk < _chunkCount ) { - startNextChunk(); - } - if (!parallelChunkUpload || _chunkCount - _currentChunk <= 0) { - emit ready(); - } -} - -void PropagateUploadFile::slotPutFinished() -{ - PUTFileJob *job = qobject_cast<PUTFileJob *>(sender()); - Q_ASSERT(job); - slotJobDestroyed(job); // remove it from the _jobs list - - qDebug() << Q_FUNC_INFO << job->reply()->request().url() << "FINISHED WITH STATUS" - << job->reply()->error() - << (job->reply()->error() == QNetworkReply::NoError ? QLatin1String("") : job->reply()->errorString()) - << job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute) - << job->reply()->attribute(QNetworkRequest::HttpReasonPhraseAttribute); - - _propagator->_activeJobList.removeOne(this); - - if (_finished) { - // We have sent the finished signal already. We don't need to handle any remaining jobs - return; - } - - QNetworkReply::NetworkError err = job->reply()->error(); - -#if QT_VERSION < QT_VERSION_CHECK(5, 4, 2) - if (err == QNetworkReply::OperationCanceledError && job->reply()->property(owncloudShouldSoftCancelPropertyName).isValid()) { - // Abort the job and try again later. - // This works around a bug in QNAM wich might reuse a non-empty buffer for the next request. - qDebug() << "Forcing job abort on HTTP connection reset with Qt < 5.4.2."; - _propagator->_anotherSyncNeeded = true; - abortWithError(SyncFileItem::SoftError, tr("Forcing job abort on HTTP connection reset with Qt < 5.4.2.")); - return; - } -#endif - - if (err != QNetworkReply::NoError) { - _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); - if(checkForProblemsWithShared(_item->_httpErrorCode, - tr("The file was edited locally but is part of a read only share. " - "It is restored and your edit is in the conflict file."))) { - return; - } - QByteArray replyContent = job->reply()->readAll(); - qDebug() << replyContent; // display the XML error in the debug - QString errorString = errorMessage(job->errorString(), replyContent); - - if (job->reply()->hasRawHeader("OC-ErrorString")) { - errorString = job->reply()->rawHeader("OC-ErrorString"); - } - - if (_item->_httpErrorCode == 412) { - // Precondition Failed: Maybe the bad etag is in the database, we need to clear the - // parent folder etag so we won't read from DB next sync. - _propagator->_journal->avoidReadFromDbOnNextSync(_item->_file); - _propagator->_anotherSyncNeeded = true; - } - - SyncFileItem::Status status = classifyError(err, _item->_httpErrorCode, - &_propagator->_anotherSyncNeeded); - abortWithError(status, errorString); - return; - } - - _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); - // The server needs some time to process the request and provide us with a poll URL - if (_item->_httpErrorCode == 202) { - _finished = true; - QString path = QString::fromUtf8(job->reply()->rawHeader("OC-Finish-Poll")); - if (path.isEmpty()) { - done(SyncFileItem::NormalError, tr("Poll URL missing")); - return; - } - startPollJob(path); - return; - } - - // 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, - // the etag has to be properly updated in the client journal, and because - // of that we can bail out here with an error. But we can reschedule a - // sync ASAP. - // But if the upload is ongoing, because not all chunks were uploaded - // yet, the upload can be stopped and an error can be displayed, because - // the server hasn't registered the new file yet. - QByteArray etag = getEtagFromReply(job->reply()); - bool finished = etag.length() > 0; - - // Check if the file still exists - const QString fullFilePath(_propagator->getFilePath(_item->_file)); - if( !FileSystem::fileExists(fullFilePath) ) { - if (!finished) { - abortWithError(SyncFileItem::SoftError, tr("The local file was removed during sync.")); - return; - } else { - _propagator->_anotherSyncNeeded = true; - } - } - - // Check whether the file changed since discovery. - if (! FileSystem::verifyFileUnchanged(fullFilePath, _item->_size, _item->_modtime)) { - _propagator->_anotherSyncNeeded = true; - if( !finished ) { - abortWithError(SyncFileItem::SoftError, tr("Local file changed during sync.")); - // FIXME: the legacy code was retrying for a few seconds. - // and also checking that after the last chunk, and removed the file in case of INSTRUCTION_NEW - return; - } - } - - if (!finished) { - // Proceed to next chunk. - if (_currentChunk >= _chunkCount) { - if (!_jobs.empty()) { - // just wait for the other job to finish. - return; - } - _finished = true; - done(SyncFileItem::NormalError, tr("The server did not acknowledge the last chunk. (No e-tag was present)")); - return; - } - - // Deletes an existing blacklist entry on successful chunk upload - if (_item->_hasBlacklistEntry) { - _propagator->_journal->wipeErrorBlacklistEntry(_item->_file); - _item->_hasBlacklistEntry = false; - } - - SyncJournalDb::UploadInfo pi; - pi._valid = true; - auto currentChunk = job->_chunk; - foreach (auto *job, _jobs) { - // Take the minimum finished one - if (auto putJob = qobject_cast<PUTFileJob*>(job)) { - currentChunk = qMin(currentChunk, putJob->_chunk - 1); - } - } - pi._chunk = (currentChunk + _startChunk + 1) % _chunkCount ; // next chunk to start with - pi._transferid = _transferId; - pi._modtime = Utility::qDateTimeFromTime_t(_item->_modtime); - _propagator->_journal->setUploadInfo(_item->_file, pi); - _propagator->_journal->commit("Upload info"); - startNextChunk(); - return; - } - - // the following code only happens after all chunks were uploaded. - _finished = true; - // the file id should only be empty for new files up- or downloaded - QByteArray fid = job->reply()->rawHeader("OC-FileID"); - if( !fid.isEmpty() ) { - if( !_item->_fileId.isEmpty() && _item->_fileId != fid ) { - qDebug() << "WARN: File ID changed!" << _item->_fileId << fid; - } - _item->_fileId = fid; - } - - _item->_etag = etag; - - _item->_responseTimeStamp = job->responseTimestamp(); - - if (job->reply()->rawHeader("X-OC-MTime") != "accepted") { - // X-OC-MTime is supported since owncloud 5.0. But not when chunking. - // Normally Owncloud 6 always puts X-OC-MTime - qWarning() << "Server does not support X-OC-MTime" << job->reply()->rawHeader("X-OC-MTime"); - // Well, the mtime was not set - done(SyncFileItem::SoftError, "Server does not support X-OC-MTime"); - } - - // performance logging - _item->_requestDuration = _stopWatch.stop(); - qDebug() << "*==* duration UPLOAD" << _item->_size - << _stopWatch.durationOfLap(QLatin1String("ContentChecksum")) - << _stopWatch.durationOfLap(QLatin1String("TransmissionChecksum")) - << _item->_requestDuration; - // The job might stay alive for the whole sync, release this tiny bit of memory. - _stopWatch.reset(); - - finalize(*_item); -} - -void PropagateUploadFile::finalize(const SyncFileItem ©) -{ - // Normally, copy == _item, but when it comes from the UpdateMTimeAndETagJob, we need to do - // some updates - _item->_etag = copy._etag; - _item->_fileId = copy._fileId; - - _item->_requestDuration = _duration.elapsed(); - - _finished = true; - - if (!_propagator->_journal->setFileRecord(SyncJournalFileRecord(*_item, _propagator->getFilePath(_item->_file)))) { - done(SyncFileItem::FatalError, tr("Error writing metadata to the database")); - return; - } - // Remove from the progress database: - _propagator->_journal->setUploadInfo(_item->_file, SyncJournalDb::UploadInfo()); - _propagator->_journal->commit("upload file start"); - - done(SyncFileItem::Success); -} - -void PropagateUploadFile::slotUploadProgress(qint64 sent, qint64 total) -{ - // Completion is signaled with sent=0, total=0; avoid accidentally - // resetting progress due to the sent being zero by ignoring it. - // finishedSignal() is bound to be emitted soon anyway. - // See https://bugreports.qt.io/browse/QTBUG-44782. - if (sent == 0 && total == 0) { - return; - } - - int progressChunk = _currentChunk + _startChunk - 1; - if (progressChunk >= _chunkCount) - progressChunk = _currentChunk - 1; - - // amount is the number of bytes already sent by all the other chunks that were sent - // not including this one. - // FIXME: this assumes all chunks have the same size, which is true only if the last chunk - // has not been finished (which should not happen because the last chunk is sent sequentially) - quint64 amount = progressChunk * chunkSize(); - - sender()->setProperty("byteWritten", sent); - if (_jobs.count() > 1) { - amount -= (_jobs.count() -1) * chunkSize(); - foreach (QObject *j, _jobs) { - amount += j->property("byteWritten").toULongLong(); - } - } else { - // sender() is the only current job, no need to look at the byteWritten properties - amount += sent; - } - emit progress(*_item, amount); -} - -void PropagateUploadFile::startPollJob(const QString& path) +void PropagateUploadFileCommon::startPollJob(const QString& path) { PollJob* job = new PollJob(_propagator->account(), path, _item, _propagator->_journal, _propagator->_localDir, this); @@ -855,7 +477,7 @@ void PropagateUploadFile::startPollJob(const QString& path) job->start(); } -void PropagateUploadFile::slotPollFinished() +void PropagateUploadFileCommon::slotPollFinished() { PollJob *job = qobject_cast<PollJob *>(sender()); Q_ASSERT(job); @@ -868,15 +490,15 @@ void PropagateUploadFile::slotPollFinished() return; } - finalize(*job->_item); + finalize(); } -void PropagateUploadFile::slotJobDestroyed(QObject* job) +void PropagateUploadFileCommon::slotJobDestroyed(QObject* job) { _jobs.erase(std::remove(_jobs.begin(), _jobs.end(), job) , _jobs.end()); } -void PropagateUploadFile::abort() +void PropagateUploadFileCommon::abort() { foreach(auto *job, _jobs) { if (job->reply()) { @@ -887,12 +509,58 @@ void PropagateUploadFile::abort() } // This function is used whenever there is an error occuring and jobs might be in progress -void PropagateUploadFile::abortWithError(SyncFileItem::Status status, const QString &error) +void PropagateUploadFileCommon::abortWithError(SyncFileItem::Status status, const QString &error) { _finished = true; abort(); done(status, error); } +QMap<QByteArray, QByteArray> PropagateUploadFileCommon::headers() +{ + QMap<QByteArray, QByteArray> headers; + headers["OC-Async"] = "1"; + headers["Content-Type"] = "application/octet-stream"; + headers["X-OC-Mtime"] = QByteArray::number(qint64(_item->_modtime)); + + if(_item->_file.contains(".sys.admin#recall#")) { + // This is a file recall triggered by the admin. Note: the + // recall list file created by the admin and downloaded by the + // client (.sys.admin#recall#) also falls into this category + // (albeit users are not supposed to mess up with it) + + // We use a special tag header so that the server may decide to store this file away in some admin stage area + // And not directly in the user's area (which would trigger redownloads etc). + headers["OC-Tag"] = ".sys.admin#recall#"; + } + + if (!_item->_etag.isEmpty() && _item->_etag != "empty_etag" + && _item->_instruction != CSYNC_INSTRUCTION_NEW // On new files never send a If-Match + && _item->_instruction != CSYNC_INSTRUCTION_TYPE_CHANGE + && !_deleteExisting + ) { + // We add quotes because the owncloud server always adds quotes around the etag, and + // csync_owncloud.c's owncloud_file_id always strips the quotes. + headers["If-Match"] = '"' + _item->_etag + '"'; + } + return headers; +} + +void PropagateUploadFileCommon::finalize() +{ + _item->_requestDuration = _duration.elapsed(); + _finished = true; + + if (!_propagator->_journal->setFileRecord(SyncJournalFileRecord(*_item, _propagator->getFilePath(_item->_file)))) { + done(SyncFileItem::FatalError, tr("Error writing metadata to the database")); + return; + } + // Remove from the progress database: + _propagator->_journal->setUploadInfo(_item->_file, SyncJournalDb::UploadInfo()); + _propagator->_journal->commit("upload file start"); + + done(SyncFileItem::Success); +} + } diff --git a/src/libsync/propagateupload.h b/src/libsync/propagateupload.h index f3375a11c..bac3e112b 100644 --- a/src/libsync/propagateupload.h +++ b/src/libsync/propagateupload.h @@ -89,12 +89,17 @@ private: QScopedPointer<QIODevice> _device; QMap<QByteArray, QByteArray> _headers; QString _errorString; + QUrl _url; public: // Takes ownership of the device explicit PUTFileJob(AccountPtr account, const QString& path, QIODevice *device, const QMap<QByteArray, QByteArray> &headers, int chunk, QObject* parent = 0) : AbstractNetworkJob(account, path, parent), _device(device), _headers(headers), _chunk(chunk) {} + explicit PUTFileJob(AccountPtr account, const QUrl& url, QIODevice *device, + const QMap<QByteArray, QByteArray> &headers, int chunk, QObject* parent = 0) + : AbstractNetworkJob(account, QString(), parent), _device(device), _headers(headers) + , _url(url), _chunk(chunk) {} ~PUTFileJob(); int _chunk; @@ -155,10 +160,90 @@ signals: }; /** - * @brief The PropagateUploadFile class + * @brief The PropagateUploadFileCommon class is the code common between all chunking algorithms * @ingroup libsync + * + * State Machine: + * + * +---> start() --> (delete job) -------+ + * | | + * +--> slotComputeContentChecksum() <---+ + * | + * v + * slotComputeTransmissionChecksum() + * | + * v + * slotStartUpload() -> doStartUpload() + * . + * . + * v + * finalize() or abortWithError() or startPollJob() */ -class PropagateUploadFile : public PropagateItemJob { +class PropagateUploadFileCommon : public PropagateItemJob { + Q_OBJECT + +protected: + QElapsedTimer _duration; + QVector<AbstractNetworkJob*> _jobs; /// network jobs that are currently in transit + bool _finished; /// Tells that all the jobs have been finished + bool _deleteExisting; + + // measure the performance of checksum calc and upload + Utility::StopWatch _stopWatch; + + QByteArray _transmissionChecksum; + QByteArray _transmissionChecksumType; + + +public: + PropagateUploadFileCommon(OwncloudPropagator* propagator,const SyncFileItemPtr& item) + : PropagateItemJob(propagator, item), _finished(false), _deleteExisting(false) {} + + /** + * Whether an existing entity with the same name may be deleted before + * the upload. + * + * Default: false. + */ + void setDeleteExisting(bool enabled); + + void start() Q_DECL_OVERRIDE; + + bool isLikelyFinishedQuickly() Q_DECL_OVERRIDE { return _item->_size < 100*1024; } + +private slots: + void slotComputeContentChecksum(); + // Content checksum computed, compute the transmission checksum + void slotComputeTransmissionChecksum(const QByteArray& contentChecksumType, const QByteArray& contentChecksum); + // transmission checksum computed, prepare the upload + void slotStartUpload(const QByteArray& transmissionChecksumType, const QByteArray& transmissionChecksum); +public: + virtual void doStartUpload() = 0; + + void startPollJob(const QString& path); + void finalize(); + void abortWithError(SyncFileItem::Status status, const QString &error); + +public slots: + void abort() Q_DECL_OVERRIDE; + void slotJobDestroyed(QObject *job); + +private slots: + void slotPollFinished(); + +protected: + // Bases headers that need to be sent with every chunk + QMap<QByteArray, QByteArray> headers(); + +}; + +/** + * @ingroup libsync + * + * Propagation job, impementing the old chunking agorithm + * + */ +class PropagateUploadFileV1 : public PropagateUploadFileCommon { Q_OBJECT private: @@ -176,51 +261,66 @@ private: int _currentChunk; int _chunkCount; /// Total number of chunks for this file int _transferId; /// transfer id (part of the url) - QElapsedTimer _duration; - QVector<AbstractNetworkJob*> _jobs; /// network jobs that are currently in transit - bool _finished; // Tells that all the jobs have been finished - // measure the performance of checksum calc and upload - Utility::StopWatch _stopWatch; + quint64 chunkSize() const { return _propagator->chunkSize(); } - QByteArray _transmissionChecksum; - QByteArray _transmissionChecksumType; - bool _deleteExisting; +public: + PropagateUploadFileV1(OwncloudPropagator* propagator,const SyncFileItemPtr& item) : + PropagateUploadFileCommon(propagator,item) {} - quint64 chunkSize() const { return _propagator->chunkSize(); } + void doStartUpload() Q_DECL_OVERRIDE; -public: - PropagateUploadFile(OwncloudPropagator* propagator,const SyncFileItemPtr& item) - : PropagateItemJob(propagator, item), _startChunk(0), _currentChunk(0), _chunkCount(0), _transferId(0), _finished(false), _deleteExisting(false) {} - void start() Q_DECL_OVERRIDE; +private slots: + void startNextChunk(); + void slotPutFinished(); + void slotUploadProgress(qint64,qint64); +}; - bool isLikelyFinishedQuickly() Q_DECL_OVERRIDE { return _item->_size < 100*1024; } +/** + * @ingroup libsync + * + * Propagation job, impementing the new chunking agorithm + * + */ +class PropagateUploadFileNG : public PropagateUploadFileCommon { + Q_OBJECT +private: + quint64 _sent; /// amount of data (bytes) that was already sent + uint _transferId; /// transfer id (part of the url) + int _currentChunk; /// Id of the next chunk that will be sent + bool _removeJobError; /// If not null, there was an error removing the job + // Map chunk number with its size from the PROPFIND on resume. + // (Only used from slotPropfindIterate/slotPropfindFinished because the LsColJob use signals to report data.) + QMap<int, quint64> _serverChunks; + + quint64 chunkSize() const { return _propagator->chunkSize(); } /** - * Whether an existing entity with the same name may be deleted before - * the upload. - * - * Default: false. + * Return the URL of a chunk. + * If chunk == -1, returns the URL of the parent folder containing the chunks */ - void setDeleteExisting(bool enabled); + QUrl chunkUrl(int chunk = -1); + +public: + PropagateUploadFileNG(OwncloudPropagator* propagator,const SyncFileItemPtr& item) : + PropagateUploadFileCommon(propagator,item) {} + void doStartUpload() Q_DECL_OVERRIDE; +private: + void startNewUpload(); + void startNextChunk(); private slots: + void slotPropfindFinished(); + void slotPropfindFinishedWithError(); + void slotPropfindIterate(const QString &name, const QMap<QString,QString> &properties); + void slotDeleteJobFinished(); + void slotMkColFinished(QNetworkReply::NetworkError); void slotPutFinished(); - void slotPollFinished(); + void slotMoveJobFinished(); void slotUploadProgress(qint64,qint64); - void abort() Q_DECL_OVERRIDE; - void startNextChunk(); - void finalize(const SyncFileItem&); - void slotJobDestroyed(QObject *job); - void slotStartUpload(const QByteArray& transmissionChecksumType, const QByteArray& transmissionChecksum); - void slotComputeTransmissionChecksum(const QByteArray& contentChecksumType, const QByteArray& contentChecksum); - void slotComputeContentChecksum(); - -private: - void startPollJob(const QString& path); - void abortWithError(SyncFileItem::Status status, const QString &error); }; + } diff --git a/src/libsync/propagateuploadng.cpp b/src/libsync/propagateuploadng.cpp new file mode 100644 index 000000000..c96164c4b --- /dev/null +++ b/src/libsync/propagateuploadng.cpp @@ -0,0 +1,488 @@ +/* + * Copyright (C) by Olivier Goffart <ogoffart@owncloud.com> + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#include "config.h" +#include "propagateupload.h" +#include "owncloudpropagator_p.h" +#include "networkjobs.h" +#include "account.h" +#include "syncjournaldb.h" +#include "syncjournalfilerecord.h" +#include "utility.h" +#include "filesystem.h" +#include "propagatorjobs.h" +#include "syncengine.h" +#include "propagateremotemove.h" +#include "propagateremotedelete.h" + + +#include <QNetworkAccessManager> +#include <QFileInfo> +#include <QDir> +#include <cmath> +#include <cstring> + +namespace OCC { + +QUrl PropagateUploadFileNG::chunkUrl(int chunk) +{ + QString path = QLatin1String("remote.php/dav/uploads/") + + _propagator->account()->user() + + QLatin1Char('/') + QString::number(_transferId); + if (chunk >= 0) { + path += QLatin1Char('/') + QString::number(chunk); + } + return Utility::concatUrlPath(_propagator->account()->url(), path); +} + +/* + State machine: + + *----> doStartUpload() + Check the db: is there an entry? + / \ + no yes + / \ + / PROPFIND + startNewUpload() <-+ +----------------------------\ + | | | \ + MKCOL + slotPropfindFinishedWithError() slotPropfindFinished() + | Is there stale files to remove? + slotMkColFinished() | | + | no yes + | | | + | | DeleteJob + | | | + +-----+<------------------------------------------------------+<--- slotDeleteJobFinished() + | + +----> startNextChunk() ---finished? --+ + ^ | | + +---------------+ | + | + +----------------------------------------+ + | + +-> MOVE ------> moveJobFinished() ---> finalize() + + + */ + +void PropagateUploadFileNG::doStartUpload() +{ + _duration.start(); + _propagator->_activeJobList.append(this); + + const SyncJournalDb::UploadInfo progressInfo = _propagator->_journal->getUploadInfo(_item->_file); + if (progressInfo._valid && Utility::qDateTimeToTime_t(progressInfo._modtime) == _item->_modtime ) { + _transferId = progressInfo._transferid; + auto url = chunkUrl(); + auto job = new LsColJob(_propagator->account(), url, this); + _jobs.append(job); + job->setProperties(QList<QByteArray>() << "resourcetype" << "getcontentlength"); + connect(job, SIGNAL(finishedWithoutError()), this, SLOT(slotPropfindFinished())); + connect(job, SIGNAL(finishedWithError(QNetworkReply*)), + this, SLOT(slotPropfindFinishedWithError())); + connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); + connect(job, SIGNAL(directoryListingIterated(QString,QMap<QString,QString>)), + this, SLOT(slotPropfindIterate(QString,QMap<QString,QString>))); + job->start(); + return; + } + + startNewUpload(); +} + +void PropagateUploadFileNG::slotPropfindIterate(const QString &name, const QMap<QString,QString> &properties) +{ + if (name == chunkUrl().path()) { + return; // skip the info about the path itself + } + bool ok = false; + auto chunkId = name.mid(name.lastIndexOf('/')+1).toUInt(&ok); + if (ok) { + _serverChunks[chunkId] = properties["getcontentlength"].toULongLong(); + } +} + +void PropagateUploadFileNG::slotPropfindFinished() +{ + auto job = qobject_cast<LsColJob *>(sender()); + slotJobDestroyed(job); // remove it from the _jobs list + _propagator->_activeJobList.removeOne(this); + + _currentChunk = 0; + _sent = 0; + while (_serverChunks.contains(_currentChunk)) { + _sent += _serverChunks[_currentChunk]; + _serverChunks.remove(_currentChunk); + ++_currentChunk; + } + + if (_sent > _item->_size) { + // Normally this can't happen because the size is xor'ed with the transfer id, and it is + // therefore impossible that there is more data on the server than on the file. + qWarning() << "Inconsistency while resuming " << _item->_file + << ": the size on the server (" << _sent << ") is bigger than the size of the file (" + << _item->_size << ")"; + startNewUpload(); + return; + } + + qDebug() << "Resuming "<< _item->_file << " from chunk " << _currentChunk << "; sent ="<< _sent; + + if (!_serverChunks.isEmpty()) { + qDebug() << "To Delete" << _serverChunks; + _propagator->_activeJobList.append(this); + _removeJobError = false; + + // Make sure that if there is a "hole" and then a few more chunks, on the server + // we should remove the later chunks. Otherwise when we do dynamic chunk sizing, we may end up + // with corruptions if there are too many chunks, or if we abort and there are still stale chunks. + for (auto it = _serverChunks.begin(); it != _serverChunks.end(); ++it) { + auto job = new DeleteJob(_propagator->account(), Utility::concatUrlPath(chunkUrl(), QString::number(it.key())), this); + QObject::connect(job, SIGNAL(finishedSignal()), this, SLOT(slotDeleteJobFinished())); + _jobs.append(job); + job->start(); + } + return; + } + + startNextChunk(); +} + +void PropagateUploadFileNG::slotPropfindFinishedWithError() +{ + auto job = qobject_cast<LsColJob *>(sender()); + slotJobDestroyed(job); // remove it from the _jobs list + QNetworkReply::NetworkError err = job->reply()->error(); + auto httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + auto status = classifyError(err, httpErrorCode, &_propagator->_anotherSyncNeeded); + if (status == SyncFileItem::FatalError) { + _propagator->_activeJobList.removeOne(this); + QString errorString = errorMessage(job->reply()->errorString(), job->reply()->readAll()); + abortWithError(status, errorString); + return; + } + startNewUpload(); +} + +void PropagateUploadFileNG::slotDeleteJobFinished() +{ + auto job = qobject_cast<DeleteJob *>(sender()); + Q_ASSERT(job); + _jobs.remove(_jobs.indexOf(job)); + + QNetworkReply::NetworkError err = job->reply()->error(); + if (err != QNetworkReply::NoError && err != QNetworkReply::ContentNotFoundError) { + const int httpStatus = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + SyncFileItem::Status status = classifyError(err, httpStatus); + if (status == SyncFileItem::FatalError) { + abortWithError(status, job->errorString()); + return; + } else { + qWarning() << "DeleteJob errored out" << job->errorString() << job->reply()->url(); + _removeJobError = true; + // Let the other jobs finish + } + } + + if (_jobs.isEmpty()) { + _propagator->_activeJobList.removeOne(this); + if (_removeJobError) { + // There was an error removing some files, just start over + startNewUpload(); + } else { + startNextChunk(); + } + } +} + + + +void PropagateUploadFileNG::startNewUpload() +{ + Q_ASSERT(_propagator->_activeJobList.count(this) == 1); + _transferId = qrand() ^ _item->_modtime ^ (_item->_size << 16) ^ qHash(_item->_file); + _sent = 0; + _currentChunk = 0; + + emit progress(*_item, 0); + + SyncJournalDb::UploadInfo pi; + pi._valid = true; + pi._transferid = _transferId; + pi._modtime = Utility::qDateTimeFromTime_t(_item->_modtime); + _propagator->_journal->setUploadInfo(_item->_file, pi); + _propagator->_journal->commit("Upload info"); + QMap<QByteArray, QByteArray> headers; + headers["OC-Total-Length"] = QByteArray::number(_item->_size); + auto job = new MkColJob(_propagator->account(), chunkUrl(), headers, this); + + connect(job, SIGNAL(finished(QNetworkReply::NetworkError)), + this, SLOT(slotMkColFinished(QNetworkReply::NetworkError))); + connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); + job->start(); +} + +void PropagateUploadFileNG::slotMkColFinished(QNetworkReply::NetworkError) +{ + _propagator->_activeJobList.removeOne(this); + auto job = qobject_cast<MkColJob *>(sender()); + slotJobDestroyed(job); // remove it from the _jobs list + QNetworkReply::NetworkError err = job->reply()->error(); + _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + + if (err != QNetworkReply::NoError || _item->_httpErrorCode != 201) { + SyncFileItem::Status status = classifyError(err, _item->_httpErrorCode, + &_propagator->_anotherSyncNeeded); + QString errorString = errorMessage(job->reply()->errorString(), job->reply()->readAll()); + if (job->reply()->hasRawHeader("OC-ErrorString")) { + errorString = job->reply()->rawHeader("OC-ErrorString"); + } + abortWithError(status, errorString); + return; + } + startNextChunk(); +} + +void PropagateUploadFileNG::startNextChunk() +{ + if (_propagator->_abortRequested.fetchAndAddRelaxed(0)) + return; + + quint64 fileSize = _item->_size; + Q_ASSERT(fileSize >= _sent); + quint64 currentChunkSize = qMin(chunkSize(), fileSize - _sent); + + if (currentChunkSize == 0) { + Q_ASSERT(_jobs.isEmpty()); // There should be no running job anymore + _finished = true; + // Finish with a MOVE + QString destination = _propagator->account()->url().path() + + QLatin1String("/remote.php/dav/files/") + _propagator->account()->user() + + _propagator->_remoteFolder + _item->_file; + auto headers = PropagateUploadFileCommon::headers(); + + // "If-Match applies to the source, but we are interested in comparing the etag of the destination + auto ifMatch = headers.take("If-Match"); + if (!ifMatch.isEmpty()) { + headers["If"] = "<" + destination.toUtf8() + "> ([" + ifMatch + "])"; + } + if (!_transmissionChecksumType.isEmpty()) { + headers[checkSumHeaderC] = makeChecksumHeader( + _transmissionChecksumType, _transmissionChecksum); + } + + auto job = new MoveJob(_propagator->account(), Utility::concatUrlPath(chunkUrl(), "/.file"), + destination, headers, this); + _jobs.append(job); + connect(job, SIGNAL(finishedSignal()), this, SLOT(slotMoveJobFinished())); + connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); + _propagator->_activeJobList.append(this); + job->start(); + return; + } + + auto device = new UploadDevice(&_propagator->_bandwidthManager); + const QString fileName = _propagator->getFilePath(_item->_file); + + if (! device->prepareAndOpen(fileName, _sent, currentChunkSize)) { + qDebug() << "ERR: Could not prepare upload device: " << device->errorString(); + + // If the file is currently locked, we want to retry the sync + // when it becomes available again. + if (FileSystem::isFileLocked(fileName)) { + emit _propagator->seenLockedFile(fileName); + } + // Soft error because this is likely caused by the user modifying his files while syncing + abortWithError( SyncFileItem::SoftError, device->errorString() ); + return; + } + + QMap<QByteArray, QByteArray> headers; + headers["OC-Chunk-Offset"] = QByteArray::number(_sent); + + _sent += currentChunkSize; + QUrl url = chunkUrl(_currentChunk); + + // job takes ownership of device via a QScopedPointer. Job deletes itself when finishing + PUTFileJob* job = new PUTFileJob(_propagator->account(), url, device, headers, _currentChunk); + _jobs.append(job); + connect(job, SIGNAL(finishedSignal()), this, SLOT(slotPutFinished())); + connect(job, SIGNAL(uploadProgress(qint64,qint64)), + this, SLOT(slotUploadProgress(qint64,qint64))); + connect(job, SIGNAL(uploadProgress(qint64,qint64)), + device, SLOT(slotJobUploadProgress(qint64,qint64))); + connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); + job->start(); + _propagator->_activeJobList.append(this); + _currentChunk++; + + // FIXME! parallel chunk? + +} + +void PropagateUploadFileNG::slotPutFinished() +{ + PUTFileJob *job = qobject_cast<PUTFileJob *>(sender()); + Q_ASSERT(job); + slotJobDestroyed(job); // remove it from the _jobs list + + qDebug() << job->reply()->request().url() << "FINISHED WITH STATUS" + << job->reply()->error() + << (job->reply()->error() == QNetworkReply::NoError ? QLatin1String("") : job->reply()->errorString()) + << job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute) + << job->reply()->attribute(QNetworkRequest::HttpReasonPhraseAttribute); + + _propagator->_activeJobList.removeOne(this); + + if (_finished) { + // We have sent the finished signal already. We don't need to handle any remaining jobs + return; + } + + QNetworkReply::NetworkError err = job->reply()->error(); + +#if QT_VERSION < QT_VERSION_CHECK(5, 4, 2) + if (err == QNetworkReply::OperationCanceledError && job->reply()->property("owncloud-should-soft-cancel").isValid()) { + // Abort the job and try again later. + // This works around a bug in QNAM wich might reuse a non-empty buffer for the next request. + qDebug() << "Forcing job abort on HTTP connection reset with Qt < 5.4.2."; + _propagator->_anotherSyncNeeded = true; + abortWithError(SyncFileItem::SoftError, tr("Forcing job abort on HTTP connection reset with Qt < 5.4.2.")); + return; + } +#endif + + if (err != QNetworkReply::NoError) { + _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + QByteArray replyContent = job->reply()->readAll(); + qDebug() << replyContent; // display the XML error in the debug + QString errorString = errorMessage(job->errorString(), replyContent); + + if (job->reply()->hasRawHeader("OC-ErrorString")) { + errorString = job->reply()->rawHeader("OC-ErrorString"); + } + + // FIXME! can this happen for the chunks? + if (_item->_httpErrorCode == 412) { + // Precondition Failed: Maybe the bad etag is in the database, we need to clear the + // parent folder etag so we won't read from DB next sync. + _propagator->_journal->avoidReadFromDbOnNextSync(_item->_file); + _propagator->_anotherSyncNeeded = true; + } + + SyncFileItem::Status status = classifyError(err, _item->_httpErrorCode, + &_propagator->_anotherSyncNeeded); + abortWithError(status, errorString); + return; + } + + Q_ASSERT(_sent <= _item->_size); + bool finished = _sent == _item->_size; + + // Check if the file still exists + const QString fullFilePath(_propagator->getFilePath(_item->_file)); + if( !FileSystem::fileExists(fullFilePath) ) { + if (!finished) { + abortWithError(SyncFileItem::SoftError, tr("The local file was removed during sync.")); + return; + } else { + _propagator->_anotherSyncNeeded = true; + } + } + + // Check whether the file changed since discovery. + if (! FileSystem::verifyFileUnchanged(fullFilePath, _item->_size, _item->_modtime)) { + _propagator->_anotherSyncNeeded = true; + if( !finished ) { + abortWithError(SyncFileItem::SoftError, tr("Local file changed during sync.")); + return; + } + } + + if (!finished) { + // Deletes an existing blacklist entry on successful chunk upload + if (_item->_hasBlacklistEntry) { + _propagator->_journal->wipeErrorBlacklistEntry(_item->_file); + _item->_hasBlacklistEntry = false; + } + } + startNextChunk(); +} + +void PropagateUploadFileNG::slotMoveJobFinished() +{ + _propagator->_activeJobList.removeOne(this); + auto job = qobject_cast<MoveJob *>(sender()); + slotJobDestroyed(job); // remove it from the _jobs list + QNetworkReply::NetworkError err = job->reply()->error(); + _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + + if (err != QNetworkReply::NoError) { + SyncFileItem::Status status = classifyError(err, _item->_httpErrorCode, + &_propagator->_anotherSyncNeeded); + QString errorString = errorMessage(job->errorString(), job->reply()->readAll()); + abortWithError(status, errorString); + return; + } + if (_item->_httpErrorCode != 201 && _item->_httpErrorCode != 204) { + abortWithError(SyncFileItem::NormalError, tr("Unexpected return code from server (%1)").arg(_item->_httpErrorCode)); + return; + } + + QByteArray fid = job->reply()->rawHeader("OC-FileID"); + if(fid.isEmpty()) { + qWarning() << "Server did not return a OC-FileID" << _item->_file; + abortWithError(SyncFileItem::NormalError, tr("Missing File ID from server")); + return; + } else { + // the old file id should only be empty for new files uploaded + if( !_item->_fileId.isEmpty() && _item->_fileId != fid ) { + qDebug() << "WARN: File ID changed!" << _item->_fileId << fid; + } + _item->_fileId = fid; + } + + _item->_etag = getEtagFromReply(job->reply());; + if (_item->_etag.isEmpty()) { + qWarning() << "Server did not return an ETAG" << _item->_file; + abortWithError(SyncFileItem::NormalError, tr("Missing ETag from server")); + return; + } + _item->_responseTimeStamp = job->responseTimestamp(); + + // performance logging + _item->_requestDuration = _stopWatch.stop(); + qDebug() << "*==* duration UPLOAD" << _item->_size + << _stopWatch.durationOfLap(QLatin1String("ContentChecksum")) + << _stopWatch.durationOfLap(QLatin1String("TransmissionChecksum")) + << _item->_requestDuration; + // The job might stay alive for the whole sync, release this tiny bit of memory. + _stopWatch.reset(); + finalize(); +} + +void PropagateUploadFileNG::slotUploadProgress(qint64 sent, qint64 total) +{ + // Completion is signaled with sent=0, total=0; avoid accidentally + // resetting progress due to the sent being zero by ignoring it. + // finishedSignal() is bound to be emitted soon anyway. + // See https://bugreports.qt.io/browse/QTBUG-44782. + if (sent == 0 && total == 0) { + return; + } + emit progress(*_item, _sent + sent - total); +} + +} diff --git a/src/libsync/propagateuploadv1.cpp b/src/libsync/propagateuploadv1.cpp new file mode 100644 index 000000000..dd075fa77 --- /dev/null +++ b/src/libsync/propagateuploadv1.cpp @@ -0,0 +1,375 @@ +/* + * Copyright (C) by Olivier Goffart <ogoffart@owncloud.com> + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#include "config.h" +#include "propagateupload.h" +#include "owncloudpropagator_p.h" +#include "networkjobs.h" +#include "account.h" +#include "syncjournaldb.h" +#include "syncjournalfilerecord.h" +#include "utility.h" +#include "filesystem.h" +#include "propagatorjobs.h" +#include "checksums.h" +#include "syncengine.h" +#include "propagateremotedelete.h" + +#include <json.h> +#include <QNetworkAccessManager> +#include <QFileInfo> +#include <QDir> +#include <cmath> +#include <cstring> + +namespace OCC { +void PropagateUploadFileV1::doStartUpload() +{ + _chunkCount = std::ceil(_item->_size / double(chunkSize())); + _startChunk = 0; + _transferId = qrand() ^ _item->_modtime ^ (_item->_size << 16); + + const SyncJournalDb::UploadInfo progressInfo = _propagator->_journal->getUploadInfo(_item->_file); + + if (progressInfo._valid && Utility::qDateTimeToTime_t(progressInfo._modtime) == _item->_modtime ) { + _startChunk = progressInfo._chunk; + _transferId = progressInfo._transferid; + qDebug() << Q_FUNC_INFO << _item->_file << ": Resuming from chunk " << _startChunk; + } + + _currentChunk = 0; + _duration.start(); + + emit progress(*_item, 0); + startNextChunk(); +} + +void PropagateUploadFileV1::startNextChunk() +{ + if (_propagator->_abortRequested.fetchAndAddRelaxed(0)) + return; + + if (! _jobs.isEmpty() && _currentChunk + _startChunk >= _chunkCount - 1) { + // Don't do parallel upload of chunk if this might be the last chunk because the server cannot handle that + // https://github.com/owncloud/core/issues/11106 + // We return now and when the _jobs are finished we will proceed with the last chunk + // NOTE: Some other parts of the code such as slotUploadProgress also assume that the last chunk + // is sent last. + return; + } + quint64 fileSize = _item->_size; + auto headers = PropagateUploadFileCommon::headers(); + headers["OC-Total-Length"] = QByteArray::number(fileSize); + headers["OC-Chunk-Size"]= QByteArray::number(quint64(chunkSize())); + + QString path = _item->_file; + + UploadDevice *device = new UploadDevice(&_propagator->_bandwidthManager); + qint64 chunkStart = 0; + qint64 currentChunkSize = fileSize; + bool isFinalChunk = false; + if (_chunkCount > 1) { + int sendingChunk = (_currentChunk + _startChunk) % _chunkCount; + // XOR with chunk size to make sure everything goes well if chunk size changes between runs + uint transid = _transferId ^ chunkSize(); + qDebug() << "Upload chunk" << sendingChunk << "of" << _chunkCount << "transferid(remote)=" << transid; + path += QString("-chunking-%1-%2-%3").arg(transid).arg(_chunkCount).arg(sendingChunk); + + headers["OC-Chunked"] = "1"; + + chunkStart = chunkSize() * quint64(sendingChunk); + currentChunkSize = chunkSize(); + if (sendingChunk == _chunkCount - 1) { // last chunk + currentChunkSize = (fileSize % chunkSize()); + if( currentChunkSize == 0 ) { // if the last chunk pretends to be 0, its actually the full chunk size. + currentChunkSize = chunkSize(); + } + isFinalChunk = true; + } + } else { + // if there's only one chunk, it's the final one + isFinalChunk = true; + } + + if (isFinalChunk && !_transmissionChecksumType.isEmpty()) { + headers[checkSumHeaderC] = makeChecksumHeader( + _transmissionChecksumType, _transmissionChecksum); + } + + const QString fileName = _propagator->getFilePath(_item->_file); + if (! device->prepareAndOpen(fileName, chunkStart, currentChunkSize)) { + qDebug() << "ERR: Could not prepare upload device: " << device->errorString(); + + // If the file is currently locked, we want to retry the sync + // when it becomes available again. + if (FileSystem::isFileLocked(fileName)) { + emit _propagator->seenLockedFile(fileName); + } + // Soft error because this is likely caused by the user modifying his files while syncing + abortWithError( SyncFileItem::SoftError, device->errorString() ); + delete device; + return; + } + + // job takes ownership of device via a QScopedPointer. Job deletes itself when finishing + PUTFileJob* job = new PUTFileJob(_propagator->account(), _propagator->_remoteFolder + path, device, headers, _currentChunk); + _jobs.append(job); + connect(job, SIGNAL(finishedSignal()), this, SLOT(slotPutFinished())); + connect(job, SIGNAL(uploadProgress(qint64,qint64)), this, SLOT(slotUploadProgress(qint64,qint64))); + connect(job, SIGNAL(uploadProgress(qint64,qint64)), device, SLOT(slotJobUploadProgress(qint64,qint64))); + connect(job, SIGNAL(destroyed(QObject*)), this, SLOT(slotJobDestroyed(QObject*))); + job->start(); + _propagator->_activeJobList.append(this); + _currentChunk++; + + bool parallelChunkUpload = true; + QByteArray env = qgetenv("OWNCLOUD_PARALLEL_CHUNK"); + if (!env.isEmpty()) { + parallelChunkUpload = env != "false" && env != "0"; + } else { + int versionNum = _propagator->account()->serverVersionInt(); + if (versionNum < 0x080003) { + // Disable parallel chunk upload severs older than 8.0.3 to avoid too many + // internal sever errors (#2743, #2938) + parallelChunkUpload = false; + } + } + + if (_currentChunk + _startChunk >= _chunkCount - 1) { + // Don't do parallel upload of chunk if this might be the last chunk because the server cannot handle that + // https://github.com/owncloud/core/issues/11106 + parallelChunkUpload = false; + } + + if (parallelChunkUpload && (_propagator->_activeJobList.count() < _propagator->maximumActiveJob()) + && _currentChunk < _chunkCount ) { + startNextChunk(); + } + if (!parallelChunkUpload || _chunkCount - _currentChunk <= 0) { + emit ready(); + } +} + +void PropagateUploadFileV1::slotPutFinished() +{ + PUTFileJob *job = qobject_cast<PUTFileJob *>(sender()); + Q_ASSERT(job); + slotJobDestroyed(job); // remove it from the _jobs list + + qDebug() << Q_FUNC_INFO << job->reply()->request().url() << "FINISHED WITH STATUS" + << job->reply()->error() + << (job->reply()->error() == QNetworkReply::NoError ? QLatin1String("") : job->reply()->errorString()) + << job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute) + << job->reply()->attribute(QNetworkRequest::HttpReasonPhraseAttribute); + + _propagator->_activeJobList.removeOne(this); + + if (_finished) { + // We have sent the finished signal already. We don't need to handle any remaining jobs + return; + } + + QNetworkReply::NetworkError err = job->reply()->error(); + +#if QT_VERSION < QT_VERSION_CHECK(5, 4, 2) + if (err == QNetworkReply::OperationCanceledError && job->reply()->property("owncloud-should-soft-cancel").isValid()) { // Abort the job and try again later. + // This works around a bug in QNAM wich might reuse a non-empty buffer for the next request. + qDebug() << "Forcing job abort on HTTP connection reset with Qt < 5.4.2."; + _propagator->_anotherSyncNeeded = true; + abortWithError(SyncFileItem::SoftError, tr("Forcing job abort on HTTP connection reset with Qt < 5.4.2.")); + return; + } +#endif + + if (err != QNetworkReply::NoError) { + _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + if(checkForProblemsWithShared(_item->_httpErrorCode, + tr("The file was edited locally but is part of a read only share. " + "It is restored and your edit is in the conflict file."))) { + return; + } + QByteArray replyContent = job->reply()->readAll(); + qDebug() << replyContent; // display the XML error in the debug + QString errorString = errorMessage(job->errorString(), replyContent); + + if (job->reply()->hasRawHeader("OC-ErrorString")) { + errorString = job->reply()->rawHeader("OC-ErrorString"); + } + + if (_item->_httpErrorCode == 412) { + // Precondition Failed: Maybe the bad etag is in the database, we need to clear the + // parent folder etag so we won't read from DB next sync. + _propagator->_journal->avoidReadFromDbOnNextSync(_item->_file); + _propagator->_anotherSyncNeeded = true; + } + + SyncFileItem::Status status = classifyError(err, _item->_httpErrorCode, + &_propagator->_anotherSyncNeeded); + abortWithError(status, errorString); + return; + } + + _item->_httpErrorCode = job->reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt(); + // The server needs some time to process the request and provide us with a poll URL + if (_item->_httpErrorCode == 202) { + _finished = true; + QString path = QString::fromUtf8(job->reply()->rawHeader("OC-Finish-Poll")); + if (path.isEmpty()) { + done(SyncFileItem::NormalError, tr("Poll URL missing")); + return; + } + startPollJob(path); + return; + } + + // 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, + // the etag has to be properly updated in the client journal, and because + // of that we can bail out here with an error. But we can reschedule a + // sync ASAP. + // But if the upload is ongoing, because not all chunks were uploaded + // yet, the upload can be stopped and an error can be displayed, because + // the server hasn't registered the new file yet. + QByteArray etag = getEtagFromReply(job->reply()); + bool finished = etag.length() > 0; + + // Check if the file still exists + const QString fullFilePath(_propagator->getFilePath(_item->_file)); + if( !FileSystem::fileExists(fullFilePath) ) { + if (!finished) { + abortWithError(SyncFileItem::SoftError, tr("The local file was removed during sync.")); + return; + } else { + _propagator->_anotherSyncNeeded = true; + } + } + + // Check whether the file changed since discovery. + if (! FileSystem::verifyFileUnchanged(fullFilePath, _item->_size, _item->_modtime)) { + _propagator->_anotherSyncNeeded = true; + if( !finished ) { + abortWithError(SyncFileItem::SoftError, tr("Local file changed during sync.")); + // FIXME: the legacy code was retrying for a few seconds. + // and also checking that after the last chunk, and removed the file in case of INSTRUCTION_NEW + return; + } + } + + if (!finished) { + // Proceed to next chunk. + if (_currentChunk >= _chunkCount) { + if (!_jobs.empty()) { + // just wait for the other job to finish. + return; + } + _finished = true; + done(SyncFileItem::NormalError, tr("The server did not acknowledge the last chunk. (No e-tag was present)")); + return; + } + + // Deletes an existing blacklist entry on successful chunk upload + if (_item->_hasBlacklistEntry) { + _propagator->_journal->wipeErrorBlacklistEntry(_item->_file); + _item->_hasBlacklistEntry = false; + } + + SyncJournalDb::UploadInfo pi; + pi._valid = true; + auto currentChunk = job->_chunk; + foreach (auto *job, _jobs) { + // Take the minimum finished one + if (auto putJob = qobject_cast<PUTFileJob*>(job)) { + currentChunk = qMin(currentChunk, putJob->_chunk - 1); + } + } + pi._chunk = (currentChunk + _startChunk + 1) % _chunkCount ; // next chunk to start with + pi._transferid = _transferId; + pi._modtime = Utility::qDateTimeFromTime_t(_item->_modtime); + _propagator->_journal->setUploadInfo(_item->_file, pi); + _propagator->_journal->commit("Upload info"); + startNextChunk(); + return; + } + + // the following code only happens after all chunks were uploaded. + _finished = true; + // the file id should only be empty for new files up- or downloaded + QByteArray fid = job->reply()->rawHeader("OC-FileID"); + if( !fid.isEmpty() ) { + if( !_item->_fileId.isEmpty() && _item->_fileId != fid ) { + qDebug() << "WARN: File ID changed!" << _item->_fileId << fid; + } + _item->_fileId = fid; + } + + _item->_etag = etag; + + _item->_responseTimeStamp = job->responseTimestamp(); + + if (job->reply()->rawHeader("X-OC-MTime") != "accepted") { + // X-OC-MTime is supported since owncloud 5.0. But not when chunking. + // Normally Owncloud 6 always puts X-OC-MTime + qWarning() << "Server does not support X-OC-MTime" << job->reply()->rawHeader("X-OC-MTime"); + // Well, the mtime was not set + done(SyncFileItem::SoftError, "Server does not support X-OC-MTime"); + } + + // performance logging + _item->_requestDuration = _stopWatch.stop(); + qDebug() << "*==* duration UPLOAD" << _item->_size + << _stopWatch.durationOfLap(QLatin1String("ContentChecksum")) + << _stopWatch.durationOfLap(QLatin1String("TransmissionChecksum")) + << _item->_requestDuration; + // The job might stay alive for the whole sync, release this tiny bit of memory. + _stopWatch.reset(); + + finalize(); +} + + +void PropagateUploadFileV1::slotUploadProgress(qint64 sent, qint64 total) +{ + // Completion is signaled with sent=0, total=0; avoid accidentally + // resetting progress due to the sent being zero by ignoring it. + // finishedSignal() is bound to be emitted soon anyway. + // See https://bugreports.qt.io/browse/QTBUG-44782. + if (sent == 0 && total == 0) { + return; + } + + int progressChunk = _currentChunk + _startChunk - 1; + if (progressChunk >= _chunkCount) + progressChunk = _currentChunk - 1; + + // amount is the number of bytes already sent by all the other chunks that were sent + // not including this one. + // FIXME: this assumes all chunks have the same size, which is true only if the last chunk + // has not been finished (which should not happen because the last chunk is sent sequentially) + quint64 amount = progressChunk * chunkSize(); + + sender()->setProperty("byteWritten", sent); + if (_jobs.count() > 1) { + amount -= (_jobs.count() -1) * chunkSize(); + foreach (QObject *j, _jobs) { + amount += j->property("byteWritten").toULongLong(); + } + } else { + // sender() is the only current job, no need to look at the byteWritten properties + amount += sent; + } + emit progress(*_item, amount); +} + +} diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 2b2e4591d..fd817c734 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -47,6 +47,7 @@ owncloud_add_test(ExcludedFiles "") if(HAVE_QT5 AND NOT BUILD_WITH_QT4) owncloud_add_test(SyncEngine "syncenginetestutils.h") owncloud_add_test(SyncFileStatusTracker "syncenginetestutils.h") + owncloud_add_test(ChunkingNg "syncenginetestutils.h") endif(HAVE_QT5 AND NOT BUILD_WITH_QT4) SET(FolderMan_SRC ../src/gui/folderman.cpp) diff --git a/test/syncenginetestutils.h b/test/syncenginetestutils.h index 471bfee5d..514a3fca4 100644 --- a/test/syncenginetestutils.h +++ b/test/syncenginetestutils.h @@ -18,6 +18,20 @@ #include <QtTest> static const QUrl sRootUrl("owncloud://somehost/owncloud/remote.php/webdav/"); +static const QUrl sRootUrl2("owncloud://somehost/owncloud/remote.php/dav/files/admin/"); +static const QUrl sUploadUrl("owncloud://somehost/owncloud/remote.php/dav/uploads/admin/"); + +inline QString getFilePathFromUrl(const QUrl &url) { + QString path = url.path(); + if (path.startsWith(sRootUrl.path())) + return path.mid(sRootUrl.path().length()); + if (path.startsWith(sRootUrl2.path())) + return path.mid(sRootUrl2.path().length()); + if (path.startsWith(sUploadUrl.path())) + return path.mid(sUploadUrl.path().length()); + return {}; +} + inline QString generateEtag() { return QString::number(QDateTime::currentDateTime().toMSecsSinceEpoch(), 16); @@ -68,10 +82,15 @@ public: QFile file{_rootDir.filePath(relativePath)}; QVERIFY(!file.exists()); file.open(QFile::WriteOnly); - file.write(QByteArray{}.fill(contentChar, size)); + QByteArray buf(1024, contentChar); + for (int x = 0; x < size/buf.size(); ++x) { + file.write(buf); + } + file.write(buf.data(), size % buf.size()); file.close(); // Set the mtime 30 seconds in the past, for some tests that need to make sure that the mtime differs. OCC::FileSystem::setModTime(file.fileName(), OCC::Utility::qDateTimeToTime_t(QDateTime::currentDateTime().addSecs(-30))); + QCOMPARE(file.size(), size); } void setContents(const QString &relativePath, char contentChar) override { QFile file{_rootDir.filePath(relativePath)}; @@ -284,6 +303,12 @@ public: setOperation(op); open(QIODevice::ReadOnly); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isNull()); // for root, it should be empty + const FileInfo *fileInfo = remoteRootFileInfo.find(fileName); + Q_ASSERT(fileInfo); + QString prefix = request.url().path().left(request.url().path().size() - fileName.size()); + // Don't care about the request and just return a full propfind const QString davUri{QStringLiteral("DAV:")}; const QString ocUri{QStringLiteral("http://owncloud.org/ns")}; @@ -297,7 +322,7 @@ public: auto writeFileResponse = [&](const FileInfo &fileInfo) { xml.writeStartElement(davUri, QStringLiteral("response")); - xml.writeTextElement(davUri, QStringLiteral("href"), "/owncloud/remote.php/webdav/" + fileInfo.path()); + xml.writeTextElement(davUri, QStringLiteral("href"), prefix + fileInfo.path()); xml.writeStartElement(davUri, QStringLiteral("propstat")); xml.writeStartElement(davUri, QStringLiteral("prop")); @@ -322,11 +347,6 @@ public: xml.writeEndElement(); // response }; - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); - const FileInfo *fileInfo = remoteRootFileInfo.find(fileName); - Q_ASSERT(fileInfo); - writeFileResponse(*fileInfo); foreach(const FileInfo &childFileInfo, fileInfo->children) writeFileResponse(childFileInfo); @@ -370,8 +390,8 @@ public: setOperation(op); open(QIODevice::ReadOnly); - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isEmpty()); if ((fileInfo = remoteRootFileInfo.find(fileName))) { fileInfo->size = putPayload.size(); fileInfo->contentChar = putPayload.at(0); @@ -388,6 +408,7 @@ public: } Q_INVOKABLE void respond() { + emit uploadProgress(fileInfo->size, fileInfo->size); setRawHeader("OC-ETag", fileInfo->etag.toLatin1()); setRawHeader("ETag", fileInfo->etag.toLatin1()); setRawHeader("X-OC-MTime", "accepted"); // Prevents Q_ASSERT(!_runningNow) since we'll call PropagateItemJob::done twice in that case. @@ -412,8 +433,8 @@ public: setOperation(op); open(QIODevice::ReadOnly); - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isEmpty()); fileInfo = remoteRootFileInfo.createDir(fileName); if (!fileInfo) { @@ -445,8 +466,8 @@ public: setOperation(op); open(QIODevice::ReadOnly); - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isEmpty()); remoteRootFileInfo.remove(fileName); QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection); } @@ -472,11 +493,10 @@ public: setOperation(op); open(QIODevice::ReadOnly); - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); - QString destPath = request.rawHeader("Destination"); - Q_ASSERT(destPath.startsWith(sRootUrl.path())); - QString dest = destPath.mid(sRootUrl.path().length()); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isEmpty()); + QString dest = getFilePathFromUrl(QUrl::fromEncoded(request.rawHeader("Destination"))); + Q_ASSERT(!dest.isEmpty()); remoteRootFileInfo.rename(fileName, dest); QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection); } @@ -505,8 +525,8 @@ public: setOperation(op); open(QIODevice::ReadOnly); - Q_ASSERT(request.url().path().startsWith(sRootUrl.path())); - QString fileName = request.url().path().mid(sRootUrl.path().length()); + QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isEmpty()); fileInfo = remoteRootFileInfo.find(fileName); QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection); } @@ -535,6 +555,79 @@ public: } }; + +class FakeChunkMoveReply : public QNetworkReply +{ + Q_OBJECT + FileInfo *fileInfo; +public: + FakeChunkMoveReply(FileInfo &uploadsFileInfo, FileInfo &remoteRootFileInfo, + QNetworkAccessManager::Operation op, const QNetworkRequest &request, + QObject *parent) : QNetworkReply{parent} { + setRequest(request); + setUrl(request.url()); + setOperation(op); + open(QIODevice::ReadOnly); + + QString source = getFilePathFromUrl(request.url()); + Q_ASSERT(!source.isEmpty()); + Q_ASSERT(source.endsWith("/.file")); + source = source.left(source.length() - qstrlen("/.file")); + auto sourceFolder = uploadsFileInfo.find(source); + Q_ASSERT(sourceFolder); + Q_ASSERT(sourceFolder->isDir); + int count = 0; + int size = 0; + char payload = '*'; + + do { + if (!sourceFolder->children.contains(QString::number(count))) + break; + auto &x = sourceFolder->children[QString::number(count)]; + Q_ASSERT(!x.isDir); + Q_ASSERT(x.size > 0); // There should not be empty chunks + size += x.size; + payload = x.contentChar; + ++count; + } while(true); + + Q_ASSERT(count > 1); // There should be at least two chunks, otherwise why would we use chunking? + QCOMPARE(sourceFolder->children.count(), count); // There should not be holes or extra files + + QString fileName = getFilePathFromUrl(QUrl::fromEncoded(request.rawHeader("Destination"))); + Q_ASSERT(!fileName.isEmpty()); + + if ((fileInfo = remoteRootFileInfo.find(fileName))) { + QCOMPARE(request.rawHeader("If"), QByteArray("<" + request.rawHeader("Destination") + "> ([\"" + fileInfo->etag.toLatin1() + "\"])")); + fileInfo->size = size; + fileInfo->contentChar = payload; + } else { + Q_ASSERT(!request.hasRawHeader("If")); + // Assume that the file is filled with the same character + fileInfo = remoteRootFileInfo.create(fileName, size, payload); + } + + if (!fileInfo) { + abort(); + return; + } + QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection); + } + + Q_INVOKABLE void respond() { + setAttribute(QNetworkRequest::HttpStatusCodeAttribute, 201); + setRawHeader("OC-ETag", fileInfo->etag.toLatin1()); + setRawHeader("ETag", fileInfo->etag.toLatin1()); + setRawHeader("OC-FileId", fileInfo->fileId); + emit metaDataChanged(); + emit finished(); + } + + void abort() override { } + qint64 readData(char *, qint64) override { return 0; } +}; + + class FakeErrorReply : public QNetworkReply { Q_OBJECT @@ -561,33 +654,41 @@ public: class FakeQNAM : public QNetworkAccessManager { FileInfo _remoteRootFileInfo; + FileInfo _uploadFileInfo; QStringList _errorPaths; public: FakeQNAM(FileInfo initialRoot) : _remoteRootFileInfo{std::move(initialRoot)} { } FileInfo ¤tRemoteState() { return _remoteRootFileInfo; } + FileInfo &uploadState() { return _uploadFileInfo; } QStringList &errorPaths() { return _errorPaths; } protected: QNetworkReply *createRequest(Operation op, const QNetworkRequest &request, QIODevice *outgoingData = 0) { - const QString fileName = request.url().path().mid(sRootUrl.path().length()); + const QString fileName = getFilePathFromUrl(request.url()); + Q_ASSERT(!fileName.isNull()); if (_errorPaths.contains(fileName)) return new FakeErrorReply{op, request, this}; + bool isUpload = request.url().path().startsWith(sUploadUrl.path()); + FileInfo &info = isUpload ? _uploadFileInfo : _remoteRootFileInfo; + auto verb = request.attribute(QNetworkRequest::CustomVerbAttribute); if (verb == QLatin1String("PROPFIND")) // Ignore outgoingData always returning somethign good enough, works for now. - return new FakePropfindReply{_remoteRootFileInfo, op, request, this}; + return new FakePropfindReply{info, op, request, this}; else if (verb == QLatin1String("GET")) - return new FakeGetReply{_remoteRootFileInfo, op, request, this}; + return new FakeGetReply{info, op, request, this}; else if (verb == QLatin1String("PUT")) - return new FakePutReply{_remoteRootFileInfo, op, request, outgoingData->readAll(), this}; + return new FakePutReply{info, op, request, outgoingData->readAll(), this}; else if (verb == QLatin1String("MKCOL")) - return new FakeMkcolReply{_remoteRootFileInfo, op, request, this}; + return new FakeMkcolReply{info, op, request, this}; else if (verb == QLatin1String("DELETE")) - return new FakeDeleteReply{_remoteRootFileInfo, op, request, this}; - else if (verb == QLatin1String("MOVE")) - return new FakeMoveReply{_remoteRootFileInfo, op, request, this}; + return new FakeDeleteReply{info, op, request, this}; + else if (verb == QLatin1String("MOVE") && !isUpload) + return new FakeMoveReply{info, op, request, this}; + else if (verb == QLatin1String("MOVE") && isUpload) + return new FakeChunkMoveReply{info, _remoteRootFileInfo, op, request, this}; else { qDebug() << verb << outgoingData; Q_UNREACHABLE(); @@ -659,6 +760,7 @@ public: } FileInfo currentRemoteState() { return _fakeQnam->currentRemoteState(); } + FileInfo &uploadState() { return _fakeQnam->uploadState(); } QStringList &serverErrorPaths() { return _fakeQnam->errorPaths(); } @@ -695,14 +797,16 @@ public: QVERIFY(false); } - void execUntilFinished() { + bool execUntilFinished() { QSignalSpy spy(_syncEngine.get(), SIGNAL(finished(bool))); - QVERIFY(spy.wait()); + bool ok = spy.wait(60000); + Q_ASSERT(ok && "Sync timed out"); + return spy[0][0].toBool(); } - void syncOnce() { + bool syncOnce() { scheduleSync(); - execUntilFinished(); + return execUntilFinished(); } private: diff --git a/test/testchunkingng.cpp b/test/testchunkingng.cpp new file mode 100644 index 000000000..21225c210 --- /dev/null +++ b/test/testchunkingng.cpp @@ -0,0 +1,78 @@ +/* + * This software is in the public domain, furnished "as is", without technical + * support, and with no warranty, express or implied, as to its usefulness for + * any purpose. + * + */ + +#include <QtTest> +#include "syncenginetestutils.h" +#include <syncengine.h> + +using namespace OCC; + +class TestChunkingNG : public QObject +{ + Q_OBJECT + +private slots: + + void testFileUpload() { + FakeFolder fakeFolder{FileInfo::A12_B12_C12_S12()}; + fakeFolder.syncEngine().account()->setCapabilities({ { "dav", QVariantMap{ {"chunking", "1.0"} } } }); + const int size = 300 * 1000 * 1000; // 300 MB + fakeFolder.localModifier().insert("A/a0", size); + QVERIFY(fakeFolder.syncOnce()); + QCOMPARE(fakeFolder.currentLocalState(), fakeFolder.currentRemoteState()); + QCOMPARE(fakeFolder.uploadState().children.count(), 1); // the transfer was done with chunking + QCOMPARE(fakeFolder.currentRemoteState().find("A/a0")->size, size); + + // Check that another upload of the same file also work. + fakeFolder.localModifier().appendByte("A/a0"); + QVERIFY(fakeFolder.syncOnce()); + QCOMPARE(fakeFolder.currentLocalState(), fakeFolder.currentRemoteState()); + QCOMPARE(fakeFolder.uploadState().children.count(), 2); // the transfer was done with chunking + } + + + void testResume () { + + FakeFolder fakeFolder{FileInfo::A12_B12_C12_S12()}; + fakeFolder.syncEngine().account()->setCapabilities({ { "dav", QVariantMap{ {"chunking", "1.0"} } } }); + const int size = 300 * 1000 * 1000; // 300 MB + fakeFolder.localModifier().insert("A/a0", size); + + // Abort when the upload is at 1/3 + int sizeWhenAbort = -1; + auto con = QObject::connect(&fakeFolder.syncEngine(), &SyncEngine::transmissionProgress, + [&](const ProgressInfo &progress) { + if (progress.completedSize() > (progress.totalSize() /3 )) { + sizeWhenAbort = progress.completedSize(); + fakeFolder.syncEngine().abort(); + } + }); + + QVERIFY(!fakeFolder.syncOnce()); // there should have been an error + QObject::disconnect(con); + QVERIFY(sizeWhenAbort > 0); + QVERIFY(sizeWhenAbort < size); + QCOMPARE(fakeFolder.uploadState().children.count(), 1); // the transfer was done with chunking + auto upStateChildren = fakeFolder.uploadState().children.first().children; + QCOMPARE(sizeWhenAbort, std::accumulate(upStateChildren.cbegin(), upStateChildren.cend(), 0, + [](int s, const FileInfo &i) { return s + i.size; })); + + + // Add a fake file to make sure it gets deleted + fakeFolder.uploadState().children.first().insert("10000", size); + QVERIFY(fakeFolder.syncOnce()); + + + + QCOMPARE(fakeFolder.currentLocalState(), fakeFolder.currentRemoteState()); + QCOMPARE(fakeFolder.uploadState().children.count(), 1); // The same chunk id was re-used + QCOMPARE(fakeFolder.currentRemoteState().find("A/a0")->size, size); + } +}; + +QTEST_GUILESS_MAIN(TestChunkingNG) +#include "testchunkingng.moc" |