qohosmtblockingcallsgateway_p.h

Go to the documentation of this file.

// 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
 
#ifndef QOHOSMTBLOCKINGCALLSGATEWAY_H
#define QOHOSMTBLOCKINGCALLSGATEWAY_H
 
//
//  W A R N I N G
//  -------------
//
// This file is not part of the Qt API.  It exists purely as an
// implementation detail.  This header file may change from version to
// version without notice, or even be removed.
//
// We mean it.
//
 
#include <QtCore/private/qohoscommon_p.h>
#include <QtCore/qglobal.h>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <optional>
 
QT_BEGIN_NAMESPACE
 
namespace QtOhos {
 
template<typename SlaveContext>
class QOhosMtBlockingCallsGateway : public std::enable_shared_from_this<QOhosMtBlockingCallsGateway<SlaveContext>>
{
public:
    enum class MasterThreadTaskResult
    {
        CancelledByDeadlock,
        TimeoutedBeforeStart,
        TimeoutedAfterStart,
        Finished,
    };
 
    static std::shared_ptr<QOhosMtBlockingCallsGateway<SlaveContext>> makeInstance(
        QOhosConsumer<std::function<void()>> masterThreadTasksExecutor,
        QOhosConsumer<std::function<void(SlaveContext &)>> slaveThreadTasksExecutor);
 
    void invokeInMasterThread(std::function<void()> &&task);
    void invokeInSlaveThread(std::function<void(SlaveContext &)> &&task);
 
    void runInSlaveThreadAndWaitForContinue(
        std::function<void(SlaveContext &, QOhosTaskPromise<>)> &&task,
        const std::string &callerContextName = std::string());
    Q_REQUIRED_RESULT MasterThreadTaskResult tryInvokeInMasterThreadAndTryWaitForContinue(
        QOhosConsumer<std::function<void()>> &&task,
        std::chrono::nanoseconds timeout);
 
private:
    struct MasterThreadTaskState
    {
        bool started = false;
        std::optional<MasterThreadTaskResult> result;
        std::condition_variable resultSetCondVar;
    };
 
    QOhosMtBlockingCallsGateway(
        QOhosConsumer<std::function<void()>> masterThreadTasksExecutor,
        QOhosConsumer<std::function<void(SlaveContext &)>> slaveThreadTasksExecutor);
 
    QOhosConsumer<std::function<void()>> m_masterThreadTasksExecutor;
    QOhosConsumer<std::function<void(SlaveContext &)>> m_slaveThreadTasksExecutor;
    std::mutex m_waitStateMutex;
    bool m_masterWaiting = false;
    std::shared_ptr<MasterThreadTaskState> m_masterThreadTaskState;
};
 
template<typename SlaveContext>
std::shared_ptr<QOhosMtBlockingCallsGateway<SlaveContext>> QOhosMtBlockingCallsGateway<SlaveContext>::makeInstance(
    QOhosConsumer<std::function<void()>> masterThreadTasksExecutor,
    QOhosConsumer<std::function<void(SlaveContext &)>> slaveThreadTasksExecutor)
{
    return std::shared_ptr<QOhosMtBlockingCallsGateway<SlaveContext>>(
        new QOhosMtBlockingCallsGateway<SlaveContext>(
            std::move(masterThreadTasksExecutor), std::move(slaveThreadTasksExecutor)));
}
 
template<typename SlaveContext>
QOhosMtBlockingCallsGateway<SlaveContext>::QOhosMtBlockingCallsGateway(
    QOhosConsumer<std::function<void()>> masterThreadTasksExecutor,
    QOhosConsumer<std::function<void(SlaveContext &)>> slaveThreadTasksExecutor)
    : m_masterThreadTasksExecutor(std::move(masterThreadTasksExecutor))
    , m_slaveThreadTasksExecutor(std::move(slaveThreadTasksExecutor))
{
}
 
template<typename SlaveContext>
void QOhosMtBlockingCallsGateway<SlaveContext>::invokeInMasterThread(std::function<void()> &&task)
{
    m_masterThreadTasksExecutor(std::move(task));
}
 
template<typename SlaveContext>
void QOhosMtBlockingCallsGateway<SlaveContext>::invokeInSlaveThread(std::function<void(SlaveContext &)> &&task)
{
    m_slaveThreadTasksExecutor(std::move(task));
}
 
template<typename SlaveContext>
void QOhosMtBlockingCallsGateway<SlaveContext>::runInSlaveThreadAndWaitForContinue(
    std::function<void(SlaveContext &, QOhosTaskPromise<>)> &&task,
    const std::string &callerContextName)
{
    const auto funcInfo = Q_FUNC_INFO;
 
    {
        std::lock_guard<std::mutex> waitStateLock(m_waitStateMutex);
        if (m_masterThreadTaskState) {
            auto taskState = std::exchange(m_masterThreadTaskState, nullptr);
            taskState->result = MasterThreadTaskResult::CancelledByDeadlock;
            taskState->resultSetCondVar.notify_all();
        }
        m_masterWaiting = true;
    }
 
    auto taskFinishedPromise = std::make_shared<std::promise<void>>();
    auto taskFinishedFuture = taskFinishedPromise->get_future();
 
    auto sharedTaskPromise = std::make_shared<QOhosTaskPromise<>>(
        [taskFinishedPromise]() {
            taskFinishedPromise->set_value();
        },
        [funcInfo, callerContextName]() {
            qOhosReportFatalErrorAndAbort(
                "%s: promise destroyed without notifying the caller: %s",
                funcInfo, callerContextName.c_str());
        },
        callerContextName);
 
    invokeInSlaveThread(
        [task = std::move(task), sharedTaskPromise](SlaveContext &context) {
            task(context, std::move(*sharedTaskPromise));
        });
    taskFinishedFuture.wait();
 
    {
        std::lock_guard<std::mutex> waitStateLock(m_waitStateMutex);
        m_masterWaiting = false;
    }
}
 
template<typename SlaveContext>
typename QOhosMtBlockingCallsGateway<SlaveContext>::MasterThreadTaskResult
QOhosMtBlockingCallsGateway<SlaveContext>::tryInvokeInMasterThreadAndTryWaitForContinue(
    QOhosConsumer<std::function<void()>> &&task,
    std::chrono::nanoseconds timeout)
{
    auto taskState = std::make_shared<MasterThreadTaskState>();
 
    {
        std::lock_guard<std::mutex> waitStateLock(m_waitStateMutex);
        if (m_masterWaiting)
            return MasterThreadTaskResult::CancelledByDeadlock;
        m_masterThreadTaskState = taskState;
    }
 
    auto weakSelf = std::weak_ptr<QOhosMtBlockingCallsGateway<SlaveContext>>(this->shared_from_this());
 
    auto continueFunc = [weakSelf, taskState]() {
        auto self = weakSelf.lock();
        if (self) {
            std::lock_guard<std::mutex> waitStateLock(self->m_waitStateMutex);
            if (self->m_masterThreadTaskState == taskState) {
                self->m_masterThreadTaskState.reset();
                taskState->result = MasterThreadTaskResult::Finished;
                taskState->resultSetCondVar.notify_all();
            }
        }
    };
 
    invokeInMasterThread(
        [weakSelf, taskState, task = std::move(task), continueFunc = std::move(continueFunc)]() mutable {
            auto self = weakSelf.lock();
            if (self) {
                bool upToDate;
                {
                    std::lock_guard<std::mutex> waitStateLock(self->m_waitStateMutex);
                    if (self->m_masterThreadTaskState == taskState)
                        self->m_masterThreadTaskState->started = true;
                    upToDate = self->m_masterThreadTaskState == taskState;
                }
                if (upToDate)
                    task(std::move(continueFunc));
            }
        });
 
    MasterThreadTaskResult result;
    {
        std::unique_lock<std::mutex> waitStateLock(m_waitStateMutex);
        taskState->resultSetCondVar.wait_for(
            waitStateLock, timeout,
            [&]() {
                return taskState->result.has_value();
            });
        result = taskState->result.value_or(
            taskState->started
                ? MasterThreadTaskResult::TimeoutedAfterStart
                : MasterThreadTaskResult::TimeoutedBeforeStart);
    }
 
    return result;
}
 
}
 
QT_END_NAMESPACE
 
#endif