9#include "qplatformdefs.h"
11#include <QtCore/qdir.h>
12#include <QtCore/qfile.h>
13#include <QtCore/private/qfilesystemengine_p.h>
21 return type == QIOOperation::Type::Flush || type == QIOOperation::Type::Open;
28template <
typename Operation,
typename ...Args>
30QRandomAccessAsyncFileNativeBackend::addOperation(QIOOperation::Type type, qint64 offset, Args &&...args)
32 auto dataStorage =
new QtPrivate::QIOOperationDataStorage(std::forward<Args>(args)...);
33 auto *priv =
new QIOOperationPrivate(dataStorage);
34 priv->offset = offset;
37 Operation *op =
new Operation(*priv, m_owner);
38 auto opId = getNextId();
39 m_operations.push_back(OperationInfo(opId, op));
40 startOperationsUntilBarrier();
45QRandomAccessAsyncFileNativeBackend::QRandomAccessAsyncFileNativeBackend(QRandomAccessAsyncFile *owner)
46 : QRandomAccessAsyncFileBackend(owner)
50QRandomAccessAsyncFileNativeBackend::~QRandomAccessAsyncFileNativeBackend()
53bool QRandomAccessAsyncFileNativeBackend::init()
58void QRandomAccessAsyncFileNativeBackend::cancelAndWait(QIOOperation *op)
60 auto it = std::find_if(m_operations.cbegin(), m_operations.cend(),
61 [op](
const auto &opInfo) {
62 return opInfo.operation.get() == op;
65 if (it == m_operations.cend())
68 const auto opInfo = m_operations.takeAt(std::distance(m_operations.cbegin(), it));
70 if (opInfo.state == OpState::Running) {
73 if (m_runningOps.contains(opInfo.opId)) {
74 m_opToCancel = opInfo.opId;
75 closeIoChannel(opInfo.channel);
76 m_cancellationCondition.wait(&m_mutex);
77 m_opToCancel = kInvalidOperationId;
83 releaseIoChannel(opInfo.channel);
84 auto *priv = QIOOperationPrivate::get(opInfo.operation);
85 priv->setError(QIOOperation::Error::Aborted);
88 startOperationsUntilBarrier();
91void QRandomAccessAsyncFileNativeBackend::close()
93 if (m_fileState == FileState::Closed)
98 m_opToCancel = kAllOperationIds;
99 for (
const auto &op : m_operations) {
101 closeIoChannel(op.channel);
104 closeIoChannel(m_ioChannel);
106 if (!m_runningOps.isEmpty() || m_ioChannel)
107 m_cancellationCondition.wait(&m_mutex);
108 m_opToCancel = kInvalidOperationId;
112 for (
auto &opInfo : m_operations) {
113 releaseIoChannel(opInfo.channel);
114 auto *priv = QIOOperationPrivate::get(opInfo.operation);
115 priv->setError(QIOOperation::Error::Aborted);
117 m_operations.clear();
119 releaseIoChannel(m_ioChannel);
126 m_fileState = FileState::Closed;
129qint64 QRandomAccessAsyncFileNativeBackend::size()
const
131 if (m_fileState != FileState::Opened)
134 QFileSystemMetaData metaData;
135 if (QFileSystemEngine::fillMetaData(m_fd, metaData))
136 return metaData.size();
142QRandomAccessAsyncFileNativeBackend::open(
const QString &path, QIODeviceBase::OpenMode mode)
144 if (m_fileState == FileState::Closed) {
149 m_fileState = FileState::OpenPending;
152 return addOperation<QIOOperation>(QIOOperation::Type::Open, 0);
155QIOOperation *QRandomAccessAsyncFileNativeBackend::flush()
157 return addOperation<QIOOperation>(QIOOperation::Type::Flush, 0);
160QIOReadOperation *QRandomAccessAsyncFileNativeBackend::read(qint64 offset, qint64 maxSize)
162 QByteArray array(maxSize, Qt::Uninitialized);
163 return addOperation<QIOReadOperation>(QIOOperation::Type::Read, offset, std::move(array));
166QIOWriteOperation *QRandomAccessAsyncFileNativeBackend::write(qint64 offset,
const QByteArray &data)
168 QByteArray copy = data;
169 return write(offset, std::move(copy));
172QIOWriteOperation *QRandomAccessAsyncFileNativeBackend::write(qint64 offset, QByteArray &&data)
174 return addOperation<QIOWriteOperation>(QIOOperation::Type::Write, offset, std::move(data));
177QIOVectoredReadOperation *
178QRandomAccessAsyncFileNativeBackend::readInto(qint64 offset, QSpan<std::byte> buffer)
180 return addOperation<QIOVectoredReadOperation>(QIOOperation::Type::Read, offset,
181 QSpan<
const QSpan<std::byte>>{buffer});
184QIOVectoredWriteOperation *
185QRandomAccessAsyncFileNativeBackend::writeFrom(qint64 offset, QSpan<
const std::byte> buffer)
187 return addOperation<QIOVectoredWriteOperation>(QIOOperation::Type::Write, offset,
188 QSpan<
const QSpan<
const std::byte>>{buffer});
191QIOVectoredReadOperation *
192QRandomAccessAsyncFileNativeBackend::readInto(qint64 offset, QSpan<
const QSpan<std::byte>> buffers)
202 return addOperation<QIOVectoredReadOperation>(QIOOperation::Type::Read, offset, buffers);
205QIOVectoredWriteOperation *
206QRandomAccessAsyncFileNativeBackend::writeFrom(qint64 offset, QSpan<
const QSpan<
const std::byte>> buffers)
208 return addOperation<QIOVectoredWriteOperation>(QIOOperation::Type::Write, offset, buffers);
211void QRandomAccessAsyncFileNativeBackend::notifyIfOperationsAreCompleted()
213 QMutexLocker locker(&m_mutex);
214 --m_numChannelsToClose;
215 if (m_opToCancel == kAllOperationIds) {
216 if (m_numChannelsToClose == 0 && m_runningOps.isEmpty())
217 m_cancellationCondition.wakeOne();
221dispatch_io_t QRandomAccessAsyncFileNativeBackend::createMainChannel(
int fd)
223 auto sharedThis =
this;
225 dispatch_io_create(DISPATCH_IO_RANDOM, fd,
226 dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0),
230 sharedThis->notifyIfOperationsAreCompleted();
233 QMutexLocker locker(&m_mutex);
234 ++m_numChannelsToClose;
239dispatch_io_t QRandomAccessAsyncFileNativeBackend::duplicateIoChannel(OperationId opId)
246 auto sharedThis =
this;
248 dispatch_io_create_with_io(DISPATCH_IO_RANDOM, m_ioChannel,
249 dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0),
251 sharedThis->notifyIfOperationsAreCompleted();
255 QMutexLocker locker(&m_mutex);
256 m_runningOps.insert(opId);
257 ++m_numChannelsToClose;
262void QRandomAccessAsyncFileNativeBackend::closeIoChannel(dispatch_io_t channel)
265 dispatch_io_close(channel, DISPATCH_IO_STOP);
268void QRandomAccessAsyncFileNativeBackend::releaseIoChannel(dispatch_io_t channel)
271 dispatch_release(channel);
276void QRandomAccessAsyncFileNativeBackend::handleOperationComplete(
const OperationResult &opResult)
279 auto onReturn = qScopeGuard([
this] {
280 startOperationsUntilBarrier();
283 auto it = std::find_if(m_operations.cbegin(), m_operations.cend(),
284 [opId = opResult.opId](
const auto &opInfo) {
285 return opInfo.opId == opId;
287 if (it == m_operations.cend())
289 qsizetype idx = std::distance(m_operations.cbegin(), it);
291 const OperationInfo info = m_operations.takeAt(idx);
292 closeIoChannel(info.channel);
293 releaseIoChannel(info.channel);
298 auto convertError = [](
int error, QIOOperation::Type type) {
300 return QIOOperation::Error::None;
301 }
else if (error == ECANCELED) {
302 return QIOOperation::Error::Aborted;
303 }
else if (error == EBADF) {
304 return QIOOperation::Error::FileNotOpen;
305 }
else if (error == EINVAL) {
307 case QIOOperation::Type::Read:
308 case QIOOperation::Type::Write:
309 return QIOOperation::Error::IncorrectOffset;
310 case QIOOperation::Type::Flush:
311 return QIOOperation::Error::Flush;
312 case QIOOperation::Type::Open:
313 return QIOOperation::Error::Open;
314 case QIOOperation::Type::Unknown:
315 Q_UNREACHABLE_RETURN(QIOOperation::Error::FileNotOpen);
319 case QIOOperation::Type::Read:
320 return QIOOperation::Error::Read;
321 case QIOOperation::Type::Write:
322 return QIOOperation::Error::Write;
323 case QIOOperation::Type::Flush:
324 return QIOOperation::Error::Flush;
325 case QIOOperation::Type::Open:
326 return QIOOperation::Error::Open;
327 case QIOOperation::Type::Unknown:
328 Q_UNREACHABLE_RETURN(QIOOperation::Error::FileNotOpen);
333 auto *priv = QIOOperationPrivate::get(info.operation);
334 switch (priv->type) {
335 case QIOOperation::Type::Read:
336 case QIOOperation::Type::Write:
337 priv->appendBytesProcessed(opResult.result);
340 if (priv->type == QIOOperation::Type::Read) {
341 auto dataStorage = priv->dataStorage.get();
342 auto processed = priv->processed;
343 if (dataStorage->containsByteArray()) {
344 QByteArray &array = dataStorage->getByteArray();
345 array.truncate(processed);
346 }
else if (dataStorage->containsReadSpans()) {
347 qint64 left = processed;
348 auto &readBuffers = dataStorage->getReadSpans();
349 for (
auto &s : readBuffers) {
350 const qint64 spanSize = qint64(s.size_bytes());
351 const qint64 newSize = (std::min)(left, spanSize);
352 if (newSize < spanSize)
353 s.chop(spanSize - newSize);
358 priv->operationComplete(convertError(opResult.error, priv->type));
360 case QIOOperation::Type::Flush: {
361 const QIOOperation::Error error = convertError(opResult.error, priv->type);
362 priv->operationComplete(error);
365 case QIOOperation::Type::Open: {
366 const QIOOperation::Error error = convertError(opResult.error, priv->type);
367 if (opResult.result >= 0 && error == QIOOperation::Error::None) {
368 m_fd = (
int)opResult.result;
369 m_ioChannel = createMainChannel(m_fd);
370 m_fileState = FileState::Opened;
372 m_fileState = FileState::Closed;
374 priv->operationComplete(error);
377 case QIOOperation::Type::Unknown:
383void QRandomAccessAsyncFileNativeBackend::queueCompletion(OperationId opId,
int error)
385 const OperationResult res = { opId, 0LL, error };
386 QMetaObject::invokeMethod(m_owner, [
this, res] {
387 handleOperationComplete(res);
388 }, Qt::QueuedConnection);
391void QRandomAccessAsyncFileNativeBackend::startOperationsUntilBarrier()
396 for (
auto &opInfo : m_operations) {
397 const bool isBarrier = isBarrierOperation(opInfo.operation->type());
398 const bool shouldExecute = (opInfo.state == OpState::Pending) && (!isBarrier || first);
401 opInfo.state = OpState::Running;
402 switch (opInfo.operation->type()) {
403 case QIOOperation::Type::Read:
406 case QIOOperation::Type::Write:
407 executeWrite(opInfo);
409 case QIOOperation::Type::Flush:
410 executeFlush(opInfo);
412 case QIOOperation::Type::Open:
415 case QIOOperation::Type::Unknown:
425void QRandomAccessAsyncFileNativeBackend::executeRead(OperationInfo &opInfo)
427 opInfo.channel = duplicateIoChannel(opInfo.opId);
428 if (!opInfo.channel) {
429 queueCompletion(opInfo.opId, EBADF);
432 auto priv = QIOOperationPrivate::get(opInfo.operation);
433 auto dataStorage = priv->dataStorage.get();
434 if (dataStorage->containsByteArray()) {
435 auto &array = dataStorage->getByteArray();
436 char *bytesPtr = array.data();
437 qint64 maxSize = array.size();
438 readOneBufferHelper(opInfo.opId, opInfo.channel, priv->offset,
442 Q_ASSERT(dataStorage->containsReadSpans());
443 auto &readBuffers = dataStorage->getReadSpans();
444 const auto totalBuffers = readBuffers.size();
445 if (totalBuffers == 0) {
446 queueCompletion(opInfo.opId, 0);
449 auto buf = readBuffers[0];
450 readOneBufferHelper(opInfo.opId, opInfo.channel, priv->offset,
451 buf.data(), buf.size(),
456void QRandomAccessAsyncFileNativeBackend::executeWrite(OperationInfo &opInfo)
458 opInfo.channel = duplicateIoChannel(opInfo.opId);
459 if (!opInfo.channel) {
460 queueCompletion(opInfo.opId, EBADF);
463 auto priv = QIOOperationPrivate::get(opInfo.operation);
464 auto dataStorage = priv->dataStorage.get();
465 if (dataStorage->containsByteArray()) {
466 const auto &array = dataStorage->getByteArray();
467 const char *dataPtr = array.constData();
468 const qint64 dataSize = array.size();
470 dispatch_queue_t queue = dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0);
475 dispatch_data_t dataToWrite = dispatch_data_create(dataPtr, dataSize, queue, ^{});
477 writeHelper(opInfo.opId, opInfo.channel, priv->offset, dataToWrite, dataSize);
479 Q_ASSERT(dataStorage->containsWriteSpans());
481 const auto &writeBuffers = dataStorage->getWriteSpans();
482 const auto totalBuffers = writeBuffers.size();
483 if (totalBuffers == 0) {
484 queueCompletion(opInfo.opId, 0);
488 dispatch_queue_t queue = dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0);
490 dispatch_data_t dataToWrite = dispatch_data_empty;
491 qint64 totalSize = 0;
493 const std::byte *dataPtr = writeBuffers[idx].data();
494 const qint64 dataSize = writeBuffers[idx].size();
495 dispatch_data_t data = dispatch_data_create(dataPtr, dataSize, queue, ^{});
496 dataToWrite = dispatch_data_create_concat(dataToWrite, data);
498 totalSize += dataSize;
499 }
while (++idx < totalBuffers);
501 writeHelper(opInfo.opId, opInfo.channel, priv->offset, dataToWrite, totalSize);
505void QRandomAccessAsyncFileNativeBackend::executeFlush(OperationInfo &opInfo)
507 opInfo.channel = duplicateIoChannel(opInfo.opId);
508 if (!opInfo.channel) {
509 queueCompletion(opInfo.opId, EBADF);
515 auto sharedThis =
this;
517 const OperationId opId = opInfo.opId;
518 dispatch_io_barrier(opInfo.channel, ^{
519 const int err = fsync(fd);
521 QMutexLocker locker(&sharedThis->m_mutex);
522 sharedThis->m_runningOps.remove(opId);
523 const auto cancelId = sharedThis->m_opToCancel;
524 if (cancelId == kAllOperationIds || cancelId == opId) {
525 if (cancelId == opId) {
526 sharedThis->m_cancellationCondition.wakeOne();
528 if (sharedThis->m_numChannelsToClose == 0
529 && sharedThis->m_runningOps.isEmpty()) {
530 sharedThis->m_cancellationCondition.wakeOne();
534 auto context = sharedThis->m_owner;
535 const OperationResult res = { opId, 0LL, err };
536 QMetaObject::invokeMethod(context, [sharedThis](
const OperationResult &r) {
537 sharedThis->handleOperationComplete(r);
538 }, Qt::QueuedConnection, res);
546 int oflags = QT_OPEN_RDONLY;
547#ifdef QT_LARGEFILE_SUPPORT
548 oflags |= QT_OPEN_LARGEFILE;
550 if ((mode & QIODevice::ReadWrite) == QIODevice::ReadWrite)
551 oflags = QT_OPEN_RDWR;
552 else if (mode & QIODevice::WriteOnly)
553 oflags = QT_OPEN_WRONLY;
554 if ((mode & QIODevice::WriteOnly)
555 && !(mode & QIODevice::ExistingOnly))
556 oflags |= QT_OPEN_CREAT;
557 if (mode & QIODevice::Truncate)
558 oflags |= QT_OPEN_TRUNC;
559 if (mode & QIODevice::Append)
560 oflags |= QT_OPEN_APPEND;
561 if (mode & QIODevice::NewOnly)
562 oflags |= QT_OPEN_EXCL;
566void QRandomAccessAsyncFileNativeBackend::executeOpen(OperationInfo &opInfo)
568 if (m_fileState != FileState::OpenPending) {
569 queueCompletion(opInfo.opId, EINVAL);
573 const QByteArray nativeName = QFile::encodeName(QDir::toNativeSeparators(m_filePath));
575 int openFlags = openModeToOpenFlags(m_openMode);
576 openFlags |= O_NONBLOCK;
578 auto sharedThis =
this;
579 const OperationId opId = opInfo.opId;
584 m_runningOps.insert(opId);
587 dispatch_async(dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0),
590 const int fd = ::open(nativeName.data(), openFlags);
594 QMutexLocker locker(&sharedThis->m_mutex);
595 sharedThis->m_runningOps.remove(opId);
596 const auto cancelId = sharedThis->m_opToCancel;
597 if (cancelId == kAllOperationIds || cancelId == opId) {
603 Q_ASSERT(sharedThis->m_runningOps.isEmpty());
604 Q_ASSERT(sharedThis->m_numChannelsToClose == 0);
605 sharedThis->m_cancellationCondition.wakeOne();
607 auto context = sharedThis->m_owner;
608 const OperationResult res = { opId, qint64(fd), err };
609 QMetaObject::invokeMethod(context,
610 [sharedThis](
const OperationResult &r) {
611 sharedThis->handleOperationComplete(r);
612 }, Qt::QueuedConnection, res);
617void QRandomAccessAsyncFileNativeBackend::readOneBuffer(OperationId opId, qsizetype bufferIdx,
623 auto it = std::find_if(m_operations.cbegin(), m_operations.cend(),
624 [opId](
const auto &opInfo) {
625 return opId == opInfo.opId;
627 if (it == m_operations.cend())
630 auto op = it->operation;
632 closeIoChannel(it->channel);
636 auto *priv = QIOOperationPrivate::get(op);
637 Q_ASSERT(priv->type == QIOOperation::Type::Read);
638 Q_ASSERT(priv->dataStorage->containsReadSpans());
640 auto &readBuffers = priv->dataStorage->getReadSpans();
641 Q_ASSERT(readBuffers.size() > bufferIdx);
643 qint64 newOffset = priv->offset;
644 for (qsizetype idx = 0; idx < bufferIdx; ++idx)
645 newOffset += readBuffers[idx].size();
647 std::byte *bytesPtr = readBuffers[bufferIdx].data();
648 qint64 maxSize = readBuffers[bufferIdx].size();
650 readOneBufferHelper(opId, it->channel, newOffset, bytesPtr, maxSize, bufferIdx,
651 readBuffers.size(), alreadyRead);
654void QRandomAccessAsyncFileNativeBackend::readOneBufferHelper(OperationId opId, dispatch_io_t channel,
655 qint64 offset,
void *bytesPtr,
656 qint64 maxSize, qsizetype currentBufferIdx,
657 qsizetype totalBuffers, qint64 alreadyRead)
659 auto sharedThis =
this;
660 __block size_t readFromBuffer = 0;
661 dispatch_io_read(channel, offset, maxSize,
662 dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0),
663 ^(
bool done, dispatch_data_t data,
int error) {
667 dispatch_data_apply(data, ^(dispatch_data_t , size_t offset,
668 const void *buffer, size_t size) {
669 const char *startPtr =
670 reinterpret_cast<
const char *>(buffer) + offset;
673 std::memcpy((std::byte *)bytesPtr + readFromBuffer,
675 readFromBuffer += size;
688 QMutexLocker locker(&sharedThis->m_mutex);
689 const auto cancelId = sharedThis->m_opToCancel;
690 if (cancelId == kAllOperationIds || cancelId == opId) {
691 sharedThis->m_runningOps.remove(opId);
692 if (cancelId == opId) {
693 sharedThis->m_cancellationCondition.wakeOne();
695 if (sharedThis->m_numChannelsToClose == 0
696 && sharedThis->m_runningOps.isEmpty()) {
697 sharedThis->m_cancellationCondition.wakeOne();
701 sharedThis->m_runningOps.remove(opId);
702 auto context = sharedThis->m_owner;
705 qint64 totalRead = qint64(readFromBuffer) + alreadyRead;
706 qsizetype nextBufferIdx = currentBufferIdx + 1;
707 if (error || nextBufferIdx == totalBuffers
708 || qint64(readFromBuffer) != maxSize) {
709 const OperationResult res = { opId, totalRead, error };
710 QMetaObject::invokeMethod(context,
711 [sharedThis](
const OperationResult &r) {
712 sharedThis->handleOperationComplete(r);
713 }, Qt::QueuedConnection, res);
716 QMetaObject::invokeMethod(context,
717 [sharedThis, opId, nextBufferIdx, totalRead] {
718 sharedThis->readOneBuffer(opId, nextBufferIdx, totalRead);
719 }, Qt::QueuedConnection);
725void QRandomAccessAsyncFileNativeBackend::writeHelper(OperationId opId, dispatch_io_t channel,
726 qint64 offset, dispatch_data_t dataToWrite,
729 auto sharedThis =
this;
730 dispatch_io_write(channel, offset, dataToWrite,
731 dispatch_get_global_queue(QOS_CLASS_USER_INTERACTIVE, 0),
732 ^(
bool done, dispatch_data_t data,
int error) {
741 QMutexLocker locker(&sharedThis->m_mutex);
742 const auto cancelId = sharedThis->m_opToCancel;
743 if (cancelId == kAllOperationIds || cancelId == opId) {
745 sharedThis->m_runningOps.remove(opId);
746 if (cancelId == opId) {
747 sharedThis->m_cancellationCondition.wakeOne();
749 if (sharedThis->m_numChannelsToClose == 0
750 && sharedThis->m_runningOps.isEmpty()) {
751 sharedThis->m_cancellationCondition.wakeOne();
755 sharedThis->m_runningOps.remove(opId);
759 const size_t toBeWritten =
760 (error == 0) ? 0 : dispatch_data_get_size(data);
761 const size_t written = dataSize - toBeWritten;
762 [dataToWrite release];
764 auto context = sharedThis->m_owner;
765 const OperationResult res = { opId, qint64(written), error };
766 QMetaObject::invokeMethod(context,
767 [sharedThis](
const OperationResult &r) {
768 sharedThis->handleOperationComplete(r);
769 }, Qt::QueuedConnection, res);
774QRandomAccessAsyncFileNativeBackend::OperationId QRandomAccessAsyncFileNativeBackend::getNextId()
777 static OperationId opId = kInvalidOperationId;
778 if (++opId == kAllOperationIds)
779 opId = kInvalidOperationId + 1;
Combined button and popup list for selecting options.
static bool isBarrierOperation(QIOOperation::Type type)
static int openModeToOpenFlags(QIODevice::OpenMode mode)