Repository navigation
Expand file tree
/
Copy pathpropagateupload.h
More file actions
396 lines (339 loc) · 11.7 KB
/
Copy pathpropagateupload.h
File metadata and controls
396 lines (339 loc) · 11.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
/*
* 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.
*/
#pragma once
#include "owncloudpropagator.h"
#include "networkjobs.h"
#include <QBuffer>
#include <QFile>
#include <QElapsedTimer>
namespace OCC {
Q_DECLARE_LOGGING_CATEGORY(lcPutJob)
Q_DECLARE_LOGGING_CATEGORY(lcPropagateUpload)
Q_DECLARE_LOGGING_CATEGORY(lcPropagateUploadV1)
Q_DECLARE_LOGGING_CATEGORY(lcPropagateUploadNG)
class BandwidthManager;
/**
* @brief The UploadDevice class
* @ingroup libsync
*/
class UploadDevice : public QIODevice
{
Q_OBJECT
public:
UploadDevice(const QString &fileName, qint64 start, qint64 size, BandwidthManager *bwm);
~UploadDevice() override;
bool open(QIODevice::OpenMode mode) override;
void close() override;
qint64 writeData(const char *, qint64) override;
qint64 readData(char *data, qint64 maxlen) override;
bool atEnd() const override;
qint64 size() const override;
qint64 bytesAvailable() const override;
bool isSequential() const override;
bool seek(qint64 pos) override;
void setBandwidthLimited(bool);
bool isBandwidthLimited() { return _bandwidthLimited; }
void setChoked(bool);
bool isChoked() { return _choked; }
void giveBandwidthQuota(qint64 bwq);
signals:
private:
/// The local file to read data from
QFile _file;
/// Start of the file data to use
qint64 _start = 0;
/// Amount of file data after _start to use
qint64 _size = 0;
/// Position between _start and _start+_size
qint64 _read = 0;
// Bandwidth manager related
QPointer<BandwidthManager> _bandwidthManager;
qint64 _bandwidthQuota;
qint64 _readWithProgress;
bool _bandwidthLimited; // if _bandwidthQuota will be used
bool _choked; // if upload is paused (readData() will return 0)
friend class BandwidthManager;
public slots:
void slotJobUploadProgress(qint64 sent, qint64 t);
};
/**
* @brief The PUTFileJob class
* @ingroup libsync
*/
class PUTFileJob : public AbstractNetworkJob
{
Q_OBJECT
private:
QIODevice *_device;
QMap<QByteArray, QByteArray> _headers;
QString _errorString;
QUrl _url;
QElapsedTimer _requestTimer;
public:
explicit PUTFileJob(AccountPtr account, const QString &path, std::unique_ptr<QIODevice> device,
const QMap<QByteArray, QByteArray> &headers, int chunk, QObject *parent = nullptr)
: AbstractNetworkJob(account, path, parent)
, _device(device.release())
, _headers(headers)
, _chunk(chunk)
{
_device->setParent(this);
}
explicit PUTFileJob(AccountPtr account, const QUrl &url, std::unique_ptr<QIODevice> device,
const QMap<QByteArray, QByteArray> &headers, int chunk, QObject *parent = nullptr)
: AbstractNetworkJob(account, QString(), parent)
, _device(device.release())
, _headers(headers)
, _url(url)
, _chunk(chunk)
{
_device->setParent(this);
}
~PUTFileJob() override;
int _chunk;
void start() override;
bool finished() override;
QIODevice *device()
{
return _device;
}
QString errorString()
{
return _errorString.isEmpty() ? AbstractNetworkJob::errorString() : _errorString;
}
std::chrono::milliseconds msSinceStart() const
{
return std::chrono::milliseconds(_requestTimer.elapsed());
}
signals:
void finishedSignal();
void uploadProgress(qint64, qint64);
};
/**
* @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()
*/
class PropagateUploadFileCommon : public PropagateItemJob
{
Q_OBJECT
protected:
QVector<AbstractNetworkJob *> _jobs; /// network jobs that are currently in transit
bool _finished BITFIELD(1); /// Tells that all the jobs have been finished
bool _deleteExisting BITFIELD(1);
/** Whether an abort is currently ongoing.
*
* Important to avoid duplicate aborts since each finishing PUTFileJob might
* trigger an abort on error.
*/
bool _aborting BITFIELD(1);
QByteArray _transmissionChecksumHeader;
public:
PropagateUploadFileCommon(OwncloudPropagator *propagator, const SyncFileItemPtr &item)
: PropagateItemJob(propagator, item)
, _finished(false)
, _deleteExisting(false)
, _aborting(false)
{
}
/**
* Whether an existing entity with the same name may be deleted before
* the upload.
*
* Default: false.
*/
void setDeleteExisting(bool enabled);
void start() override;
bool isLikelyFinishedQuickly() override { return _item->_size < propagator()->smallFileSize(); }
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 finalize();
void abortWithError(SyncFileItem::Status status, const QString &error);
public slots:
void slotJobDestroyed(QObject *job);
protected:
void done(SyncFileItem::Status status, const QString &errorString = QString()) override;
/**
* Aborts all running network jobs, except for the ones that mayAbortJob
* returns false on and, for async aborts, emits abortFinished when done.
*/
void abortNetworkJobs(
AbortType abortType,
const std::function<bool(AbstractNetworkJob *job)> &mayAbortJob);
/**
* Checks whether the current error is one that should reset the whole
* transfer if it happens too often. If so: Bump UploadInfo::errorCount
* and maybe perform the reset.
*/
void checkResettingErrors();
/**
* Error handling functionality that is shared between jobs.
*/
void commonErrorHandling(AbstractNetworkJob *job);
/**
* Increases the timeout for the final MOVE/PUT for large files.
*
* This is an unfortunate workaround since the drawback is not being able to
* detect real disconnects in a timely manner. Shall go away when the server
* response starts coming quicker, or there is some sort of async api.
*
* See #6527, enterprise#2480
*/
static void adjustLastJobTimeout(AbstractNetworkJob *job, qint64 fileSize);
/** Bases headers that need to be sent on the PUT, or in the MOVE for chunking-ng */
QMap<QByteArray, QByteArray> headers();
private:
bool _quotaUpdated = false;
};
/**
* @ingroup libsync
*
* Propagation job, impementing the old chunking agorithm
*
*/
class PropagateUploadFileV1 : public PropagateUploadFileCommon
{
Q_OBJECT
private:
/**
* That's the start chunk that was stored in the database for resuming.
* In the non-resuming case it is 0.
* If we are resuming, this is the first chunk we need to send
*/
int _startChunk = 0;
/**
* This is the next chunk that we need to send. Starting from 0 even if _startChunk != 0
* (In other words, _startChunk + _currentChunk is really the number of the chunk we need to send next)
* (In other words, _currentChunk is the number of the chunk that we already sent or started sending)
*/
int _currentChunk = 0;
int _chunkCount = 0; /// Total number of chunks for this file
uint _transferId = 0; /// transfer id (part of the url)
qint64 chunkSize() const {
// Old chunking does not use dynamic chunking algorithm, and does not adjusts the chunk size respectively,
// thus this value should be used as the one classifing item to be chunked
return propagator()->syncOptions()._initialChunkSize;
}
public:
PropagateUploadFileV1(OwncloudPropagator *propagator, const SyncFileItemPtr &item)
: PropagateUploadFileCommon(propagator, item)
{
}
void doStartUpload() override;
public slots:
void abort(PropagatorJob::AbortType abortType) override;
private slots:
void startNextChunk();
void slotPutFinished();
void slotUploadProgress(qint64, qint64);
};
/**
* @ingroup libsync
*
* Propagation job, impementing the new chunking agorithm
*
*/
class PropagateUploadFileNG : public PropagateUploadFileCommon
{
Q_OBJECT
private:
/** Amount of data that was already sent in bytes.
*
* If this job is resuming an upload, this number includes bytes that were
* sent in previous jobs.
*/
qint64 _sent = 0;
/** Amount of data that needs to be sent to the server in bytes.
*
* For normal uploads this will be the file size.
*
* This value is intended to be comparable to _sent: it's always the total
* amount of data that needs to be present at the server to finish the upload -
* regardless of whether previous jobs have already sent something.
*/
qint64 _bytesToUpload;
uint _transferId = 0; /// transfer id (part of the url)
qint64 _currentChunkOffset = 0; /// byte offset of the next chunk data that will be sent
qint64 _currentChunkSize = 0; /// current chunk size
bool _removeJobError = false; /// 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.)
struct ServerChunkInfo
{
qint64 size;
QString originalName;
};
QMap<qint64, ServerChunkInfo> _serverChunks;
// Vector with expected PUT ranges.
struct UploadRangeInfo
{
qint64 start;
qint64 size;
qint64 end() const { return start + size; }
};
QVector<UploadRangeInfo> _rangesToUpload;
/**
* Return the URL of a chunk.
* If chunkOffset == -1, returns the URL of the parent folder containing the chunks
*/
QUrl chunkUrl(qint64 chunkOffset = -1);
/**
* Finds the range starting at 'start' in _rangesToUpload and removes the first
* 'size' bytes from it. If it becomes empty, remove the range.
*
* Retuns false if no matching range was found.
*/
bool markRangeAsDone(qint64 start, qint64 size);
public:
PropagateUploadFileNG(OwncloudPropagator *propagator, const SyncFileItemPtr &item);
void doStartUpload() override;
private:
void doStartUploadNext();
void startNewUpload();
void startNextChunk();
void doFinalMove();
public slots:
void abort(AbortType abortType) override;
private slots:
void slotPropfindFinished();
void slotPropfindFinishedWithError();
void slotPropfindIterate(const QString &name, const QMap<QString, QString> &properties);
void slotDeleteJobFinished();
void slotMkColFinished();
void slotPutFinished();
void slotMoveJobFinished();
void slotUploadProgress(qint64, qint64);
};
}