[frameworks/kio] /: KIO: hand what a message carries over as it is within a process
Méven Car <[email protected]>
| Newsgroups | gmane.comp.kde.cvs |
|---|---|
| Message-ID | <[email protected]> |
Git commit 94c9b2843064d7470ba785e77866196158ebc726 by Méven Car.
Committed on 18/08/2026 at 09:58.
Pushed by meven into branch 'master'.
KIO: hand what a message carries over as it is within a process
A worker running in a thread of the application, which is how kio_file and
kio_admin run, wrote what every message carries down with QDataStream and the
application read it back, over memory both ends already reach. The application
wrote what a command carries down for the worker the same way.
A task now holds one thing, what the message carries: the bytes, a number, a
string, a url, an entry, the entries of a listing, the meta data, or an error.
ConnectionBackend::sendPayload() writes a kind down and sends the bytes, which is
what a peer in another process reads back, and the backend whose peer shares this
process hands it over as it is. The listing, the stat, the sizes, the mime type,
the meta data, the redirection, the resume offset and the error go this way, and
so do the url of a get, a stat, a listing, a mime type query or a free space
query, and the meta data every job sends ahead of its command. A message box and
an ssl error stay written down. Nothing changes on the wire. The limit of what
one message can carry moved to the socket, whose header holds it.
The packed arguments of a job are that payload rather than a byte array beside
it, so a job which knows its one argument hands it over and a job which packs
several values still packs them. DataWorker, which runs inside the application,
reads either.
A socket reads what a command carries into the buffer that command is collected
in, rather than into a buffer of its own that is then copied in. Without that,
a peer in another process pays one allocation per message for the copy, which it
did not pay before.
Counted with heaptrack over a whole run: listing 22974 files asks the allocator
636248 times before and 535410 after. 4000 stats, 560682 before, 524178 with the
worker's side alone converted, 420188 with both. Reading 256 MB through get is
unchanged at about 70000, those bytes already travelled as they are.
Copying 1000 files of 256 bytes, best of ten per round over fifteen pinned
rounds: in this process 166.6 ms before and 158.7 ms after, faster in all
fifteen. Through a socket, 206.9 ms before and 207.3 ms after, which the buffer
above is what keeps level.
M +3 -3 autotests/connectionbackendtest.cpp
M +81 -8 autotests/threadconnectionbackendtest.cpp
M +33 -13 src/core/connection.cpp
M +13 -1 src/core/connection_p.h
M +37 -3 src/core/connectionbackend.cpp
M +52 -1 src/core/connectionbackend_p.h
M +13 -3 src/core/dataworker.cpp
M +1 -0 src/core/dataworker_p.h
M +7 -4 src/core/davjob.cpp
M +3 -3 src/core/filesystemfreespacejob.cpp
M +1 -1 src/core/job.cpp
M +12 -8 src/core/job_p.h
M +2 -5 src/core/listjob.cpp
M +4 -6 src/core/mimetypejob.cpp
M +5 -4 src/core/mkdirjob.cpp
M +2 -3 src/core/simplejob.cpp
M +40 -48 src/core/slavebase.cpp
M +4 -0 src/core/slavebase.h
M +19 -5 src/core/socketconnectionbackend.cpp
M +1 -0 src/core/socketconnectionbackend_p.h
M +2 -2 src/core/specialjob.cpp
M +4 -7 src/core/statjob.cpp
M +2 -2 src/core/storedtransferjob.cpp
M +15 -2 src/core/threadconnectionbackend.cpp
M +5 -0 src/core/threadconnectionbackend_p.h
M +7 -10 src/core/transferjob.cpp
M +5 -0 src/core/worker.cpp
M +5 -0 src/core/worker_p.h
M +29 -32 src/core/workerinterface.cpp
M +3 -0 src/core/workerinterface_p.h
https://invent.kde.org/frameworks/kio/-/commit/94c9b2843064d7470ba785e77866196158ebc726
diff --git a/autotests/connectionbackendtest.cpp b/autotests/connectionbackendtest.cpp
index 71aa344c6f..c178f0abad 100644
--- a/autotests/connectionbackendtest.cpp
+++ b/autotests/connectionbackendtest.cpp
@@ -41,7 +41,7 @@ private Q_SLOTS:
QVERIFY(!spy->isEmpty());
auto task = spy->at(0).at(0).value<KIO::Task>();
- QCOMPARE(task.data.size(), data.size());
+ QCOMPARE(task.bytes().size(), data.size());
}
// Resuming a backend whose socket is not open must stay quiet. connectToRemote sets the
@@ -111,7 +111,7 @@ private Q_SLOTS:
QVERIFY(sendOk);
auto task = spy.at(0).at(0).value<KIO::Task>();
- QCOMPARE(task.data, data);
+ QCOMPARE(task.bytes(), data);
}
// Commands keep flowing intact and in order across repeated suspend and
@@ -157,7 +157,7 @@ private Q_SLOTS:
QCOMPARE(spy.size(), payloads.size());
for (qsizetype i = 0; i < payloads.size(); ++i) {
auto task = spy.at(i).at(0).value<KIO::Task>();
- QCOMPARE(task.data, payloads.at(i));
+ QCOMPARE(task.bytes(), payloads.at(i));
}
}
diff --git a/autotests/threadconnectionbackendtest.cpp b/autotests/threadconnectionbackendtest.cpp
index 433758e94a..69e759bdd1 100644
--- a/autotests/threadconnectionbackendtest.cpp
+++ b/autotests/threadconnectionbackendtest.cpp
@@ -51,8 +51,25 @@ private Q_SLOTS:
void testOwnedPayloadSurvivesSenderScope();
void testManyTasksPreserveOrderUnderBackPressure();
void testWorkerCloseDisconnectsApplication();
+ void testEntriesTravelAsObjects();
+ void testEntriesBuiltOnTheWorkerThread();
};
+// The entries of a listing, of the shape kio_file sends.
+static UDSEntryList makeEntries(int count)
+{
+ UDSEntryList entries;
+ entries.reserve(count);
+ for (int i = 0; i < count; ++i) {
+ UDSEntry entry;
+ entry.fastInsert(UDSEntry::UDS_NAME, QStringLiteral("file%1").arg(i));
+ entry.fastInsert(UDSEntry::UDS_FILE_TYPE, S_IFREG);
+ entry.fastInsert(UDSEntry::UDS_SIZE, i);
+ entries.append(entry);
+ }
+ return entries;
+}
+
void ThreadConnectionBackendTest::testApplicationReceivesFromWorker()
{
auto [appBackend, workerBackend] = ThreadConnectionBackend::createPair();
@@ -71,9 +88,9 @@ void ThreadConnectionBackendTest::testApplicationReceivesFromWorker()
QTRY_COMPARE(sink.count(), 3); // delivered via the application event loop
const QList<Task> tasks = sink.tasks();
QCOMPARE(tasks.at(0).cmd, 1);
- QCOMPARE(tasks.at(0).data, QByteArrayLiteral("a"));
+ QCOMPARE(tasks.at(0).bytes(), QByteArrayLiteral("a"));
QCOMPARE(tasks.at(2).cmd, 3);
- QCOMPARE(tasks.at(2).data, QByteArrayLiteral("ccc"));
+ QCOMPARE(tasks.at(2).bytes(), QByteArrayLiteral("ccc"));
}
void ThreadConnectionBackendTest::testSuspendedApplicationBuffersUntilResumed()
@@ -101,11 +118,11 @@ void ThreadConnectionBackendTest::testSuspendedApplicationBuffersUntilResumed()
QTRY_COMPARE(sink.count(), 3);
const QList<Task> tasks = sink.tasks();
QCOMPARE(tasks.at(0).cmd, 1);
- QCOMPARE(tasks.at(0).data, QByteArrayLiteral("a"));
+ QCOMPARE(tasks.at(0).bytes(), QByteArrayLiteral("a"));
QCOMPARE(tasks.at(1).cmd, 2);
- QCOMPARE(tasks.at(1).data, QByteArrayLiteral("bb"));
+ QCOMPARE(tasks.at(1).bytes(), QByteArrayLiteral("bb"));
QCOMPARE(tasks.at(2).cmd, 3);
- QCOMPARE(tasks.at(2).data, QByteArrayLiteral("ccc"));
+ QCOMPARE(tasks.at(2).bytes(), QByteArrayLiteral("ccc"));
}
void ThreadConnectionBackendTest::testWorkerReceivesFromApplication()
@@ -139,7 +156,7 @@ void ThreadConnectionBackendTest::testWorkerReceivesFromApplication()
const QList<Task> tasks = sink.tasks();
QCOMPARE(tasks.at(0).cmd, 10);
- QCOMPARE(tasks.at(1).data, QByteArrayLiteral("yy"));
+ QCOMPARE(tasks.at(1).bytes(), QByteArrayLiteral("yy"));
}
void ThreadConnectionBackendTest::testOwnedPayloadSurvivesSenderScope()
@@ -161,7 +178,7 @@ void ThreadConnectionBackendTest::testOwnedPayloadSurvivesSenderScope()
QTRY_COMPARE(sink.count(), 1);
QCOMPARE(sink.tasks().at(0).cmd, 7);
- QCOMPARE(sink.tasks().at(0).data, QByteArray(64, 'z'));
+ QCOMPARE(sink.tasks().at(0).bytes(), QByteArray(64, 'z'));
}
void ThreadConnectionBackendTest::testManyTasksPreserveOrderUnderBackPressure()
@@ -190,7 +207,7 @@ void ThreadConnectionBackendTest::testManyTasksPreserveOrderUnderBackPressure()
const QList<Task> tasks = sink.tasks();
for (int i = 0; i < total; ++i) {
QCOMPARE(tasks.at(i).cmd, i);
- QCOMPARE(tasks.at(i).data, QByteArray::number(i));
+ QCOMPARE(tasks.at(i).bytes(), QByteArray::number(i));
}
}
@@ -207,6 +224,62 @@ void ThreadConnectionBackendTest::testWorkerCloseDisconnectsApplication()
QCOMPARE(appBackend->state, ConnectionBackend::Idle);
}
+void ThreadConnectionBackendTest::testEntriesTravelAsObjects()
+{
+ auto [appBackend, workerBackend] = ThreadConnectionBackend::createPair();
+
+ TaskSink sink;
+ connect(appBackend.get(), &ConnectionBackend::commandReceived, this, [&sink](const Task &task) {
+ sink.add(task);
+ });
+
+ QVERIFY(workerBackend->sendPayload(42, makeEntries(3)));
+
+ QTRY_COMPARE(sink.count(), 1);
+ const Task task = sink.tasks().constFirst();
+ QCOMPARE(task.cmd, 42);
+ QVERIFY(task.bytes().isEmpty()); // nothing was written down, the entries came as they are
+ const UDSEntryList *entries = std::get_if<UDSEntryList>(&task.payload);
+ QVERIFY(entries);
+ QCOMPARE(entries->count(), 3);
+ QCOMPARE(entries->at(2).stringValue(UDSEntry::UDS_NAME), QStringLiteral("file2"));
+ QCOMPARE(entries->at(2).numberValue(UDSEntry::UDS_SIZE), 2);
+}
+
+void ThreadConnectionBackendTest::testEntriesBuiltOnTheWorkerThread()
+{
+ // What a listing does: the entries are built on the worker thread and read on the
+ // application thread, so whatever an entry holds must be safe to hand over that way.
+ auto [appBackend, workerBackend] = ThreadConnectionBackend::createPair();
+
+ TaskSink sink;
+ connect(appBackend.get(), &ConnectionBackend::commandReceived, this, [&sink](const Task &task) {
+ sink.add(task);
+ });
+
+ const int batches = 20;
+ QThread *workerThread = QThread::create([worker = workerBackend.get()] {
+ for (int i = 0; i < batches; ++i) {
+ worker->sendPayload(i, makeEntries(100));
+ }
+ });
+ workerThread->start();
+
+ QTRY_COMPARE_WITH_TIMEOUT(sink.count(), batches, 30000);
+ QVERIFY(workerThread->wait(5000));
+ delete workerThread;
+
+ const QList<Task> tasks = sink.tasks();
+ for (int i = 0; i < batches; ++i) {
+ QCOMPARE(tasks.at(i).cmd, i);
+ const UDSEntryList *entries = std::get_if<UDSEntryList>(&tasks.at(i).payload);
+ QVERIFY(entries);
+ QCOMPARE(entries->count(), 100);
+ QCOMPARE(entries->at(99).stringValue(UDSEntry::UDS_NAME), QStringLiteral("file99"));
+ QCOMPARE(entries->at(99).numberValue(UDSEntry::UDS_SIZE), 99);
+ }
+}
+
QTEST_MAIN(ThreadConnectionBackendTest)
#include "threadconnectionbackendtest.moc"
diff --git a/src/core/connection.cpp b/src/core/connection.cpp
index 6c95fa3dbc..0c3e906abc 100644
--- a/src/core/connection.cpp
+++ b/src/core/connection.cpp
@@ -12,7 +12,9 @@
#include "connectionbackend_p.h"
#include "kiocoredebug.h"
#include "socketconnectionbackend_p.h"
+#include <QDataStream>
#include <QDebug>
+#include <QIODevice>
#include <cerrno>
@@ -25,7 +27,7 @@ void ConnectionPrivate::dequeue()
}
for (const Task &task : std::as_const(outgoingTasks)) {
- q->sendnow(task.cmd, task.data);
+ q->sendnow(task.cmd, task.payload);
}
outgoingTasks.clear();
@@ -174,25 +176,40 @@ bool Connection::send(int cmd, const QByteArray &data)
return false;
}
if (!inited() || !d->outgoingTasks.isEmpty()) {
- Task task;
- task.cmd = cmd;
- task.data = data;
- d->outgoingTasks.append(std::move(task));
+ d->outgoingTasks.append(Task{.cmd = cmd, .payload = data});
return true;
} else {
return sendnow(cmd, data);
}
}
-bool Connection::sendnow(int cmd, const QByteArray &data)
+bool Connection::send(int cmd, const TaskPayload &payload)
{
- if (!d->backend) {
- qCWarning(KIO_CORE) << "Connection::sendnow has no backend";
+ if (m_type == Type::Worker && !inited()) {
+ qCWarning(KIO_CORE) << "Connection::send() called with connection not inited";
+ return false;
+ }
+ if (!inited() || !d->outgoingTasks.isEmpty()) {
+ d->outgoingTasks.append(Task{.cmd = cmd, .payload = payload});
+ return true;
+ }
+ return sendnow(cmd, payload);
+}
+
+bool Connection::sendnow(int cmd, const TaskPayload &payload)
+{
+ if (!isConnected()) {
+ qCWarning(KIO_CORE) << "Connection::sendnow not connected";
return false;
}
- if (data.size() > 0xffffff) {
- qCWarning(KIO_CORE) << "Connection::sendnow too much data";
+ return d->backend->sendPayload(cmd, payload);
+}
+
+bool Connection::sendnow(int cmd, const QByteArray &data)
+{
+ if (!d->backend) {
+ qCWarning(KIO_CORE) << "Connection::sendnow has no backend";
return false;
}
@@ -222,7 +239,7 @@ bool Connection::waitForIncomingTask(int ms)
return false;
}
-int Connection::read(int *_cmd, QByteArray &data)
+int Connection::read(int *_cmd, QByteArray &data, TaskPayload *payload)
{
// if it's still empty, then it's an error
if (d->incomingTasks.isEmpty()) {
@@ -230,9 +247,12 @@ int Connection::read(int *_cmd, QByteArray &data)
return -1;
}
const Task &task = d->incomingTasks.constFirst();
- // qDebug() << this << "Command" << task.cmd << "removed from the queue (size" << task.data.size() << ")";
+ // qDebug() << this << "Command" << task.cmd << "removed from the queue";
*_cmd = task.cmd;
- data = task.data;
+ data = task.bytes();
+ if (payload) {
+ *payload = task.payload;
+ }
d->incomingTasks.removeFirst();
diff --git a/src/core/connection_p.h b/src/core/connection_p.h
index dbec14cf15..f5bc47a12f 100644
--- a/src/core/connection_p.h
+++ b/src/core/connection_p.h
@@ -77,6 +77,12 @@ public:
*/
bool send(int cmd, const QByteArray &arr = QByteArray());
+ /*!
+ * Queues what a message carries. It is handed over as it is when the peer lives in another
+ * thread of this process, and written down otherwise.
+ */
+ bool send(int cmd, const TaskPayload &payload);
+
/*!
* Sends the given command immediately.
* \a _cmd the command to set
@@ -85,6 +91,12 @@ public:
*/
bool sendnow(int _cmd, const QByteArray &data);
+ /*!
+ * Sends what a message carries. It is handed over as it is when the peer lives in another thread
+ * of this process, and written down otherwise.
+ */
+ bool sendnow(int _cmd, const TaskPayload &payload);
+
/*!
* Returns true if there are packets to be read immediately,
* false if waitForIncomingTask must be called before more data
@@ -109,7 +121,7 @@ public:
* Returns >=0 indicates the received data size upon success
* -1 indicates error
*/
- int read(int *_cmd, QByteArray &data);
+ int read(int *_cmd, QByteArray &data, TaskPayload *payload = nullptr);
/*!
* Don't handle incoming data until resumed.
diff --git a/src/core/connectionbackend.cpp b/src/core/connectionbackend.cpp
index d165a58e78..0d6d53966c 100644
--- a/src/core/connectionbackend.cpp
+++ b/src/core/connectionbackend.cpp
@@ -10,8 +10,42 @@
#include "connectionbackend_p.h"
-// ConnectionBackend is an abstract base. This translation unit only exists to compile its
-// meta-object (signals). The concrete backends live in socketconnectionbackend.cpp and
-// threadconnectionbackend.cpp.
+#include <QDataStream>
+#include <QIODevice>
+
+// ConnectionBackend is an abstract base. The concrete backends live in
+// socketconnectionbackend.cpp and threadconnectionbackend.cpp.
+
+using namespace KIO;
+
+// What a message carries, written down. This is what a peer in another process is given, and what
+// it reads back, so every kind is written the way the sender used to write it.
+bool ConnectionBackend::sendPayload(int command, const TaskPayload &payload)
+{
+ // Bytes are what the peer is given, so they go as they are and nothing is allocated for them.
+ if (const QByteArray *bytes = std::get_if<QByteArray>(&payload)) {
+ return sendCommand(command, *bytes);
+ }
+
+ QByteArray data;
+ QDataStream stream(&data, QIODevice::WriteOnly);
+ std::visit(
+ [&stream](const auto &value) {
+ using Type = std::decay_t<decltype(value)>;
+ if constexpr (std::is_same_v<Type, std::monostate> || std::is_same_v<Type, QByteArray>) {
+ // Nothing to write, the bytes went above.
+ } else if constexpr (std::is_same_v<Type, UDSEntryList>) {
+ for (const UDSEntry &entry : value) {
+ stream << entry;
+ }
+ } else if constexpr (std::is_same_v<Type, TaskError>) {
+ stream << value.code << value.text;
+ } else {
+ stream << value;
+ }
+ },
+ payload);
+ return sendCommand(command, data);
+}
#include "moc_connectionbackend_p.cpp"
diff --git a/src/core/connectionbackend_p.h b/src/core/connectionbackend_p.h
index d31ad8bd58..838969dc8a 100644
--- a/src/core/connectionbackend_p.h
+++ b/src/core/connectionbackend_p.h
@@ -9,14 +9,58 @@
#ifndef KIO_CONNECTIONBACKEND_P_H
#define KIO_CONNECTIONBACKEND_P_H
+#include "metadata.h"
+#include "udsentry.h"
+
+#include <QDataStream>
#include <QObject>
#include <QUrl>
+#include <utility>
+#include <variant>
+
namespace KIO
{
+// The two values an error carries.
+struct TaskError {
+ qint32 code = 0;
+ QString text;
+};
+
+// What a task carries. A peer sharing this process is handed the objects themselves, a peer in
+// another process is handed the bytes they are written to, which is all a socket can carry, so the
+// bytes are one of the kinds. std::monostate is a message carrying nothing.
+using TaskPayload = std::variant<std::monostate, QByteArray, quint32, quint64, QString, QUrl, UDSEntry, UDSEntryList, MetaData, TaskError>;
+
+// The bytes a payload carries, which is what it carries whenever it came off a socket. Empty when it
+// carries an object instead.
+inline QByteArray payloadBytes(const TaskPayload &payload)
+{
+ const QByteArray *value = std::get_if<QByteArray>(&payload);
+ return value ? *value : QByteArray();
+}
+
+// What a message carries: the object the peer handed over, when it shares this process, and what the
+// bytes it sent stand for otherwise. Taking the object leaves the payload empty of it.
+template<typename T>
+T carried(TaskPayload &payload, QDataStream &stream)
+{
+ if (T *value = std::get_if<T>(&payload)) {
+ return std::exchange(*value, T{});
+ }
+ T value{};
+ stream >> value;
+ return value;
+}
+
struct Task {
int cmd = -1;
- QByteArray data{};
+ TaskPayload payload{};
+
+ QByteArray bytes() const
+ {
+ return payloadBytes(payload);
+ }
};
/*!
@@ -52,6 +96,13 @@ public:
virtual bool waitForIncomingTask(int ms) = 0;
virtual bool sendCommand(int command, const QByteArray &data) = 0;
+ /*!
+ * Hands what a message carries to the peer. This writes it down and sends the bytes, which is
+ * what a peer in another process can be given, and bytes are sent as they are. A backend whose
+ * peer shares this process overrides this to hand every kind over as it is.
+ */
+ virtual bool sendPayload(int command, const TaskPayload &payload);
+
Q_SIGNALS:
void disconnected();
void commandReceived(const KIO::Task &task);
diff --git a/src/core/dataworker.cpp b/src/core/dataworker.cpp
index 573e8ea857..20de1ee344 100644
--- a/src/core/dataworker.cpp
+++ b/src/core/dataworker.cpp
@@ -126,18 +126,28 @@ void DataWorker::dispatchNext()
void DataWorker::send(int cmd, const QByteArray &arr)
{
- QDataStream stream(arr);
+ send(cmd, TaskPayload{arr});
+}
+
+// This worker runs in the application itself, so a command reaches it without a transport under it
+// and carries whatever the job handed over.
+void DataWorker::send(int cmd, const TaskPayload &payload)
+{
+ QDataStream stream(payloadBytes(payload));
QUrl url;
+ if (const QUrl *carried = std::get_if<QUrl>(&payload)) {
+ url = *carried;
+ } else {
+ stream >> url;
+ }
switch (cmd) {
case CMD_GET: {
- stream >> url;
get(url);
break;
}
case CMD_MIMETYPE: {
- stream >> url;
mimetype(url);
break;
}
diff --git a/src/core/dataworker_p.h b/src/core/dataworker_p.h
index 9eea184a45..50d620f2a4 100644
--- a/src/core/dataworker_p.h
+++ b/src/core/dataworker_p.h
@@ -46,6 +46,7 @@ public:
void resume() override;
bool suspended() override;
void send(int cmd, const QByteArray &arr = QByteArray()) override;
+ void send(int cmd, const TaskPayload &payload) override;
// pure virtual methods that are defined by the actual protocol
virtual void get(const QUrl &url) = 0;
diff --git a/src/core/davjob.cpp b/src/core/davjob.cpp
index 783d8eb737..7abee84225 100644
--- a/src/core/davjob.cpp
+++ b/src/core/davjob.cpp
@@ -46,8 +46,10 @@ DavJob::DavJob(DavJobPrivate &dd, int method, const QString &request)
// We couldn't set the args when calling the parent constructor,
// so do it now.
Q_D(DavJob);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
+ QByteArray args;
+ QDataStream stream(&args, QIODevice::WriteOnly);
stream << (int)7 << d->m_url << method;
+ d->m_packedArgs = args;
// Same for static data
if (!request.isEmpty()) {
if (!request.startsWith(QLatin1StringView("<?xml"))) {
@@ -85,7 +87,7 @@ void DavJob::slotFinished()
Q_D(DavJob);
// qDebug() << d->str_response;
if (!d->m_redirectionURL.isEmpty() && d->m_redirectionURL.isValid() && (d->m_command == CMD_SPECIAL)) {
- QDataStream istream(d->m_packedArgs);
+ QDataStream istream(payloadBytes(d->m_packedArgs));
int s_cmd;
int s_method;
qint64 s_size;
@@ -96,9 +98,10 @@ void DavJob::slotFinished()
istream >> s_size;
// PROPFIND
if ((s_cmd == 7) && (s_method == (int)KIO::DAV_PROPFIND)) {
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
+ QByteArray args;
+ QDataStream stream(&args, QIODevice::WriteOnly);
stream << (int)7 << d->m_redirectionURL << (int)KIO::DAV_PROPFIND << s_size;
+ d->m_packedArgs = args;
}
}
TransferJob::slotFinished();
diff --git a/src/core/filesystemfreespacejob.cpp b/src/core/filesystemfreespacejob.cpp
index 588d097db6..6e279753a0 100644
--- a/src/core/filesystemfreespacejob.cpp
+++ b/src/core/filesystemfreespacejob.cpp
@@ -17,7 +17,7 @@ using namespace KIO;
class KIO::FileSystemFreeSpaceJobPrivate : public SimpleJobPrivate
{
public:
- FileSystemFreeSpaceJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ FileSystemFreeSpaceJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: SimpleJobPrivate(url, command, packedArgs)
{
}
@@ -32,7 +32,7 @@ public:
Q_DECLARE_PUBLIC(FileSystemFreeSpaceJob)
- static inline FileSystemFreeSpaceJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs)
+ static inline FileSystemFreeSpaceJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs)
{
FileSystemFreeSpaceJob *job = new FileSystemFreeSpaceJob(*new FileSystemFreeSpaceJobPrivate(url, command, packedArgs));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
@@ -84,7 +84,7 @@ void FileSystemFreeSpaceJob::slotFinished()
KIO::FileSystemFreeSpaceJob *KIO::fileSystemFreeSpace(const QUrl &url)
{
- KIO_ARGS << url;
+ const TaskPayload packedArgs = url;
return FileSystemFreeSpaceJobPrivate::newJob(url, CMD_FILESYSTEMFREESPACE, packedArgs);
}
diff --git a/src/core/job.cpp b/src/core/job.cpp
index 7331637c66..19b8d847d7 100644
--- a/src/core/job.cpp
+++ b/src/core/job.cpp
@@ -260,7 +260,7 @@ MetaData Job::outgoingMetaData() const
class KIO::DirectCopyJobPrivate : public KIO::SimpleJobPrivate
{
public:
- DirectCopyJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ DirectCopyJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: SimpleJobPrivate(url, command, packedArgs)
{
}
diff --git a/src/core/job_p.h b/src/core/job_p.h
index 5283161d4f..6bf10017fa 100644
--- a/src/core/job_p.h
+++ b/src/core/job_p.h
@@ -13,6 +13,7 @@
#define KIO_JOB_P_H
#include "commands_p.h"
+#include "connectionbackend_p.h" // for KIO::TaskPayload
#include "global.h"
#include "jobtracker.h"
#include "kiocoredebug.h"
@@ -35,6 +36,7 @@
namespace KIO
{
+
static constexpr filesize_t invalidFilesize = static_cast<KIO::filesize_t>(-1);
// Exported for KIOWidgets jobs
@@ -107,7 +109,7 @@ public:
* \a command the command of the job
* \a packedArgs the arguments
*/
- SimpleJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ SimpleJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: m_worker(nullptr)
, m_packedArgs(packedArgs)
, m_url(url)
@@ -118,7 +120,9 @@ public:
}
QPointer<Worker> m_worker;
- QByteArray m_packedArgs;
+ // What the command carries. The packed bytes for most commands, the object itself for those a
+ // worker sharing this process can be handed as it is.
+ TaskPayload m_packedArgs;
QUrl m_url;
int m_command;
@@ -200,12 +204,12 @@ public:
{
return job->d_func();
}
- static inline SimpleJob *newJobNoUi(const QUrl &url, int command, const QByteArray &packedArgs)
+ static inline SimpleJob *newJobNoUi(const QUrl &url, int command, const TaskPayload &packedArgs)
{
SimpleJob *job = new SimpleJob(*new SimpleJobPrivate(url, command, packedArgs));
return job;
}
- static inline SimpleJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs, JobFlags flags = HideProgressInfo)
+ static inline SimpleJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs, JobFlags flags = HideProgressInfo)
{
SimpleJob *job = new SimpleJob(*new SimpleJobPrivate(url, command, packedArgs));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
@@ -219,7 +223,7 @@ public:
class TransferJobPrivate : public SimpleJobPrivate
{
public:
- inline TransferJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs, const QByteArray &_staticData)
+ inline TransferJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs, const QByteArray &_staticData)
: SimpleJobPrivate(url, command, packedArgs)
, m_internalSuspended(false)
, staticData(_staticData)
@@ -228,7 +232,7 @@ public:
{
}
- inline TransferJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs, QIODevice *ioDevice)
+ inline TransferJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs, QIODevice *ioDevice)
: SimpleJobPrivate(url, command, packedArgs)
, m_internalSuspended(false)
, m_isMimetypeEmitted(false)
@@ -274,7 +278,7 @@ public:
void slotPostRedirection();
Q_DECLARE_PUBLIC(TransferJob)
- static inline TransferJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs, const QByteArray &_staticData, JobFlags flags)
+ static inline TransferJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs, const QByteArray &_staticData, JobFlags flags)
{
TransferJob *job = new TransferJob(*new TransferJobPrivate(url, command, packedArgs, _staticData));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
@@ -285,7 +289,7 @@ public:
return job;
}
- static inline TransferJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs, QIODevice *ioDevice, JobFlags flags)
+ static inline TransferJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs, QIODevice *ioDevice, JobFlags flags)
{
TransferJob *job = new TransferJob(*new TransferJobPrivate(url, command, packedArgs, ioDevice));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
diff --git a/src/core/listjob.cpp b/src/core/listjob.cpp
index e57cbb63e1..e9f60286d9 100644
--- a/src/core/listjob.cpp
+++ b/src/core/listjob.cpp
@@ -75,8 +75,7 @@ ListJob::ListJob(ListJobPrivate &dd)
Q_D(ListJob);
// We couldn't set the args when calling the parent constructor,
// so do it now.
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_url;
+ d->m_packedArgs = d->m_url;
}
ListJob::~ListJob()
@@ -251,9 +250,7 @@ void ListJob::slotFinished()
}
if (d->m_redirectionHandlingEnabled) {
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_redirectionURL;
+ d->m_packedArgs = d->m_redirectionURL;
d->restartAfterRedirection(&d->m_redirectionURL);
return;
diff --git a/src/core/mimetypejob.cpp b/src/core/mimetypejob.cpp
index 0afc77000b..37af1bca82 100644
--- a/src/core/mimetypejob.cpp
+++ b/src/core/mimetypejob.cpp
@@ -14,14 +14,14 @@ using namespace KIO;
class KIO::MimetypeJobPrivate : public KIO::TransferJobPrivate
{
public:
- MimetypeJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ MimetypeJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: TransferJobPrivate(url, command, packedArgs, QByteArray())
{
}
Q_DECLARE_PUBLIC(MimetypeJob)
- static inline MimetypeJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs, JobFlags flags)
+ static inline MimetypeJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs, JobFlags flags)
{
MimetypeJob *job = new MimetypeJob(*new MimetypeJobPrivate(url, command, packedArgs));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
@@ -66,9 +66,7 @@ void MimetypeJob::slotFinished()
if (d->m_redirectionHandlingEnabled) {
d->staticData.truncate(0);
d->m_internalSuspended = false;
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_redirectionURL;
+ d->m_packedArgs = d->m_redirectionURL;
d->restartAfterRedirection(&d->m_redirectionURL);
return;
@@ -81,7 +79,7 @@ void MimetypeJob::slotFinished()
MimetypeJob *KIO::mimetype(const QUrl &url, JobFlags flags)
{
- KIO_ARGS << url;
+ const TaskPayload packedArgs = url;
return MimetypeJobPrivate::newJob(url, CMD_MIMETYPE, packedArgs, flags);
}
diff --git a/src/core/mkdirjob.cpp b/src/core/mkdirjob.cpp
index 560ec2ff4b..9955626306 100644
--- a/src/core/mkdirjob.cpp
+++ b/src/core/mkdirjob.cpp
@@ -18,7 +18,7 @@ using namespace KIO;
class KIO::MkdirJobPrivate : public SimpleJobPrivate
{
public:
- MkdirJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ MkdirJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: SimpleJobPrivate(url, command, packedArgs)
{
}
@@ -91,12 +91,13 @@ void MkdirJob::slotFinished()
if (d->m_redirectionHandlingEnabled) {
QUrl dummyUrl;
int permissions;
- QDataStream istream(d->m_packedArgs);
+ QDataStream istream(payloadBytes(d->m_packedArgs));
istream >> dummyUrl >> permissions;
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
+ QByteArray args;
+ QDataStream stream(&args, QIODevice::WriteOnly);
stream << d->m_redirectionURL << permissions;
+ d->m_packedArgs = args;
d->restartAfterRedirection(&d->m_redirectionURL);
return;
diff --git a/src/core/simplejob.cpp b/src/core/simplejob.cpp
index 3cd55b8583..e580f78daf 100644
--- a/src/core/simplejob.cpp
+++ b/src/core/simplejob.cpp
@@ -163,8 +163,7 @@ void SimpleJobPrivate::start(Worker *worker)
}
if (!m_outgoingMetaData.isEmpty()) {
- KIO_ARGS << m_outgoingMetaData;
- worker->send(CMD_META_DATA, packedArgs);
+ worker->send(CMD_META_DATA, m_outgoingMetaData);
}
worker->send(m_command, m_packedArgs);
@@ -204,7 +203,7 @@ void SimpleJob::slotFinished()
} else { /*if ( m_command == CMD_RENAME )*/
QUrl src;
QUrl dst;
- QDataStream str(d->m_packedArgs);
+ QDataStream str(payloadBytes(d->m_packedArgs));
str >> src >> dst;
if (src.adjusted(QUrl::RemoveFilename) == dst.adjusted(QUrl::RemoveFilename) // For the user, moving isn't
// renaming. Only renaming is.
diff --git a/src/core/slavebase.cpp b/src/core/slavebase.cpp
index d1bbcbcb6d..dd4e95523a 100644
--- a/src/core/slavebase.cpp
+++ b/src/core/slavebase.cpp
@@ -110,6 +110,8 @@ public:
UDSEntryList pendingListEntries;
QElapsedTimer m_timeSinceLastBatch;
Connection appConnection{Connection::Type::Worker};
+ // What the last command carried, when the application shares this process with the worker.
+ TaskPayload incomingPayload;
bool isConnectedToApp;
QString slaveid;
@@ -320,7 +322,7 @@ void SlaveBase::dispatchLoop()
// dispatch application messages
int cmd;
QByteArray data;
- ret = d->appConnection.read(&cmd, data);
+ ret = d->appConnection.read(&cmd, data, &d->incomingPayload);
if (ret != -1) {
if (d->inOpenLoop) {
@@ -432,9 +434,7 @@ void SlaveBase::sendMetaData()
void SlaveBase::sendAndKeepMetaData()
{
if (!mOutgoingMetaData.isEmpty()) {
- KIO_DATA << mOutgoingMetaData;
-
- send(INF_META_DATA, data);
+ send(INF_META_DATA, mOutgoingMetaData);
}
}
@@ -499,9 +499,7 @@ void SlaveBase::error(int _errid, const QString &_text)
mIncomingMetaData.clear(); // Clear meta data
d->rebuildConfig();
mOutgoingMetaData.clear();
- KIO_DATA << static_cast<qint32>(_errid) << _text;
-
- send(MSG_ERROR, data);
+ send(MSG_ERROR, TaskError{static_cast<qint32>(_errid), _text});
// reset
d->totalSize = 0;
d->inOpenLoop = false;
@@ -573,8 +571,7 @@ void SlaveBase::canResume()
void SlaveBase::totalSize(KIO::filesize_t _bytes)
{
- KIO_DATA << static_cast<quint64>(_bytes);
- send(INF_TOTAL_SIZE, data);
+ send(INF_TOTAL_SIZE, static_cast<quint64>(_bytes));
// this one is usually called before the first item is listed in listDir()
d->totalSize = _bytes;
@@ -595,8 +592,7 @@ void SlaveBase::processedSize(KIO::filesize_t _bytes)
}
if (emitSignal) {
- KIO_DATA << static_cast<quint64>(_bytes);
- send(INF_PROCESSED_SIZE, data);
+ send(INF_PROCESSED_SIZE, static_cast<quint64>(_bytes));
d->lastTimeout.start();
}
@@ -605,20 +601,17 @@ void SlaveBase::processedSize(KIO::filesize_t _bytes)
void SlaveBase::written(KIO::filesize_t _bytes)
{
- KIO_DATA << static_cast<quint64>(_bytes);
- send(MSG_WRITTEN, data);
+ send(MSG_WRITTEN, static_cast<quint64>(_bytes));
}
void SlaveBase::position(KIO::filesize_t _pos)
{
- KIO_DATA << static_cast<quint64>(_pos);
- send(INF_POSITION, data);
+ send(INF_POSITION, static_cast<quint64>(_pos));
}
void SlaveBase::truncated(KIO::filesize_t _length)
{
- KIO_DATA << static_cast<quint64>(_length);
- send(INF_TRUNCATED, data);
+ send(INF_TRUNCATED, static_cast<quint64>(_length));
}
void SlaveBase::processedPercent(float /* percent */)
@@ -628,14 +621,12 @@ void SlaveBase::processedPercent(float /* percent */)
void SlaveBase::speed(unsigned long _bytes_per_second)
{
- KIO_DATA << static_cast<quint32>(_bytes_per_second);
- send(INF_SPEED, data);
+ send(INF_SPEED, static_cast<quint32>(_bytes_per_second));
}
void SlaveBase::redirection(const QUrl &_url)
{
- KIO_DATA << _url;
- send(INF_REDIRECTION, data);
+ send(INF_REDIRECTION, _url);
}
static bool isSubCommand(int cmd)
@@ -659,16 +650,15 @@ void SlaveBase::mimeType(const QString &_type)
// Send the meta-data each time we send the MIME type.
if (!mOutgoingMetaData.isEmpty()) {
qCDebug(KIO_CORE) << "sending mimetype meta data";
- KIO_DATA << mOutgoingMetaData;
- send(INF_META_DATA, data);
+ send(INF_META_DATA, mOutgoingMetaData);
}
- KIO_DATA << _type;
- send(INF_MIME_TYPE, data);
+ send(INF_MIME_TYPE, _type);
+ QByteArray data;
while (true) {
cmd = 0;
int ret = -1;
if (d->appConnection.hasTaskAvailable() || d->appConnection.waitForIncomingTask(-1)) {
- ret = d->appConnection.read(&cmd, data);
+ ret = d->appConnection.read(&cmd, data, &d->incomingPayload);
}
if (ret == -1) {
qCDebug(KIO_CORE) << "read error on app connection while sending mimetype";
@@ -705,20 +695,17 @@ void SlaveBase::exit() // possibly called from another thread, only use atomics
void SlaveBase::warning(const QString &_msg)
{
- KIO_DATA << _msg;
- send(INF_WARNING, data);
+ send(INF_WARNING, _msg);
}
void SlaveBase::infoMessage(const QString &_msg)
{
- KIO_DATA << _msg;
- send(INF_INFOMESSAGE, data);
+ send(INF_INFOMESSAGE, _msg);
}
void SlaveBase::statEntry(const UDSEntry &entry)
{
- KIO_DATA << entry;
- send(MSG_STAT_ENTRY, data);
+ send(MSG_STAT_ENTRY, entry);
}
void SlaveBase::listEntry(const UDSEntry &entry)
@@ -755,14 +742,11 @@ void SlaveBase::listEntry(const UDSEntry &entry)
void SlaveBase::listEntries(const UDSEntryList &list)
{
- QByteArray data;
- QDataStream stream(&data, QIODevice::WriteOnly);
-
- for (const UDSEntry &entry : list) {
- stream << entry;
+ // A worker running in a thread of the application hands the entries over as they are, which
+ // leaves them to be written down and read back only when the two are in different processes.
+ if (!d->appConnection.send(MSG_LIST_ENTRIES, list)) {
+ exit();
}
-
- send(MSG_LIST_ENTRIES, data);
}
static void sigpipe_handler(int)
@@ -993,9 +977,9 @@ bool SlaveBase::canResume(KIO::filesize_t offset)
{
// qDebug() << "offset=" << KIO::number(offset);
d->needSendCanResume = false;
- KIO_DATA << static_cast<quint64>(offset);
- send(MSG_RESUME, data);
+ send(MSG_RESUME, static_cast<quint64>(offset));
if (offset) {
+ QByteArray data;
int cmd;
if (waitForAnswer(CMD_RESUMEANSWER, CMD_NONE, data, &cmd) != -1) {
// qDebug() << "returning" << (cmd == CMD_RESUMEANSWER);
@@ -1014,7 +998,7 @@ int SlaveBase::waitForAnswer(int expected1, int expected2, QByteArray &data, int
int result = -1;
for (;;) {
if (d->appConnection.hasTaskAvailable() || d->appConnection.waitForIncomingTask(-1)) {
- result = d->appConnection.read(&cmd, data);
+ result = d->appConnection.read(&cmd, data, &d->incomingPayload);
}
if (result == -1) {
// qDebug() << "read error.";
@@ -1102,7 +1086,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
break;
}
case CMD_GET: {
- stream >> url;
+ url = carried<QUrl>(d->incomingPayload, stream);
d->m_state = d->InsideMethod;
get(url);
d->verifyState("get()");
@@ -1142,7 +1126,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
break;
}
case CMD_STAT: {
- stream >> url;
+ url = carried<QUrl>(d->incomingPayload, stream);
d->m_state = d->InsideMethod;
stat(url); // krazy:exclude=syscalls
d->verifyState("stat()");
@@ -1150,7 +1134,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
break;
}
case CMD_MIMETYPE: {
- stream >> url;
+ url = carried<QUrl>(d->incomingPayload, stream);
d->m_state = d->InsideMethod;
mimetype(url);
d->verifyState("mimetype()");
@@ -1158,7 +1142,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
break;
}
case CMD_LISTDIR: {
- stream >> url;
+ url = carried<QUrl>(d->incomingPayload, stream);
d->m_state = d->InsideMethod;
listDir(url);
d->verifyState("listDir()");
@@ -1261,7 +1245,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
}
case CMD_META_DATA: {
// qDebug() << "(" << getpid() << ") Incoming meta-data...";
- stream >> mIncomingMetaData;
+ mIncomingMetaData = carried<MetaData>(d->incomingPayload, stream);
d->rebuildConfig();
break;
}
@@ -1270,7 +1254,7 @@ void SlaveBase::dispatch(int command, const QByteArray &data)
break;
}
case CMD_FILESYSTEMFREESPACE: {
- stream >> url;
+ url = carried<QUrl>(d->incomingPayload, stream);
void *data = static_cast<void *>(&url);
@@ -1409,6 +1393,14 @@ void SlaveBase::setKillFlag()
d->wasKilled = true;
}
+void SlaveBase::send(int cmd, const TaskPayload &payload)
+{
+ if (!d->appConnection.send(cmd, payload)) {
+ qCWarning(KIO_CORE) << "An error occurred during write. The worker terminates now.";
+ exit();
+ }
+}
+
void SlaveBase::send(int cmd, const QByteArray &arr)
{
if (d->runInThread) {
diff --git a/src/core/slavebase.h b/src/core/slavebase.h
index efe1d026ca..57049769a4 100644
--- a/src/core/slavebase.h
+++ b/src/core/slavebase.h
@@ -7,6 +7,7 @@
#ifndef SLAVEBASE_H
#define SLAVEBASE_H
+#include "connectionbackend_p.h" // for KIO::TaskPayload
#include "job_base.h" // for KIO::JobFlags
#include <kio/authinfo.h>
#include <kio/global.h>
@@ -936,6 +937,9 @@ private:
// This helps catching missing tr()/i18n() calls in error().
void error(int _errid, const QByteArray &_text);
void send(int cmd, const QByteArray &arr = QByteArray());
+ // Hands what a message carries to the application as it is, when the application is in this
+ // process, and as the bytes standing for it otherwise.
+ void send(int cmd, const TaskPayload &payload);
std::unique_ptr<SlaveBasePrivate> const d;
friend class SlaveBasePrivate;
diff --git a/src/core/socketconnectionbackend.cpp b/src/core/socketconnectionbackend.cpp
index 197c06f61b..7818726b4a 100644
--- a/src/core/socketconnectionbackend.cpp
+++ b/src/core/socketconnectionbackend.cpp
@@ -185,6 +185,12 @@ bool SocketConnectionBackend::sendCommand(int cmd, const QByteArray &data)
Q_ASSERT(state == Connected);
Q_ASSERT(socket);
+ // The header holds the size in six hex digits, so that is as much as one message can carry.
+ if (data.size() > 0xffffff) {
+ qCWarning(KIO_CORE_CONNECTION) << "Message of" << data.size() << "bytes is too much for one command";
+ return false;
+ }
+
char buffer[HeaderSize + 2];
sprintf(buffer, "%6zx_%2x_", static_cast<size_t>(data.size()), cmd);
socket->write(buffer, HeaderSize);
@@ -268,22 +274,30 @@ void SocketConnectionBackend::socketReadyRead()
pendingTask = Task{.cmd = static_cast<int>(cmd)};
pendingLen = len;
+ pendingData.clear();
+ pendingData.reserve(len);
qCDebug(KIO_CORE_CONNECTION) << this << "Beginning of command" << pendingTask->cmd << "of size" << pendingLen;
}
QPointer<ConnectionBackend> that = this;
- const auto toRead = std::min<off_t>(socket->bytesAvailable(), pendingLen - pendingTask->data.size());
- qCDebug(KIO_CORE_CONNECTION) << socket << "Want to read" << toRead << "bytes; appending to already existing bytes" << pendingTask->data.size();
- pendingTask->data += socket->read(toRead);
+ const auto alreadyRead = pendingData.size();
+ const auto toRead = std::min<off_t>(socket->bytesAvailable(), pendingLen - alreadyRead);
+ qCDebug(KIO_CORE_CONNECTION) << socket << "Want to read" << toRead << "bytes; appending to already existing bytes" << alreadyRead;
+ // Read into the buffer the command is collected in, rather than into one of its own.
+ pendingData.resize(alreadyRead + toRead);
+ const auto wasRead = socket->read(pendingData.data() + alreadyRead, toRead);
+ pendingData.resize(alreadyRead + std::max<qint64>(wasRead, 0));
- if (pendingTask->data.size() == pendingLen) { // read all data of this task -> emit it and reset
+ if (pendingData.size() == pendingLen) { // read all data of this task -> emit it and reset
signalEmitted = true;
- qCDebug(KIO_CORE_CONNECTION) << "emitting task" << pendingTask->cmd << pendingTask->data.size();
+ qCDebug(KIO_CORE_CONNECTION) << "emitting task" << pendingTask->cmd << pendingData.size();
+ pendingTask->payload = std::move(pendingData);
Q_EMIT commandReceived(pendingTask.value());
pendingTask = {};
+ pendingData.clear();
}
// If we're dead, better don't try anything.
diff --git a/src/core/socketconnectionbackend_p.h b/src/core/socketconnectionbackend_p.h
index 988ea77a57..71cd007b4a 100644
--- a/src/core/socketconnectionbackend_p.h
+++ b/src/core/socketconnectionbackend_p.h
@@ -56,6 +56,7 @@ private:
QLocalServer *localServer;
std::optional<Task> pendingTask = std::nullopt;
long pendingLen = 0; // expected payload size, valid while pendingTask has a value
+ QByteArray pendingData; // the bytes read so far for pendingTask
bool signalEmitted;
Q_SIGNALS:
diff --git a/src/core/specialjob.cpp b/src/core/specialjob.cpp
index 19f7099078..14ade439aa 100644
--- a/src/core/specialjob.cpp
+++ b/src/core/specialjob.cpp
@@ -13,7 +13,7 @@ using namespace KIO;
class KIO::SpecialJobPrivate : public TransferJobPrivate
{
- SpecialJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs, const QByteArray &_staticData)
+ SpecialJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs, const QByteArray &_staticData)
: TransferJobPrivate(url, command, packedArgs, _staticData)
{
}
@@ -36,7 +36,7 @@ void SpecialJob::setArguments(const QByteArray &data)
QByteArray SpecialJob::arguments() const
{
- return d_func()->m_packedArgs;
+ return payloadBytes(d_func()->m_packedArgs);
}
#include "moc_specialjob.cpp"
diff --git a/src/core/statjob.cpp b/src/core/statjob.cpp
index 0e6bcd9935..8cf5663e14 100644
--- a/src/core/statjob.cpp
+++ b/src/core/statjob.cpp
@@ -20,7 +20,7 @@ using namespace KIO;
class KIO::StatJobPrivate : public SimpleJobPrivate
{
public:
- inline StatJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs)
+ inline StatJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs)
: SimpleJobPrivate(url, command, packedArgs)
, m_bSource(true)
, m_details(KIO::StatDefaultDetails)
@@ -44,7 +44,7 @@ public:
Q_DECLARE_PUBLIC(StatJob)
- static inline StatJob *newJob(const QUrl &url, int command, const QByteArray &packedArgs, JobFlags flags)
+ static inline StatJob *newJob(const QUrl &url, int command, const TaskPayload &packedArgs, JobFlags flags)
{
StatJob *job = new StatJob(*new StatJobPrivate(url, command, packedArgs));
job->setUiDelegate(KIO::createDefaultJobUiDelegate());
@@ -157,9 +157,7 @@ void StatJob::slotFinished()
}
if (d->m_redirectionHandlingEnabled) {
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_redirectionURL;
+ d->m_packedArgs = d->m_redirectionURL;
d->restartAfterRedirection(&d->m_redirectionURL);
return;
@@ -209,8 +207,7 @@ StatJob *KIO::stat(const QUrl &url, JobFlags flags)
StatJob *KIO::stat(const QUrl &url, KIO::StatJob::StatSide side, KIO::StatDetails details, JobFlags flags)
{
// qCDebug(KIO_CORE) << "stat" << url;
- KIO_ARGS << url;
- StatJob *job = StatJobPrivate::newJob(url, CMD_STAT, packedArgs, flags);
+ StatJob *job = StatJobPrivate::newJob(url, CMD_STAT, url, flags);
job->setSide(side);
job->setDetails(details);
return job;
diff --git a/src/core/storedtransferjob.cpp b/src/core/storedtransferjob.cpp
index f386c50e33..c560c8cbfe 100644
--- a/src/core/storedtransferjob.cpp
+++ b/src/core/storedtransferjob.cpp
@@ -18,12 +18,12 @@ using namespace KIO;
class KIO::StoredTransferJobPrivate : public TransferJobPrivate
{
public:
- StoredTransferJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs, const QByteArray &_staticData)
+ StoredTransferJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs, const QByteArray &_staticData)
: TransferJobPrivate(url, command, packedArgs, _staticData)
, m_uploadOffset(0)
{
}
- StoredTransferJobPrivate(const QUrl &url, int command, const QByteArray &packedArgs, QIODevice *ioDevice)
+ StoredTransferJobPrivate(const QUrl &url, int command, const TaskPayload &packedArgs, QIODevice *ioDevice)
: TransferJobPrivate(url, command, packedArgs, ioDevice)
, m_uploadOffset(0)
{
diff --git a/src/core/threadconnectionbackend.cpp b/src/core/threadconnectionbackend.cpp
index db716e5a7c..ff4bcb093c 100644
--- a/src/core/threadconnectionbackend.cpp
+++ b/src/core/threadconnectionbackend.cpp
@@ -92,7 +92,7 @@ void ThreadConnectionBackend::drainIncoming()
}
}
-bool ThreadConnectionBackend::sendCommand(int cmd, const QByteArray &data)
+bool ThreadConnectionBackend::queueTask(Task &&task)
{
if (!m_channel) {
return false;
@@ -116,7 +116,7 @@ bool ThreadConnectionBackend::sendCommand(int cmd, const QByteArray &data)
// Share the payload (no copy): in-process workers must hand us an owned QByteArray,
// since we deliver it to the application asynchronously (after the worker has moved
// on). Its refcount keeps the bytes alive until the application has consumed them.
- out.queue.append(Task{.cmd = cmd, .data = data});
+ out.queue.append(std::move(task));
out.dataAvailable.wakeOne(); // one consumer per direction
// Post the wakeup to the event-loop consumer while still holding the mutex: the application
@@ -130,6 +130,19 @@ bool ThreadConnectionBackend::sendCommand(int cmd, const QByteArray &data)
return true;
}
+bool ThreadConnectionBackend::sendCommand(int cmd, const QByteArray &data)
+{
+ return queueTask(Task{.cmd = cmd, .payload = data});
+}
+
+// The peer lives in another thread of this process, so what the message carries is handed over as it
+// is. The objects are shared, so what travels is a reference, and the worker letting go of its own
+// leaves them to the application.
+bool ThreadConnectionBackend::sendPayload(int cmd, const TaskPayload &payload)
+{
+ return queueTask(Task{.cmd = cmd, .payload = payload});
+}
+
bool ThreadConnectionBackend::waitForIncomingTask(int ms)
{
if (!m_channel) {
diff --git a/src/core/threadconnectionbackend_p.h b/src/core/threadconnectionbackend_p.h
index 9be342e16b..05a8751378 100644
--- a/src/core/threadconnectionbackend_p.h
+++ b/src/core/threadconnectionbackend_p.h
@@ -62,6 +62,8 @@ public:
bool waitForIncomingTask(int ms) override;
bool sendCommand(int command, const QByteArray &data) override;
+ bool sendPayload(int command, const TaskPayload &payload) override;
+
private Q_SLOTS:
/// Emit commandReceived() for every queued task (drives the event-loop side).
void drainIncoming();
@@ -93,6 +95,9 @@ private:
std::atomic<bool> appDrainScheduled{false};
};
+ /// Hands \a task to the peer, waiting while the peer is HighWaterMark tasks behind.
+ bool queueTask(Task &&task);
+
Direction &incoming(); // the direction this backend reads
Direction &outgoing(); // the direction this backend writes
void emitQueued(Direction &dir); // pop+emit commandReceived while not suspended
diff --git a/src/core/transferjob.cpp b/src/core/transferjob.cpp
index bae688d40e..ca0732290b 100644
--- a/src/core/transferjob.cpp
+++ b/src/core/transferjob.cpp
@@ -120,14 +120,12 @@ void TransferJob::slotFinished()
d->m_internalSuspended = false;
// The very tricky part is the packed arguments business
QUrl dummyUrl;
- QDataStream istream(d->m_packedArgs);
+ QDataStream istream(payloadBytes(d->m_packedArgs));
switch (d->m_command) {
case CMD_GET:
case CMD_STAT:
case CMD_DEL: {
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_redirectionURL;
+ d->m_packedArgs = d->m_redirectionURL;
break;
}
case CMD_PUT: {
@@ -135,9 +133,10 @@ void TransferJob::slotFinished()
qint8 iOverwrite;
qint8 iResume;
istream >> dummyUrl >> iOverwrite >> iResume >> permissions;
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
+ QByteArray args;
+ QDataStream stream(&args, QIODevice::WriteOnly);
stream << d->m_redirectionURL << iOverwrite << iResume << permissions;
+ d->m_packedArgs = args;
break;
}
case CMD_SPECIAL: {
@@ -146,9 +145,7 @@ void TransferJob::slotFinished()
if (specialcmd == 1) { // HTTP POST
d->m_outgoingMetaData.remove(QStringLiteral("content-type"));
addMetaData(QStringLiteral("cache"), QStringLiteral("reload"));
- d->m_packedArgs.truncate(0);
- QDataStream stream(&d->m_packedArgs, QIODevice::WriteOnly);
- stream << d->m_redirectionURL;
+ d->m_packedArgs = d->m_redirectionURL;
d->m_command = CMD_GET;
}
break;
@@ -390,7 +387,7 @@ void TransferJob::setModificationTime(const QDateTime &mtime)
TransferJob *KIO::get(const QUrl &url, LoadType reload, JobFlags flags)
{
// Send decoded path and encoded query
- KIO_ARGS << url;
+ const TaskPayload packedArgs = url;
TransferJob *job = TransferJobPrivate::newJob(url, CMD_GET, packedArgs, QByteArray(), flags);
if (reload == Reload) {
job->addMetaData(QStringLiteral("cache"), QStringLiteral("reload"));
diff --git a/src/core/worker.cpp b/src/core/worker.cpp
index ff607da970..88d14a8604 100644
--- a/src/core/worker.cpp
+++ b/src/core/worker.cpp
@@ -297,6 +297,11 @@ void Worker::send(int cmd, const QByteArray &arr)
m_connection->send(cmd, arr);
}
+void Worker::send(int cmd, const TaskPayload &payload)
+{
+ m_connection->send(cmd, payload);
+}
+
void Worker::gotInput()
{
if (m_dead) { // already dead? then workerDied was emitted and we are done
diff --git a/src/core/worker_p.h b/src/core/worker_p.h
index 4781286827..f790525e00 100644
--- a/src/core/worker_p.h
+++ b/src/core/worker_p.h
@@ -50,6 +50,11 @@ public:
* \a arr byte array containing data
*/
virtual void send(int cmd, const QByteArray &arr = QByteArray());
+ /*!
+ * Sends what the command carries. The worker is handed the object as it is when it runs in a
+ * thread of this process, and the bytes it is written to otherwise.
+ */
+ virtual void send(int cmd, const TaskPayload &payload);
/*!
* Returns Host this worker is (was?) connected to
diff --git a/src/core/workerinterface.cpp b/src/core/workerinterface.cpp
index 26f74a3f29..bffc05be54 100644
--- a/src/core/workerinterface.cpp
+++ b/src/core/workerinterface.cpp
@@ -49,7 +49,7 @@ bool WorkerInterface::dispatch()
int cmd;
QByteArray data;
- int ret = m_connection->read(&cmd, data);
+ int ret = m_connection->read(&cmd, data, &m_incomingPayload);
if (ret == -1) {
return false;
}
@@ -124,13 +124,16 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
m_speed_timer.stop();
Q_EMIT finished();
break;
- case MSG_STAT_ENTRY: {
- UDSEntry entry;
- stream >> entry;
- Q_EMIT statEntry(entry);
+ case MSG_STAT_ENTRY:
+ Q_EMIT statEntry(carried<UDSEntry>(m_incomingPayload, stream));
break;
- }
case MSG_LIST_ENTRIES: {
+ if (UDSEntryList *entries = std::get_if<UDSEntryList>(&m_incomingPayload)) {
+ // The worker lives in a thread of this process and handed its entries over as they are.
+ Q_EMIT listEntries(std::exchange(*entries, UDSEntryList{}));
+ break;
+ }
+
UDSEntryList list;
UDSEntry entry;
@@ -143,7 +146,7 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
break;
}
case MSG_RESUME: { // From the put job
- m_offset = readFilesize_t(stream);
+ m_offset = carried<quint64>(m_incomingPayload, stream);
Q_EMIT canResume(m_offset);
break;
}
@@ -152,20 +155,23 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
Q_EMIT canResume(0); // the arg doesn't matter
break;
case MSG_ERROR:
- stream >> i >> str1;
- // qDebug() << "error " << i << " " << str1;
- Q_EMIT error(i, str1);
+ if (TaskError *reported = std::get_if<TaskError>(&m_incomingPayload)) {
+ Q_EMIT error(reported->code, reported->text);
+ } else {
+ stream >> i >> str1;
+ Q_EMIT error(i, str1);
+ }
break;
case MSG_CONNECTED:
Q_EMIT connected();
break;
case MSG_WRITTEN: {
- KIO::filesize_t size = readFilesize_t(stream);
+ const KIO::filesize_t size = carried<quint64>(m_incomingPayload, stream);
Q_EMIT written(size);
break;
}
case INF_TOTAL_SIZE: {
- KIO::filesize_t size = readFilesize_t(stream);
+ const KIO::filesize_t size = carried<quint64>(m_incomingPayload, stream);
m_start_time = QDateTime::currentMSecsSinceEpoch();
m_last_time = 0;
m_filesize = m_offset;
@@ -178,45 +184,40 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
break;
}
case INF_PROCESSED_SIZE: {
- KIO::filesize_t size = readFilesize_t(stream);
+ const KIO::filesize_t size = carried<quint64>(m_incomingPayload, stream);
Q_EMIT processedSize(size);
m_filesize = size;
break;
}
case INF_POSITION: {
- KIO::filesize_t pos = readFilesize_t(stream);
+ const KIO::filesize_t pos = carried<quint64>(m_incomingPayload, stream);
Q_EMIT position(pos);
break;
}
case INF_TRUNCATED: {
- KIO::filesize_t length = readFilesize_t(stream);
+ const KIO::filesize_t length = carried<quint64>(m_incomingPayload, stream);
Q_EMIT truncated(length);
break;
}
case INF_SPEED:
- stream >> ul;
+ ul = carried<quint32>(m_incomingPayload, stream);
m_worker_calcs_speed = true;
m_speed_timer.stop();
Q_EMIT speed(ul);
break;
case INF_ERROR_PAGE:
break;
- case INF_REDIRECTION: {
- QUrl url;
- stream >> url;
- Q_EMIT redirection(url);
+ case INF_REDIRECTION:
+ Q_EMIT redirection(carried<QUrl>(m_incomingPayload, stream));
break;
- }
case INF_MIME_TYPE:
- stream >> str1;
- Q_EMIT mimeType(str1);
+ Q_EMIT mimeType(carried<QString>(m_incomingPayload, stream));
if (!m_connection->suspended()) {
m_connection->sendnow(CMD_NONE, QByteArray());
}
break;
case INF_WARNING:
- stream >> str1;
- Q_EMIT warning(str1);
+ Q_EMIT warning(carried<QString>(m_incomingPayload, stream));
break;
case INF_MESSAGEBOX: {
// qDebug() << "needs a msg box";
@@ -235,12 +236,9 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
}
break;
}
- case INF_INFOMESSAGE: {
- QString msg;
- stream >> msg;
- Q_EMIT infoMessage(msg);
+ case INF_INFOMESSAGE:
+ Q_EMIT infoMessage(carried<QString>(m_incomingPayload, stream));
break;
- }
case INF_SSLERROR: {
QVariantMap sslErrorData;
stream >> sslErrorData;
@@ -248,8 +246,7 @@ bool WorkerInterface::dispatch(int _cmd, const QByteArray &rawdata)
break;
}
case INF_META_DATA: {
- MetaData m;
- stream >> m;
+ const MetaData m = carried<MetaData>(m_incomingPayload, stream);
if (auto it = m.constFind(QStringLiteral("privilege_conf_details")); it != m.cend()) {
// see WORKER_MESSAGEBOX_DETAILS_HACK
m_messageBoxDetails = it.value();
diff --git a/src/core/workerinterface_p.h b/src/core/workerinterface_p.h
index 8c20e758d8..5369dbb5b6 100644
--- a/src/core/workerinterface_p.h
+++ b/src/core/workerinterface_p.h
@@ -144,6 +144,9 @@ protected:
virtual bool dispatch();
virtual bool dispatch(int _cmd, const QByteArray &data);
+ // What the last task carried, when the worker lives in a thread of this process.
+ TaskPayload m_incomingPayload;
+
void messageBox(int type, const QString &text, const QString &title, const QString &primaryActionText, const QString &secondaryActionText);
void messageBox(int type,