Add a write buffer to QLocalSocket/Win

Commit 0307c008 removed the buffering of data-to-be-written from
QWindowsPipeWriter, because it was assumed that users of this class
(QProcess and QLocalSocket) already buffer data internally.

This assumption was wrong for QLocalSocket. The following sequence
    localSocket->write(someData);
    localSocket->write(someMoreData);
would not write anything on the second write.

Add a write buffer to the Windows implementation of QLocalSocket.

Task-number: QTBUG-52073
Change-Id: I6d0f03a722ec48138cbde3e2f69aae7dafe790d3
Reviewed-by: Oswald Buddenhagen <oswald.buddenhagen@theqtcompany.com>

src/corelib/io/qwindowspipewriter_p.h

@@ -109,6 +109,7 @@ public:
     qint64 write(const char *data, qint64 maxlen);
     void stop();
     bool waitForWrite(int msecs);
+    bool isWriteOperationActive() const { return writeSequenceStarted; }
     qint64 bytesToWrite() const;
 
 Q_SIGNALS:

src/network/socket/qlocalsocket.h

@@ -124,7 +124,7 @@ private:
     Q_PRIVATE_SLOT(d_func(), void _q_stateChanged(QAbstractSocket::SocketState))
     Q_PRIVATE_SLOT(d_func(), void _q_error(QAbstractSocket::SocketError))
 #elif defined(Q_OS_WIN)
-    Q_PRIVATE_SLOT(d_func(), void _q_canWrite())
+    Q_PRIVATE_SLOT(d_func(), void _q_bytesWritten(qint64))
     Q_PRIVATE_SLOT(d_func(), void _q_pipeClosed())
     Q_PRIVATE_SLOT(d_func(), void _q_winError(ulong, const QString &))
 #else

src/network/socket/qlocalsocket_p.h

@@ -55,6 +55,7 @@
 #if defined(QT_LOCALSOCKET_TCP)
 #   include "qtcpsocket.h"
 #elif defined(Q_OS_WIN)
+#   include <private/qringbuffer_p.h>
 #   include "private/qwindowspipereader_p.h"
 #   include "private/qwindowspipewriter_p.h"
 #   include <qwineventnotifier.h>
@@ -123,10 +124,12 @@ public:
     ~QLocalSocketPrivate();
     void destroyPipeHandles();
     void setErrorString(const QString &function);
-    void _q_canWrite();
+    void startNextWrite();
+    void _q_bytesWritten(qint64 bytes);
     void _q_pipeClosed();
     void _q_winError(ulong windowsError, const QString &function);
     HANDLE handle;
+    QRingBuffer writeBuffer;
     QWindowsPipeWriter *pipeWriter;
     QWindowsPipeReader *pipeReader;
     QLocalSocket::LocalSocketError error;

src/network/socket/qlocalsocket_win.cpp

@@ -207,15 +207,21 @@ qint64 QLocalSocket::readData(char *data, qint64 maxSize)
     }
 }
 
-qint64 QLocalSocket::writeData(const char *data, qint64 maxSize)
+qint64 QLocalSocket::writeData(const char *data, qint64 len)
 {
     Q_D(QLocalSocket);
+    if (len == 0)
+        return 0;
+    char *dest = d->writeBuffer.reserve(len);
+    memcpy(dest, data, len);
     if (!d->pipeWriter) {
         d->pipeWriter = new QWindowsPipeWriter(d->handle, this);
-        connect(d->pipeWriter, SIGNAL(canWrite()), this, SLOT(_q_canWrite()));
-        connect(d->pipeWriter, SIGNAL(bytesWritten(qint64)), this, SIGNAL(bytesWritten(qint64)));
+        QObjectPrivate::connect(d->pipeWriter, &QWindowsPipeWriter::bytesWritten,
+                                d, &QLocalSocketPrivate::_q_bytesWritten);
     }
-    return d->pipeWriter->write(data, maxSize);
+    if (!d->pipeWriter->isWriteOperationActive())
+        d->startNextWrite();
+    return len;
 }
 
 void QLocalSocket::abort()
@@ -224,6 +230,7 @@ void QLocalSocket::abort()
     if (d->pipeWriter) {
         delete d->pipeWriter;
         d->pipeWriter = 0;
+        d->writeBuffer.clear();
     }
     close();
 }
@@ -266,7 +273,7 @@ qint64 QLocalSocket::bytesAvailable() const
 qint64 QLocalSocket::bytesToWrite() const
 {
     Q_D(const QLocalSocket);
-    return (d->pipeWriter) ? d->pipeWriter->bytesToWrite() : 0;
+    return d->writeBuffer.size();
 }
 
 bool QLocalSocket::canReadLine() const
@@ -298,9 +305,12 @@ void QLocalSocket::close()
 bool QLocalSocket::flush()
 {
     Q_D(QLocalSocket);
-    if (d->pipeWriter)
-        return d->pipeWriter->waitForWrite(0);
-    return false;
+    bool written = false;
+    if (d->pipeWriter) {
+        while (d->pipeWriter->waitForWrite(0))
+            written = true;
+    }
+    return written;
 }
 
 void QLocalSocket::disconnectFromServer()
@@ -313,10 +323,11 @@ void QLocalSocket::disconnectFromServer()
         // It must be destroyed before close() to prevent an infinite loop.
         delete d->pipeWriter;
         d->pipeWriter = 0;
+        d->writeBuffer.clear();
     }
 
     flush();
-    if (d->pipeWriter && d->pipeWriter->bytesToWrite() != 0) {
+    if (bytesToWrite() != 0) {
         d->state = QLocalSocket::ClosingState;
         emit stateChanged(d->state);
     } else {
@@ -345,11 +356,24 @@ bool QLocalSocket::setSocketDescriptor(qintptr socketDescriptor,
     return true;
 }
 
-void QLocalSocketPrivate::_q_canWrite()
+void QLocalSocketPrivate::startNextWrite()
 {
     Q_Q(QLocalSocket);
-    if (state == QLocalSocket::ClosingState)
-        q->close();
+    if (writeBuffer.isEmpty()) {
+        if (state == QLocalSocket::ClosingState)
+            q->close();
+    } else {
+        Q_ASSERT(pipeWriter);
+        pipeWriter->write(writeBuffer.readPointer(), writeBuffer.nextDataBlockSize());
+    }
+}
+
+void QLocalSocketPrivate::_q_bytesWritten(qint64 bytes)
+{
+    Q_Q(QLocalSocket);
+    writeBuffer.free(bytes);
+    startNextWrite();
+    emit q->bytesWritten(bytes);
 }
 
 qintptr QLocalSocket::socketDescriptor() const

tests/auto/network/socket/qlocalsocket/tst_qlocalsocket.cpp

@@ -102,7 +102,10 @@ private slots:
 
     void multiConnect();
     void writeOnlySocket();
+
+    void writeToClientAndDisconnect_data();
     void writeToClientAndDisconnect();
+
     void debug();
     void bytesWrittenSignal();
     void syncDisconnectNotify();
@@ -1034,8 +1037,16 @@ void tst_QLocalSocket::writeOnlySocket()
     QCOMPARE(client.state(), QLocalSocket::ConnectedState);
 }
 
+void tst_QLocalSocket::writeToClientAndDisconnect_data()
+{
+    QTest::addColumn<int>("chunks");
+    QTest::newRow("one chunk") << 1;
+    QTest::newRow("several chunks") << 20;
+}
+
 void tst_QLocalSocket::writeToClientAndDisconnect()
 {
+    QFETCH(int, chunks);
     QLocalServer server;
     QLocalSocket client;
     QSignalSpy readChannelFinishedSpy(&client, SIGNAL(readChannelFinished()));
@@ -1049,14 +1060,17 @@ void tst_QLocalSocket::writeToClientAndDisconnect()
 
     char buffer[100];
     memset(buffer, 0, sizeof(buffer));
-    QCOMPARE(clientSocket->write(buffer, sizeof(buffer)), (qint64)sizeof(buffer));
-    clientSocket->waitForBytesWritten();
+    for (int i = 0; i < chunks; ++i)
+        QCOMPARE(clientSocket->write(buffer, sizeof(buffer)), qint64(sizeof(buffer)));
+    while (clientSocket->bytesToWrite())
+        QVERIFY(clientSocket->waitForBytesWritten());
     clientSocket->close();
     server.close();
 
     client.waitForDisconnected();
     QCOMPARE(readChannelFinishedSpy.count(), 1);
-    QCOMPARE(client.read(buffer, sizeof(buffer)), (qint64)sizeof(buffer));
+    const QByteArray received = client.readAll();
+    QCOMPARE(received.size(), qint64(sizeof(buffer) * chunks));
     QCOMPARE(client.state(), QLocalSocket::UnconnectedState);
 }