partially migrate to IndexedTimer

migrate to PIP 3.21
fix server restartings

src/CMakeLists.txt

@@ -11,7 +11,7 @@ if("${SHS_QT_VERSION}" STREQUAL "")
 	set(SHS_QT_VERSION 5 CACHE STRING "Qt version")
 endif()
 
-find_package(PIP 3.15 REQUIRED)
+find_package(PIP 3.21 REQUIRED)
 find_package(QAD 2.8.4 REQUIRED)
 qad_find_qt(Qt${SHS_QT_VERSION} Core Gui Widgets)
 set(TARGET_SUFFIX_Qt${SHS_QT_VERSION} "")

src/server/server_client.cpp

@@ -3,13 +3,14 @@
 #include "SH_network_types.h"
 #include "SH_packages.h"
 #include "ccm_SHS_shared.h"
-#include "piqt.h"
 #include "server.h"
 
 #include <QCryptographicHash>
 #include <QDebug>
 #include <QMetaObject>
 #include <piethernet.h>
+#include <piliterals_time.h>
+#include <piqt.h>
 
 
 ServerClient::ServerClient(ServerCore * s, SHNetworkTypes::ChannelType c): PIThread(), server(s), channel_type(c) {
@@ -44,10 +45,13 @@ void ServerClient::destroy() {
 	PIThread::stop();
 	// piCout << "~ServerClient stop";
 	if (device) device->stop();
-	if (!PIThread::waitForFinish(10000)) PIThread::terminate();
+	if (!PIThread::waitForFinish(5_s)) PIThread::terminate();
 	if (device) {
-		device->stopAndWait(10000);
+		device->stopAndWait(5_s);
-		if (device->isThreadedRead()) piCout << "[ServerClient::destroy]" << device << "can`t stop ...";
+		if (device->isThreadedRead()) {
+			piCout << "[ServerClient::destroy]" << device << "can`t stop ...";
+			device->terminateThreadedRead();
+		}
 	}
 	// piCout << "~ServerClient stop done";
 	piDeleteSafety(tcp);
@@ -208,6 +212,7 @@ void ServerClient::recGraphicsRequest(PIStringList gl) {
 
 
 void ServerClient::packetRec(PIByteArray data) {
+	if (closing) return;
 	// piCout << "packetRec" << data.toHex() << "...";
 	if (diag) diag->received(data.size_s());
 	// piCout << "packetRec" << data.toHex();

src/server/server_core.cpp

@@ -1,22 +1,33 @@
 #include "server_core.h"
-#include "piqt.h"
+
-#include "SH_packages.h"
 #include "SH_network_types.h"
+#include "SH_packages.h"
 #include "SH_plugins_manager.h"
-#include <QDebug>
-#include <QMetaEnum>
-#include <QCryptographicHash>
-#include <QResource>
-#include <qad_locations.h>
-#include <piethernet.h>
 #include "ccm_SHS_shared.h"
 
-//const double keepAliveInterval_s = 2.;
+#include <QCryptographicHash>
+#include <QDebug>
+#include <QMetaEnum>
+#include <QResource>
+#include <piethernet.h>
+#include <piliterals_time.h>
+#include <piqt.h>
+#include <qad_locations.h>
+
+// const double keepAliveInterval_s = 2.;
+
+
+enum TimerRole {
+	trSaveState,
+	trAdmin,
+	trServerSync,
+	trClients
+};
 
 
 ServerCore::ServerCore(bool reservation): QObject(), SHMulticast(), eth_tcp(PIEthernet::TCP_Server) {
 	PACKAGES;
-	file_ui_state     = QAD::userPath(QAD::ltCache, "ui_state"    );
+	file_ui_state     = QAD::userPath(QAD::ltCache, "ui_state");
 	file_ui_state_new = QAD::userPath(QAD::ltCache, "ui_state_new");
 	with_reservation  = reservation;
 	auto il           = PIEthernet::interfaces();
@@ -27,17 +38,27 @@ ServerCore::ServerCore(bool reservation): QObject(), SHMulticast(), eth_tcp(PIEt
 			break;
 		}
 	}
-	seed = generateID();//PINetworkAddress(my_ip, 0).ip();
+	seed = generateID(); // PINetworkAddress(my_ip, 0).ip();
 	addressesChanged();
 	piCoutObj << PICoutManipulators::Hex << seed << PINetworkAddress(seed, 0);
-	connect(&engine, SIGNAL(variableHistoryAdded(QString,QPair<QDateTime,double>)), this, SLOT(variableHistoryAdded(QString,QPair<QDateTime,double>)), Qt::DirectConnection);
+	connect(&engine,
+			SIGNAL(variableHistoryAdded(QString, QPair<QDateTime, double>)),
+			this,
+			SLOT(variableHistoryAdded(QString, QPair<QDateTime, double>)),
+			Qt::DirectConnection);
 	connect(&engine, SIGNAL(debug(QString)), this, SLOT(scriptDebug(QString)), Qt::DirectConnection);
 	CONNECTU(&eth_tcp, newConnection, this, newTCPConnection);
 	CONNECTU(&cloud_server, newConnection, this, newCloudConnection);
+	bindIndexedTimer(trSaveState, [this] { saveUIState(); });
+	bindIndexedTimer(trAdmin, [this] { sendAdminData(); });
+	bindIndexedTimer(trServerSync, [this] { syncServers(); });
+	bindIndexedTimer(trClients, [this] { procClients(); });
+	delete_clients_thread.start([this] { deleteOldClients(); }, 2_Hz);
 }
 
 
 ServerCore::~ServerCore() {
+	piCoutObj << "~ServerCore";
 	cloud_server.close();
 	cloud_server.stop();
 	engine.remove();
@@ -45,25 +66,29 @@ ServerCore::~ServerCore() {
 	PIByteArray sba = SHNetworkTypes::makeHeader(SHNetworkTypes::ServerRemove);
 	sba << seed;
 	piCoutObj << "send ServerRemove seed =" << seed;
-	send(sba);
-	killTimer(timer_save_state);
-	killTimer(timer_admin);
-	killTimer(timer_sync);
-	killTimer(timer);
 	stopRead();
-	PIVector<ServerClient * > rl = remotes.values();
+	send(sba);
+	stopIndexedTimer(trSaveState);
+	stopIndexedTimer(trAdmin);
+	stopIndexedTimer(trServerSync);
+	stopIndexedTimer(trClients);
+	delete_clients_thread.stopAndWait();
+	deleteOldClients();
+	PIVector<ServerClient *> rl = remotes.values();
 	for (ServerClient * r: rl)
 		if (r) {
+			piCoutObj << "delete" << (uintptr_t)r << "...";
 			r->destroy();
+			delete r->device;
 			delete r;
+			piCoutObj << "delete" << (uintptr_t)r << "done";
 		}
 	remotes.clear();
 	eth_tcp.stop();
 	for (auto * c: eth_tcp.clients())
 		c->stop();
 	piDeleteSafety(form);
-	if (bqr_content.data())
+	if (bqr_content.data()) QResource::unregisterResource((uchar *)bqr_content.data());
-		QResource::unregisterResource((uchar*)bqr_content.data());
 }
 
 
@@ -78,8 +103,8 @@ void ServerCore::startEth() {
 		cloud_server.startThreadedRead();
 		log(QString("Connect to cloud %1@%2").arg(sa.server, PI2QString(cloud_server.serverName())));
 	}
-	timer_sync = startTimer(100);
+	startIndexedTimer(trServerSync, 10_Hz);
-	timer_admin = startTimer(1000);
+	startIndexedTimer(trAdmin, 1_Hz);
 	SHMulticast::startRead();
 }
 
@@ -102,8 +127,8 @@ void ServerCore::loadRCC(const QString & p) {
 	bqr_content = Q2PIByteArray(f.readAll());
 	rcc_hash    = bqr_content.hash();
 	f.close();
-	if(bqr_content.isEmpty()) return;
+	if (bqr_content.isEmpty()) return;
-	if (!QResource::registerResource((uchar*)bqr_content.data())) {
+	if (!QResource::registerResource((uchar *)bqr_content.data())) {
 		log("Can`t open resource \"" + QFileInfo(f).absoluteFilePath() + "\"!", SHServer::LogWarning);
 		return;
 	}
@@ -150,11 +175,11 @@ void ServerCore::loadQS(const QString & p) {
 	}
 	log("Loaded script \"" + QFileInfo(f).absoluteFilePath() + "\"", SHServer::LogSuccess);
 	script_dir  = QFileInfo(f).absoluteDir();
-	//script = QString::fromUtf8(f.readAll());
+	// script = QString::fromUtf8(f.readAll());
 	file_script = f.fileName();
-	QObject::connect(&engine, SIGNAL(error(int,QString,QString,QString)), this, SLOT(scriptError(int,QString,QString,QString)));
+	QObject::connect(&engine, SIGNAL(error(int, QString, QString, QString)), this, SLOT(scriptError(int, QString, QString, QString)));
-	QObject::connect(&engine, SIGNAL(log(QString,int)), this, SLOT(scriptLog(QString,int)));
+	QObject::connect(&engine, SIGNAL(log(QString, int)), this, SLOT(scriptLog(QString, int)));
-	QObject::connect(&engine, SIGNAL(message(QString,int)), this, SLOT(scriptMessage(QString,int)));
+	QObject::connect(&engine, SIGNAL(message(QString, int)), this, SLOT(scriptMessage(QString, int)));
 	QObject::connect(&engine, SIGNAL(logIO(QString)), this, SLOT(IOLog(QString)));
 }
 
@@ -171,7 +196,7 @@ void ServerCore::loadDAT(const QString & p) {
 	}
 	log("Loaded database \"" + QFileInfo(f).absoluteFilePath() + "\"", SHServer::LogSuccess);
 	RUNTIME->setServerData(loadFromFile<SHS::ServerData>(p));
-	server_key_  = Q2PIString(RUNTIME->server_data.project.propertyValueByName("key" ).toString());
+	server_key_  = Q2PIString(RUNTIME->server_data.project.propertyValueByName("key").toString());
 	server_name_ = Q2PIString(RUNTIME->server_data.project.propertyValueByName("name").toString());
 	piCout << "key" << server_key_;
 	loadedDAT();
@@ -186,7 +211,7 @@ void ServerCore::loadCM(const QString & p) {
 	}
 	log("Loaded connection \"" + QFileInfo(f).absoluteFilePath() + "\"", SHServer::LogSuccess);
 	RUNTIME->connection.loadFromCMFile(p);
-	connect(&RUNTIME->connection, SIGNAL(qDataReceivedEvent(QString,QByteArray)), this, SLOT(connDataReceivedEvent(QString,QByteArray)));
+	connect(&RUNTIME->connection, SIGNAL(qDataReceivedEvent(QString, QByteArray)), this, SLOT(connDataReceivedEvent(QString, QByteArray)));
 	loadedCM();
 }
 
@@ -225,8 +250,7 @@ void ServerCore::loadBAF(const PIString & p) {
 	if (!cmp.isEmpty()) loadCM(cmp);
 	if (!dbp.isEmpty()) loadDAT(dbp);
 	RUNTIME->server_data.setRootPath(QDir::tempPath());
-	if (!varp.isEmpty())
+	if (!varp.isEmpty()) loadVAR(varp);
-		loadVAR(varp);
 	if (!sdpp.isEmpty())
 		loadProject(sdpp);
 	else if (qspl.size() == 1) {
@@ -259,221 +283,9 @@ void ServerCore::loadProject(const QString & p) {
 }
 
 
-void ServerCore::timerEvent(QTimerEvent * e) {
-	if (e->timerId() == timer_save_state) {
-		saveUIState();
-		return;
-	}
-	if (e->timerId() == timer_sync) {
-		if (with_reservation) {
-			smutex.lock();
-			if (server_state > ServerCore::ssSync) {
-				for (int i = 0; i < servers.size_s(); ++i) {
-					servers[i].missed++;
-					if (servers[i].missed > 10) {
-						piCoutObj << "remove by timeout" << servers[i].seed;
-						if (servers[i].seed == seed_main)
-							selectNewMain();
-						servers.remove(i);
-						i--;
-					}
-				}
-			}
-			smutex.unlock();
-			if (sync_skip > 0) {
-				sync_skip--;
-				return;
-			}
-			switch (server_state) {
-			case ServerCore::ssStart: {
-				PIByteArray sba = SHNetworkTypes::makeHeader(SHNetworkTypes::ServerNew);
-				sba << seed;
-				piCoutObj << "send ServerNew seed =" << seed << ", ip =" << my_ip;
-				send(sba);
-				selectNewMain();
-				} break;
-			case ServerCore::ssSync:
-				if (seed_main == 0) {
-					int mseed = seed;
-					int mi = -1;
-					smutex.lock();
-					for (int i = 0; i < servers.size_s(); ++i) {
-						if (mseed > servers[i].seed) {
-							mseed = servers[i].seed;
-							mi = i;
-						}
-					}
-					smutex.unlock();
-					seed_main = mseed;
-					main_ip = PINetworkAddress(mseed, 0).ipString();
-					if (mi < 0)
-						startAsMain();
-					else
-						startAsReserve();
-				}
-				break;
-			case ServerCore::ssWorkReserve:
-				main_missed++;
-				if (main_missed > 10) {
-					main_missed = 0;
-					selectNewMain();
-				}
-			case ServerCore::ssWorkMain:
-				{
-				PIByteArray sba = SHNetworkTypes::makeHeader(SHNetworkTypes::ServerPing);
-				sba << seed;
-				//piCoutObj << "send ServerPing seed =" << seed;
-				send(sba);
-				}
-				break;
-			}
-		} else startAsMain();
-		return;
-	}
-	if (e->timerId() == timer_admin) {
-
-		lmutex.lock();
-		PacketServerLog pck_log, pck_log_hist;
-		pck_log.entries = std::move(admin_log);
-		admin_log.clear();
-		auto mq = std::move(message_queue);
-		message_queue.clear();
-		lmutex.unlock();
-
-		rmutex.lock();
-		state_dynamic.clients.clear();
-		PIVector<ServerClient * > rl = remotes.values();
-		for (ServerClient * r: rl) {
-			if (!r) continue;
-			r->updateInfo();
-			state_dynamic.clients << r->info;
-		}
-		state_dynamic.devices.clear();
-		PIVector<PIIODevice * > devs = RUNTIME->connection.boundedDevices();
-		for (PIIODevice * d: devs) {
-			if (!d) continue;
-			ServerDeviceInfo di;
-			di.classname = QLatin1String(d->className());
-			di.path = PI2QString(d->path());
-			di.mode = d->mode();
-			PIDiagnostics * diag = RUNTIME->connection.diagnostic(d);
-			if (diag) {
-				di.sended_bytes = diag->state().sended_bytes;
-				di.received_bytes = diag->state().received_bytes;
-				di.send_speed = PI2QString(diag->sendSpeed());
-				di.receive_speed = PI2QString(diag->receiveSpeed());
-			}
-			state_dynamic.devices << di;
-		}
-		for (auto & v: state_dynamic.variables) {
-			auto val = engine.getVariableValue(v.name);
-			v.changed = val.first;
-			v.value = val.second;
-		}
-
-		PIByteArray msg_state;
-		msg_state = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminDynamicState);
-		msg_state << state_dynamic;
-
-		for (ServerClient * r: rl) {
-			if (!r) continue;
-
-			if (r->info.role == SHServer::RoleMessages) {
-				for (const auto & m: mq) {
-					PIByteArray ba = SHNetworkTypes::makeHeader(SHNetworkTypes::Message);
-					ba << m;
-					r->queueSend(ba);
-				}
-			}
-
-			if (r->info.role == SHServer::RoleAdmin) {
-				if (!r->log_sended) {
-					r->log_sended = true;
-					pck_log_hist.entries = admin_log_history;
-					PIByteArray msg = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminLog);
-					msg << pck_log_hist;
-					r->queueSend(msg);
-				}
-				if (!msg_state.isEmpty())
-					r->queueSend(msg_state);
-				if (!pck_log.entries.isEmpty() || !r->admin_io_log.isEmpty()) {
-					PacketServerLog pckt;
-					pckt.entries << pck_log.entries << r->admin_io_log;
-					pckt.entries.sort([](const LogEntry & t0, const LogEntry & t1){return t0.time < t1.time;});
-					r->admin_io_log.clear();
-					PIByteArray msg = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminLog);
-					msg << pckt;
-					r->queueSend(msg);
-				}
-			}
-		}
-		rmutex.unlock();
-
-		for (const auto & e: pck_log.entries) {
-			admin_log_cur_size += e.text.size();
-			admin_log_history << e;
-			while (admin_log_cur_size > admin_log_max_size)
-				admin_log_cur_size -= admin_log_history.take_front().text.size();
-		}
-
-		return;
-	}
-	rmutex.lock();
-
-	sync.buildDiffForRoles();
-
-	PIVector<ServerClient * > rl = remotes.values(), to_remove;
-	for (ServerClient * r: rl) {
-		bool disconn = !r->isConnected();
-		if (r) {if (r->tcp->wasError()) disconn = true;}
-		if (disconn) {
-			log("Client " + PI2QString(r->device->path()) + " disconnected", SHServer::LogInfo);
-			//piCout << "disconn ...";
-			r->close();
-			//piCout << "disconn close";
-			if (df_client == r) {
-				df_client = 0;
-				df_device.clear();
-				engine.connection()->setInactive(false);
-			}
-			to_remove << r;
-			//piCout << "disconn done";
-		}
-		r->maybeCallQueuedEvents();
-		r->updateInfo();
-		if (!r->tcp || !r->active || (r->info.role != SHServer::RoleGui)) continue;
-		PIByteArray diff;
-		if (r->role_changed) {
-			diff = Q2PIByteArray(sync.buildTransitionDiff(r->user_role, r->new_user_role));
-			//qDebug() << "send transitional diff" << r->user_role << "->" << r->new_user_role << diff.size();
-			r->newRoleDone();
-			if (!diff.isEmpty()) {
-				diff.insert(0, SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncDiff));
-				r->queueSend(diff);
-			}
-			diff.clear();
-		}
-		diff = sync.diffForRole(r->user_role);
-		if (!diff.isEmpty()) {
-			diff.insert(0, SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncDiff));
-			r->queueSend(diff);
-		}
-		//piCout << "diff role" << r->user_role << diff.size();
-	}
-	for (ServerClient * r: to_remove) {
-		remotes.remove(r->device);
-		r->destroy();
-		delete r->device;
-		delete r;
-	}
-	rmutex.unlock();
-}
-
-
 bool ServerCore::hasServer(const PIString & ip) const {
 	for (const auto & s: servers)
-		if (s.ip == ip)
+		if (s.ip == ip) return true;
-			return true;
 	return false;
 }
 
@@ -504,10 +316,8 @@ void ServerCore::startAsMain() {
 	if (server_state == ServerCore::ssWorkMain) return;
 	piCoutObj << "startAsMain";
 	server_state = ServerCore::ssWorkMain;
-	if (timer > 0) killTimer(timer);
+	startIndexedTimer(trSaveState, 10_s);
-	timer = startTimer(100);
+	startIndexedTimer(trClients, 10_Hz);
-	if (timer_save_state > 0) killTimer(timer_save_state);
-	timer_save_state = startTimer(10000);
 	sync_skip = 0;
 	start();
 	engine.setDir(script_dir);
@@ -531,8 +341,7 @@ void ServerCore::startAsReserve() {
 	if (server_state == ServerCore::ssWorkReserve) return;
 	piCoutObj << "startAsReserve";
 	server_state = ServerCore::ssWorkReserve;
-	if (timer > 0) killTimer(timer);
+	stopIndexedTimer(trClients);
-	timer = 0;
 	sync_skip = 0;
 	start();
 	startedAsReserve();
@@ -566,8 +375,7 @@ void ServerCore::restoreUIState() {
 	QFile f(file_ui_state);
 	if (!f.open(QIODevice::ReadOnly)) {
 		f.setFileName(file_ui_state_new);
-		if (!f.open(QIODevice::ReadOnly))
+		if (!f.open(QIODevice::ReadOnly)) return;
-			return;
 	}
 	if (f.size() < 8) return;
 	QDataStream s(&f);
@@ -644,9 +452,230 @@ void ServerCore::gatherFixedState() {
 		PIString server = Q2PIString(RUNTIME->server_data.project.propertyValueByName("cloud_address").toString()) + "@";
 		server += getThisCloudName();
 		state_fixed.cloud_address = PI2QString(server);
-		state_fixed.cloud_href = QString("https://") + SHServerURL + "/pult/?address=" + QString(QUrl::toPercentEncoding(PI2QString(server + getThisCloudName())));
+		state_fixed.cloud_href    = QString("https://") + SHServerURL +
+								 "/pult/?address=" + QString(QUrl::toPercentEncoding(PI2QString(server + getThisCloudName())));
+	}
+}
+
+
+void ServerCore::sendAdminData() {
+	lmutex.lock();
+	PacketServerLog pck_log, pck_log_hist;
+	pck_log.entries = std::move(admin_log);
+	admin_log.clear();
+	auto mq = std::move(message_queue);
+	message_queue.clear();
+	lmutex.unlock();
+
+	rmutex.lock();
+	state_dynamic.clients.clear();
+	PIVector<ServerClient *> rl = remotes.values();
+	for (ServerClient * r: rl) {
+		if (!r) continue;
+		r->updateInfo();
+		state_dynamic.clients << r->info;
+	}
+	state_dynamic.devices.clear();
+	PIVector<PIIODevice *> devs = RUNTIME->connection.boundedDevices();
+	for (PIIODevice * d: devs) {
+		if (!d) continue;
+		ServerDeviceInfo di;
+		di.classname         = QLatin1String(d->className());
+		di.path              = PI2QString(d->path());
+		di.mode              = d->mode();
+		PIDiagnostics * diag = RUNTIME->connection.diagnostic(d);
+		if (diag) {
+			di.sended_bytes   = diag->state().sended_bytes;
+			di.received_bytes = diag->state().received_bytes;
+			di.send_speed     = PI2QString(diag->sendSpeed());
+			di.receive_speed  = PI2QString(diag->receiveSpeed());
+		}
+		state_dynamic.devices << di;
+	}
+	for (auto & v: state_dynamic.variables) {
+		auto val  = engine.getVariableValue(v.name);
+		v.changed = val.first;
+		v.value   = val.second;
 	}
 
+	PIByteArray msg_state;
+	msg_state = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminDynamicState);
+	msg_state << state_dynamic;
+
+	for (ServerClient * r: rl) {
+		if (!r) continue;
+
+		if (r->info.role == SHServer::RoleMessages) {
+			for (const auto & m: mq) {
+				PIByteArray ba = SHNetworkTypes::makeHeader(SHNetworkTypes::Message);
+				ba << m;
+				r->queueSend(ba);
+			}
+		}
+
+		if (r->info.role == SHServer::RoleAdmin) {
+			if (!r->log_sended) {
+				r->log_sended        = true;
+				pck_log_hist.entries = admin_log_history;
+				PIByteArray msg      = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminLog);
+				msg << pck_log_hist;
+				r->queueSend(msg);
+			}
+			if (!msg_state.isEmpty()) r->queueSend(msg_state);
+			if (!pck_log.entries.isEmpty() || !r->admin_io_log.isEmpty()) {
+				PacketServerLog pckt;
+				pckt.entries << pck_log.entries << r->admin_io_log;
+				pckt.entries.sort([](const LogEntry & t0, const LogEntry & t1) { return t0.time < t1.time; });
+				r->admin_io_log.clear();
+				PIByteArray msg = SHNetworkTypes::makeHeader(SHNetworkTypes::AdminLog);
+				msg << pckt;
+				r->queueSend(msg);
+			}
+		}
+	}
+	rmutex.unlock();
+
+	for (const auto & e: pck_log.entries) {
+		admin_log_cur_size += e.text.size();
+		admin_log_history << e;
+		while (admin_log_cur_size > admin_log_max_size)
+			admin_log_cur_size -= admin_log_history.take_front().text.size();
+	}
+}
+
+
+void ServerCore::syncServers() {
+	if (with_reservation) {
+		smutex.lock();
+		if (server_state > ServerCore::ssSync) {
+			for (int i = 0; i < servers.size_s(); ++i) {
+				servers[i].missed++;
+				if (servers[i].missed > 10) {
+					piCoutObj << "remove by timeout" << servers[i].seed;
+					if (servers[i].seed == seed_main) selectNewMain();
+					servers.remove(i);
+					i--;
+				}
+			}
+		}
+		smutex.unlock();
+		if (sync_skip > 0) {
+			sync_skip--;
+			return;
+		}
+		switch (server_state) {
+		case ServerCore::ssStart: {
+			PIByteArray sba = SHNetworkTypes::makeHeader(SHNetworkTypes::ServerNew);
+			sba << seed;
+			piCoutObj << "send ServerNew seed =" << seed << ", ip =" << my_ip;
+			send(sba);
+			selectNewMain();
+		} break;
+		case ServerCore::ssSync:
+			if (seed_main == 0) {
+				int mseed = seed;
+				int mi    = -1;
+				smutex.lock();
+				for (int i = 0; i < servers.size_s(); ++i) {
+					if (mseed > servers[i].seed) {
+						mseed = servers[i].seed;
+						mi    = i;
+					}
+				}
+				smutex.unlock();
+				seed_main = mseed;
+				main_ip   = PINetworkAddress(mseed, 0).ipString();
+				if (mi < 0)
+					startAsMain();
+				else
+					startAsReserve();
+			}
+			break;
+		case ServerCore::ssWorkReserve:
+			main_missed++;
+			if (main_missed > 10) {
+				main_missed = 0;
+				selectNewMain();
+			}
+		case ServerCore::ssWorkMain: {
+			PIByteArray sba = SHNetworkTypes::makeHeader(SHNetworkTypes::ServerPing);
+			sba << seed;
+			// piCoutObj << "send ServerPing seed =" << seed;
+			send(sba);
+		} break;
+		}
+	} else
+		startAsMain();
+}
+
+
+void ServerCore::procClients() {
+	PIMutexLocker ml(rmutex);
+
+	sync.buildDiffForRoles();
+
+	PIVector<ServerClient *> rl = remotes.values(), to_remove;
+	for (ServerClient * r: rl) {
+		bool disconn = !r->isConnected();
+		if (r) {
+			if (r->tcp->wasError()) disconn = true;
+		}
+		if (disconn) {
+			log("Client " + PI2QString(r->device->path()) + " disconnected", SHServer::LogInfo);
+			// piCout << "disconn ...";
+			r->close();
+			// piCout << "disconn close";
+			if (df_client == r) {
+				df_client = 0;
+				df_device.clear();
+				engine.connection()->setInactive(false);
+			}
+			to_remove << r;
+			// piCout << "disconn done";
+		}
+		r->maybeCallQueuedEvents();
+		r->updateInfo();
+		if (!r->tcp || !r->active || (r->info.role != SHServer::RoleGui)) continue;
+		PIByteArray diff;
+		if (r->role_changed) {
+			diff = Q2PIByteArray(sync.buildTransitionDiff(r->user_role, r->new_user_role));
+			// qDebug() << "send transitional diff" << r->user_role << "->" << r->new_user_role << diff.size();
+			r->newRoleDone();
+			if (!diff.isEmpty()) {
+				diff.insert(0, SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncDiff));
+				r->queueSend(diff);
+			}
+			diff.clear();
+		}
+		diff = sync.diffForRole(r->user_role);
+		if (!diff.isEmpty()) {
+			diff.insert(0, SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncDiff));
+			r->queueSend(diff);
+		}
+		// piCout << "diff role" << r->user_role << diff.size();
+	}
+	for (ServerClient * r: to_remove) {
+		remotes.remove(r->device);
+		r->close();
+		rcq_mutex.lock();
+		delete_clients_queue.enqueue(r);
+		rcq_mutex.unlock();
+	}
+}
+
+
+void ServerCore::deleteOldClients() {
+	rcq_mutex.lock();
+	auto del_queue = delete_clients_queue;
+	delete_clients_queue.clear();
+	rcq_mutex.unlock();
+	for (auto * r: del_queue) {
+		piCoutObj << "delete" << (uintptr_t)r << "...";
+		r->destroy();
+		delete r->device;
+		delete r;
+		piCoutObj << "delete" << (uintptr_t)r << "done";
+	}
 }
 
 
@@ -668,14 +697,14 @@ void ServerCore::connDataReceivedEvent(QString from, QByteArray data) {
 
 
 void ServerCore::variableHistoryAdded(QString vname, QPair<QDateTime, double> value) {
-	//qDebug() << "history add" << vname;
+	// qDebug() << "history add" << vname;
 	SHS::PacketGraphicData gd(vname);
 	ullong time = value.first.toMSecsSinceEpoch();
 	gd.data << PIPair<ullong, double>(time, value.second);
 	PIByteArray ba = SHNetworkTypes::makeHeader(SHNetworkTypes::GraphicData);
 	ba << gd;
 	PIMutexLocker _ml(rmutex);
-	PIVector<ServerClient * > rl = remotes.values();
+	PIVector<ServerClient *> rl = remotes.values();
 	for (auto * r: rl)
 		if (r->tcp && r->active && r->info.role == SHServer::RoleGraphics) {
 			if (r->graphics.contains(vname)) {
@@ -687,14 +716,14 @@ void ServerCore::variableHistoryAdded(QString vname, QPair<QDateTime, double> va
 
 
 void ServerCore::receivedSH(PIByteArray data, uint rec_id) {
-	//piCout << "received" << data;
+	// piCout << "received" << data;
 	PacketHeader hdr = SHNetworkTypes::takeHeader(data);
 	if (hdr.type < 0) return;
-	//piCout << "received type" << hdr.type << rec_id << data.size();
+	// piCout << "received type" << hdr.type << rec_id << data.size();
 	int _seed;
 	PIString _ip;
 	PIByteArray ba;
-	//piCout << "received" << type;
+	// piCout << "received" << type;
 	if (with_reservation) {
 		switch ((SHNetworkTypes::MulticastType)hdr.type) {
 		case SHNetworkTypes::ServerNew:
@@ -729,8 +758,7 @@ void ServerCore::receivedSH(PIByteArray data, uint rec_id) {
 			smutex.lock();
 			for (int i = 0; i < servers.size_s(); ++i)
 				if (servers[i].seed == _seed) {
-					if (seed_main == _seed)
+					if (seed_main == _seed) selectNewMain();
-						selectNewMain();
 					servers.remove(i);
 					i--;
 				}
@@ -745,23 +773,21 @@ void ServerCore::receivedSH(PIByteArray data, uint rec_id) {
 			break;
 		case SHNetworkTypes::ServerPing:
 			data >> _seed;
-			if (seed_main == _seed)
+			if (seed_main == _seed) main_missed = 0;
-				main_missed = 0;
 			smutex.lock();
 			for (int i = 0; i < servers.size_s(); ++i)
-				if (servers[i].seed == _seed)
+				if (servers[i].seed == _seed) servers[i].missed = 0;
-					servers[i].missed = 0;
 			smutex.unlock();
-			//seed_main = _seed;
+			// seed_main = _seed;
-			//piCoutObj << "recv ServerPing seed =" << _seed;
+			// piCoutObj << "recv ServerPing seed =" << _seed;
 			break;
 		default: break;
 		}
 	}
 	if (server_state != ServerCore::ssWorkMain) return;
-	//piCout << "received type" << type << rec_id << data.size();
+	// piCout << "received type" << type << rec_id << data.size();
 	if (hdr.type == SHNetworkTypes::Request) {
-		//PIMutexLocker locker(mcast_mutex);
+		// PIMutexLocker locker(mcast_mutex);
 		PIStringList addresses;
 		data >> addresses;
 		PIString addr;
@@ -772,10 +798,9 @@ void ServerCore::receivedSH(PIByteArray data, uint rec_id) {
 					break;
 				}
 			}
-			if (!addr.isEmpty())
+			if (!addr.isEmpty()) break;
-				break;
 		}
-		//piCout << "return address" << addr;
+		// piCout << "return address" << addr;
 		PacketServerInfo info;
 		info.identify                 = PacketIdentification::makeLocal(0, SHS_NETWORK_VERSION, SHS_UISYNC_VERSION);
 		info.identify.hashes.hash_ui  = ui_hash;
@@ -799,36 +824,36 @@ void ServerCore::addressesChanged() {
 
 
 void ServerCore::newTCPConnection(PIEthernet * client) {
-	//piCout << "newConnection" << client;
+	// piCout << "newConnection" << client;
 	log("Client connected from " + PI2QString(client->path()) + " (TCP)", SHServer::LogInfo);
 	client->setDebug(false);
 	client->setReadTimeout(60000.);
 	client->setWriteTimeout(60000.);
 	client->setParameter(PIEthernet::DisonnectOnTimeout, false);
 	ServerClient * r = new ServerClient(this, SHNetworkTypes::TCP);
-	//CONNECTU(r, packetReceived, this, packetReceived)
+	// CONNECTU(r, packetReceived, this, packetReceived)
 	r->init(client);
-	//piCout << "newConnection lock ...";
+	// piCout << "newConnection lock ...";
 	rmutex.lock();
 	remotes[client] = r;
 	rmutex.unlock();
 	client->startThreadedRead();
-	//piCout << "newConnection lock done";
+	// piCout << "newConnection lock done";
 }
 
 
 void ServerCore::newCloudConnection(PICloudServer::Client * client) {
 	piCout << "newCloudConnection" << client;
 	log("Client connected from " + PI2QString(client->path()) + " (Cloud)", SHServer::LogInfo);
-	//client->setDebug(false);
+	// client->setDebug(false);
 	ServerClient * r = new ServerClient(this, SHNetworkTypes::Cloud);
-	//CONNECTU(r, packetReceived, this, packetReceived)
+	// CONNECTU(r, packetReceived, this, packetReceived)
 	r->init(client);
-	//piCout << "newConnection lock ...";
+	// piCout << "newConnection lock ...";
 	rmutex.lock();
 	remotes[client] = r;
 	rmutex.unlock();
-	//piCout << "newConnection lock done";
+	// piCout << "newConnection lock done";
 	client->startThreadedRead();
 }
 
@@ -879,13 +904,13 @@ void ServerCore::receivedIdentification(ServerClient * r, PIByteArray data) {
 		sendInitialAdminData(r);
 	}
 
-	//piCoutObj << "my" << ui_hash << rcc_hash;
+	// piCoutObj << "my" << ui_hash << rcc_hash;
-	//piCoutObj << "cl" << p.hashes.hash_ui << p.hashes.hash_rcc;
+	// piCoutObj << "cl" << p.hashes.hash_ui << p.hashes.hash_rcc;
 	r->upload_ui  = (p.hashes.hash_ui != ui_hash);
 	r->upload_rcc = (p.hashes.hash_rcc != rcc_hash);
 	r->tcp->setCheckHeader(true);
 	r->tcp->setID(r->id);
-	//qDebug() << "ident" << int(r->info.OS_type) << r->info.OS_arch << r->info.OS_version;
+	// qDebug() << "ident" << int(r->info.OS_type) << r->info.OS_arch << r->info.OS_version;
 }
 
 
@@ -908,11 +933,10 @@ void ServerCore::receivedKey(ServerClient * r, PIByteArray data) {
 	piCout << "received key" << key;
 	r->tcp->send(ba);
 	if (ok) {
-
 		if (r->upload_rcc) {
 			ba = SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncRCC);
 			ba.append(bqr_content);
-			//qDebug() << "send rcc" << int(SHNetworkTypes::GuiSyncRCC) << bqr_content.size();
+			// qDebug() << "send rcc" << int(SHNetworkTypes::GuiSyncRCC) << bqr_content.size();
 		} else {
 			ba = SHNetworkTypes::makeHeader(SHNetworkTypes::GuiRestoreRCC);
 		}
@@ -924,11 +948,11 @@ void ServerCore::receivedKey(ServerClient * r, PIByteArray data) {
 		} else {
 			ba = SHNetworkTypes::makeHeader(SHNetworkTypes::GuiRestoreUI);
 		}
-		//qDebug() << "send ui" << int(SHNetworkTypes::GuiSyncUI) << ba.size();
+		// qDebug() << "send ui" << int(SHNetworkTypes::GuiSyncUI) << ba.size();
 		r->queueSend(ba);
 
 		ba = Q2PIByteArray(sync.buildFullDiffForRole(r->user_role));
-		//qDebug() << "send init diff" << r->user_role << ba.size();
+		// qDebug() << "send init diff" << r->user_role << ba.size();
 		if (!ba.isEmpty()) {
 			PIByteArray hdr = SHNetworkTypes::makeHeader(SHNetworkTypes::GuiSyncDiff);
 			ba.insert(0, hdr);
@@ -938,14 +962,14 @@ void ServerCore::receivedKey(ServerClient * r, PIByteArray data) {
 		ba = SHNetworkTypes::makeHeader(SHNetworkTypes::ConnectingDone);
 		r->queueSend(ba);
 
-		//r->active = true;
+		// r->active = true;
 	}
 }
 
 
 void ServerCore::sendToAll(const PIByteArray & data, SHServer::ClientRole role, bool no_check) {
 	PIMutexLocker _ml(rmutex);
-	PIVector<ServerClient * > rl = remotes.values();
+	PIVector<ServerClient *> rl = remotes.values();
 	for (ServerClient * r: rl) {
 		qDebug() << r << r->tcp << r->active << r->info.role;
 		if (!r->tcp || r->info.role != role) continue;
@@ -1040,7 +1064,7 @@ void ServerCore::IOLog(QString text) {
 	e.text = text;
 	e.type = SHServer::LogIO;
 	rmutex.lock();
-	PIVector<ServerClient * > rl = remotes.values();
+	PIVector<ServerClient *> rl = remotes.values();
 	for (ServerClient * r: rl) {
 		if (!r) continue;
 		if ((r->info.role != SHServer::RoleAdmin) || !r->admin_log_io) continue;

src/server/server_core.h

@@ -1,49 +1,54 @@
 #ifndef SERVER_CORE_H
 #define SERVER_CORE_H
 
+#include "SH_assembly.h"
+#include "SH_ui_loader.h"
+#include "SH_ui_sync.h"
+#include "script_engine.h"
+#include "server_client.h"
+
 #include <QTimer>
 #include <picli.h>
 #include <picloudserver.h>
 #include <piliterals_bytes.h>
-#include "script_engine.h"
+#include <qad_timers.h>
-#include "server_client.h"
-#include "SH_assembly.h"
-#include "SH_ui_loader.h"
-#include "SH_ui_sync.h"
 
 class Server;
 class ServerClient;
 class ServerStarter;
 
-class ServerCore: public QObject, public SHMulticast {
+class ServerCore
+	: public QObject
+	, public SHMulticast
+	, public IndexedTimer<> {
 	Q_OBJECT
 	PIOBJECT_SUBCLASS(ServerCore, SHMulticast)
 	friend class Server;
 	friend class ServerClient;
 	friend class ServerStarter;
+
 public:
 	ServerCore(bool reservation = false);
 	~ServerCore();
 
 	void loadRCC(const QString & p);
-	void loadUI (const QString & p);
+	void loadUI(const QString & p);
-	void loadQS (const QString & p);
+	void loadQS(const QString & p);
 	void loadDAT(const QString & p);
-	void loadCM (const QString & p);
+	void loadCM(const QString & p);
 	void loadVAR(const QString & p);
 	void loadBAF(const PIString & p);
 	void loadProject(const QString & p);
 	void startEth();
 	void start();
 
-	bool isPluginsLoaded() const {return plugins_ok;}
+	bool isPluginsLoaded() const { return plugins_ok; }
 
 	SHSScriptEngine engine;
 	PIByteArray admin_password;
 	uint admin_log_max_size = 65_KiB;
 
 private:
-	void timerEvent(QTimerEvent *) override;
 	void receivedSH(PIByteArray data, uint rec_id) override;
 	void addressesChanged() override;
 	void sendToAll(const PIByteArray & data, SHServer::ClientRole role, bool no_check = false);
@@ -64,11 +69,20 @@ private:
 	void updatePlugins(const QStringList & names, ServerClient * r);
 	void gatherVariables();
 	void gatherFixedState();
+	void sendAdminData();
+	void syncServers();
+	void procClients();
+	void deleteOldClients();
 	PacketParameterList adminParameters() const;
-	EVENT_HANDLER1(void, newTCPConnection, PIEthernet * , client);
+	EVENT_HANDLER1(void, newTCPConnection, PIEthernet *, client);
-	EVENT_HANDLER1(void, newCloudConnection, PICloudServer::Client * , client);
+	EVENT_HANDLER1(void, newCloudConnection, PICloudServer::Client *, client);
 
-	enum ServerState {ssStart, ssSync, ssWorkMain, ssWorkReserve};
+	enum ServerState {
+		ssStart,
+		ssSync,
+		ssWorkMain,
+		ssWorkReserve
+	};
 
 	struct ServerSync {
 		int seed;
@@ -90,15 +104,17 @@ private:
 	SHAssembly ass;
 	PIEthernet eth_tcp;
 	PICloudServer cloud_server;
-	PIMutex rmutex, lmutex, smutex;
+	PIMutex rmutex, lmutex, smutex, rcq_mutex;
 	QQueue<LogEntry> log_queue;
+	PIThread delete_clients_thread;
 	PIDeque<LogEntry> admin_log, admin_log_history;
 	QQueue<PacketMessage> message_queue;
-	PIVector<PIEthernet*> eth_mcast;
+	PIVector<PIEthernet *> eth_mcast;
 	PIByteArray ui_content, bqr_content;
 	PacketServerDynamicState state_dynamic;
 	PacketServerFixedState state_fixed;
-	PIMap<PIIODevice * , ServerClient * > remotes;
+	PIMap<PIIODevice *, ServerClient *> remotes;
+	PIQueue<ServerClient *> delete_clients_queue;
 	QDir script_dir;
 	QString lqs_err_msg, script, file_ui_state, file_ui_state_new, file_script;
 	QList<SHS::ScriptVariable> vars;
@@ -110,30 +126,29 @@ private:
 	SHSUiLoader ui_loader;
 	uint ui_hash = 0, rcc_hash = 0;
 	uint admin_log_cur_size = 0;
-	int timer = 0, timer_sync = 0, timer_admin = 0, timer_save_state = 0, lqs_err_line = -1, sync_skip = 0, main_missed = 0;
+	int timer = 0, lqs_err_line = -1, sync_skip = 0, main_missed = 0;
-	int seed = 0 , seed_main = 0;
+	int seed = 0, seed_main = 0;
 	bool with_reservation = false, plugins_ok = false;
 
 private slots:
 	void connDataReceivedEvent(QString from, QByteArray data);
 	void variableHistoryAdded(QString vname, QPair<QDateTime, double> value);
 	void scriptError(int line, QString file, QString message, QString stack);
-	void scriptLog(QString text, int type) {log(text, (SHServer::LogEntryType)type);}
+	void scriptLog(QString text, int type) { log(text, (SHServer::LogEntryType)type); }
 	void scriptDebug(QString text);
-	void scriptMessage(QString text, int type) {message(text, type);}
+	void scriptMessage(QString text, int type) { message(text, type); }
 	void IOLog(QString text);
 
 signals:
 	void restart(QString baf_path);
-	void loadedUI ();
+	void loadedUI();
 	void loadedDAT();
-	void loadedCM ();
+	void loadedCM();
 	void loadedVAR();
 	void startedAsMain();
 	void startedAsReserve();
 	void zeroFailure();
 	void updatePluginsRequest(QStringList);
-
 };
 
 #endif // SERVER_CORE_H

src/server/server_starter.cpp

@@ -1,8 +1,9 @@
 #include "server_starter.h"
+
 #include "SH_base.h"
 #include "SH_plugins_manager.h"
-#include <qad_locations.h>
 
+#include <qad_locations.h>
 
 
 ServerStarter::ServerStarter(): QObject() {
@@ -13,8 +14,7 @@ ServerStarter::ServerStarter(): QObject() {
 	if (QFile::exists(confpath)) {
 		QPIConfig conf(confpath, QIODevice::ReadOnly);
 		QString baf = conf.getValue("baf_path", "").toString().trimmed();
-		if (!baf.isEmpty())
+		if (!baf.isEmpty()) baf_path = baf;
-			baf_path = baf;
 		max_failures = conf.getValue("max_failures", max_failures).toInt();
 		log_add_time = conf.getValue("log_add_time", log_add_time).toBool();
 		log_add_dev  = conf.getValue("log_add_dev", log_add_dev).toBool();
@@ -27,17 +27,17 @@ ServerStarter::ServerStarter(): QObject() {
 		admin_password = passwordHash("");
 	else
 		admin_password = PIByteArray::fromHex(Q2PIString(ph));
-	//piCout << "last admin_passwd" << admin_passwd.toHex();
+	// piCout << "last admin_passwd" << admin_passwd.toHex();
 }
 
 
 ServerStarter::~ServerStarter() {
-	destroy();
+	// destroy();
 }
 
 
 void ServerStarter::writePassword(const PIByteArray & ph) {
-	//piCout << " set admin_passwd" << ph.toHex();
+	// piCout << " set admin_passwd" << ph.toHex();
 	QPIConfig conf(confpath, QIODevice::ReadWrite);
 	conf.setValue("admin_passwd", PI2QString(ph.toHex()));
 }
@@ -50,8 +50,7 @@ void ServerStarter::init() {
 	server->core.admin_log_max_size = piRound(max_log_kb * 1024);
 	server->core.admin_password     = admin_password;
 	connect(server, SIGNAL(restartRequest()), this, SLOT(restart()));
-	if (no_graphics)
+	if (no_graphics) server->disableGraphics();
-		server->disableGraphics();
 	server->core.engine.connection()->setLogAddTime(log_add_time);
 	server->core.engine.connection()->setLogAddDevice(log_add_dev);
 	server->core.engine.connection()->setLogIO(log_io);
@@ -75,8 +74,7 @@ void ServerStarter::start() {
 		RUNTIME->connection.start();
 	}
 	server->core.startEth();
-	if (!no_graphics)
+	if (!no_graphics) server->show();
-		server->show();
 	QTimer::singleShot(10000, &(server->core), SIGNAL(zeroFailure()));
 }
 
@@ -90,8 +88,7 @@ bool ServerStarter::checkFailure() {
 	}
 	int fails = 0;
 	QTextStream s(&f);
-	if (f.size() >= 1)
+	if (f.size() >= 1) s >> fails;
-		s >> fails;
 	if (fails >= max_failures) {
 		qDebug() << "[ServerStarter] Error: too many failures, standby ...";
 		return true;
@@ -104,7 +101,8 @@ bool ServerStarter::checkFailure() {
 }
 
 
-#define CHECK_CONF(e) if (!conf.isEntryExists(#e)) s << #e << " = " << e << "\n";
+#define CHECK_CONF(e) \
+	if (!conf.isEntryExists(#e)) s << #e << " = " << e << "\n";
 
 void ServerStarter::checkConfig() {
 	QPIConfig conf(confpath, QIODevice::ReadOnly);
@@ -117,13 +115,13 @@ void ServerStarter::checkConfig() {
 	f.seek(f.size());
 	QTextStream s(&f);
 	auto baf_path = src_baf_path;
-	CHECK_CONF(baf_path    );
+	CHECK_CONF(baf_path);
 	CHECK_CONF(max_failures);
 	CHECK_CONF(log_add_time);
-	CHECK_CONF(log_add_dev );
+	CHECK_CONF(log_add_dev);
-	CHECK_CONF(log_io      );
+	CHECK_CONF(log_io);
-	CHECK_CONF(reservation );
+	CHECK_CONF(reservation);
-	CHECK_CONF(max_log_kb  );
+	CHECK_CONF(max_log_kb);
 }
 
 #undef CHECK_CONF
@@ -150,13 +148,17 @@ void ServerStarter::updatePlugins(QStringList names) {
 		return;
 	}
 	RUNTIME->plugins_manager->updateInfo();
-	qDebug() << "[ServerStarter::updatePlugins]" << "Got info from server, shutdown server ...";
+	qDebug() << "[ServerStarter::updatePlugins]"
+			 << "Got info from server, shutdown server ...";
 	destroy();
-	qDebug() << "[ServerStarter::updatePlugins]" << "Update plugins ...";
+	qDebug() << "[ServerStarter::updatePlugins]"
+			 << "Update plugins ...";
 	RUNTIME->plugins_manager->update(names);
-	qDebug() << "[ServerStarter::updatePlugins]" << "Update done, start server ...";
+	qDebug() << "[ServerStarter::updatePlugins]"
+			 << "Update done, start server ...";
 	restart({});
-	qDebug() << "[ServerStarter::updatePlugins]" << "Restarted";
+	qDebug() << "[ServerStarter::updatePlugins]"
+			 << "Restarted";
 }