QThread + QtSql the right way22.12.2015 23:03

Сегодняшняя статья вдохновила меня поделиться своим способом вынесения баз данных в отдельный тред. Способ подходит не только для БД, но и для любых взаимодействий, описываемых паттерном «в отдельном потоке живёт какой-то объект, надо у него что-то спрашивать и что-то с ним делать». Кроме того, способ хорош тем, что он пытается быть типобезопасным и расширяемым: никаких stringly-typed QMetaObject::invokeMethod(), никаких передач результатов дёрганья объекта в потоке через сигналы. Только прямой вызов функций, только QFuture!
Disclaimer: приведенный здесь код является куском одного моего большого проекта, поэтому в нём будут использоваться некоторые вспомогательные библиотечные функции из этого проекта. Я, однако, постараюсь не упустить случаи такого использования и опишу их семантику.

Итак, начнём с самого главного: как бы нам хотелось работать с объектом в отдельном потоке? В идеале мы просто дёргаем какие-то методы у какого-то объекта, методы возвращают QFuture, готовность которых будет означать, что соответствующий асинхронный метод закончил выполняться, и у него появились результаты типа T.

Вспомним, что декомпозиция — наш друг, поэтому возьмём нашу исходную задачу «надо дёргать что-то в отдельном треде» и рассмотрим её кусочек «надо держать отдельный поток и обеспечивать потокобезопасный вызов чего-то в нём с возвратом QFuture».

Решим эту задачу следующим образом: объект-наследник QThread, отвечающий за управление потоком, имеет метод ScheduleImpl(), вызываемый из основного потока (да и из других тоже), принимающий некоторый функтор, оборачивающий этот функтор в QFuture и сохраняющий всё, что нужно, в специальную очередь, которая затем обработается внутри QThread::run().

Получится что-то такое:

class WorkerThreadBase : public QThread
{
        Q_OBJECT

        QMutex FunctionsMutex_;
        QList> Functions_;
public:
        using QThread::QThread;
protected:
        void run () override;

        virtual void Initialize () = 0;
        virtual void Cleanup () = 0;

        template
        QFuture> ScheduleImpl (const F& func)
        {
                QFutureInterface> iface;
                iface.reportStarted ();

                auto reporting = [func, iface] () mutable
                {
                        ReportFutureResult (iface, func);
                };

                {
                        QMutexLocker locker { &FunctionsMutex_ };
                        Functions_ << reporting;
                }

                emit rotateFuncs ();

                return iface.future ();
        }
private:
        void RotateFuncs ();
signals:
        void rotateFuncs ();
};

template
using ResultOf_t = typename std::result_of::type;

template
EnableIf_t::value>
        ReportFutureResult (QFutureInterface& iface, F&& f, Args... args)
{
        try
        {
                const auto result = f (args...);
                iface.reportFinished (&result);
        }
        catch (const QtException_t& e)
        {
                iface.reportException (e);
                iface.reportFinished ();
        }
        catch (const std::exception& e)
        {
                iface.reportException (ConcurrentStdException { e });
                iface.reportFinished ();
        }
}

template
void ReportFutureResult (QFutureInterface& iface, F&& f, Args... args)
{
        try
        {
                f (args...);
        }
        catch (const QtException_t& e)
        {
                iface.reportException (e);
        }
        catch (const std::exception& e)
        {
                iface.reportException (ConcurrentStdException { e });
        }

        iface.reportFinished ();
}

#if QT_VERSION < 0x050000
        using QtException_t = QtConcurrent::Exception;
#else
        using QtException_t = QException;
#endif

То есть, ScheduleImpl() принимает некий функтор с сигнатурой типа T (), возвращает QFuture, оборачивает функтор в специальную функцию, теперь уже с сигнатурой void (), связанную с возвращаемым QFuture, и которая по выполнению функтора этот QFuture пометит готовым, и добавляет эту обёртку в очередь

После этого излучается сигнал rotateFuncs(), который внутри run() соединяется с методом RotateFuncs(), как раз ответственным за обработку очереди сохраненных обёрток функторов.

Посмотрим теперь на реализацию методов run() и RotateFuncs():

void WorkerThreadBase::run ()
{
        SlotClosure rotator
        {
                [this] { RotateFuncs (); },
                this,
                SIGNAL (rotateFuncs ()),
                nullptr
        };

        Initialize ();

        QThread::run ();

        Cleanup ();
}

void WorkerThreadBase::RotateFuncs ()
{
        decltype (Functions_) funcs;

        {
                QMutexLocker locker { &FunctionsMutex_ };

                using std::swap;
                swap (funcs, Functions_);
        }

        for (const auto& func : funcs)
                func ();
}

С этими функциями всё просто: подсоединяем сигнал rotateFuncs() к функции RotateFuncs() (хм, интересно, сколько комментариев будет на тему стиля именования?), вызываем функцию инициализации объектов потока, определённую где-то в наследнике, и начинаем крутить поток. Когда владелец потока сделает потоку quit(), QThread::run() вернёт управление, и наследник сможет подчистить за собой в Cleanup().

Отметим, что именно Qt’шный механизм сигналов-слотов ответственен за то, чтобы rotateFuncs(), излучаемый из основного потока, вызывал RotateFuncs() в потоке нашего WorkerThreadBase.

RotateFuncs() же ненадолго блокирует основную очередь, перемещая её к себе, после чего начинает её последовательно выполнять.

Собственно, на этом и всё. В качестве примера использования можно привести, например, кусочек системы хранения аватарок на диске в IM-клиенте:

class AvatarsStorageThread final : public Util::WorkerThreadBase
{
        std::unique_ptr Storage_;
public:
        using Util::WorkerThreadBase::WorkerThreadBase;

        QFuture SetAvatar (const QString& entryId, IHaveAvatars::Size size, const QByteArray& imageData);
        QFuture> GetAvatar (const QString& entryId, IHaveAvatars::Size size);
        QFuture DeleteAvatars (const QString& entryId);
protected:
        void Initialize () override;
        void Cleanup () override;
};

QFuture AvatarsStorageThread::SetAvatar (const QString& entryId,
                IHaveAvatars::Size size, const QByteArray& imageData)
{
        return ScheduleImpl ([=] { Storage_->SetAvatar (entryId, size, imageData); });
}

QFuture> AvatarsStorageThread::GetAvatar (const QString& entryId, IHaveAvatars::Size size)
{
        return ScheduleImpl ([=] { return Storage_->GetAvatar (entryId, size); });
}

QFuture AvatarsStorageThread::DeleteAvatars (const QString& entryId)
{
        return ScheduleImpl ([=] { Storage_->DeleteAvatars (entryId); });
}

void AvatarsStorageThread::Initialize ()
{
        Storage_.reset (new AvatarsStorageOnDisk);
}

void AvatarsStorageThread::Cleanup ()
{
        Storage_.reset ();
}

А вот реализация AvatarsStorageOnDisk — отдельная интересная тема, связанная с моим доморощенным недо-ORM-фреймворком, позволяющим генерировать таблички, SQL-запросы и соответствующие функции для вставки/удаления/обновления по описанию структуры с данными через Boost.Fusion. Впрочем, конкретно к вопросу многопоточности эта реализация никак не относится, что, в общем-то, и хорошо, особенно с точки зрения декомпозиции исходной задачи.

Ну и напоследок, отметим недостатки предложенного решения:

1. Нужно дублировать в публичном API класса-наследника WorkerThreadBase все методы, которые захочется вызвать у выносимого в отдельный поток объекта. Как достаточно эффективно решить эту проблему, я сходу не придумал.
2. Initialize() и Cleanup() прямо просятся обернуться в какое-нибудь RAII. Стоит придумать что-нибудь на эту тему.

© Habrahabr.ru