strawberry-audio-player-win.../ext/libstrawberry-common/core/workerpool.h

430 lines
13 KiB
C
Raw Normal View History

2018-02-27 18:06:05 +01:00
/* This file is part of Strawberry.
Copyright 2011, David Sansome <me@davidsansome.com>
2021-04-10 07:32:38 +02:00
Copyright 2018-2021, Jonas Kvinge <jonas@jkvinge.net>
2018-02-27 18:06:05 +01:00
Strawberry is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
Strawberry is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with Strawberry. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef WORKERPOOL_H
#define WORKERPOOL_H
2020-06-14 23:54:18 +02:00
#include <cstddef>
#include <QtGlobal>
#include <QObject>
2018-02-27 18:06:05 +01:00
#include <QCoreApplication>
#include <QThread>
2018-02-27 18:06:05 +01:00
#include <QMutex>
2020-02-08 03:40:30 +01:00
#include <QLocalServer>
2018-02-27 18:06:05 +01:00
#include <QProcess>
#include <QFile>
#include <QList>
2020-02-08 03:40:30 +01:00
#include <QQueue>
#include <QString>
#include <QStringList>
#include <QAtomicInt>
#if QT_VERSION >= QT_VERSION_CHECK(5, 10, 0)
# include <QRandomGenerator>
#endif
2018-02-27 18:06:05 +01:00
#include "core/logging.h"
class QLocalSocket;
// Base class containing signals and slots - required because moc doesn't do templated objects.
2018-02-27 18:06:05 +01:00
class _WorkerPoolBase : public QObject {
Q_OBJECT
public:
2020-06-26 22:41:38 +02:00
explicit _WorkerPoolBase(QObject *parent = nullptr);
2018-02-27 18:06:05 +01:00
2020-11-19 18:22:57 +01:00
signals:
// Emitted when a worker failed to start. This usually happens when the worker wasn't found, or couldn't be executed.
2018-02-27 18:06:05 +01:00
void WorkerFailedToStart();
2020-11-19 18:22:57 +01:00
protected slots:
2018-02-27 18:06:05 +01:00
virtual void DoStart() {}
virtual void NewConnection() {}
virtual void ProcessError(QProcess::ProcessError) {}
virtual void SendQueuedMessages() {}
};
// Manages a pool of one or more external processes.
// A local socket server is started for each process, and the address is passed to the process as argv[1].
// The process is expected to connect back to the socket server, and when it does a HandlerType is created for it.
2018-02-27 18:06:05 +01:00
// Instances of HandlerType are created in the WorkerPool's thread.
template <typename HandlerType>
class WorkerPool : public _WorkerPoolBase {
public:
2020-06-26 22:41:38 +02:00
explicit WorkerPool(QObject *parent = nullptr);
2020-06-26 23:01:57 +02:00
~WorkerPool() override;
2018-02-27 18:06:05 +01:00
typedef typename HandlerType::MessageType MessageType;
typedef typename HandlerType::ReplyType ReplyType;
// Sets the name of the worker executable. This is looked for first in the current directory, and then in $PATH.
// You must call this before calling Start().
void SetExecutableName(const QString &executable_name);
2018-02-27 18:06:05 +01:00
// Sets the number of worker process to use. Defaults to 1 <= (processors / 2) <= 2.
2020-11-19 18:22:57 +01:00
void SetWorkerCount(const int count);
2018-02-27 18:06:05 +01:00
// Sets the prefix to use for the local server (on unix this is a named pipe in /tmp).
// Defaults to QApplication::applicationName().
// A random number is appended to this name when creating each server.
void SetLocalServerName(const QString &local_server_name);
2018-02-27 18:06:05 +01:00
// Starts all workers.
void Start();
// Fills in the message's "id" field and creates a reply future.
// The message is queued and the WorkerPool's thread will send it to the next available worker.
// Can be called from any thread.
ReplyType *SendMessageWithReply(MessageType *message);
2018-02-27 18:06:05 +01:00
2020-11-19 18:22:57 +01:00
protected:
// These are all reimplemented slots, they are called on the WorkerPool's thread.
2020-06-26 23:01:57 +02:00
void DoStart() override;
void NewConnection() override;
void ProcessError(QProcess::ProcessError error) override;
void SendQueuedMessages() override;
2018-02-27 18:06:05 +01:00
2020-11-19 18:22:57 +01:00
private:
2018-02-27 18:06:05 +01:00
struct Worker {
2019-02-20 21:26:14 +01:00
Worker() : local_server_(nullptr), local_socket_(nullptr), process_(nullptr), handler_(nullptr) {}
2018-02-27 18:06:05 +01:00
QLocalServer *local_server_;
QLocalSocket *local_socket_;
QProcess *process_;
HandlerType *handler_;
2018-02-27 18:06:05 +01:00
};
// Must only ever be called on my thread.
void StartOneWorker(Worker *worker);
2018-02-27 18:06:05 +01:00
template <typename T>
Worker *FindWorker(T Worker::*member, T value) {
2019-04-12 19:55:19 +02:00
for (typename QList<Worker>::iterator it = workers_.begin() ; it != workers_.end() ; ++it) {
2018-02-27 18:06:05 +01:00
if ((*it).*member == value) {
return &(*it);
}
}
2019-02-20 21:26:14 +01:00
return nullptr;
2018-02-27 18:06:05 +01:00
}
template <typename T>
void DeleteQObjectPointerLater(T **p) {
2018-02-27 18:06:05 +01:00
if (*p) {
(*p)->deleteLater();
2019-02-20 21:26:14 +01:00
*p = nullptr;
2018-02-27 18:06:05 +01:00
}
}
// Creates a new reply future for the request with the next sequential ID,
// and sets the request's ID to the ID of the reply. Can be called from any thread
ReplyType *NewReply(MessageType *message);
2018-02-27 18:06:05 +01:00
2019-02-20 21:26:14 +01:00
// Returns the next handler, or nullptr if there isn't one. Must be called from my thread.
HandlerType *NextHandler() const;
2018-02-27 18:06:05 +01:00
2020-11-19 18:22:57 +01:00
private:
2018-02-27 18:06:05 +01:00
QString local_server_name_;
QString executable_name_;
QString executable_path_;
int worker_count_;
mutable int next_worker_;
QList<Worker> workers_;
QAtomicInt next_id_;
2018-02-27 18:06:05 +01:00
QMutex message_queue_mutex_;
QQueue<ReplyType*> message_queue_;
};
template <typename HandlerType>
WorkerPool<HandlerType>::WorkerPool(QObject *parent)
2018-02-27 18:06:05 +01:00
: _WorkerPoolBase(parent),
next_worker_(0),
2019-04-12 19:55:19 +02:00
next_id_(0) {
2019-10-19 15:09:18 +02:00
2019-11-27 22:57:51 +01:00
worker_count_ = qBound(1, QThread::idealThreadCount() / 2, 4);
2018-02-27 18:06:05 +01:00
local_server_name_ = qApp->applicationName().toLower();
2021-06-20 19:04:08 +02:00
if (local_server_name_.isEmpty()) {
2018-02-27 18:06:05 +01:00
local_server_name_ = "workerpool";
2021-06-20 19:04:08 +02:00
}
2019-10-19 15:09:18 +02:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
WorkerPool<HandlerType>::~WorkerPool() {
2020-11-19 18:22:57 +01:00
for (const Worker &worker : workers_) {
2018-02-27 18:06:05 +01:00
if (worker.local_socket_ && worker.process_) {
2021-01-26 16:48:04 +01:00
QObject::disconnect(worker.process_, &QProcess::errorOccurred, this, &WorkerPool::ProcessError);
2018-02-27 18:06:05 +01:00
// The worker is connected. Close his socket and wait for him to exit.
qLog(Debug) << "Closing worker socket";
worker.local_socket_->close();
worker.process_->waitForFinished(500);
}
if (worker.process_ && worker.process_->state() == QProcess::Running) {
// The worker is still running - kill it.
qLog(Debug) << "Killing worker process";
worker.process_->terminate();
if (!worker.process_->waitForFinished(500)) {
worker.process_->kill();
}
}
}
for (ReplyType *reply : message_queue_) {
2018-02-27 18:06:05 +01:00
reply->Abort();
}
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
2020-11-19 18:22:57 +01:00
void WorkerPool<HandlerType>::SetWorkerCount(const int count) {
2018-02-27 18:06:05 +01:00
Q_ASSERT(workers_.isEmpty());
worker_count_ = count;
}
template <typename HandlerType>
void WorkerPool<HandlerType>::SetLocalServerName(const QString &local_server_name) {
2018-02-27 18:06:05 +01:00
Q_ASSERT(workers_.isEmpty());
local_server_name_ = local_server_name;
}
template <typename HandlerType>
void WorkerPool<HandlerType>::SetExecutableName(const QString &executable_name) {
2018-02-27 18:06:05 +01:00
Q_ASSERT(workers_.isEmpty());
executable_name_ = executable_name;
}
template <typename HandlerType>
void WorkerPool<HandlerType>::Start() {
2021-09-09 21:53:14 +02:00
QMetaObject::invokeMethod(this, "DoStart");
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
void WorkerPool<HandlerType>::DoStart() {
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
Q_ASSERT(workers_.isEmpty());
Q_ASSERT(!executable_name_.isEmpty());
Q_ASSERT(QThread::currentThread() == thread());
// Find the executable if we can, default to searching $PATH
executable_path_ = executable_name_;
QStringList search_path;
search_path << QCoreApplication::applicationDirPath();
2019-01-03 21:19:07 +01:00
#if defined(Q_OS_MACOS) && defined(USE_BUNDLE)
search_path << QCoreApplication::applicationDirPath() + "/" + USE_BUNDLE_DIR;
2018-02-27 18:06:05 +01:00
#endif
for (const QString &path_prefix : search_path) {
2018-02-27 18:06:05 +01:00
const QString executable_path = path_prefix + "/" + executable_name_;
if (QFile::exists(executable_path)) {
executable_path_ = executable_path;
2020-12-24 20:44:16 +01:00
qLog(Debug) << "Using worker" << executable_name_ << "from" << path_prefix;
2018-02-27 18:06:05 +01:00
break;
}
}
2020-12-24 20:44:16 +01:00
if (executable_path_ == executable_name_) {
qLog(Debug) << "Using worker" << executable_name_;
}
2018-02-27 18:06:05 +01:00
// Start all the workers
for (int i = 0; i < worker_count_; ++i) {
Worker worker;
StartOneWorker(&worker);
workers_ << worker;
}
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
void WorkerPool<HandlerType>::StartOneWorker(Worker *worker) {
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
Q_ASSERT(QThread::currentThread() == thread());
DeleteQObjectPointerLater(&worker->local_server_);
DeleteQObjectPointerLater(&worker->local_socket_);
DeleteQObjectPointerLater(&worker->process_);
DeleteQObjectPointerLater(&worker->handler_);
worker->local_server_ = new QLocalServer(this);
worker->process_ = new QProcess(this);
2021-01-26 16:48:04 +01:00
QObject::connect(worker->local_server_, &QLocalServer::newConnection, this, &WorkerPool::NewConnection);
QObject::connect(worker->process_, &QProcess::errorOccurred, this, &WorkerPool::ProcessError);
2018-02-27 18:06:05 +01:00
// Create a server, find an unused name and start listening
forever {
#if QT_VERSION >= QT_VERSION_CHECK(5, 10, 0)
2021-04-10 08:21:58 +02:00
const quint32 unique_number = QRandomGenerator::global()->bounded(static_cast<quint32>(quint64(this) & 0xFFFFFFFF));
#else
2021-04-10 08:21:58 +02:00
const quint32 unique_number = qrand() ^ (static_cast<quint32>(quint64(this) & 0xFFFFFFFF));
#endif
2018-02-27 18:06:05 +01:00
const QString name = QString("%1_%2").arg(local_server_name_).arg(unique_number);
if (worker->local_server_->listen(name)) {
break;
}
}
qLog(Debug) << "Starting worker" << worker << executable_path_ << worker->local_server_->fullServerName();
// Start the process
#if defined(Q_OS_WIN32) && defined(QT_NO_DEBUG_OUTPUT) && !defined(ENABLE_WIN32_CONSOLE)
worker->process_->setProcessChannelMode(QProcess::SeparateChannels);
#else
2018-02-27 18:06:05 +01:00
worker->process_->setProcessChannelMode(QProcess::ForwardedChannels);
#endif
2018-02-27 18:06:05 +01:00
worker->process_->start(executable_path_, QStringList() << worker->local_server_->fullServerName());
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
void WorkerPool<HandlerType>::NewConnection() {
Q_ASSERT(QThread::currentThread() == thread());
QLocalServer *server = qobject_cast<QLocalServer*>(sender());
// Find the worker with this server.
Worker *worker = FindWorker(&Worker::local_server_, server);
2018-02-27 18:06:05 +01:00
if (!worker) return;
qLog(Debug) << "Worker" << worker << "connected to" << server->fullServerName();
// Accept the connection.
worker->local_socket_ = server->nextPendingConnection();
// We only ever accept one connection per worker, so destroy the server now.
worker->local_socket_->setParent(this);
worker->local_server_->deleteLater();
2019-02-20 21:26:14 +01:00
worker->local_server_ = nullptr;
2018-02-27 18:06:05 +01:00
// Create the handler.
worker->handler_ = new HandlerType(worker->local_socket_, this);
SendQueuedMessages();
}
template <typename HandlerType>
void WorkerPool<HandlerType>::ProcessError(QProcess::ProcessError error) {
Q_ASSERT(QThread::currentThread() == thread());
QProcess *process = qobject_cast<QProcess*>(sender());
// Find the worker with this process.
Worker *worker = FindWorker(&Worker::process_, process);
if (!worker) return;
switch (error) {
case QProcess::FailedToStart:
// Failed to start errors are bad - it usually means the worker isn't installed.
// Don't restart the process, but tell our owner, who will probably want to do something fatal.
2018-02-27 18:06:05 +01:00
qLog(Error) << "Worker failed to start";
emit WorkerFailedToStart();
break;
default:
// On any other error we just restart the process.
2019-02-20 21:26:14 +01:00
qLog(Debug) << "Worker" << worker << "failed with error" << error << "- restarting";
2018-02-27 18:06:05 +01:00
StartOneWorker(worker);
break;
}
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
typename WorkerPool<HandlerType>::ReplyType*
WorkerPool<HandlerType>::NewReply(MessageType *message) {
2020-11-19 18:22:57 +01:00
const int id = next_id_.fetchAndAddOrdered(1);
2018-02-27 18:06:05 +01:00
message->set_id(id);
return new ReplyType(*message);
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
typename WorkerPool<HandlerType>::ReplyType*
WorkerPool<HandlerType>::SendMessageWithReply(MessageType *message) {
2020-11-19 18:22:57 +01:00
ReplyType *reply = NewReply(message);
2018-02-27 18:06:05 +01:00
// Add the pending reply to the queue
{
QMutexLocker l(&message_queue_mutex_);
message_queue_.enqueue(reply);
}
// Wake up the main thread
2021-09-09 21:53:14 +02:00
QMetaObject::invokeMethod(this, "SendQueuedMessages", Qt::QueuedConnection);
2018-02-27 18:06:05 +01:00
return reply;
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
void WorkerPool<HandlerType>::SendQueuedMessages() {
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
QMutexLocker l(&message_queue_mutex_);
while (!message_queue_.isEmpty()) {
ReplyType *reply = message_queue_.dequeue();
// Find a worker for this message
HandlerType *handler = NextHandler();
2018-02-27 18:06:05 +01:00
if (!handler) {
// No available handlers - put the message on the front of the queue.
message_queue_.prepend(reply);
qLog(Debug) << "No available handlers to process request";
break;
}
handler->SendRequest(reply);
}
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
template <typename HandlerType>
HandlerType *WorkerPool<HandlerType>::NextHandler() const {
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
for (int i = 0; i < workers_.count(); ++i) {
const int worker_index = (next_worker_ + i) % workers_.count();
2019-02-20 21:26:14 +01:00
if (workers_[worker_index].handler_ && !workers_[worker_index].handler_->is_device_closed()) {
2018-02-27 18:06:05 +01:00
next_worker_ = (worker_index + 1) % workers_.count();
return workers_[worker_index].handler_;
}
}
2019-02-20 21:26:14 +01:00
return nullptr;
2020-11-19 18:22:57 +01:00
2018-02-27 18:06:05 +01:00
}
#endif // WORKERPOOL_H