pip/libs/main/io_devices/pipeer.h

//! \~\file pipeer.h
//! \~\ingroup IO
//! \~\brief
//! \~english Peer-to-peer network node
//! \~russian Узел одноранговой сети
/*
    PIP - Platform Independent Primitives
    Peer - named I/O ethernet node, forming self-organized peering network
    Ivan Pelipenko peri4ko@yandex.ru

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program 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 Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#ifndef PIPEER_H
#define PIPEER_H

#include "pidiagnostics.h"
#include "piethernet.h"

//! \~\ingroup IO
//! \~\brief
//! \~english Named network peer built on top of %PIIODevice.
//! \~russian Именованный сетевой пир, построенный поверх %PIIODevice.
//! \~\details
//! \~english
//! The class discovers peers, routes packets by peer name and can expose a trusted-peer stream through inherited \a read() and \a write().
//! \~russian
//! Класс обнаруживает пиры, маршрутизирует пакеты по имени пира и может предоставлять поток trusted-peer через унаследованные \a read() и
//! \a write().
class PIP_EXPORT PIPeer: public PIIODevice {
	PIIODEVICE(PIPeer, "peer");

private:
	class PeerData;

public:
	//! \~english Constructs a peer node with local name "name".
	//! \~russian Создает пиринговый узел с локальным именем "name".
	explicit PIPeer(const PIString & name = PIString());

	//! \~english Destroys the peer node.
	//! \~russian Уничтожает пиринговый узел.
	virtual ~PIPeer();

	//! \~\ingroup IO
	//! \~\brief
	//! \~english Public information about a discovered peer.
	//! \~russian Общедоступная информация об обнаруженном пире.
	class PIP_EXPORT PeerInfo {
		friend class PIPeer;
		BINARY_STREAM_FRIEND(PIPeer::PeerInfo);

	public:
		//! \~english Constructs an empty peer description.
		//! \~russian Создает пустое описание пира.
		PeerInfo() {
			dist = sync = cnt = 0;
			trace             = -1;
			was_update        = false;
			_data             = 0;
		}

		//! \~english Destroys the peer description.
		//! \~russian Уничтожает описание пира.
		~PeerInfo() {}

		//! \~\ingroup IO
		//! \~\brief
		//! \~english Network address of a peer endpoint.
		//! \~russian Сетевой адрес конечной точки пира.
		struct PIP_EXPORT PeerAddress {
			//! \~english Constructs a peer address with address and netmask.
			//! \~russian Создает адрес пира с адресом и маской сети.
			PeerAddress(const PINetworkAddress & a = PINetworkAddress(), const PINetworkAddress & m = PINetworkAddress("255.255.255.0"));

			//! \~english Returns whether this address has a valid measured ping.
			//! \~russian Возвращает, есть ли для этого адреса валидный измеренный ping.
			bool isAvailable() const { return ping > 0; }

			//! \~english Peer address.
			//! \~russian Адрес пира.
			PINetworkAddress address;

			//! \~english Netmask for the address.
			//! \~russian Маска сети для адреса.
			PINetworkAddress netmask;

			//! \~english Last measured ping in milliseconds, or a negative value if unknown.
			//! \~russian Последний измеренный ping в миллисекундах, либо отрицательное значение если он неизвестен.
			double ping; // ms

			//! \~english Returns whether a ping request is currently pending.
			//! \~russian Показывает, ожидается ли сейчас ответ на ping-запрос.
			bool wait_ping;

			//! \~english Timestamp of the last ping request or reply.
			//! \~russian Временная метка последнего ping-запроса или ответа.
			PISystemTime last_ping;
		};

		//! \~english Peer name.
		//! \~russian Имя пира.
		PIString name;

		//! \~english Known addresses of the peer.
		//! \~russian Известные адреса пира.
		PIVector<PeerAddress> addresses;

		//! \~english Distance in hops from the local peer (0 for direct neighbours).
		//! \~russian Расстояние в хопах от локального пира (0 для прямых соседей).
		int dist;

		//! \~english Names of direct neighbours for this peer.
		//! \~russian Имена прямых соседей этого пира.
		PIStringList neighbours;

		//! \~english Returns whether the peer is a direct neighbour.
		//! \~russian Возвращает, является ли пир прямым соседом.
		bool isNeighbour() const { return dist == 0; }

		//! \~english Returns the best known ping in milliseconds.
		//! \~russian Возвращает наилучший известный ping в миллисекундах.
		int ping() const;

		//! \~english Returns the fastest known address of the peer.
		//! \~russian Возвращает самый быстрый известный адрес пира.
		PINetworkAddress fastestAddress() const;

	protected:
		void addNeighbour(const PIString & n);
		void addNeighbours(const PIStringList & l);
		void removeNeighbour(const PIString & n);
		void resetPing();
		void init();
		void destroy();

		int sync, cnt, trace;
		bool was_update;
		PISystemTime time;
		PeerData * _data;
	};

	BINARY_STREAM_FRIEND(PIPeer::PeerInfo);

	//! \~english Sends byte array "data" to peer "to".
	//! \~russian Отправляет массив байт "data" пиру "to".
	//! \~\details
	//! \~english Sends the specified byte array to the peer identified by name using the most efficient route.
	//! \~russian Отправляет указанный байтовый массив пиру, идентифицируемому по имени, используя наиболее эффективный маршрут.
	bool send(const PIString & to, const PIByteArray & data) { return send(to, data.data(), data.size_s()); }

	//! \~english Sends string "data" to peer "to".
	//! \~russian Отправляет строку "data" пиру "to".
	//! \~\details
	//! \~english Sends the specified string to the peer identified by name using the most efficient route.
	//! \~russian Отправляет указанную строку пиру, идентифицируемому по имени, используя наиболее эффективный маршрут.
	bool send(const PIString & to, const PIString & data) { return send(to, data.data(), data.size_s()); }

	//! \~english Sends raw buffer to peer "to".
	//! \~russian Отправляет сырой буфер пиру "to".
	bool send(const PIString & to, const void * data, int size);

	//! \~english Sends byte array "data" to peer described by "to".
	//! \~russian Отправляет массив байт "data" пиру, описанному в "to".
	bool send(const PeerInfo & to, const PIByteArray & data) { return send(to.name, data.data(), data.size_s()); }

	//! \~english Sends string "data" to peer described by "to".
	//! \~russian Отправляет строку "data" пиру, описанному в "to".
	bool send(const PeerInfo & to, const PIString & data) { return send(to.name, data.data(), data.size_s()); }

	//! \~english Sends raw buffer to peer described by "to".
	//! \~russian Отправляет сырой буфер пиру, описанному в "to".
	bool send(const PeerInfo & to, const void * data, int size) { return send(to.name, data, size); }

	//! \~english Sends byte array "data" to peer pointer "to".
	//! \~russian Отправляет массив байт "data" пиру по указателю "to".
	bool send(const PeerInfo * to, const PIByteArray & data);

	//! \~english Sends string "data" to peer pointer "to".
	//! \~russian Отправляет строку "data" пиру по указателю "to".
	bool send(const PeerInfo * to, const PIString & data);

	//! \~english Sends raw buffer to peer pointer "to".
	//! \~russian Отправляет сырой буфер пиру по указателю "to".
	bool send(const PeerInfo * to, const void * data, int size);

	//! \~english Sends byte array "data" to all known peers.
	//! \~russian Отправляет массив байт "data" всем известным пирам.
	void sendToAll(const PIByteArray & data);

	//! \~english Sends string "data" to all known peers.
	//! \~russian Отправляет строку "data" всем известным пирам.
	void sendToAll(const PIString & data);

	//! \~english Sends raw buffer to all known peers.
	//! \~russian Отправляет сырой буфер всем известным пирам.
	void sendToAll(const void * data, int size);

	//! \~english Returns whether multicast reception is active.
	//! \~russian Возвращает, активно ли получение multicast-пакетов.
	bool isMulticastReceive() const { return !eths_mcast.isEmpty(); }

	//! \~english Returns whether broadcast reception is active.
	//! \~russian Возвращает, активно ли получение broadcast-пакетов.
	bool isBroadcastReceive() const { return !eths_bcast.isEmpty(); }

	//! \~english Returns service-channel diagnostics.
	//! \~russian Возвращает диагностику служебного канала.
	PIDiagnostics & diagnosticService() { return diag_s; }

	//! \~english Returns payload-channel diagnostics.
	//! \~russian Возвращает диагностику канала данных.
	PIDiagnostics & diagnosticData() { return diag_d; }

	//! \~english Returns all currently known peers.
	//! \~russian Возвращает всех известных на данный момент пиров.
	const PIVector<PIPeer::PeerInfo> & allPeers() const { return peers; }

	//! \~english Returns whether a peer with name "name" is known.
	//! \~russian Возвращает, известен ли пир с именем "name".
	bool isPeerExists(const PIString & name) const { return getPeerByName(name) != 0; }

	//! \~english Returns peer information by name, or null if absent.
	//! \~russian Возвращает информацию о пире по имени, либо null если пир не найден.
	const PeerInfo * getPeerByName(const PIString & name) const { return peers_map.value(name, 0); }

	//! \~english Returns information about the local peer.
	//! \~russian Возвращает информацию о локальном пире.
	const PeerInfo & selfInfo() const { return self_info; }

	//! \~english Returns routing map used to reach known peers.
	//! \~russian Возвращает карту маршрутов, используемую для доступа к известным пирам.
	const PIMap<PIString, PIVector<PeerInfo *>> & _peerMap() const { return addresses_map; }

	//! \~english Rebuilds the peer network state and restarts discovery sockets.
	//! \~russian Перестраивает состояние сети пиров и перезапускает сокеты обнаружения.
	void reinit();

	//! \~english Locks the peer list for manual external access.
	//! \~russian Блокирует список пиров для внешнего ручного доступа.
	void lock() { peers_mutex.lock(); }

	//! \~english Unlocks the peer list after external access.
	//! \~russian Снимает блокировку списка пиров после внешнего доступа.
	void unlock() { peers_mutex.unlock(); }

	//! \~english Changes local peer name and updates related diagnostics names.
	//! \~russian Изменяет имя локального пира и обновляет связанные диагностические имена.
	void changeName(const PIString & new_name);

	//! \~english Returns trusted peer name used by inherited \a read() and \a write().
	//! \~russian Возвращает имя доверенного пира, используемое унаследованными \a read() и \a write().
	const PIString & trustPeerName() const { return trust_peer; }

	//! \~english Sets trusted peer name for inherited \a read() and \a write().
	//! \~russian Устанавливает имя доверенного пира для унаследованных \a read() и \a write().
	void setTrustPeerName(const PIString & peer_name) { trust_peer = peer_name; }

	//! \~english Sets TCP server address used for peer discovery fallback.
	//! \~russian Устанавливает адрес TCP-сервера, используемого как резервный канал обнаружения пиров.
	void setTcpServerIP(const PIString & ip);

	//! \~english Returns size of the next buffered payload from the trusted peer stream.
	//! \~russian Возвращает размер следующей буферизованной полезной нагрузки из trusted-peer потока.
	ssize_t bytesAvailable() const override;

	//! \events
	//! \{

	//! \fn void dataReceivedEvent(const PIString & from, const PIByteArray & data)
	//! \~english Raised when payload data is delivered from peer "from".
	//! \~russian Вызывается, когда полезные данные доставлены от пира "from".
	EVENT2(dataReceivedEvent, const PIString &, from, const PIByteArray &, data);

	//! \fn void peerConnectedEvent(const PIString & name)
	//! \~english Raised when a new peer becomes available.
	//! \~russian Вызывается, когда становится доступен новый пир.
	EVENT1(peerConnectedEvent, const PIString &, name);

	//! \fn void peerDisconnectedEvent(const PIString & name)
	//! \~english Raised when a known peer disappears from the network.
	//! \~russian Вызывается, когда известный пир исчезает из сети.
	EVENT1(peerDisconnectedEvent, const PIString &, name);

	//! \}

	//	bool lockedEth() const {return eth_mutex.isLocked();}
	//	bool lockedPeers() const {return peers_mutex.isLocked();}
	//	bool lockedMBcasts() const {return mc_mutex.isLocked();}
	//	bool lockedSends() const {return send_mutex.isLocked();}
	//	bool lockedMCSends() const {return send_mc_mutex.isLocked();}

protected:
	//! \~english Reimplement to handle incoming payload data.
	//! \~russian Переопределите для обработки входящих полезных данных.
	virtual void dataReceived(const PIString & from, const PIByteArray & data) { ; }

	//! \~english Reimplement to react to peer appearance.
	//! \~russian Переопределите для реакции на появление пира.
	virtual void peerConnected(const PIString & name) { ; }

	//! \~english Reimplement to react to peer disappearance.
	//! \~russian Переопределите для реакции на исчезновение пира.
	virtual void peerDisconnected(const PIString & name) { ; }

	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);
	EVENT_HANDLER2(void, dtReceived, const PIString &, from, const PIByteArray &, data);
	EVENT_HANDLER1(void, newTcpClient, PIEthernet *, client);
	EVENT_HANDLER(void, tcpClientReconnect);

	bool hasPeer(const PIString & name);
	bool removePeer(const PIString & name);
	void removeNeighbour(const PIString & name);
	void addPeer(const PeerInfo & pd);

	void sendPeerInfo(const PeerInfo & info);
	void sendPeerRemove(const PIString & peer);
	void sendSelfInfo() { sendPeerInfo(self_info); }
	void sendSelfRemove() { sendPeerRemove(self_info.name); }
	void syncPeers();
	void checkNetwork();
	void initNetwork();
	void buildMap();
	void initEths(PIStringList al);
	void initMBcasts(PIStringList al);
	void destroyEths();
	void destroyMBcasts();
	void sendMBcast(const PIByteArray & ba);
	void pingNeighbours();
	void addToRemoved(const PeerInfo & pi) { removed[pi.name] = PIPair<int, PISystemTime>(pi.cnt, pi.time); }
	bool isRemoved(const PeerInfo & pi) const { return (removed.value(pi.name) == PIPair<int, PISystemTime>(pi.cnt, pi.time)); }

	bool openDevice() override;
	bool closeDevice() override;
	PIString constructFullPathDevice() const override;
	void configureFromFullPathDevice(const PIString & full_path) override;
	PIPropertyStorage constructVariantDevice() const override;
	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);
	bool sendToNeighbour(PeerInfo * peer, const PIByteArray & ba);
	inline static bool isPeerRecent(const PeerInfo & my, const PeerInfo & income) {
		return (my.cnt < income.cnt) || (my.time < income.time);
	}

	// 1 - new peer, 2 - remove peer, 3 - sync peers, 4 - data, 5 - ping request, 6 - ping reply
	// Data packet: 4, from, to, ticks, data_size, data

protected:
	bool inited__; // for internal use

private:
	PIVector<PIEthernet *> eths_traffic, eths_mcast, eths_bcast;
	PIEthernet::InterfaceList prev_ifaces;
	PIEthernet eth_send, eth_lo, eth_tcp_srv, eth_tcp_cli;
	PITimer sync_timer;
	PeerInfo self_info;
	PIVector<PeerInfo> peers;
	PIMap<PIString, PeerInfo *> peers_map;
	PIMap<PIString, PIVector<PeerInfo *>> addresses_map; // map {"to" = list of nearest peers}
	PIMap<PIString, PIPair<int, PISystemTime>> removed;
	PIDiagnostics diag_s, diag_d;
	bool destroyed, no_timer;
	PIString trust_peer;
	PIString server_ip;
	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;
};

inline PICout operator<<(PICout c, const PIPeer::PeerInfo::PeerAddress & v) {
	c.space();
	c << "PeerAddress(" << v.address << ", " << v.netmask << ", " << v.ping << ")";
	return c;
}
inline PICout operator<<(PICout c, const PIPeer::PeerInfo & v) {
	c.space();
	c << "PeerInfo(" << v.name << ", " << v.dist << ", " << v.addresses << ")";
	return c;
}


//! \relatesalso PIBinaryStream
//! \~english Store operator.
//! \~russian Оператор сохранения.
BINARY_STREAM_WRITE(PIPeer::PeerInfo::PeerAddress) {
	s << v.address << v.netmask << v.ping;
	return s;
}

//! \relatesalso PIBinaryStream
//! \~english Restore operator.
//! \~russian Оператор извлечения.
BINARY_STREAM_READ(PIPeer::PeerInfo::PeerAddress) {
	s >> v.address >> v.netmask >> v.ping;
	return s;
}


//! \relatesalso PIBinaryStream
//! \~english Store operator.
//! \~russian Оператор сохранения.
BINARY_STREAM_WRITE(PIPeer::PeerInfo) {
	s << v.name << v.addresses << v.dist << v.neighbours << v.cnt << v.time;
	return s;
}

//! \relatesalso PIBinaryStream
//! \~english Restore operator.
//! \~russian Оператор извлечения.
BINARY_STREAM_READ(PIPeer::PeerInfo) {
	s >> v.name >> v.addresses >> v.dist >> v.neighbours >> v.cnt >> v.time;
	return s;
}


#endif // PIPEER_H