WIP fix exiting...

Author: Skycoder42

src/datasync/exchangeengine.cpp

@@ -9,7 +9,8 @@
 #include "syncmanager_p.h"
 #include "changeemitter_p.h"
 
-#include <QtCore/QThread>
+#include <QtCore/QDebug>
+#include <QtCore/QCoreApplication>
 
 using namespace QtDataSync;
 
@@ -26,6 +27,11 @@ ExchangeEngine::ExchangeEngine(const QString &setupName, Setup::FatalErrorHandle
 	_emitter{new ChangeEmitter(_defaults, this)} //must be created here, because of access
 {}
 
+ExchangeEngine::~ExchangeEngine()
+{
+	logDebug() << "Finalization completed";
+}
+
 void ExchangeEngine::enterFatalState(const QString &error, const char *file, int line, const char *function, const char *category)
 {
 	QMessageLogContext context {file, line, function, category};
@@ -168,10 +174,8 @@ void ExchangeEngine::finalize()
 
 	//remoteconnector is the only one asynchronous (for now)
 	connect(_remoteConnector, &RemoteConnector::finalized,
-			this, [this](){
-		logDebug() << "Finalization completed";
-		thread()->quit();
-	});
+			thread(), &QThread::quit,
+			Qt::DirectConnection); //TODO unsafe?
 
 	_syncController->finalize();
 	_changeController->finalize();
@@ -335,3 +339,97 @@ ExchangeEngine::ImportData::ImportData(QJsonObject data, QString password, bool
 	keepData{keepData},
 	allowFailure{allowFailure}
 {}
+
+// ------------- Engine Thread -------------
+
+EngineThread::EngineThread(QString setupName, ExchangeEngine *engine, QLockFile *lockFile) :
+	_name{std::move(setupName)},
+	_engine{engine},
+	_lockFile{lockFile}
+{
+	setTerminationEnabled(true);
+	_engine.load()->moveToThread(this);
+
+	connect(thread(), &QThread::finished,
+			this, &EngineThread::stopSelf);
+	//TODO dangerous? find better way
+//	connect(this, &QThread::finished,
+//			this, &QThread::deleteLater);
+}
+
+EngineThread::~EngineThread()
+{
+	Q_ASSERT_X(!isRunning(), Q_FUNC_INFO, "Engine thread destroyed while engine is still running");
+}
+
+QString EngineThread::name() const
+{
+	return _name;
+}
+
+ExchangeEngine *EngineThread::engine() const
+{
+	return _engine.load();
+}
+
+bool EngineThread::isRunning() const
+{
+	return _running;
+}
+
+bool EngineThread::startEngine()
+{
+	if(!_running.testAndSetRelaxed(false, true))
+		return false;
+	start();
+	return true;
+}
+
+bool EngineThread::stopEngine()
+{
+	if(!_running.testAndSetRelaxed(true, false))
+		return false;
+	qCDebug(qdssetup) << "Finalizing engine of setup" << _name;
+	QMetaObject::invokeMethod(_engine, "finalize", Qt::QueuedConnection);
+	return true;
+}
+
+void EngineThread::waitAndTerminate(unsigned long timeout)
+{
+	if(!isRunning())
+		return;
+	if(!wait(timeout)) {
+		qCWarning(qdssetup) << "Workerthread of setup" << _name << "did not finish before the timout. Terminating...";
+		terminate();
+		auto wRes = wait(500);
+		qCDebug(qdssetup) << "Terminate result for setup" << _name << ":"
+						  << wRes;
+	}
+}
+
+void EngineThread::run()
+{
+	try {
+		qCDebug(qdssetup) << "Engine thread for setup" << _name << "has started";
+		_engine.load()->initialize();
+		exec();
+		_running = false;
+		delete _engine;
+		_lockFile->unlock();
+		delete _lockFile;
+		qCDebug(qdssetup) << "Engine thread for setup" << _name << "has stopped";
+	} catch(QException &e) {
+		if(_lockFile)
+			_lockFile->unlock();
+		if(_engine)
+			_engine.load()->enterFatalState(QString::fromUtf8(e.what()), QT_MESSAGELOG_FILE, QT_MESSAGELOG_LINE, QT_MESSAGELOG_FUNC, "qtdatasync.setup");
+		else
+			qFatal("%s", e.what());
+	}
+}
+
+void EngineThread::stopSelf()
+{
+	stopEngine();
+	waitAndTerminate(SetupPrivate::currentTimeout());
+}

src/datasync/exchangeengine_p.h

@@ -3,6 +3,8 @@
 
 #include <QtCore/QObject>
 #include <QtCore/QAtomicPointer>
+#include <QtCore/QThread>
+#include <QtCore/QLockFile>
 
 #include <QtRemoteObjects/QRemoteObjectHost>
 
@@ -45,6 +47,7 @@ class Q_DATASYNC_EXPORT ExchangeEngine : public QObject
 
 	explicit ExchangeEngine(const QString &setupName,
 							Setup::FatalErrorHandler errorHandler);
+	~ExchangeEngine();
 
 	Q_NORETURN void enterFatalState(const QString &error,
 									const char *file,
@@ -116,6 +119,36 @@ private Q_SLOTS:
 	void resetProgress(Controller *controller = nullptr);
 };
 
+class EngineThread : public QThread
+{
+	Q_OBJECT
+
+public:
+	EngineThread(QString setupName, ExchangeEngine *engine, QLockFile *lockFile);
+	~EngineThread() override;
+
+	QString name() const;
+	ExchangeEngine *engine() const;
+	bool isRunning() const;
+
+	bool startEngine();
+	bool stopEngine();
+
+	void waitAndTerminate(unsigned long timeout);
+
+protected:
+	void run() override;
+
+private Q_SLOTS:
+	void stopSelf();
+
+private:
+	const QString _name;
+	QAtomicPointer<ExchangeEngine> _engine;
+	QLockFile *_lockFile;
+	QAtomicInteger<quint16> _running = false;
+};
+
 }
 
 #endif // QTDATASYNC_EXCHANGEENGINE_P_H

src/datasync/setup.cpp

@@ -26,28 +26,13 @@ void Setup::setCleanupTimeout(unsigned long timeout)
 void Setup::removeSetup(const QString &name, bool waitForFinished)
 {
 	QMutexLocker _(&SetupPrivate::setupMutex);
-	if(SetupPrivate::engines.contains(name)) {
-		auto &info = SetupPrivate::engines[name];
-		if(info.engine) {
-			qCDebug(qdssetup) << "Finalizing engine of setup" << name;
-			QMetaObject::invokeMethod(info.engine, "finalize", Qt::QueuedConnection);
-			info.engine = nullptr;
-		} else
-			qCDebug(qdssetup) << "Engine of setup" << name << "already finalizing";
-
-		if(waitForFinished) {
-			if(!info.thread->wait(SetupPrivate::timeout)) {
-				qCWarning(qdssetup) << "Workerthread of setup" << name << "did not finish before the timout. Terminating...";
-				info.thread->terminate();
-				auto wRes = info.thread->wait(100);
-				qCDebug(qdssetup) << "Terminate result for setup" << name << ":"
-								  << wRes;
-			}
-			info.thread->deleteLater();
-
-			_.unlock(); //unlock first
-			QCoreApplication::processEvents();//required to perform queued events
-		}
+	auto mThread = SetupPrivate::engines.value(name);
+	_.unlock(); //unlock first
+
+	if(mThread) {
+		mThread->stopEngine();
+		if(waitForFinished)
+			mThread->waitAndTerminate(SetupPrivate::timeout);
 	} else //no there -> remove defaults (either already removed does nothing, or remove passive)
 		DefaultsPrivate::removeDefaults(name);
 }
@@ -443,6 +428,9 @@ Setup &Setup::setAccountTrusted(const QByteArray &importData, const QString &pas
 
 void Setup::create(const QString &name)
 {
+	if(QThread::currentThread() != qApp->thread())
+		qCWarning(qdssetup) << "Setup::create should only be called from the main thread";
+
 	QMutexLocker _(&SetupPrivate::setupMutex);
 	if(SetupPrivate::engines.contains(name))
 		throw SetupExistsException(name);
@@ -463,36 +451,22 @@ void Setup::create(const QString &name)
 		engine->prepareInitialAccount(d->initialImport);
 
 	// create and connect the new thread
-	auto thread = new QThread();
-	engine->moveToThread(thread);
-	QObject::connect(thread, &QThread::started,
-					 engine, &ExchangeEngine::initialize);
-	QObject::connect(thread, &QThread::finished,
-					 engine, &ExchangeEngine::deleteLater);
-	// unlock as soon as the engine has been destroyed
-	QObject::connect(engine, &ExchangeEngine::destroyed, qApp, [lockFile, name](){
-		qCDebug(qdssetup) << "Engine for setup" << name << "destroyed - removing lockfile";
-		lockFile->unlock();
-		delete lockFile;
-	}, Qt::DirectConnection); //direct connection required
+	auto thread = new EngineThread{name, engine, lockFile};
 	// once the thread finished, clear the engine from the cache of known ones
-	QObject::connect(thread, &QThread::finished, thread, [name, thread](){
-		qCDebug(qdssetup) << "Thread for setup" << name << "stopped - setup completly removed";
+	QObject::connect(thread, &QThread::finished, [name](){
 		QMutexLocker _cn(&SetupPrivate::setupMutex);
 		SetupPrivate::engines.remove(name);
+		_cn.unlock();
 		DefaultsPrivate::removeDefaults(name);
-		thread->deleteLater();
-	}, Qt::QueuedConnection); //queued connection, sent from within the thread to thread object on current thread
+	}); //queued connection, sent from within the thread to thread object on current thread
 
 	// start the thread and cache engine data
-	thread->start();
-	SetupPrivate::engines.insert(name, {thread, engine});
+	SetupPrivate::engines.insert(name, thread);
+	thread->startEngine();
 }
 
 bool Setup::createPassive(const QString &name, int timeout)
 {
-	QMutexLocker _(&SetupPrivate::setupMutex);
-
 	// create storage dir and defaults
 	auto storageDir = d->createStorageDir(name);
 	d->createDefaults(name, storageDir, true);
@@ -527,30 +501,27 @@ bool Setup::createPassive(const QString &name, int timeout)
 
 const QString SetupPrivate::DefaultLocalDir = QStringLiteral("./qtdatasync/default");
 QMutex SetupPrivate::setupMutex(QMutex::Recursive);
-QHash<QString, SetupPrivate::SetupInfo> SetupPrivate::engines;
+QHash<QString, QPointer<EngineThread>> SetupPrivate::engines;
 unsigned long SetupPrivate::timeout = ULONG_MAX;
 
 void SetupPrivate::cleanupHandler()
 {
 	QMutexLocker _(&setupMutex);
-	for (auto it = engines.begin(); it != engines.end(); it++) {
-		if(it->engine) {
-			qCDebug(qdssetup) << "Finalizing engine of setup" << it.key() << "because of app quit";
-			QMetaObject::invokeMethod(it->engine, "finalize", Qt::QueuedConnection);
-			it->engine = nullptr;
-		}
-	}
-	for (auto it = engines.constBegin(); it != engines.constEnd(); it++) {
-		if(!it->thread->wait(timeout)) {
-			qCWarning(qdssetup) << "Workerthread of setup" << it.key() << "did not finish before the timout. Terminating...";
-			it->thread->terminate();
-			auto wRes = it->thread->wait(100);
-			qCDebug(qdssetup) << "Terminate result for setup" << it.key() << ":"
-							  << wRes;
+	auto threads = engines.values();
+	engines.clear();
+	_.unlock();
+
+	for(auto &thread : threads) {
+		if(thread->stopEngine() &&
+		   thread->thread() != QThread::currentThread()) {
+			qCCritical(qdssetup) << "Stopping datasync thread" << thread->name()
+								 << "from the main thread even though it was created on a different thread!"
+								 << "This can lead to synchronization errors. You should always stop all other threads before quitting the application.";
 		}
-		it->thread->deleteLater();
 	}
-	engines.clear();
+	for(auto &thread : threads)
+		thread->waitAndTerminate(timeout);
+
 	DefaultsPrivate::clearDefaults();
 }
 
@@ -562,7 +533,7 @@ unsigned long SetupPrivate::currentTimeout()
 ExchangeEngine *SetupPrivate::engine(const QString &setupName)
 {
 	QMutexLocker _(&setupMutex);
-	auto engine = engines.value(setupName).engine;
+	auto engine = engines.value(setupName)->engine();
 	if(!engine)
 		throw SetupDoesNotExistException(setupName);
 	else
@@ -627,13 +598,6 @@ SetupPrivate::SetupPrivate() :
 	}
 {}
 
-SetupPrivate::SetupInfo::SetupInfo() = default;
-
-SetupPrivate::SetupInfo::SetupInfo(QThread *thread, ExchangeEngine *engine) :
-	thread{thread},
-	engine{engine}
-{}
-
 // ------------- Exceptions -------------
 
 SetupException::SetupException(const QString &setupName, const QString &message) :

src/datasync/setup_p.h

@@ -3,6 +3,7 @@
 
 #include <QtCore/QMutex>
 #include <QtCore/QThread>
+#include <QtCore/QPointer>
 
 #include <QtJsonSerializer/QJsonSerializer>
 
@@ -31,18 +32,10 @@ class Q_DATASYNC_EXPORT SetupPrivate
 	static QJsonObject parseObj(const QByteArray &data);
 
 private:
-	struct SetupInfo {
-		QThread *thread = nullptr;
-		ExchangeEngine *engine = nullptr;
-
-		SetupInfo();
-		SetupInfo(QThread *thread, ExchangeEngine *engine);
-	};
-
 	static const QString DefaultLocalDir;
 
 	static QMutex setupMutex;
-	static QHash<QString, SetupInfo> engines;
+	static QHash<QString, QPointer<EngineThread>> engines;
 	static unsigned long timeout;
 
 	QString localDir;
@@ -58,4 +51,6 @@ class Q_DATASYNC_EXPORT SetupPrivate
 
 }
 
+Q_DECLARE_LOGGING_CATEGORY(qdssetup)
+
 #endif // QTDATASYNC_SETUP_P_H