From 21199a2bb0b5d6c4d419157656c975c16c947b8f Mon Sep 17 00:00:00 2001
From: Markus Goetz <Markus.Goetz@nokia.com>
Date: Wed, 13 May 2009 13:16:04 +0200
Subject: QNAM: Upload architecture change: Backend and QNetworkReplyImpl
changes
Reviewed-by: Thiago Macieira
Reviewed-by: Peter Hartmann
---
src/network/access/qnetworkaccessbackend.cpp | 57 +++++---
src/network/access/qnetworkaccessbackend_p.h | 38 ++---
src/network/access/qnetworkaccessmanager.cpp | 8 +-
src/network/access/qnetworkreplyimpl.cpp | 202 ++++++++++++++++-----------
src/network/access/qnetworkreplyimpl_p.h | 17 ++-
5 files changed, 188 insertions(+), 134 deletions(-)
diff --git a/src/network/access/qnetworkaccessbackend.cpp b/src/network/access/qnetworkaccessbackend.cpp
index df468b8..b9d1b85 100644
--- a/src/network/access/qnetworkaccessbackend.cpp
+++ b/src/network/access/qnetworkaccessbackend.cpp
@@ -50,6 +50,8 @@
#include "qnetworkaccesscachebackend_p.h"
#include "qabstractnetworkcache.h"
+#include "private/qnoncontiguousbytedevice_p.h"
+
QT_BEGIN_NAMESPACE
static bool factoryDataShutdown = false;
@@ -109,17 +111,43 @@ QNetworkAccessBackend *QNetworkAccessManagerPrivate::findBackend(QNetworkAccessM
return 0;
}
-QNetworkAccessBackend::QNetworkAccessBackend()
+
+QNonContiguousByteDevice* QNetworkAccessBackend::createUploadByteDevice()
{
+ QNonContiguousByteDevice* device = 0;
+
+ if (reply->outgoingDataBuffer)
+ device = QNonContiguousByteDeviceFactory::create(reply->outgoingDataBuffer);
+ else
+ device = QNonContiguousByteDeviceFactory::create(reply->outgoingData);
+
+ bool bufferDisallowed =
+ reply->request.attribute(QNetworkRequest::DoNotBufferUploadDataAttribute,
+ QVariant(false)) == QVariant(true);
+ if (bufferDisallowed)
+ device->disableReset();
+
+ // make sure we delete this later
+ device->setParent(this);
+
+ connect(device, SIGNAL(readProgress(qint64,qint64)), this, SLOT(emitReplyUploadProgress(qint64,qint64)));
+
+ return device;
}
-QNetworkAccessBackend::~QNetworkAccessBackend()
+// need to have this function since the reply is a private member variable
+// and the special backends need to access this.
+void QNetworkAccessBackend::emitReplyUploadProgress(qint64 bytesSent, qint64 bytesTotal)
{
+ reply->emitUploadProgress(bytesSent, bytesTotal);
}
-void QNetworkAccessBackend::upstreamReadyRead()
+QNetworkAccessBackend::QNetworkAccessBackend()
+{
+}
+
+QNetworkAccessBackend::~QNetworkAccessBackend()
{
- // do nothing
}
void QNetworkAccessBackend::downstreamReadyWrite()
@@ -184,23 +212,6 @@ bool QNetworkAccessBackend::isCachingEnabled() const
return reply->isCachingEnabled();
}
-qint64 QNetworkAccessBackend::upstreamBytesAvailable() const
-{
- return reply->writeBuffer.size();
-}
-
-void QNetworkAccessBackend::upstreamBytesConsumed(qint64 count)
-{
- // remove count bytes from the write buffer
- reply->consume(count);
-}
-
-QByteArray QNetworkAccessBackend::readUpstream()
-{
- // ### this is expensive. Consider making QRingBuffer::peekAll keep the buffer it returns
- return reply->writeBuffer.peek(upstreamBytesAvailable());
-}
-
qint64 QNetworkAccessBackend::nextDownstreamBlockSize() const
{
return reply->nextDownstreamBlockSize();
@@ -213,12 +224,12 @@ qint64 QNetworkAccessBackend::downstreamBytesToConsume() const
void QNetworkAccessBackend::writeDownstreamData(const QByteArray &data)
{
- reply->feed(data);
+ reply->appendDownstreamData(data);
}
void QNetworkAccessBackend::writeDownstreamData(QIODevice *data)
{
- reply->feed(data);
+ reply->appendDownstreamData(data);
}
QVariant QNetworkAccessBackend::header(QNetworkRequest::KnownHeaders header) const
diff --git a/src/network/access/qnetworkaccessbackend_p.h b/src/network/access/qnetworkaccessbackend_p.h
index 9012396..6035f3a 100644
--- a/src/network/access/qnetworkaccessbackend_p.h
+++ b/src/network/access/qnetworkaccessbackend_p.h
@@ -70,6 +70,8 @@ class QNetworkAccessManagerPrivate;
class QNetworkReplyImplPrivate;
class QAbstractNetworkCache;
class QNetworkCacheMetaData;
+class QNetworkAccessBackendUploadIODevice;
+class QNonContiguousByteDevice;
// Should support direct file upload from disk or download to disk.
//
@@ -86,14 +88,13 @@ public:
// have different names. The Connection has two streams:
//
// - Upstream:
- // Upstream is data that is being written into this connection,
- // from the user. Upstream operates in a "pull" mechanism: the
- // connection will be notified that there is more data available
- // by a call to "upstreamReadyRead". The number of bytes
- // available is given by upstreamBytesAvailable(). A call to
- // readUpstream() always yields the entire upstream buffer. When
- // the connection has processed a certain amount of bytes from
- // that buffer, it should call upstreamBytesConsumed().
+ // The upstream uses a QNonContiguousByteDevice provided
+ // by the backend. This device emits the usual readyRead()
+ // signal when the backend has data available for the connection
+ // to write. The different backends can listen on this signal
+ // and then pull upload data from the QNonContiguousByteDevice and
+ // deal with it.
+ //
//
// - Downstream:
// Downstream is the data that is being read from this
@@ -111,12 +112,9 @@ public:
virtual void open() = 0;
virtual void closeDownstreamChannel() = 0;
- virtual void closeUpstreamChannel() = 0;
virtual bool waitForDownstreamReadyRead(int msecs) = 0;
- virtual bool waitForUpstreamBytesWritten(int msecs) = 0;
// slot-like:
- virtual void upstreamReadyRead();
virtual void downstreamReadyWrite();
virtual void copyFinished(QIODevice *);
virtual void ignoreSslErrors();
@@ -155,18 +153,24 @@ public:
QVariant attribute(QNetworkRequest::Attribute code) const;
void setAttribute(QNetworkRequest::Attribute code, const QVariant &value);
+ // return true if the QNonContiguousByteDevice of the upload
+ // data needs to support reset(). Currently needed for HTTP.
+ // This will possibly enable buffering of the upload data.
+ virtual bool needsResetableUploadData() {return false;};
+
protected:
- // these functions control the upstream mechanism
- // that is, data coming into the backend and out via the connection
- qint64 upstreamBytesAvailable() const;
- void upstreamBytesConsumed(qint64 count);
- QByteArray readUpstream();
+ // Create the device used for reading the upload data
+ QNonContiguousByteDevice* createUploadByteDevice();
+
// these functions control the downstream mechanism
// that is, data that has come via the connection and is going out the backend
qint64 nextDownstreamBlockSize() const;
qint64 downstreamBytesToConsume() const;
void writeDownstreamData(const QByteArray &data);
+
+public slots:
+ // for task 251801, needs to be a slot to be called asynchronously
void writeDownstreamData(QIODevice *data);
protected slots:
@@ -179,10 +183,12 @@ protected slots:
void metaDataChanged();
void redirectionRequested(const QUrl &destination);
void sslErrors(const QList<QSslError> &errors);
+ void emitReplyUploadProgress(qint64 bytesSent, qint64 bytesTotal);
private:
friend class QNetworkAccessManager;
friend class QNetworkAccessManagerPrivate;
+ friend class QNetworkAccessBackendUploadIODevice;
QNetworkAccessManagerPrivate *manager;
QNetworkReplyImplPrivate *reply;
};
diff --git a/src/network/access/qnetworkaccessmanager.cpp b/src/network/access/qnetworkaccessmanager.cpp
index bcbeef1..bf06ede 100644
--- a/src/network/access/qnetworkaccessmanager.cpp
+++ b/src/network/access/qnetworkaccessmanager.cpp
@@ -686,7 +686,10 @@ QNetworkReply *QNetworkAccessManager::createRequest(QNetworkAccessManager::Opera
priv->urlForLastAuthentication = url;
}
- // third step: setup the reply
+ // third step: find a backend
+ priv->backend = d->findBackend(op, request);
+
+ // fourth step: setup the reply
priv->setup(op, request, outgoingData);
if (request.attribute(QNetworkRequest::CacheLoadControlAttribute, QNetworkRequest::PreferNetwork).toInt() !=
QNetworkRequest::AlwaysNetwork)
@@ -695,9 +698,6 @@ QNetworkReply *QNetworkAccessManager::createRequest(QNetworkAccessManager::Opera
QList<QNetworkProxy> proxyList = d->queryProxy(QNetworkProxyQuery(request.url()));
priv->proxyList = proxyList;
#endif
-
- // fourth step: find a backend
- priv->backend = d->findBackend(op, request);
if (priv->backend) {
priv->backend->setParent(reply);
priv->backend->reply = priv;
diff --git a/src/network/access/qnetworkreplyimpl.cpp b/src/network/access/qnetworkreplyimpl.cpp
index 79c3d1a..749a462 100644
--- a/src/network/access/qnetworkreplyimpl.cpp
+++ b/src/network/access/qnetworkreplyimpl.cpp
@@ -46,13 +46,15 @@
#include "QtCore/qcoreapplication.h"
#include "QtCore/qdatetime.h"
#include "QtNetwork/qsslconfiguration.h"
+#include "qnetworkaccesshttpbackend_p.h"
#include <QtCore/QCoreApplication>
QT_BEGIN_NAMESPACE
inline QNetworkReplyImplPrivate::QNetworkReplyImplPrivate()
- : copyDevice(0), networkCache(0),
+ : backend(0), outgoingData(0), outgoingDataBuffer(0),
+ copyDevice(0), networkCache(0),
cacheEnabled(false), cacheSaveDevice(0),
bytesDownloaded(0), lastBytesDownloaded(-1), bytesUploaded(-1),
state(Idle)
@@ -61,8 +63,13 @@ inline QNetworkReplyImplPrivate::QNetworkReplyImplPrivate()
void QNetworkReplyImplPrivate::_q_startOperation()
{
- // This function is called exactly once
+ // ensure this function is only being called once
+ if (state == Working) {
+ qDebug("QNetworkReplyImpl::_q_startOperation was called more than once");
+ return;
+ }
state = Working;
+
if (!backend) {
error(QNetworkReplyImpl::ProtocolUnknownError,
QCoreApplication::translate("QNetworkReply", "Protocol \"%1\" is unknown").arg(url.scheme())); // not really true!;
@@ -74,57 +81,11 @@ void QNetworkReplyImplPrivate::_q_startOperation()
if (state != Finished) {
if (operation == QNetworkAccessManager::GetOperation)
pendingNotifications.append(NotifyDownstreamReadyWrite);
- if (outgoingData) {
- _q_sourceReadyRead();
-#if 0 // ### FIXME
- if (outgoingData->atEndOfStream() && writeBuffer.isEmpty())
- // empty upload
- emit q->uploadProgress(0, 0);
-#endif
- }
handleNotifications();
}
}
-void QNetworkReplyImplPrivate::_q_sourceReadyRead()
-{
- // read data from the outgoingData QIODevice into our internal buffer
- enum { DesiredBufferSize = 32 * 1024 };
-
- if (writeBuffer.size() >= DesiredBufferSize)
- return; // don't grow the buffer too much
-
- // read as many bytes are available or up until we fill up the buffer
- // but always read at least one byte
- qint64 bytesToRead = qBound<qint64>(1, outgoingData->bytesAvailable(),
- DesiredBufferSize - writeBuffer.size());
- char *ptr = writeBuffer.reserve(bytesToRead);
- qint64 bytesActuallyRead = outgoingData->read(ptr, bytesToRead);
- if (bytesActuallyRead == -1) {
- // EOF
- writeBuffer.chop(bytesToRead);
- backendNotify(NotifyCloseUpstreamChannel);
- return;
- }
-
- if (bytesActuallyRead < bytesToRead)
- writeBuffer.chop(bytesToRead - bytesActuallyRead);
-
- // if we did read anything, let the backend know and handle it
- if (bytesActuallyRead)
- backendNotify(NotifyUpstreamReadyRead);
-
- // check for EOF again
- if (!outgoingData->isSequential() && outgoingData->atEnd())
- backendNotify(NotifyCloseUpstreamChannel);
-}
-
-void QNetworkReplyImplPrivate::_q_sourceReadChannelFinished()
-{
- _q_sourceReadyRead();
-}
-
void QNetworkReplyImplPrivate::_q_copyReadyRead()
{
Q_Q(QNetworkReplyImpl);
@@ -143,7 +104,7 @@ void QNetworkReplyImplPrivate::_q_copyReadyRead()
if (bytesActuallyRead == -1) {
readBuffer.chop(bytesToRead);
backendNotify(NotifyCopyFinished);
- return;
+ break;
}
if (bytesActuallyRead != bytesToRead)
@@ -151,6 +112,7 @@ void QNetworkReplyImplPrivate::_q_copyReadyRead()
if (!copyDevice->isSequential() && copyDevice->atEnd()) {
backendNotify(NotifyCopyFinished);
+ bytesDownloaded += bytesActuallyRead;
break;
}
@@ -174,6 +136,67 @@ void QNetworkReplyImplPrivate::_q_copyReadChannelFinished()
_q_copyReadyRead();
}
+void QNetworkReplyImplPrivate::_q_bufferOutgoingDataFinished()
+{
+ Q_Q(QNetworkReplyImpl);
+
+ // make sure this is only called once, ever.
+ //_q_bufferOutgoingData may call it or the readChannelFinished emission
+ if (state != Buffering)
+ return;
+
+ // disconnect signals
+ QObject::disconnect(outgoingData, SIGNAL(readyRead()), q, SLOT(_q_bufferOutgoingData()));
+ QObject::disconnect(outgoingData, SIGNAL(readChannelFinished()), q, SLOT(_q_bufferOutgoingDataFinished()));
+
+ // finally, start the request
+ QMetaObject::invokeMethod(q, "_q_startOperation", Qt::QueuedConnection);
+}
+
+void QNetworkReplyImplPrivate::_q_bufferOutgoingData()
+{
+ Q_Q(QNetworkReplyImpl);
+
+ if (!outgoingDataBuffer) {
+ // first call, create our buffer
+ outgoingDataBuffer = new QRingBuffer();
+
+ QObject::connect(outgoingData, SIGNAL(readyRead()), q, SLOT(_q_bufferOutgoingData()));
+ QObject::connect(outgoingData, SIGNAL(readChannelFinished()), q, SLOT(_q_bufferOutgoingDataFinished()));
+ }
+
+ qint64 bytesBuffered = 0;
+ qint64 bytesToBuffer = 0;
+
+ // read data into our buffer
+ forever {
+ bytesToBuffer = outgoingData->bytesAvailable();
+ // unknown? just try 2 kB, this also ensures we always try to read the EOF
+ if (bytesToBuffer <= 0)
+ bytesToBuffer = 2*1024;
+
+ char *dst = outgoingDataBuffer->reserve(bytesToBuffer);
+ bytesBuffered = outgoingData->read(dst, bytesToBuffer);
+
+ if (bytesBuffered == -1) {
+ // EOF has been reached.
+ outgoingDataBuffer->chop(bytesToBuffer);
+
+ _q_bufferOutgoingDataFinished();
+ break;
+ } else if (bytesBuffered == 0) {
+ // nothing read right now, just wait until we get called again
+ outgoingDataBuffer->chop(bytesToBuffer);
+
+ break;
+ } else {
+ // don't break, try to read() again
+ outgoingDataBuffer->chop(bytesToBuffer - bytesBuffered);
+ }
+ }
+}
+
+
void QNetworkReplyImplPrivate::setup(QNetworkAccessManager::Operation op, const QNetworkRequest &req,
QIODevice *data)
{
@@ -184,13 +207,42 @@ void QNetworkReplyImplPrivate::setup(QNetworkAccessManager::Operation op, const
url = request.url();
operation = op;
- if (outgoingData) {
- q->connect(outgoingData, SIGNAL(readyRead()), SLOT(_q_sourceReadyRead()));
- q->connect(outgoingData, SIGNAL(readChannelFinished()), SLOT(_q_sourceReadChannelFinished()));
+ if (outgoingData && backend) {
+ // there is data to be uploaded, e.g. HTTP POST.
+
+ if (!backend->needsResetableUploadData() || !outgoingData->isSequential()) {
+ // backend does not need upload buffering or
+ // fixed size non-sequential
+ // just start the operation
+ QMetaObject::invokeMethod(q, "_q_startOperation", Qt::QueuedConnection);
+ } else {
+ bool bufferingDisallowed =
+ req.attribute(QNetworkRequest::DoNotBufferUploadDataAttribute,
+ false).toBool();
+
+ if (bufferingDisallowed) {
+ // if a valid content-length header for the request was supplied, we can disable buffering
+ // if not, we will buffer anyway
+ if (req.header(QNetworkRequest::ContentLengthHeader).isValid()) {
+ QMetaObject::invokeMethod(q, "_q_startOperation", Qt::QueuedConnection);
+ } else {
+ state = Buffering;
+ QMetaObject::invokeMethod(q, "_q_bufferOutgoingData", Qt::QueuedConnection);
+ }
+ } else {
+ // _q_startOperation will be called when the buffering has finished.
+ state = Buffering;
+ QMetaObject::invokeMethod(q, "_q_bufferOutgoingData", Qt::QueuedConnection);
+ }
+ }
+ } else {
+ // No outgoing data (e.g. HTTP GET request)
+ // or no backend
+ // if no backend, _q_startOperation will handle the error of this
+ QMetaObject::invokeMethod(q, "_q_startOperation", Qt::QueuedConnection);
}
q->QIODevice::open(QIODevice::ReadOnly);
- QMetaObject::invokeMethod(q, "_q_startOperation", Qt::QueuedConnection);
}
void QNetworkReplyImplPrivate::setNetworkCache(QAbstractNetworkCache *nc)
@@ -226,18 +278,10 @@ void QNetworkReplyImplPrivate::handleNotifications()
backend->downstreamReadyWrite();
break;
- case NotifyUpstreamReadyRead:
- backend->upstreamReadyRead();
- break;
-
case NotifyCloseDownstreamChannel:
backend->closeDownstreamChannel();
break;
- case NotifyCloseUpstreamChannel:
- backend->closeUpstreamChannel();
- break;
-
case NotifyCopyFinished: {
QIODevice *dev = copyDevice;
copyDevice = 0;
@@ -299,29 +343,14 @@ void QNetworkReplyImplPrivate::completeCacheSave()
cacheEnabled = false;
}
-void QNetworkReplyImplPrivate::consume(qint64 count)
+void QNetworkReplyImplPrivate::emitUploadProgress(qint64 bytesSent, qint64 bytesTotal)
{
Q_Q(QNetworkReplyImpl);
- if (count <= 0) {
- qWarning("QNetworkConnection: backend signalled that it consumed %ld bytes", long(count));
- return;
- }
-
- if (outgoingData)
- // schedule another read from the source
- QMetaObject::invokeMethod(q_func(), "_q_sourceReadyRead", Qt::QueuedConnection);
-
- writeBuffer.skip(count);
- if (bytesUploaded == -1)
- bytesUploaded = count;
- else
- bytesUploaded += count;
-
- QVariant totalSize = request.header(QNetworkRequest::ContentLengthHeader);
- emit q->uploadProgress(bytesUploaded,
- totalSize.isNull() ? Q_INT64_C(-1) : totalSize.toLongLong());
+ bytesUploaded = bytesSent;
+ emit q->uploadProgress(bytesSent, bytesTotal);
}
+
qint64 QNetworkReplyImplPrivate::nextDownstreamBlockSize() const
{
enum { DesiredBufferSize = 32 * 1024 };
@@ -331,7 +360,9 @@ qint64 QNetworkReplyImplPrivate::nextDownstreamBlockSize() const
return qMax<qint64>(0, readBufferMaxSize - readBuffer.size());
}
-void QNetworkReplyImplPrivate::feed(const QByteArray &data)
+// we received downstream data and send this to the cache
+// and to our readBuffer (which in turn gets read by the user of QNetworkReply)
+void QNetworkReplyImplPrivate::appendDownstreamData(const QByteArray &data)
{
Q_Q(QNetworkReplyImpl);
if (!q->isOpen())
@@ -379,7 +410,8 @@ void QNetworkReplyImplPrivate::feed(const QByteArray &data)
}
}
-void QNetworkReplyImplPrivate::feed(QIODevice *data)
+// this is used when it was fetched from the cache, right?
+void QNetworkReplyImplPrivate::appendDownstreamData(QIODevice *data)
{
Q_Q(QNetworkReplyImpl);
Q_ASSERT(q->isOpen());
@@ -409,9 +441,11 @@ void QNetworkReplyImplPrivate::finished()
pendingNotifications.clear();
QVariant totalSize = cookedHeaders.value(QNetworkRequest::ContentLengthHeader);
- if (bytesDownloaded != lastBytesDownloaded || totalSize.isNull())
+ if (totalSize.isNull() || totalSize == -1) {
emit q->downloadProgress(bytesDownloaded, bytesDownloaded);
- if (bytesUploaded == -1 && outgoingData)
+ }
+
+ if (bytesUploaded == -1 && (outgoingData || outgoingDataBuffer))
emit q->uploadProgress(0, 0);
completeCacheSave();
diff --git a/src/network/access/qnetworkreplyimpl_p.h b/src/network/access/qnetworkreplyimpl_p.h
index ad06f78..8d3c90e 100644
--- a/src/network/access/qnetworkreplyimpl_p.h
+++ b/src/network/access/qnetworkreplyimpl_p.h
@@ -59,6 +59,7 @@
#include "qnetworkproxy.h"
#include "QtCore/qmap.h"
#include "QtCore/qqueue.h"
+#include "QtCore/qbuffer.h"
#include "private/qringbuffer_p.h"
QT_BEGIN_NAMESPACE
@@ -91,10 +92,10 @@ public:
Q_DECLARE_PRIVATE(QNetworkReplyImpl)
Q_PRIVATE_SLOT(d_func(), void _q_startOperation())
- Q_PRIVATE_SLOT(d_func(), void _q_sourceReadyRead())
- Q_PRIVATE_SLOT(d_func(), void _q_sourceReadChannelFinished())
Q_PRIVATE_SLOT(d_func(), void _q_copyReadyRead())
Q_PRIVATE_SLOT(d_func(), void _q_copyReadChannelFinished())
+ Q_PRIVATE_SLOT(d_func(), void _q_bufferOutgoingData())
+ Q_PRIVATE_SLOT(d_func(), void _q_bufferOutgoingDataFinished())
};
class QNetworkReplyImplPrivate: public QNetworkReplyPrivate
@@ -102,15 +103,13 @@ class QNetworkReplyImplPrivate: public QNetworkReplyPrivate
public:
enum InternalNotifications {
NotifyDownstreamReadyWrite,
- NotifyUpstreamReadyRead,
NotifyCloseDownstreamChannel,
- NotifyCloseUpstreamChannel,
NotifyCopyFinished
};
enum State {
Idle,
- Opening,
+ Buffering,
Working,
Finished,
Aborted
@@ -125,6 +124,8 @@ public:
void _q_sourceReadChannelFinished();
void _q_copyReadyRead();
void _q_copyReadChannelFinished();
+ void _q_bufferOutgoingData();
+ void _q_bufferOutgoingDataFinished();
void setup(QNetworkAccessManager::Operation op, const QNetworkRequest &request,
QIODevice *outgoingData);
@@ -138,9 +139,10 @@ public:
void setCachingEnabled(bool enable);
bool isCachingEnabled() const;
void consume(qint64 count);
+ void emitUploadProgress(qint64 bytesSent, qint64 bytesTotal);
qint64 nextDownstreamBlockSize() const;
- void feed(const QByteArray &data);
- void feed(QIODevice *data);
+ void appendDownstreamData(const QByteArray &data);
+ void appendDownstreamData(QIODevice *data);
void finished();
void error(QNetworkReply::NetworkError code, const QString &errorString);
void metaDataChanged();
@@ -149,6 +151,7 @@ public:
QNetworkAccessBackend *backend;
QIODevice *outgoingData;
+ QRingBuffer *outgoingDataBuffer;
QIODevice *copyDevice;
QAbstractNetworkCache *networkCache;
--
cgit v0.12