Revision control
++ // Copyright (C) 2025 The Qt Company Ltd.
++ // SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
++ // Qt-Security score:significant reason:default
++
++ #include "qrandomaccessasyncfile_p_p.h"
++
++ #include "qiooperation_p.h"
++ #include "qiooperation_p_p.h"
++
++ #include <QtCore/qfuture.h>
++ #include <QtCore/qthread.h>
++ #include <QtCore/qthreadpool.h>
++
++ QT_REQUIRE_CONFIG(thread);
++ QT_REQUIRE_CONFIG(future);
++
++ QT_BEGIN_NAMESPACE
++
++ namespace {
++
++ // We cannot use Q_GLOBAL_STATIC(QThreadPool, foo) because the Windows
++ // implementation raises a qWarning in its destructor when used as a global
++ // static, and this warning leads to a crash on Windows CI. Cannot reproduce
++ // the crash locally, so cannot really fix the issue :(
++ // This class should act like a global thread pool, but it'll have a sort of
++ // ref counting, and will be created/destroyed by QRAAFP instances.
++ class SharedThreadPool
++ {
++ public:
++ void ref()
++ {
++ QMutexLocker locker(&m_mutex);
++ if (m_refCount == 0) {
++ Q_ASSERT(!m_pool);
++ m_pool = new QThreadPool;
++ }
++ ++m_refCount;
++ }
++
++ void deref()
++ {
++ QMutexLocker locker(&m_mutex);
++ Q_ASSERT(m_refCount);
++ if (--m_refCount == 0) {
++ delete m_pool;
++ m_pool = nullptr;
++ }
++ }
++
++ QThreadPool *operator()()
++ {
++ QMutexLocker locker(&m_mutex);
++ Q_ASSERT(m_refCount > 0);
++ return m_pool;
++ }
++
++ private:
++ QBasicMutex m_mutex;
++ QThreadPool *m_pool = nullptr;
++ quint64 m_refCount = 0;
++ };
++
++ static SharedThreadPool asyncFileThreadPool;
++
++ } // anonymous namespace
++
++ QRandomAccessAsyncFileThreadPoolBackend::QRandomAccessAsyncFileThreadPoolBackend(QRandomAccessAsyncFile *owner) :
++ QRandomAccessAsyncFileBackend(owner)
++ {
++ asyncFileThreadPool.ref();
++ }
++
++ QRandomAccessAsyncFileThreadPoolBackend::~QRandomAccessAsyncFileThreadPoolBackend()
++ {
++ asyncFileThreadPool.deref();
++ }
++
++ bool QRandomAccessAsyncFileThreadPoolBackend::init()
++ {
++ QObject::connect(&m_watcher, &QFutureWatcherBase::finished, m_owner, [this]{
++ operationComplete();
++ });
++ QObject::connect(&m_watcher, &QFutureWatcherBase::canceled, m_owner, [this]{
++ operationComplete();
++ });
++ return true;
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::cancelAndWait(QIOOperation *op)
++ {
++ if (op == m_currentOperation) {
++ m_currentOperation = nullptr; // to discard the result
++ m_watcher.cancel(); // might have no effect
++ m_watcher.waitForFinished();
++ } else {
++ m_operations.removeAll(op);
++ }
++ }
++
++ QIOOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::open(const QString &path, QIODeviceBase::OpenMode mode)
++ {
++ // We generate the command in any case. But if the file is already opened,
++ // it will finish with an error
++ if (m_fileState == FileState::Closed) {
++ m_filePath = path;
++ m_openMode = mode;
++ m_fileState = FileState::OpenPending;
++ }
++
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage();
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->type = QIOOperation::Type::Open;
++
++ auto *op = new QIOOperation(*priv, m_owner);
++
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::close()
++ {
++ // all the operations should be aborted
++ for (const auto &op : std::as_const(m_operations)) {
++ if (op) {
++ auto *priv = QIOOperationPrivate::get(op.get());
++ priv->setError(QIOOperation::Error::Aborted);
++ }
++ }
++ m_operations.clear();
++
++ // Wait until the current operation is complete
++ if (m_currentOperation) {
++ auto *priv = QIOOperationPrivate::get(m_currentOperation.get());
++ priv->setError(QIOOperation::Error::Aborted);
++ cancelAndWait(m_currentOperation.get());
++ }
++
++ {
++ QMutexLocker locker(&m_engineMutex);
++ if (m_engine) {
++ m_engine->close();
++ m_engine.reset();
++ }
++ }
++
++ m_fileState = FileState::Closed;
++ }
++
++ qint64 QRandomAccessAsyncFileThreadPoolBackend::size() const
++ {
++ QMutexLocker locker(&m_engineMutex);
++ if (m_engine)
++ return m_engine->size();
++ return -1;
++ }
++
++ QIOOperation *QRandomAccessAsyncFileThreadPoolBackend::flush()
++ {
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage();
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->type = QIOOperation::Type::Flush;
++
++ auto *op = new QIOOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOReadOperation *QRandomAccessAsyncFileThreadPoolBackend::read(qint64 offset, qint64 maxSize)
++ {
++ QByteArray array;
++ array.resizeForOverwrite(maxSize);
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage(std::move(array));
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Read;
++
++ auto *op = new QIOReadOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOWriteOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::write(qint64 offset, const QByteArray &data)
++ {
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage(data);
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Write;
++
++ auto *op = new QIOWriteOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOWriteOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::write(qint64 offset, QByteArray &&data)
++ {
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage(std::move(data));
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Write;
++
++ auto *op = new QIOWriteOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOVectoredReadOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::readInto(qint64 offset, QSpan<std::byte> buffer)
++ {
++ auto *dataStorage =
++ new QtPrivate::QIOOperationDataStorage(QSpan<const QSpan<std::byte>>{buffer});
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Read;
++
++ auto *op = new QIOVectoredReadOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOVectoredWriteOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::writeFrom(qint64 offset, QSpan<const std::byte> buffer)
++ {
++ auto *dataStorage =
++ new QtPrivate::QIOOperationDataStorage(QSpan<const QSpan<const std::byte>>{buffer});
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Write;
++
++ auto *op = new QIOVectoredWriteOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOVectoredReadOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::readInto(qint64 offset, QSpan<const QSpan<std::byte>> buffers)
++ {
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage(buffers);
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Read;
++
++ auto *op = new QIOVectoredReadOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ QIOVectoredWriteOperation *
++ QRandomAccessAsyncFileThreadPoolBackend::writeFrom(qint64 offset, QSpan<const QSpan<const std::byte>> buffers)
++ {
++ auto *dataStorage = new QtPrivate::QIOOperationDataStorage(buffers);
++
++ auto *priv = new QIOOperationPrivate(dataStorage);
++ priv->offset = offset;
++ priv->type = QIOOperation::Type::Write;
++
++ auto *op = new QIOVectoredWriteOperation(*priv, m_owner);
++ m_operations.append(op);
++ executeNextOperation();
++ return op;
++ }
++
++ static QRandomAccessAsyncFileThreadPoolBackend::OperationResult
++ executeRead(QFSFileEngine *engine, QBasicMutex *mutex, qint64 offset, char *buffer, qint64 maxSize)
++ {
++ QRandomAccessAsyncFileThreadPoolBackend::OperationResult result{0, QIOOperation::Error::None};
++
++ QMutexLocker locker(mutex);
++ if (engine) {
++ if (engine->seek(offset)) {
++ qint64 bytesRead = engine->read(buffer, maxSize);
++ if (bytesRead >= 0)
++ result.bytesProcessed = bytesRead;
++ else
++ result.error = QIOOperation::Error::Read;
++ } else {
++ result.error = QIOOperation::Error::IncorrectOffset;
++ }
++ } else {
++ result.error = QIOOperation::Error::FileNotOpen;
++ }
++ return result;
++ }
++
++ static QRandomAccessAsyncFileThreadPoolBackend::OperationResult
++ executeWrite(QFSFileEngine *engine, QBasicMutex *mutex, qint64 offset,
++ const char *buffer, qint64 size)
++ {
++ QRandomAccessAsyncFileThreadPoolBackend::OperationResult result{0, QIOOperation::Error::None};
++
++ QMutexLocker locker(mutex);
++ if (engine) {
++ if (engine->seek(offset)) {
++ qint64 written = engine->write(buffer, size);
++ if (written >= 0)
++ result.bytesProcessed = written;
++ else
++ result.error = QIOOperation::Error::Write;
++ } else {
++ result.error = QIOOperation::Error::IncorrectOffset;
++ }
++ } else {
++ result.error = QIOOperation::Error::FileNotOpen;
++ }
++ return result;
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::executeNextOperation()
++ {
++ if (m_currentOperation.isNull()) {
++ // start next
++ if (!m_operations.isEmpty()) {
++ m_currentOperation = m_operations.takeFirst();
++ switch (m_currentOperation->type()) {
++ case QIOOperation::Type::Read:
++ case QIOOperation::Type::Write:
++ numProcessedBuffers = 0;
++ processBufferAt(numProcessedBuffers);
++ break;
++ case QIOOperation::Type::Flush:
++ processFlush();
++ break;
++ case QIOOperation::Type::Open:
++ processOpen();
++ break;
++ case QIOOperation::Type::Unknown:
++ Q_ASSERT_X(false, "executeNextOperation", "Operation of type Unknown!");
++ // For release builds - directly complete the operation
++ m_watcher.setFuture(QtFuture::makeReadyValueFuture(OperationResult{}));
++ operationComplete();
++ break;
++ }
++ }
++ }
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::processBufferAt(qsizetype idx)
++ {
++ Q_ASSERT(!m_currentOperation.isNull());
++ auto *priv = QIOOperationPrivate::get(m_currentOperation.get());
++ auto &dataStorage = priv->dataStorage;
++ // if we do not use span buffers, we have only one buffer
++ Q_ASSERT(dataStorage->containsReadSpans()
++ || dataStorage->containsWriteSpans()
++ || idx == 0);
++ if (priv->type == QIOOperation::Type::Read) {
++ qint64 maxSize = -1;
++ char *buf = nullptr;
++ if (dataStorage->containsReadSpans()) {
++ auto &readBuffers = dataStorage->getReadSpans();
++ Q_ASSERT(readBuffers.size() > idx);
++ maxSize = readBuffers[idx].size_bytes();
++ buf = reinterpret_cast<char *>(readBuffers[idx].data());
++ } else {
++ Q_ASSERT(dataStorage->containsByteArray());
++ auto &array = dataStorage->getByteArray();
++ maxSize = array.size();
++ buf = array.data();
++ }
++ Q_ASSERT(maxSize >= 0);
++
++ qint64 offset = priv->offset;
++ if (idx != 0)
++ offset += priv->processed;
++ QBasicMutex *mutexPtr = &m_engineMutex;
++ auto op = [engine = m_engine.get(), buf, maxSize, offset, mutexPtr] {
++ return executeRead(engine, mutexPtr, offset, buf, maxSize);
++ };
++
++ QFuture<OperationResult> f =
++ QtFuture::makeReadyVoidFuture().then(asyncFileThreadPool(), op);
++ m_watcher.setFuture(f);
++ } else if (priv->type == QIOOperation::Type::Write) {
++ qint64 size = -1;
++ const char *buf = nullptr;
++ if (dataStorage->containsWriteSpans()) {
++ const auto &writeBuffers = dataStorage->getWriteSpans();
++ Q_ASSERT(writeBuffers.size() > idx);
++ size = writeBuffers[idx].size_bytes();
++ buf = reinterpret_cast<const char *>(writeBuffers[idx].data());
++ } else {
++ Q_ASSERT(dataStorage->containsByteArray());
++ const auto &array = dataStorage->getByteArray();
++ size = array.size();
++ buf = array.constData();
++ }
++ Q_ASSERT(size >= 0);
++
++ qint64 offset = priv->offset;
++ if (idx != 0)
++ offset += priv->processed;
++ QBasicMutex *mutexPtr = &m_engineMutex;
++ auto op = [engine = m_engine.get(), buf, size, offset, mutexPtr] {
++ return executeWrite(engine, mutexPtr, offset, buf, size);
++ };
++
++ QFuture<OperationResult> f =
++ QtFuture::makeReadyVoidFuture().then(asyncFileThreadPool(), op);
++ m_watcher.setFuture(f);
++ }
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::processFlush()
++ {
++ Q_ASSERT(!m_currentOperation.isNull());
++ auto *priv = QIOOperationPrivate::get(m_currentOperation.get());
++ auto &dataStorage = priv->dataStorage;
++ Q_ASSERT(dataStorage->isEmpty());
++
++ QBasicMutex *mutexPtr = &m_engineMutex;
++ auto op = [engine = m_engine.get(), mutexPtr] {
++ QMutexLocker locker(mutexPtr);
++ QRandomAccessAsyncFileThreadPoolBackend::OperationResult result{0, QIOOperation::Error::None};
++ if (engine) {
++ if (!engine->flush())
++ result.error = QIOOperation::Error::Flush;
++ } else {
++ result.error = QIOOperation::Error::FileNotOpen;
++ }
++ return result;
++ };
++
++ QFuture<OperationResult> f =
++ QtFuture::makeReadyVoidFuture().then(asyncFileThreadPool(), op);
++ m_watcher.setFuture(f);
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::processOpen()
++ {
++ Q_ASSERT(!m_currentOperation.isNull());
++ auto *priv = QIOOperationPrivate::get(m_currentOperation.get());
++ auto &dataStorage = priv->dataStorage;
++ Q_ASSERT(dataStorage->isEmpty());
++
++ QFuture<OperationResult> f;
++ if (m_fileState == FileState::OpenPending) {
++ // create the engine
++ m_engineMutex.lock();
++ m_engine = std::make_unique<QFSFileEngine>(m_filePath);
++ m_engineMutex.unlock();
++ QBasicMutex *mutexPtr = &m_engineMutex;
++ auto op = [engine = m_engine.get(), mutexPtr, mode = m_openMode] {
++ QRandomAccessAsyncFileThreadPoolBackend::OperationResult result{0, QIOOperation::Error::None};
++ QMutexLocker locker(mutexPtr);
++ const bool res =
++ engine && engine->open(mode | QIODeviceBase::Unbuffered, std::nullopt);
++ if (!res)
++ result.error = QIOOperation::Error::Open;
++ return result;
++ };
++ f = QtFuture::makeReadyVoidFuture().then(asyncFileThreadPool(), op);
++ } else {
++ f = QtFuture::makeReadyVoidFuture().then(asyncFileThreadPool(), [] {
++ return QRandomAccessAsyncFileThreadPoolBackend::OperationResult{0, QIOOperation::Error::Open};
++ });
++ }
++ m_watcher.setFuture(f);
++ }
++
++ void QRandomAccessAsyncFileThreadPoolBackend::operationComplete()
++ {
++ // TODO: if one of the buffers was read/written with an error,
++ // stop processing immediately
++
++ auto scheduleNextOperation = qScopeGuard([this]{
++ m_currentOperation = nullptr;
++ executeNextOperation();
++ });
++
++ if (m_currentOperation && !m_watcher.isCanceled()) {
++ OperationResult res = m_watcher.future().result();
++ auto *priv = QIOOperationPrivate::get(m_currentOperation.get());
++ auto &dataStorage = priv->dataStorage;
++
++ switch (priv->type) {
++ case QIOOperation::Type::Read: {
++ qsizetype expectedBuffersCount = 1;
++ if (dataStorage->containsReadSpans()) {
++ auto &readBuffers = dataStorage->getReadSpans();
++ expectedBuffersCount = readBuffers.size();
++ Q_ASSERT(numProcessedBuffers < expectedBuffersCount);
++ const qsizetype unusedBytes =
++ readBuffers[numProcessedBuffers].size_bytes() - res.bytesProcessed;
++ readBuffers[numProcessedBuffers].chop(unusedBytes);
++ } else {
++ Q_ASSERT(dataStorage->containsByteArray());
++ Q_ASSERT(numProcessedBuffers == 0);
++ auto &array = dataStorage->getByteArray();
++ array.resize(res.bytesProcessed);
++ }
++ priv->appendBytesProcessed(res.bytesProcessed);
++ if (++numProcessedBuffers < expectedBuffersCount) {
++ // keep executing this command
++ processBufferAt(numProcessedBuffers);
++ scheduleNextOperation.dismiss();
++ } else {
++ priv->operationComplete(res.error);
++ }
++ break;
++ }
++ case QIOOperation::Type::Write: {
++ qsizetype expectedBuffersCount = 1;
++ if (dataStorage->containsWriteSpans())
++ expectedBuffersCount = dataStorage->getWriteSpans().size();
++ Q_ASSERT(numProcessedBuffers < expectedBuffersCount);
++ priv->appendBytesProcessed(res.bytesProcessed);
++ if (++numProcessedBuffers < expectedBuffersCount) {
++ // keep executing this command
++ processBufferAt(numProcessedBuffers);
++ scheduleNextOperation.dismiss();
++ } else {
++ priv->operationComplete(res.error);
++ }
++ break;
++ }
++ case QIOOperation::Type::Flush:
++ priv->operationComplete(res.error);
++ break;
++ case QIOOperation::Type::Open:
++ if (m_fileState == FileState::OpenPending) {
++ if (res.error == QIOOperation::Error::None) {
++ m_fileState = FileState::Opened;
++ } else {
++ m_fileState = FileState::Closed;
++ m_engineMutex.lock();
++ m_engine.reset();
++ m_engineMutex.unlock();
++ }
++ }
++ priv->operationComplete(res.error);
++ break;
++ case QIOOperation::Type::Unknown:
++ priv->setError(QIOOperation::Error::Aborted);
++ break;
++ }
++ }
++ }
++
++ QT_END_NAMESPACE