important:

* PIThread::~PIThread() now unregister itself from introspection, if terminates than show warning
 * PISystemMonitor now correctly stops
 * PIPeer now can correctly stopAndWait
 * PIPeer::destroy(), protected method for close all eths and threads
 * new PIINTROSPECTION_STOP macro
 * Introspection now can be correctly stopped by macro, more safety

ClientServer:
 * ClientBase::close() stop and disconnect channel
 * Server clients clean-up now event-based
 * No warnings on client destructor

libs/client_server/piclientserver_client.cpp

@@ -18,13 +18,15 @@
 
 #include "piclientserver_client.h"
 
+#include "piclientserver_server.h"
 #include "piethernet.h"
 
 
-void PIClientServer::ServerClient::createForServer(PIEthernet * tcp_) {
+void PIClientServer::ServerClient::createForServer(Server * parent, PIEthernet * tcp_) {
 	tcp = tcp_;
 	tcp->setParameter(PIEthernet::KeepConnection, false);
 	init();
+	CONNECTL(tcp, disconnected, ([this, parent](bool) { parent->clientDisconnected(this); }));
 }
 
 
@@ -37,13 +39,14 @@ PIClientServer::Client::Client() {
 
 
 PIClientServer::Client::~Client() {
-	stop();
+	if (tcp) tcp->setDebug(false);
+	close();
 }
 
 
 void PIClientServer::Client::connect(PINetworkAddress addr) {
 	if (!tcp || !own_tcp) return;
-	stop();
+	close();
 	tcp->connect(addr, true);
 	tcp->startThreadedRead();
 	piCout << "Connect to" << addr.toString();

libs/client_server/piclientserver_client_base.cpp

@@ -26,17 +26,18 @@ PIClientServer::ClientBase::ClientBase() {}
 
 
 PIClientServer::ClientBase::~ClientBase() {
-	stop();
+	close();
 	if (own_tcp) piDeleteSafety(tcp);
 }
 
 
-void PIClientServer::ClientBase::stop() {
+void PIClientServer::ClientBase::close() {
 	if (!tcp) return;
 	can_write = false;
 	tcp->interrupt();
 	tcp->stopAndWait(10_s);
 	if (tcp->isThreadedRead()) tcp->terminateThreadedRead();
+	tcp->close();
 	stream.clear();
 }
 

libs/client_server/piclientserver_server.cpp

@@ -39,12 +39,12 @@ PIClientServer::Server::Server() {
 			piCout << "ClientFactory returns nullptr!";
 			return;
 		}
-		sc->createForServer(c);
+		sc->createForServer(this, c);
 		newClient(sc);
 	});
 
-	clean_thread->start(
+	clean_thread->start([this]() {
-		[this]() {
+		clean_notifier.wait();
 		PIVector<ServerClient *> to_delete;
 		clients_mutex.lock();
 		for (auto c: clients) {
@@ -62,13 +62,14 @@ PIClientServer::Server::Server() {
 			c->destroy();
 			delete c;
 		}
-		},
+	});
-		5_Hz);
 }
 
 
 PIClientServer::Server::~Server() {
-	clean_thread->stopAndWait();
+	clean_thread->stop();
+	clean_notifier.notify();
+	clean_thread->waitForFinish();
 	piDeleteSafety(clean_thread);
 	stopServer();
 	for (auto c: clients) {
@@ -129,3 +130,8 @@ void PIClientServer::Server::newClient(ServerClient * c) {
 	c->connected();
 	piCout << "New client";
 }
+
+
+void PIClientServer::Server::clientDisconnected(ServerClient * c) {
+	clean_notifier.notify();
+}

libs/main/client_server/piclientserver_client.h

@@ -44,7 +44,7 @@ protected:
 	virtual void aboutDelete() {}
 
 private:
-	void createForServer(PIEthernet * tcp_);
+	void createForServer(Server * parent, PIEthernet * tcp_);
 };
 
 

libs/main/client_server/piclientserver_client_base.h

@@ -45,7 +45,7 @@ public:
 
 	const PIEthernet * getTCP() const { return tcp; }
 
-	void stop();
+	void close();
 
 	int write(const void * d, const size_t s);
 	int write(const PIByteArray & ba) { return write(ba.data(), ba.size()); }

libs/main/client_server/piclientserver_server.h

@@ -28,6 +28,7 @@
 #include "pimutex.h"
 #include "pinetworkaddress.h"
 #include "pip_client_server_export.h"
+#include "pithreadnotifier.h"
 
 class PIEthernet;
 class PIThread;
@@ -37,6 +38,9 @@ namespace PIClientServer {
 class ServerClient;
 
 class PIP_CLIENT_SERVER_EXPORT Server {
+	friend class ServerClient;
+	NO_COPY_CLASS(Server);
+
 public:
 	Server();
 	virtual ~Server();
@@ -56,11 +60,13 @@ public:
 private:
 	void stopServer();
 	void newClient(ServerClient * c);
+	void clientDisconnected(ServerClient * c);
 
 	std::function<ServerClient *()> client_factory;
 	std::atomic_bool is_closing = {false};
 	PIEthernet * tcp_server     = nullptr;
 	PIThread * clean_thread     = nullptr;
+	PIThreadNotifier clean_notifier;
 	PIByteArray crypt_key;
 	PIVector<ServerClient *> clients;
 	mutable PIMutex clients_mutex;

libs/main/core/pibase_macros.h

@@ -328,7 +328,7 @@ typedef long long ssize_t;
 		  __PrivateInitializer__(const __PrivateInitializer__ & o);             \
 		  ~__PrivateInitializer__();                                            \
 		  __PrivateInitializer__ & operator=(const __PrivateInitializer__ & o); \
-		  __Private__ * p;                                                      \
+		  __Private__ * p = nullptr;                                            \
 	  };                                                                        \
 	  __PrivateInitializer__ __privateinitializer__;
 
@@ -343,11 +343,10 @@ typedef long long ssize_t;
 		  p = new c::__Private__();                                                                               \
 	  }                                                                                                           \
 	  c::__PrivateInitializer__::~__PrivateInitializer__() {                                                      \
-		  delete p;                                                                                               \
+		  piDeleteSafety(p);                                                                                      \
-		  p = 0;                                                                                                  \
 	  }                                                                                                           \
 	  c::__PrivateInitializer__ & c::__PrivateInitializer__::operator=(const c::__PrivateInitializer__ &) {       \
-		  if (p) delete p;                                                                                        \
+		  piDeleteSafety(p);                                                                                      \
 		  p = new c::__Private__();                                                                               \
 		  return *this;                                                                                           \
 	  }

libs/main/introspection/piintrospection_server.cpp

@@ -33,15 +33,11 @@ PRIVATE_DEFINITION_END(PIIntrospectionServer)
 
 PIIntrospectionServer::PIIntrospectionServer(): PIPeer(genName()) {
 	PRIVATE->process_info = PIIntrospection::getInfo();
-	sysmon                = 0;
 }
 
 
 PIIntrospectionServer::~PIIntrospectionServer() {
-	PIPeer::stop();
+	// stop();
-	if (sysmon)
-		if (sysmon->property("__iserver__").toBool()) delete sysmon;
-	sysmon = 0;
 }
 
 
@@ -69,6 +65,15 @@ void PIIntrospectionServer::start(const PIString & server_name) {
 }
 
 
+void PIIntrospectionServer::stop() {
+	PIPeer::stopAndWait();
+	PIPeer::destroy();
+	if (sysmon)
+		if (sysmon->property("__iserver__").toBool()) delete sysmon;
+	sysmon = nullptr;
+}
+
+
 PIString PIIntrospectionServer::genName() {
 	randomize();
 	return "__introspection__server_" + PIString::fromNumber(randomi() % 1000);
@@ -102,7 +107,7 @@ void PIIntrospectionServer::dataReceived(const PIString & from, const PIByteArra
 
 void PIIntrospectionServer::sysmonDeleted() {
 	PIMutexLocker _ml(sysmon_mutex);
-	sysmon = 0;
+	sysmon = nullptr;
 }
 
 #endif // PIP_INTROSPECTION

libs/main/introspection/piintrospection_server.h

@@ -33,6 +33,11 @@
 //! \~russian Запускает сервер интроспекции с именем "name"
 #  define PIINTROSPECTION_START(name)
 
+//! \ingroup Introspection
+//! \~english Stop introspection server
+//! \~russian Останавливает сервер интроспекции
+#  define PIINTROSPECTION_STOP
+
 #else
 
 #  if defined(PIP_INTROSPECTION) && !defined(PIP_FORCE_NO_PIINTROSPECTION)
@@ -44,6 +49,7 @@ class PISystemMonitor;
 
 #	define PIINTROSPECTION_SERVER      (PIIntrospectionServer::instance())
 #	define PIINTROSPECTION_START(name) PIINTROSPECTION_SERVER->start(#name);
+#	define PIINTROSPECTION_STOP        PIINTROSPECTION_SERVER->stop();
 
 class PIP_EXPORT PIIntrospectionServer: public PIPeer {
 	PIOBJECT_SUBCLASS(PIIntrospectionServer, PIPeer);
@@ -52,6 +58,7 @@ public:
 	static PIIntrospectionServer * instance();
 
 	void start(const PIString & server_name);
+	void stop();
 
 private:
 	PIIntrospectionServer();
@@ -59,17 +66,17 @@ private:
 	NO_COPY_CLASS(PIIntrospectionServer);
 
 	PIString genName();
-	virtual void dataReceived(const PIString & from, const PIByteArray & data);
+	void dataReceived(const PIString & from, const PIByteArray & data) override;
 	EVENT_HANDLER(void, sysmonDeleted);
 
 	PRIVATE_DECLARATION(PIP_EXPORT)
-	PITimer itimer;
+	PISystemMonitor * sysmon = nullptr;
-	PISystemMonitor * sysmon;
 	PIMutex sysmon_mutex;
 };
 
 #  else
 #	define PIINTROSPECTION_START(name)
+#	define PIINTROSPECTION_STOP
 #  endif
 
 #endif // DOXYGEN

libs/main/introspection/piintrospection_server_p.cpp

@@ -139,9 +139,11 @@ PIByteArray PIIntrospection::packThreads() {
 		PIMap<PIThread *, PIIntrospectionThreads::ThreadInfo> & tm(p->threads);
 		auto it = tm.makeIterator();
 		while (it.next()) {
+			if (it.key()->isPIObject()) {
 				it.value().classname = PIStringAscii(it.key()->className());
 				it.value().name      = it.key()->name();
 			}
+		}
 		ret << tm.values();
 		p->mutex.unlock();
 	} else {

libs/main/io_devices/pipeer.cpp

@@ -201,33 +201,7 @@ PIPeer::~PIPeer() {
 	stop();
 	if (destroyed) return;
 	destroyed = true;
-	sync_timer.stopAndWait();
+	destroy();
-	diag_s.stopAndWait();
-	diag_d.stopAndWait();
-	PIMutexLocker ml(peers_mutex);
-	piForeach(PeerInfo & p, peers)
-		if (p._data) {
-			p._data->dt_in.stop();
-			p._data->dt_out.stop();
-			p._data->t.stopAndWait();
-		}
-	destroyEths();
-	piForeach(PIEthernet * i, eths_mcast) {
-		if (!i) continue;
-		i->stopAndWait();
-	}
-	piForeach(PIEthernet * i, eths_bcast) {
-		if (!i) continue;
-		i->stopAndWait();
-	}
-	eth_lo.stopAndWait();
-	eth_tcp_srv.stopAndWait();
-	eth_tcp_cli.stopAndWait();
-	sendSelfRemove();
-	destroyMBcasts();
-	eth_send.close();
-	piForeach(PeerInfo & p, peers)
-		p.destroy();
 }
 
 
@@ -363,6 +337,42 @@ void PIPeer::initMBcasts(PIStringList al) {
 }
 
 
+void PIPeer::destroy() {
+	sync_timer.stopAndWait();
+	diag_s.stopAndWait();
+	diag_d.stopAndWait();
+	PIMutexLocker ml(peers_mutex);
+	for (auto & p: peers)
+		if (p._data) {
+			p._data->dt_in.stop();
+			p._data->dt_out.stop();
+			p._data->t.stopAndWait();
+		}
+	destroyEths();
+	for (auto * i: eths_mcast) {
+		if (!i) continue;
+		i->stopAndWait();
+	}
+	for (auto * i: eths_bcast) {
+		if (!i) continue;
+		i->stopAndWait();
+	}
+	eth_lo.stopAndWait();
+	eth_tcp_srv.stopAndWait();
+	eth_tcp_cli.stopAndWait();
+	sendSelfRemove();
+	eth_lo.close();
+	eth_tcp_srv.close();
+	eth_tcp_cli.close();
+	destroyMBcasts();
+	eth_send.close();
+	for (auto & p: peers)
+		p.destroy();
+	peers.clear();
+	destroyed = true;
+}
+
+
 void PIPeer::destroyEths() {
 	for (auto * i: eths_traffic) {
 		if (!i) continue;
@@ -930,6 +940,7 @@ ssize_t PIPeer::bytesAvailable() const {
 
 
 ssize_t PIPeer::readDevice(void * read_to, ssize_t max_size) {
+	iterrupted = false;
 	read_buffer_mutex.lock();
 	bool empty = read_buffer.isEmpty();
 	read_buffer_mutex.unlock();
@@ -937,6 +948,9 @@ ssize_t PIPeer::readDevice(void * read_to, ssize_t max_size) {
 		read_buffer_mutex.lock();
 		empty = read_buffer.isEmpty();
 		read_buffer_mutex.unlock();
+		if (iterrupted) {
+			return 0;
+		}
 		piMSleep(10);
 	}
 	read_buffer_mutex.lock();
@@ -945,6 +959,9 @@ ssize_t PIPeer::readDevice(void * read_to, ssize_t max_size) {
 		read_buffer_mutex.unlock();
 		ssize_t sz = piMini(ba.size_s(), max_size);
 		memcpy(read_to, ba.data(), sz);
+		if (iterrupted) {
+			return 0;
+		}
 		return sz;
 	}
 	read_buffer_mutex.unlock();
@@ -964,6 +981,11 @@ ssize_t PIPeer::writeDevice(const void * data, ssize_t size) {
 }
 
 
+void PIPeer::interrupt() {
+	iterrupted = true;
+}
+
+
 void PIPeer::newTcpClient(PIEthernet * client) {
 	client->setName("__S__PIPeer_eth_TCP_ServerClient" + client->path());
 	piCoutObj << "client" << client->path();

libs/main/io_devices/pipeer.h

@@ -171,6 +171,8 @@ protected:
 	EVENT_HANDLER2(bool, dataRead, const uchar *, readed, ssize_t, size);
 	EVENT_HANDLER2(bool, mbcastRead, const uchar *, readed, ssize_t, size);
 
+	void destroy();
+
 private:
 	EVENT_HANDLER1(void, timerEvent, int, delim);
 	EVENT_HANDLER2(bool, sendInternal, const PIString &, to, const PIByteArray &, data);
@@ -212,6 +214,7 @@ private:
 	void configureFromVariantDevice(const PIPropertyStorage & d) override;
 	ssize_t readDevice(void * read_to, ssize_t max_size) override;
 	ssize_t writeDevice(const void * data, ssize_t size) override;
+	void interrupt() override;
 	DeviceInfoFlags deviceInfoFlags() const override { return PIIODevice::Reliable; }
 
 	PeerInfo * quickestPeer(const PIString & to);
@@ -243,6 +246,7 @@ private:
 	mutable PIMutex read_buffer_mutex;
 	PIQueue<PIByteArray> read_buffer;
 	int read_buffer_size;
+	std::atomic_bool iterrupted = {false};
 	PIMutex mc_mutex, eth_mutex, peers_mutex, send_mutex, send_mc_mutex;
 };
 

libs/main/system/pisystemmonitor.cpp

@@ -150,7 +150,7 @@ void PISystemMonitor::setStatistic(const PISystemMonitor::ProcessStats & s) {
 
 
 void PISystemMonitor::stop() {
-	PIThread::stop();
+	PIThread::stopAndWait();
 #ifdef WINDOWS
 	if (PRIVATE->hProc != 0) {
 		CloseHandle(PRIVATE->hProc);

libs/main/thread/pithread.cpp

@@ -568,6 +568,7 @@ PIThread::PIThread(bool startNow, PISystemTime loop_delay): PIObject() {
 PIThread::~PIThread() {
 	PIINTROSPECTION_THREAD_DELETE(this);
 	if (!running_ || PRIVATE->thread == 0) return;
+	piCout << "[PIThread \"" << name() << "\"] Warning, terminate on destructor!";
 #ifdef FREERTOS
 	// void * ret(0);
 	// PICout(PICoutManipulators::DefaultControls) << "~PIThread" << PRIVATE->thread;
@@ -587,6 +588,8 @@ PIThread::~PIThread() {
 	CloseHandle(PRIVATE->thread);
 #  endif
 #endif
+	UNREGISTER_THREAD(this);
+	PIINTROSPECTION_THREAD_STOP(this);
 	terminating = running_ = false;
 }
 

libs/main/thread/pithreadnotifier.cpp

@@ -72,9 +72,9 @@
 //! 		w->startOnce();
 //!
 //! 	piMSleep(500);
-//! 	notifier.notifyOnce(); // notify one of them after 500 ms
+//! 	notifier.notify(); // notify one of them after 500 ms
 //! 	piMSleep(500);
-//! 	notifier.notifyOnce(); // notify one of them after 1000 ms
+//! 	notifier.notify(); // notify one of them after 1000 ms
 //!
 //! 	for (auto * w: workers)
 //! 		w->waitForFinish();
@@ -93,17 +93,17 @@ PIThreadNotifier::PIThreadNotifier(): cnt(0) {}
 
 //! \~\details
 //! \~english
-//! If \a notifyOnce() has been called before, then returns immediately.\n
+//! If \a notify() has been called before, then returns immediately.\n
-//! If \a notifyOnce() has been called "n" times, then returns immediately "n" times,
+//! If \a notify() has been called "n" times, then returns immediately "n" times,
 //! but only if wait in one thread.\n
-//! If many threads waiting, then if \a notifyOnce() has been called "n" times,
+//! If many threads waiting, then if \a notify() has been called "n" times,
 //! all threads total returns "n" times in undefined sequence.
 //!
 //! \~russian
-//! Если ранее был вызван \a notifyOnce(), то возвращает управление немедленно.\n
+//! Если ранее был вызван \a notify(), то возвращает управление немедленно.\n
-//! Если ранее был вызван \a notifyOnce() "n" раз, то возвращает управление немедленно "n" раз,
+//! Если ранее был вызван \a notify() "n" раз, то возвращает управление немедленно "n" раз,
 //! но только если ожидать одним потоком.\n
-//! Если ожидают несколько потоков, и \a notifyOnce() был вызван "n" раз,
+//! Если ожидают несколько потоков, и \a notify() был вызван "n" раз,
 //! то все потоки суммарно вернут управление "n" раз в неопределенной последовательности.
 //!
 void PIThreadNotifier::wait() {

libs/main/thread/pithreadnotifier.h

@@ -33,8 +33,8 @@ class PIP_EXPORT PIThreadNotifier {
 public:
 	PIThreadNotifier();
 
-	//! \~english Start waiting, return if other thread call \a notifyOnce()
+	//! \~english Start waiting, return if other thread call \a notify()
-	//! \~russian Начать ожидание, продолжает когда другой поток вызовет \a notifyOnce()
+	//! \~russian Начать ожидание, продолжает когда другой поток вызовет \a notify()
 	void wait();
 
 	//! \~english Notify one waiting thread, which waiting on \a wait() function

main.cpp

@@ -2,6 +2,7 @@
 #include "piclientserver_client.h"
 #include "piclientserver_server.h"
 #include "picodeparser.h"
+#include "piintrospection_server.h"
 #include "piiostream.h"
 #include "pijson.h"
 #include "pimathbase.h"
@@ -22,12 +23,15 @@ PIKbdListener kbd;
 
 
 class MyServerClient: public PIClientServer::ServerClient {
+public:
+	~MyServerClient() { send_thread.stopAndWait(); }
+
 protected:
-	void readed(PIByteArray data) override { piCout << "readed" << (data.size()); }
+	// void readed(PIByteArray data) override { piCout << "readed" << (data.size()); }
-	void aboutDelete() override { piCout << "aboutDelete"; }
+	// void aboutDelete() override { piCout << "aboutDelete"; }
-	void disconnected() override { piCout << "disconnected"; }
+	// void disconnected() override { piCout << "disconnected"; }
 	void connected() override {
-		piCout << "connected";
+		// piCout << "connected";
 		send_thread.start(
 			[this] {
 				// write((PIString::fromNumber(++counter)).toUTF8());
@@ -42,16 +46,18 @@ protected:
 
 
 class MyClient: public PIClientServer::Client {
+public:
+	~MyClient() { send_thread.stopAndWait(); }
+
 protected:
-	void readed(PIByteArray data) override { piCout << "readed" << (data.size()); }
+	// void readed(PIByteArray data) override { piCout << "readed" << (data.size()); }
 	void disconnected() override { piCout << "disconnected"; }
 	void connected() override {
 		piCout << "connected";
 		send_thread.start(
 			[this] {
 				// write((PIString::fromNumber(++counter)).toUTF8());
-				PIByteArray ba(64_KiB);
+				write(PIByteArray(64_KiB));
-				write(ba);
 			},
 			2_Hz);
 	}
@@ -61,11 +67,34 @@ protected:
 
 
 int main(int argc, char * argv[]) {
+	/*PIPeer p("123");
+
+	piCout << "start ...";
+	p.start();
+	piCout << "start ok";
+
+	piSleep(1.);
+
+	piCout << "stop ...";
+	p.stopAndWait();
+	piCout << "stop ok";
+
+	piSleep(1.);
+
+	piCout << "exit";
+	return 0;*/
+
+	if (argc > 1) {
+		PIINTROSPECTION_START(server);
+	} else {
+		PIINTROSPECTION_START(client);
+	}
+
 	kbd.enableExitCapture();
 
 
 	PIClientServer::Server * s = nullptr;
-	PIThread s_thread;
+	PIThread * s_thread        = new PIThread();
 	PIVector<PIClientServer::Client *> cv;
 
 	if (argc > 1) {
@@ -74,21 +103,22 @@ int main(int argc, char * argv[]) {
 		s->setClientFactory([] { return new MyServerClient(); });
 		s->enableSymmetricEncryption("1122334455667788"_hex);
 		s->listenAll(12345);
-		s_thread.start(
+		s_thread->start(
 			[s] {
 				piCout << "*** clients" << s->clientsCount();
 				int i = 0;
 				s->forEachClient([&i](PIClientServer::ServerClient * c) {
-					piCout << "client" << ++i << c;
+					// piCout << "client" << ++i << c;
-					c->write(PIByteArray(16_KiB));
+					c->write(PIByteArray(1_KiB));
-					piMSleep(200);
+					c->close();
+					// piMSleep(200);
 				});
 			},
-			0.5_Hz);
+			1._Hz);
 	} else {
 		piCout << "Client";
-		piForTimes(5) {
+		piForTimes(20) {
-			piMSleep(50);
+			piMSleep(25);
 			auto c = new MyClient();
 			c->enableSymmetricEncryption("1122334455667788"_hex);
 			c->connect(PINetworkAddress::resolve("127.0.0.1", 12345));
@@ -98,11 +128,13 @@ int main(int argc, char * argv[]) {
 
 	WAIT_FOR_EXIT;
 
-	s_thread.stopAndWait();
+	s_thread->stopAndWait();
 
 	piDeleteSafety(s);
 	piDeleteAllAndClear(cv);
 
+	PIINTROSPECTION_STOP
+
 	return 0;
 	/*PIPackedTCP * tcp_s =
 		PIIODevice::createFromFullPath("ptcp://s::8000")->cast<PIPackedTCP>(); // new PIPackedTCP(PIPackedTCP::Server, {"0.0.0.0:8000"});