PICA_node.c

/*
	(c) Copyright  2012 - 2018 Anton Sviridenko
	https://picapica.im

	This program 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, version 3.

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

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

#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
#include <string.h>
#include <time.h>
#include <getopt.h>
#include <signal.h>
#include "PICA_security.h"
#include "PICA_node.h"
#include "PICA_proto.h"
#include "PICA_msgproc.h"
#include "PICA_nodeaddrlist.h"
#include "PICA_netconf.h"
#include "PICA_common.h"
#include "PICA_nodeconfig.h"
#include "PICA_log.h"
#include "PICA_nodewait.h"
#include "PICA_signverify.h"

#ifdef NO_RAND_DEV
#include "PICA_rand_seed.h"
#endif

#ifdef WIN32

#include <shlobj.h>

#ifdef __MINGW32__
#include <stdint.h>
#endif

typedef u_long in_addr_t;
typedef u_short in_port_t;
typedef u_short uint16_t;
#endif

#define MAX_NEWCONNS 64

#define NEWCONN_TIMEOUT 10 //sec
#define SKYNET_REFRESH_TIMEOUT 93
#define SELECT_TIMEOUT 1

#define NOREPLY_TIMEOUT 15
#define KEEPALIVE_TIMEOUT 60

time_t skynet_refresh_tmst;

clock_t TMO_CCLINK_WAITACTIVE = 15; //CONF

char *my_addr;//TEMP CONF FIXME

static struct event_base *pica_event_base;
SOCKET listen_comm_sck;//*

static int stop_node_loop = 0;

SSL_CTX *ctx;
SSL_CTX *anon_ctx;
struct newconn newconns[MAX_NEWCONNS];
unsigned int newconns_pos = 0;


struct client *client_tree_root;
struct client *client_list_head;
struct client *client_list_end;

struct cclink *cclink_list_head;
struct cclink *cclink_list_end;

unsigned int cclink_list_count;

struct nodelink *nodelink_list_head;
struct nodelink *nodelink_list_end;
unsigned int nodelink_list_count;


const char msg_INITRESPOK[] = {PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK, 'O', 'K'};
const char msg_VERDIFFER[] = {PICA_PROTO_VERDIFFER, PICA_PROTO_VERDIFFER, PICA_PROTO_VER_HIGH, PICA_PROTO_VER_LOW};

unsigned int procmsg_INITREQ(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_CONNID(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_NODECONNREQ(unsigned char* buf, unsigned int size, void* ptr);

unsigned int procmsg_CONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_CONNALLOW(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_CONNDENY(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_NODERESP(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_NEWNODE(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_NODELISTREQ(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_NODELIST(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_N2NFOUND(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_SEARCH(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_N2NCONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_N2NALLOW(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_N2NNOTFOUND(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_N2NMSG(unsigned char* buf, unsigned int size, void* ptr);

unsigned int procmsg_PINGREQ_client(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_PINGREP_client(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_PINGREQ_node(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_PINGREP_node(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_dummy_multi_c2n(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_MULTILOGIN_client(unsigned char* buf, unsigned int size, void* ptr);
unsigned int procmsg_MULTILOGIN_node(unsigned char* buf, unsigned int size, void* ptr);

struct cclink* cclink_list_add(struct client *clr, struct client *cle);
struct cclink* cclink_list_search(const unsigned char *caller_id, const unsigned char *callee_id);
struct cclink* cclink_list_findwaitsearch(const unsigned char *callee_id);
void cclink_list_addlocal(struct client *clr, struct client *cle);
void cclink_list_delete(struct cclink *l);
void cclink_list_addn2nclr(struct client *clr, const unsigned char *callee_id);
void cclink_setwaitconn(struct cclink *ccl);
void cclink_list_addn2ncle(const unsigned char *caller_id, struct nodelink *caller_node, struct client *cle);
void cclink_activate(struct cclink *ccl);
void cclink_setwaitanontlsshutdown(struct cclink *link, struct newconn *nc);
void cclink_attach_remotecle_node(struct cclink *ccl, struct nodelink *callee_node);

void cclink_list_delete_by_client(struct client *cl);
void cclink_list_delete_by_nodelink(struct nodelink *node);

struct client* client_tree_search(const unsigned char *id);//
struct client *client_list_addnew(struct newconn *nc);
void client_list_delete(struct client* ci);
struct PICA_proto_msg* client_wbuf_push(struct client *c, unsigned int msgid, unsigned int size);
int client_rbuf_grow(struct client *c);

struct PICA_proto_msg* nodelink_wbuf_push(struct nodelink *nl, unsigned int msgid, unsigned int size);
int nodelink_rbuf_grow(struct nodelink *nl);
int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size);
struct nodelink *nodelink_list_addnew(struct newconn *nc);
void nodelink_list_delete(struct nodelink *n);
struct nodelink *nodelink_search_by_ipv4addr(in_addr_t addr, in_port_t port); /* arguments are in network byte order */

void newconn_close(struct newconn* nc);
void newconn_free(struct newconn* nc);
struct newconn* newconn_add(struct newconn *ncs, int *pos);
void newconns_init();

void PICA_node_joinskynet(const char* addrlistfilename, const char *my_addr);

void process_listen_read(evutil_socket_t listener, short event, void *arg);
void process_timeout_event(evutil_socket_t listener, short event, void *arg);
void process_newconn_read(evutil_socket_t s, short event, void *arg);
void process_c2n_read(evutil_socket_t s, short event, void *arg);
void process_c2n_write(evutil_socket_t s, short event, void *arg);
void process_n2n_read(evutil_socket_t s, short event, void *arg);
void process_n2n_write(evutil_socket_t s, short event, void *arg);
void process_c2c_read(evutil_socket_t s, short event, void *arg);
void process_c2c_write(evutil_socket_t s, short event, void *arg);
void process_nodewait();

#define MSGINFO_MSGSNUM(arr) (sizeof(arr)/sizeof(struct PICA_msginfo))

struct PICA_msginfo  _msginfo_comm[] =
{
	{PICA_PROTO_CONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNREQOUTG_SIZE, procmsg_CONNREQOUTG},
	{PICA_PROTO_CONNALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNALLOW_SIZE, procmsg_CONNALLOW},
	{PICA_PROTO_CONNDENY, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNDENY_SIZE, procmsg_CONNDENY},
	{PICA_PROTO_CLNODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_CLNODELISTREQ_SIZE, procmsg_NODELISTREQ},
	{PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_client},
	{PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_client},
	{PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_client}
};

struct PICA_msginfo _msginfo_node[] =
{
	{PICA_PROTO_INITRESPOK, PICA_MSG_FIXED_SIZE, PICA_PROTO_INITRESPOK_SIZE, procmsg_NODERESP},
	{PICA_PROTO_VERDIFFER, PICA_MSG_FIXED_SIZE, PICA_PROTO_VERDIFFER_SIZE, procmsg_NODERESP},
	{PICA_PROTO_NEWNODE_IPV4, PICA_MSG_FIXED_SIZE, PICA_PROTO_NEWNODE_IPV4_SIZE, procmsg_NEWNODE},
	{PICA_PROTO_NODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_NODELISTREQ_SIZE, procmsg_NODELISTREQ},
	{PICA_PROTO_NODELIST, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_NODELIST},

	{PICA_PROTO_SEARCH, PICA_MSG_FIXED_SIZE, PICA_PROTO_SEARCH_SIZE, procmsg_SEARCH},
	{PICA_PROTO_N2NFOUND, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NFOUND_SIZE, procmsg_N2NFOUND},
	{PICA_PROTO_N2NFOUNDCACHE, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_N2NFOUND},
	{PICA_PROTO_N2NCONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NCONNREQOUTG_SIZE, procmsg_N2NCONNREQOUTG},
	{PICA_PROTO_N2NALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NALLOW_SIZE, procmsg_N2NALLOW},
	{PICA_PROTO_N2NNOTFOUND, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NNOTFOUND_SIZE, procmsg_N2NNOTFOUND},
	{PICA_PROTO_N2NMSG, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_N2NMSG},
	{PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_node},
	{PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_node},
	{PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_node}
};

struct PICA_msginfo _msginfo_newconn[] =
{
	{PICA_PROTO_INITREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_INITREQ_SIZE, procmsg_INITREQ},
	{PICA_PROTO_CONNID, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNID_SIZE, procmsg_CONNID},
	{PICA_PROTO_NODECONNREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_NODECONNREQ_SIZE, procmsg_NODECONNREQ}
};

struct PICA_msginfo  _msginfo_multi[] =
{
	{PICA_PROTO_CONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNREQOUTG_SIZE, procmsg_dummy_multi_c2n},
	{PICA_PROTO_CONNALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNALLOW_SIZE, procmsg_dummy_multi_c2n},
	{PICA_PROTO_CONNDENY, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNDENY_SIZE, procmsg_dummy_multi_c2n},
	{PICA_PROTO_CLNODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_CLNODELISTREQ_SIZE, procmsg_NODELISTREQ},
	{PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_client},
	{PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_client},
	{PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_client}
};

static int process_async_ssl_readwrite(SSL *ssl, int ret)
{
	if (ret == 0)
		return 0;

	if (ret < 0)
	{
		switch(SSL_get_error(ssl, ret))
		{
		case SSL_ERROR_WANT_WRITE:
		case SSL_ERROR_WANT_READ:
			break;
		default:
			return 0;
		}
	}

	return 1;
}

static int verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
{
	//return 1 for self-signed certificates
	char    buf[256];
	X509   *err_cert;
	int     err, depth;

	err_cert = X509_STORE_CTX_get_current_cert(ctx);
	err = X509_STORE_CTX_get_error(ctx);
	depth = X509_STORE_CTX_get_error_depth(ctx);

	X509_NAME_oneline(X509_get_subject_name(err_cert), buf, 256);

	if (err == X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT && depth == 0)
		preverify_ok = 1;

	if (!preverify_ok)
	{
		PICA_info("certificate verification error:num=%d:%s:depth=%d:%s\n", err, X509_verify_cert_error_string(err), depth, buf);
	}

	return preverify_ok;
}

unsigned int procmsg_INITREQ(unsigned char* buf, unsigned int size, void* ptr)
{
	struct newconn *nc = (struct newconn*)ptr;
	struct PICA_proto_msg *mp;

	PICA_debug1("received INITREQ");

	nc->type = NEWCONN_C2N;

	if (buf[2] == PICA_PROTO_VER_HIGH && buf[3] == PICA_PROTO_VER_LOW)
	{
		struct client *c;
		c = client_list_addnew(nc);
		if (!c)
			return 0;

		c->state = PICA_CLSTATE_WAITANONTLSSHUTDOWN;
		nc->iconn.cl = c;
		nc->state = PICA_NEWCONN_SENDINGOK;
		/*
		newconn_free(nc);

		if ((mp = client_wbuf_push(c, PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK_SIZE)))
		{
			mp->tail[0] = 'O';
			mp->tail[1] = 'K';
		}

		c->state = PICA_CLSTATE_SENDINGRESP;*/
	}///////---------<<<<<<<<<<< new client
	else
	{
		nc->state = PICA_NEWCONN_SENDINGREJECT;

		if (buf[2] > PICA_PROTO_VER_HIGH || (buf[2] == PICA_PROTO_VER_HIGH && buf[3] > PICA_PROTO_VER_LOW))
		{
			PICA_warn("Client has newer protocol version than node. Please check if update for pica-node is available");
		}
		else
		{
			PICA_info("Client has older protocol version. Disconnecting ...");
		}
	}

	return 1;
}

void cclink_sendcleconnected(struct cclink *link)
{
	struct PICA_proto_msg *mp;

	if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN)
	{
		if((mp = client_wbuf_push(link->p1, PICA_PROTO_FOUND, PICA_PROTO_FOUND_SIZE)))
		{
			memcpy(mp->tail, link->callee_id, PICA_ID_SIZE);
		}
		else
		{
			cclink_list_delete(link);
			return;
		}
	}

	if (link->state == PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN)
	{
		if((mp = nodelink_wbuf_push(link->caller_node, PICA_PROTO_N2NALLOW, PICA_PROTO_N2NALLOW_SIZE)))
		{
			memcpy(mp->tail, link->caller_id, PICA_ID_SIZE);
			memcpy(mp->tail + PICA_ID_SIZE, link->callee_id, PICA_ID_SIZE);
		}
		else
		{
			cclink_list_delete(link);
			return;
		}
	}

}

unsigned int procmsg_CONNID(unsigned char* buf, unsigned int size, void* ptr)
{
	struct PICA_proto_msg *mp;
	struct newconn *nc = (struct newconn*)ptr;
	unsigned char *caller_id, *callee_id;
	struct cclink *link;

	PICA_debug1("received CONNID");

	nc->type = NEWCONN_C2C;

	caller_id = buf + 2;
	callee_id = buf + 2 + PICA_ID_SIZE;

	{
		char caller_id_buf[2 * PICA_ID_SIZE], callee_id_buf[2 * PICA_ID_SIZE];
		PICA_debug1("CONNID: caller_id=%s callee_id=%s", PICA_id_to_base64(caller_id, caller_id_buf), PICA_id_to_base64(callee_id, callee_id_buf));
	}

	link = cclink_list_search(caller_id, callee_id);

	if (!link)
	{
		PICA_debug3("corresponding cclink not found");
		return 0;
	}

	if (link->state != PICA_CCLINK_LOCAL_WAITCONNCLE && link->state != PICA_CCLINK_LOCAL_WAITCONNCLR
	        && link->state != PICA_CCLINK_N2NCLR_WAITCONNCLR  && link->state != PICA_CCLINK_N2NCLE_WAITCONNCLE)
	{
		PICA_debug2("found cclink is not in expected state");
		return 0;
	}

	{
		struct client *c;
//проверка IP адреса.
//IP адрес должен совпадать с адресом  клиента в управляющем соединении

		if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLE || link->state == PICA_CCLINK_N2NCLE_WAITCONNCLE)
			c = link->p2;
		else if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLR || link->state == PICA_CCLINK_N2NCLR_WAITCONNCLR)
			c = link->p1;

		if (nc->addr.sin_addr.s_addr != c->addr.sin_addr.s_addr)
		{
			PICA_warn("CONNID: IP check failed. IP address of sender does not match IP in c2n connection");
			return 0;
		}
	}

	PICA_debug3("CONNID processed successfully");
	cclink_setwaitanontlsshutdown(link, nc);

	return 1;
}

unsigned int procmsg_NODECONNREQ(unsigned char* buf, unsigned int size, void* ptr)
{
	struct newconn *nc = (struct newconn*)ptr;
	struct PICA_proto_msg *mp;

	PICA_debug1("received NODECONNREQ");

	nc->type = NEWCONN_N2N;

	if (buf[2] == PICA_PROTO_VER_HIGH &&
	        buf[3] == PICA_PROTO_VER_LOW)
	{
		////-прием входящего подключения
		//--------------------------
		nc->state = PICA_NEWCONN_SENDINGOK;
		/*struct nodelink *nl;
		nl = nodelink_list_addnew(nc);
		newconn_free(nc);
		if (!nl)
			return 0;

		if ((mp = nodelink_wbuf_push(nl, PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK_SIZE)))
		{
			mp->tail[0] = 'O';
			mp->tail[1] = 'K';
		}*/
	}
	else
	{
		nc->state = PICA_NEWCONN_SENDINGREJECT;
		/*
		//// PICA_PROTO_VERDIFFER
		send(nc->sck, msg_VERDIFFER, msg_VERDIFFER_len, 0);

		if (buf[2] > PICA_PROTO_VER_HIGH || (buf[2] == PICA_PROTO_VER_HIGH && buf[3] > PICA_PROTO_VER_LOW))
		{
			PICA_warn("Node %s has newer protocol version [%u, %u]. Please check if update for pica-node is available",
			          inet_ntoa(nc->addr.sin_addr), buf[2], buf[3]);
		}
		else
		{
			PICA_info("Node %s has older protocol version. Disconnecting ...", inet_ntoa(nc->addr.sin_addr));
		}

		return 0;*/
	}

	return 1;
}


void mapfunc_sendmultilogin(struct nodelink *p, const unsigned char *msgbuf)
{
	struct PICA_proto_msg *mp;
	uint16_t payload_size = *(uint16_t*)(msgbuf + 2);

	mp = nodelink_wbuf_push(p, PICA_PROTO_MULTILOGIN, payload_size + 4);
	if (mp)
		memcpy(mp->tail, msgbuf + 2, payload_size + 2);
}

void mapfunc_sendsearchreq(struct nodelink *p, const unsigned char *id)
{
	struct PICA_proto_msg *mp;

	mp = nodelink_wbuf_push(p, PICA_PROTO_SEARCH, PICA_PROTO_SEARCH_SIZE);

	if (mp)
		memcpy(mp->tail, id, PICA_ID_SIZE);
}

int nodelink_send_searchreq(const unsigned char *id)
{
	if (!nodelink_list_count)
		return 0;

	LISTMAP(nodelink, nodelink, mapfunc_sendsearchreq, id);

	return 1;
}

unsigned int procmsg_CONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *i, *o; //i - указатель на вызывающего клиента, o - вызываемого(если найден)
	unsigned char *callee_id;
	struct PICA_proto_msg *mp;

	i = (struct client *)ptr;
	callee_id = buf + 2;

	char caller_id_buf[2 * PICA_ID_SIZE], callee_id_buf[2 * PICA_ID_SIZE];
	PICA_debug1("received CONNREQOUTG from %s\n  searching for %s", PICA_id_to_base64(i->id, caller_id_buf),
			PICA_id_to_base64(callee_id, callee_id_buf));

	if ((o = client_tree_search(callee_id)))
	{
		if (cclink_list_search(i->id, callee_id))
		{
			PICA_info("duplicate request for existing c2c connection is ignored");
			return 1;
		}

		if (memcmp(i->id, callee_id, PICA_ID_SIZE) == 0)
		{
			PICA_info("request to create c2c connection to self is ignored");
			return 1;
		}

		//отправить o _CONNREQINC
		if ((mp = client_wbuf_push(o, PICA_PROTO_CONNREQINC, PICA_PROTO_CONNREQINC_SIZE)))
		{
			memcpy(mp->tail, i->id, PICA_ID_SIZE);
			cclink_list_addlocal(i, o);

			o->tmst = time(0);
			o->disconnect_ticking = 1;//if client does not reply in given time period, client is disconnected
			return 1;
		}

		return 1;//сообщение отправить не удалось, но все равно возвращаем 1
	}

	if (cclink_list_findwaitsearch(callee_id)/*search request for this id is already in progress*/
	        || nodelink_send_searchreq(callee_id) /*make new search*/)
	{
		cclink_list_addn2nclr(i, callee_id);
		return 1;
	}

//отправить i _NOTFOUND
	if ((mp = client_wbuf_push(i, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE)))
	{
		memcpy(mp->tail, callee_id, PICA_ID_SIZE);
	}

	return 1;
}

unsigned int procmsg_CONNALLOW(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *callee;
	struct cclink *link;
	unsigned char *caller_id;

	PICA_debug1("received CONALLOW");

	callee = (struct client *)ptr;
	caller_id = buf + 2;

	callee->disconnect_ticking = 0;
	callee->tmst = time(0);

	link = cclink_list_search(caller_id, callee->id);

	if (!link)
		return 0;

	if (link->state != PICA_CCLINK_LOCAL_WAITREP &&
	        link->state != PICA_CCLINK_N2NCLE_WAITREP)
		return 0;

	cclink_setwaitconn(link);

	return 1;
}

unsigned int procmsg_CONNDENY(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *caller, *callee;
	struct cclink *link;
	unsigned char *caller_id;
	struct PICA_proto_msg *mp;

	PICA_debug1("received CONNDENY");

	callee = (struct client *)ptr;
	caller_id = buf + 2;

	callee->disconnect_ticking = 0;
	callee->tmst = time(0);

	link = cclink_list_search(caller_id, callee->id);

	if (!link)
		return 0;

	caller = link -> p1;//check for NULL if not LOCAL

	if (link->state != PICA_CCLINK_LOCAL_WAITREP &&
	        link->state != PICA_CCLINK_N2NCLE_WAITREP)
	{
		cclink_list_delete(link);
		return 0;
	}

	if (link->state == PICA_CCLINK_LOCAL_WAITREP)
	{
		//отправить caller _NOTFOUND
		if ((mp = client_wbuf_push(caller, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE)))
		{
			memcpy(mp->tail, callee->id, PICA_ID_SIZE);
		}
	}
	else if (link->state == PICA_CCLINK_N2NCLE_WAITREP)
	{
		//отправить caller N2NNOTFOUND
		if ((mp = nodelink_wbuf_push(link->caller_node, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE)))
		{
			memcpy(mp->tail, caller_id, PICA_ID_SIZE);
			memcpy(mp->tail + PICA_ID_SIZE, callee->id, PICA_ID_SIZE);
		}
	}

	cclink_list_delete(link);

	return 1;
}

unsigned int procmsg_PINGREQ_client(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *c = (struct client*)ptr;
	struct PICA_proto_msg *mp;

	PICA_debug3("PINGREQ");

	if ((mp = client_wbuf_push(c, PICA_PROTO_PINGREP, PICA_PROTO_PINGREP_SIZE)))
	{
		RAND_pseudo_bytes(mp->tail, 2);
	}
	else
		return 0;

	return 1;
}

unsigned int procmsg_PINGREP_client(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *c = (struct client*)ptr;

	PICA_debug3("PINGREP c2n");

	c->disconnect_ticking = 0;
	c->tmst = time(0);

	return 1;
}

unsigned int procmsg_MULTILOGIN_client(unsigned char* buf, unsigned int size, void* ptr)
{
	struct client *other, *me = (struct client*)ptr;

	PICA_debug3("MULTILOGIN c2n");

	//send to other instances connected to this node
	other = client_tree_search(me->id);

	while(other)
	{
		if (other != me)
		{
			struct PICA_proto_msg *mp;

			if ((mp = client_wbuf_push(other, PICA_PROTO_MULTILOGIN, size)))
			{
				memcpy(mp->tail, buf + 2, size - 2);
			}
		}

		other = other->next_multi;
	}
	//relay to other nodes
	LISTMAP(nodelink, nodelink, mapfunc_sendmultilogin, buf);

	return 1;
}

unsigned int procmsg_MULTILOGIN_node(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	struct client *c = client_list_head;
	uint64_t timestamp;
	void *sigdatas[4];
	size_t sigdatalengths[4];
	uint16_t payload_len = *(uint16_t*)(buf + 2);
	int ret;
	size_t siglen;
	unsigned char *sig;

	PICA_debug3("MULTILOGIN n2n");

	if (payload_len <= sizeof timestamp + PICA_PROTO_NODELIST_ITEM_IPV4_SIZE)
		return 0;

	timestamp = *(uint64_t*)(buf + 4);
	//TODO compare timestamps

	sigdatalengths[0] = PICA_ID_SIZE;

	//add timestamp to signature data
	sigdatas[1] = buf + 4;
	sigdatalengths[1] = sizeof(uint64_t);

	//add sender node address to signature data
	sigdatas[2] = buf + 4 + sizeof(uint64_t);

	switch (buf[4 + sizeof(uint64_t)])
	{
	case PICA_PROTO_NEWNODE_IPV4:
		sigdatalengths[2] = PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;
		break;

	case PICA_PROTO_NEWNODE_IPV6:
		sigdatalengths[2] = PICA_PROTO_NODELIST_ITEM_IPV6_SIZE;
		break;

	case PICA_PROTO_NEWNODE_DNS:
		sigdatalengths[2] = 4 + buf[5 + sizeof(uint64_t)];
	default:
		PICA_warn("Received MULTILOGIN message with invalid node address");
		return 1;
	}

	siglen = payload_len - sizeof(uint64_t) - sigdatalengths[2];
	sig = buf + 4 + sizeof(uint64_t) + sigdatalengths[2];

	sigdatas[3] = NULL;
	sigdatalengths[3] = 0;

	while(c)
	{
		//check signature against current client ID and data in MULTILOGIN packet
		if (c->state == PICA_CLSTATE_CONNECTED)
		{
			EVP_PKEY *pubkey;
			X509* x;

			sigdatas[0] = c->id;

			if ((x = SSL_get_peer_certificate(c->ssl_comm)) &&
				(pubkey = X509_get_pubkey(x)))
			{
				ret = PICA_signverify(pubkey, sigdatas, sigdatalengths, sig, siglen);

				EVP_PKEY_free(pubkey);
				X509_free(x);

				if (ret == 1)
				{
					PICA_debug3("found matching client connection for incoming n2n MULTILOGIN message");
					break;
				}
			}
		}

		c = c->next;
	}

	if (!c)
	{
		PICA_debug3("no matching client found for receved n2n MULTILOGIN");
		return 1;
	}

	while(c) //relay received MULTILOGIN to each client with same id
	{
		struct PICA_proto_msg *mp;

		mp = client_wbuf_push(c, PICA_PROTO_MULTILOGIN, size);

		if (mp)
		{
			memcpy(mp, buf, size);
		}

		c = c->next_multi;
	}

	return 1;
}

unsigned int procmsg_dummy_multi_c2n(unsigned char* buf, unsigned int size, void* ptr)
{
	PICA_debug3("message %X from %p is handled by dummy procmsg", buf[0], ptr);

	return 1;
}

unsigned int procmsg_PINGREQ_node(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	struct PICA_proto_msg *mp;

	PICA_debug3("PINGREQ");

	if ((mp = nodelink_wbuf_push(n, PICA_PROTO_PINGREP, PICA_PROTO_PINGREP_SIZE)))
	{
		RAND_pseudo_bytes(mp->tail, 2);
	}
	else
		return 0;

	return 1;
}

unsigned int procmsg_PINGREP_node(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;

	PICA_debug3("PINGREP n2n");

	n->disconnect_ticking = 0;
	n->tmst = time(0);

	return 1;
}

unsigned int procmsg_NODERESP(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	struct PICA_proto_msg *mp;

	PICA_debug1("received NODERESP");

	switch(buf[0])
	{
	case PICA_PROTO_INITRESPOK:
		if (buf[2] != 'O' || buf[3] != 'K')
			return 0;


		mp = nodelink_wbuf_push(n, PICA_PROTO_NEWNODE_IPV4, PICA_PROTO_NEWNODE_IPV4_SIZE);//IPv6
		if (mp)
		{
			*((in_addr_t*)(mp->tail)) = inet_addr(nodecfg.announced_addr);
			*((in_port_t*)(mp->tail + 4)) = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) :  htons(PICA_COMM_PORT)); //CONF;
		}
		else
			return 0;

		mp = nodelink_wbuf_push(n, PICA_PROTO_NODELISTREQ, PICA_PROTO_NODELISTREQ_SIZE);
		if (mp)
		{
			RAND_pseudo_bytes(mp->tail, 2);
		}
		else
			return 0;

		break;
	case PICA_PROTO_VERDIFFER:
		if (buf[2] > PICA_PROTO_VER_HIGH ||  (buf[3] > PICA_PROTO_VER_LOW && buf[2] == PICA_PROTO_VER_HIGH))
		{
			PICA_warn("make update of node software - peer node has newer protocol version");
		}
		else
		{
			//peer node has older software version
			PICA_info("peer node has older protocol version");
		}
		return 0;
		break;
	}

	return 1;
}

unsigned int procmsg_NEWNODE(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	struct PICA_nodeaddr na;

	PICA_debug1("received NEWNODE");

	switch (buf[0])
	{
	case PICA_PROTO_NEWNODE_IPV4:
		PICA_debug1("NEWNODE_IPV4: ip %.16s port %hu\n", inet_ntoa(*(struct in_addr*)(buf + 2)), ntohs(*(in_port_t*)(buf + 6)));
		{
			struct PICA_nodeaddr_ipv4 na_ipv4;

			na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4;
			na_ipv4.addr = *(in_addr_t*)(buf + 2);
			na_ipv4.port = *(in_port_t*)(buf + 6);

			if (nodelink_search_by_ipv4addr(na_ipv4.addr, na_ipv4.port)) //connection with this node is already established
			{
				PICA_debug1("disconnecting node %.16s:%hu - connection with this node is already established", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port));
				return 0;
			}

			if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(na_ipv4.addr))
			{
				PICA_debug1("disconnecting node %.16s:%hu - private IP ranges are rejected by node configuration",
				            inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port)
				           );
				return 0;
			}

			sprintf(na.addr, "%.16s", inet_ntoa(*(struct in_addr*)&na_ipv4.addr));
			na.port = ntohs(na_ipv4.port);

			//int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size)
			nodelink_attach_nodeaddr(n, na_ipv4.magick, &na_ipv4, sizeof(struct PICA_nodeaddr_ipv4));
		}
		break;
	case PICA_PROTO_NEWNODE_IPV6:
#warning "TO DO"
		break;
	case PICA_PROTO_NEWNODE_DNS:
#warning "TODO"
		break;
	}

	PICA_nodeaddr_save(nodecfg.nodes_db_file, &na);//CONF filename

	return 1;
}

unsigned int procmsg_NODELISTREQ(unsigned char* buf, unsigned int size, void* ptr)
{
	struct PICA_proto_msg *mp;
	struct nodelink *nlp;
	unsigned int nl_size = 0;
	struct nodelink *n = 0;
	struct client *c = 0;
	unsigned char *nl_buf = 0;

	PICA_debug1("received NODELISTREQ");

	switch(buf[0])
	{
	case PICA_PROTO_CLNODELISTREQ:
		c = (struct client *)ptr;
		break;
	case PICA_PROTO_NODELISTREQ:
		n = (struct nodelink *)ptr;
		break;
	}

	nl_buf = (unsigned char*)malloc(65536);

	if (!nl_buf)
		return 0;//ERR


	{
		//TEMP FIXME CONF
//собственный адрес узла
		nl_buf[nl_size] = PICA_PROTO_NEWNODE_IPV4;
		*((in_addr_t*)(nl_buf + nl_size + 1)) = inet_addr(my_addr);
		*((in_port_t*)(nl_buf + nl_size + 5)) = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) :  htons(PICA_COMM_PORT)); //CONF
		nl_size += PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;
	}

	nlp = nodelink_list_head;
	while(nlp && nl_size < 65532)//MAP
	{
		switch(nlp->addr_type)
		{
		case PICA_PROTO_NEWNODE_IPV4:
			if ( PICA_PROTO_NODELISTREQ == buf[0] &&
			        ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->addr == ((struct PICA_nodeaddr_ipv4*)n->node_addr)->addr    &&
			        ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->port == ((struct PICA_nodeaddr_ipv4*)n->node_addr)->port )
				break;//exclude requester's address

			nl_buf[nl_size] = PICA_PROTO_NEWNODE_IPV4;
			*((in_addr_t*)(nl_buf + nl_size + 1)) = ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->addr;
			*((in_port_t*)(nl_buf + nl_size + 5)) = ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->port;
			nl_size += PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;//CONF
			break;
		}
//#warning "set limit on list size, <65532 bytes!!!"
		nlp = nlp->next;
	}

	switch(buf[0])
	{
	case PICA_PROTO_CLNODELISTREQ:
		mp =  client_wbuf_push( c, PICA_PROTO_CLNODELIST, nl_size + 4);
		break;
	case PICA_PROTO_NODELISTREQ:
		mp = nodelink_wbuf_push( n, PICA_PROTO_NODELIST, nl_size + 4);
		break;
	}

	if (mp)
	{
		*((uint16_t*)mp->tail) = nl_size;
		memcpy(mp->tail + 2, nl_buf, nl_size);
	}

	free(nl_buf);
	return 1;
}

unsigned int procmsg_NODELIST(unsigned char* buf, unsigned int size, void* ptr)
{
	struct PICA_nodeaddr_ipv4 na_ipv4;

	struct nodelink *n = (struct nodelink *)ptr;
	unsigned int listsize = size - 2 - PICA_MSG_VARSIZE_INT16;
	unsigned int listleft = listsize;

	PICA_debug1("received NODELIST");

	while(listleft)
	{
#warning "strict error checking!!! -- listleft overflow, unknown list items, check listleft >= possible item size"
		if (listleft == listsize) //first entry, sender node's address
		{
			switch(buf[size - listleft])
			{
			case PICA_PROTO_NEWNODE_IPV4:
				na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4;
				na_ipv4.addr = *(in_addr_t*)(buf + size - listleft + 1);
				na_ipv4.port = *(in_port_t*)(buf + size - listleft + 5);
				nodelink_attach_nodeaddr(n, na_ipv4.magick, &na_ipv4, sizeof(struct PICA_nodeaddr_ipv4));

				listleft -= PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;
				break;
			}
			continue;
		}
		else
		{
			struct PICA_nodeaddr na;
			//char addr[256];
			struct newconn nc;
			struct nodelink *nlp;
			int skip_save = 0;

			memset(na.addr, 0, 256);
			memset(&nc, 0, sizeof(struct newconn));

			switch(buf[size - listleft])
			{
			case PICA_PROTO_NEWNODE_IPV4:
				na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4;
				na_ipv4.addr = *(in_addr_t*)(buf + size - listleft + 1);
				na_ipv4.port = *(in_port_t*)(buf + size - listleft + 5);

				sprintf(na.addr, "%.16s", inet_ntoa(*(struct in_addr*)&na_ipv4.addr));
				na.port = ntohs(na_ipv4.port);

				//nodewait_start_connection(&na);
// 			if (try_connect_to_node(na.addr,ntohs(na_ipv4.port),&nc))
// 				if (try_get_reply(&nc))
// 					{
// 					nlp=nodelink_list_addnew(&nc);
// 					send_newnode(nlp,my_addr);
// 					nodelink_attach_nodeaddr(nlp,na_ipv4.magick,&na_ipv4,sizeof(struct PICA_nodeaddr_ipv4));
// 					PICA_debug1("LINKED NODE:%s\n",inet_ntoa(*(struct in_addr*)&na_ipv4.addr));
// 					}

				PICA_debug3("NODELIST item %.16s:%hu", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port));

				if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(na_ipv4.addr))
				{
					skip_save = 1;
					PICA_debug1("Received NODELIST item %.16s:%hu is in private range, skipping",
					            inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port)
					           );
				}
				listleft -= PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;
				break;
			}
			if (!skip_save)
				PICA_nodeaddr_save(nodecfg.nodes_db_file, &na);//CONF filename
		}
	}

	return 1;
}

unsigned int procmsg_N2NFOUND(unsigned char* buf, unsigned int size, void* ptr)
{
	struct cclink *cc;
	struct nodelink *n = (struct nodelink *)ptr;
	unsigned char *callee_id = buf + 2;
	struct PICA_proto_msg *mp;

	PICA_debug1("received N2NFOUND");

	while((cc = cclink_list_findwaitsearch(callee_id)))
	{
		cclink_attach_remotecle_node(cc, n);
		if((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NCONNREQOUTG, PICA_PROTO_N2NCONNREQOUTG_SIZE)))
		{
			memcpy(mp->tail, cc->caller_id, PICA_ID_SIZE);
			memcpy(mp->tail + PICA_ID_SIZE, callee_id, PICA_ID_SIZE);

		}
	}
	return 1;
}

unsigned int procmsg_SEARCH(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	unsigned char *callee_id = buf + 2;
	struct PICA_proto_msg *mp;
	struct client *callee;

	PICA_debug1("received SEARCH");

	if ((callee = client_tree_search(callee_id)))
	{
		if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NFOUND, PICA_PROTO_N2NFOUND_SIZE)))
		{
			memcpy(mp->tail, callee_id, PICA_ID_SIZE);
		}
	}
	return 1;
}

unsigned int procmsg_N2NCONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	unsigned char *caller_id, *callee_id;
	struct PICA_proto_msg *mp;
	struct client *callee;

	PICA_debug1("received N2NCONNREQOUTG");

	caller_id = buf + 2;
	callee_id = buf + 2 + PICA_ID_SIZE;

	if ((callee = client_tree_search(callee_id)))
	{
		if ((mp = client_wbuf_push(callee, PICA_PROTO_CONNREQINC, PICA_PROTO_CONNREQINC_SIZE)))
		{
			memcpy(mp->tail, caller_id, PICA_ID_SIZE);
			cclink_list_addn2ncle(caller_id, n, callee);
		}
	}
	else
	{
		if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE)))
		{
			memcpy(mp->tail, caller_id, PICA_ID_SIZE);
			memcpy(mp->tail + PICA_ID_SIZE, callee_id, PICA_ID_SIZE);
		}
	}
	return 1;
}

unsigned int procmsg_N2NALLOW(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	unsigned char *caller_id, *callee_id;
	struct PICA_proto_msg *mp;
	struct cclink *cc;

	PICA_debug1("received N2NALLOW");

	caller_id = buf + 2;
	callee_id = buf + 2 + PICA_ID_SIZE;

	cc = cclink_list_search(caller_id, callee_id);

	if (cc)
	{
		if ((mp = client_wbuf_push(cc->p1, PICA_PROTO_FOUND, PICA_PROTO_FOUND_SIZE)))
		{
			memcpy(mp->tail, callee_id, PICA_ID_SIZE);
			cclink_setwaitconn(cc);
		}
	}
	return 1;
}

unsigned int procmsg_N2NNOTFOUND(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	unsigned char *caller_id, *callee_id;
	struct PICA_proto_msg *mp;
	struct cclink *cc;

	PICA_debug1("received N2NNOTFOUND");

	caller_id = buf + 2;
	callee_id = buf + 2 + PICA_ID_SIZE;

	cc = cclink_list_search(caller_id, callee_id);

	if (cc)
	{
		if ((mp = client_wbuf_push(cc->p1, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE)))
		{
			memcpy(mp->tail, callee_id, PICA_ID_SIZE);
		}
		cclink_list_delete(cc);
	}

	return 1;
}

unsigned int procmsg_N2NMSG(unsigned char* buf, unsigned int size, void* ptr)
{
	struct nodelink *n = (struct nodelink *)ptr;
	struct PICA_proto_msg *mp;
	unsigned char *sender_id, *receiver_id;
	struct cclink *cc;
	unsigned int len, datalen;

	PICA_debug1("received N2NMSG");

	len = *(unsigned short*)(buf + 2);
	sender_id = buf + 4;
	receiver_id = buf + 4 + PICA_ID_SIZE;

	if (len < 2 * PICA_ID_SIZE)
		return 0;

	datalen = len - 2 * PICA_ID_SIZE;

	cc = cclink_list_search(sender_id, receiver_id);

	if (cc && cc->state == PICA_CCLINK_N2NCLE_ACTIVE)
	{
		if (datalen + cc->bufpos_p1p2 > cc->buflen_p1p2)
		{
			cclink_list_delete(cc);//CLOSING CCLINK DUE TO BUFER OVERRUN --------- FIXME FIX FIX FIX
			return 1;
		}

		memcpy(cc->buf_p1p2 + cc->bufpos_p1p2, buf + 4 + 2 * PICA_ID_SIZE, datalen);
		cc->bufpos_p1p2 += datalen;
		return 1;
	}
///////////////
	cc = cclink_list_search(receiver_id, sender_id);

	if (cc && cc->state == PICA_CCLINK_N2NCLR_ACTIVE)
	{
		if (datalen + cc->bufpos_p2p1 > cc->buflen_p2p1)
		{
			cclink_list_delete(cc);//CLOSING CCLINK DUE TO BUFER OVERRUN --------- FIXME FIX FIX FIX
			return 1;
		}

		memcpy(cc->buf_p2p1 + cc->bufpos_p2p1, buf + 4 + 2 * PICA_ID_SIZE, datalen);
		cc->bufpos_p2p1 += datalen;
		return 1;
	}

	if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE)))
	{
		memcpy(mp->tail, sender_id, PICA_ID_SIZE);
		memcpy(mp->tail + PICA_ID_SIZE, receiver_id, PICA_ID_SIZE);
	}

	return 1;
}


struct cclink* cclink_list_findwaitsearch(const unsigned char *callee_id)
{
	struct cclink *c;
	c = cclink_list_head;
	while(c)
	{
		if (memcmp(c->callee_id, callee_id, PICA_ID_SIZE) == 0
		        &&  c->state == PICA_CCLINK_N2NCLR_WAITSEARCH)
			return c;
		c = c->next;
	}
	return 0;
}

struct cclink* cclink_list_search(const unsigned char *caller_id, const unsigned char *callee_id)
{
	struct cclink *c;
	c = cclink_list_head;

	while(c)
	{
		if (memcmp(c->caller_id, caller_id, PICA_ID_SIZE) == 0
		        &&   memcmp(c->callee_id, callee_id, PICA_ID_SIZE) == 0)
			return c;
		c = c->next;
	}
	return 0;
}

void cclink_list_delete(struct cclink *l)
{
	PICA_debug1("cclink_list_delete(%p)", (void*)l);
//удаление из глобального списка
	if (l->prev)
		l->prev->next = l->next;
	else
		cclink_list_head = l->next;

	if (l->next)
		l->next->prev = l->prev;
	else
		cclink_list_end = l->prev;

	if (l->p1)
	{
		if (l->ev_read_p1)
		{
			event_del(l->ev_read_p1);
			event_free(l->ev_read_p1);
		}

		if (l->ev_write_p1)
		{
			event_del(l->ev_write_p1);
			event_free(l->ev_write_p1);
		}

		if (l->sck_p1 != -1)
		{
			SHUTDOWN(l->sck_p1);
			CLOSE(l->sck_p1);
			PICA_debug2("closed c2c socket %i", l->sck_p1);
		}

	}//caller_id is stored in p1 structure
	else
		free(l->caller_id);//caller_id was allocated by malloc

	if (l->p2)
	{
		if (l->ev_read_p2)
		{
			event_del(l->ev_read_p2);
			event_free(l->ev_read_p2);
		}

		if (l->ev_write_p2)
		{
			event_del(l->ev_write_p2);
			event_free(l->ev_write_p2);
		}

		if (l->sck_p2 != -1)
		{
			SHUTDOWN(l->sck_p2);
			CLOSE(l->sck_p2);
			PICA_debug2("closed c2c socket %i", l->sck_p2);
		}
	}
	else
		free(l->callee_id);

	if (l->buf_p1p2)
		free(l->buf_p1p2);

	if (l->buf_p2p1)
		free(l->buf_p2p1);

	free(l);
	cclink_list_count--;
}

void cclink_list_delete_by_client(struct client *cl)
{
	struct cclink *kill_ptr = 0, *l = cclink_list_head;
	PICA_debug1("cclink_list_delete_by_client(%p)", (void*)cl);

	while(l)
	{
		if (l->p1 == cl || l->p2 == cl)
			kill_ptr = l;

		l = l->next;

		if (kill_ptr)
		{
			cclink_list_delete(kill_ptr);
			kill_ptr = 0;
		}
	}

}

void cclink_list_delete_by_nodelink(struct nodelink *node)
{
	struct cclink *kill_ptr = 0, *l = cclink_list_head;

	PICA_debug1("cclink_list_delete_by_nodelink(%p)", (void*)node);

	while(l)
	{
		if (l->caller_node == node || l->callee_node == node)
			kill_ptr = l;

		l = l->next;

		if (kill_ptr)
		{
			cclink_list_delete(kill_ptr);
			kill_ptr = 0;
		}
	}
}


//clr - caller
//cle - callee
// null value of clr or cle means that corresponding client is on remote node, is not local client
struct cclink* cclink_list_add(struct client *clr, struct client *cle)
{
	struct cclink *l;

	l = (struct cclink*)calloc(sizeof(struct cclink), 1);

	if (!l)
		return 0;

//l->state=PICA_LINKSTATE_WAITREP;

	l->p1 = clr;
	l->p2 = cle;

//добавление в глобальный список
	if (cclink_list_head)
	{
		cclink_list_end->next = l;
		l->prev = cclink_list_end;
		cclink_list_end = l;
	}
	else
	{
		cclink_list_head = l;
		cclink_list_end = l;
	}

//timestamp
	l->tmst = time(0);

	cclink_list_count++;

	l->sck_p1 = -1;
	l->sck_p2 = -1;

	return l;
}


void cclink_list_addlocal(struct client *clr, struct client *cle)
{
	struct cclink *l;

	l = cclink_list_add(clr, cle);

	if (l)
	{
		l->state = PICA_CCLINK_LOCAL_WAITREP;
		l->caller_id = l->p1->id;
		l->callee_id = l->p2->id;
	}

}

void cclink_list_addn2nclr(struct client *clr, const unsigned char *callee_id)
{
	struct cclink *l;

	l = cclink_list_add(clr, 0);

	if (l)
	{
		l->state = PICA_CCLINK_N2NCLR_WAITSEARCH;
		l->caller_id = l->p1->id;
		l->callee_id = (unsigned char*)malloc(PICA_ID_SIZE);

		if (l->callee_id == NULL)
			PICA_fatal("memory allocation failed");

		memcpy(l->callee_id, callee_id, PICA_ID_SIZE);
	}
}

void cclink_attach_remotecle_node(struct cclink *ccl, struct nodelink *callee_node)
{
	if (ccl && ccl->state == PICA_CCLINK_N2NCLR_WAITSEARCH)
	{
		ccl->state = PICA_CCLINK_N2NCLR_WAITREP;
		ccl->callee_node = callee_node;
	}
}

void cclink_list_addn2ncle(const unsigned char *caller_id, struct nodelink *caller_node, struct client *cle)
{
	struct cclink *l;

	l = cclink_list_add(0, cle);

	if (l)
	{
		l->state = PICA_CCLINK_N2NCLE_WAITREP;
		l->caller_id = (unsigned char*)malloc(PICA_ID_SIZE);

		if (l->caller_id == NULL)
			PICA_fatal("memory allocation failed");

		memcpy(l->caller_id, caller_id, PICA_ID_SIZE);
		l->callee_id = l->p2->id;
		l->caller_node = caller_node;
	}

}

void cclink_setwaitconn(struct cclink *ccl)
{
	switch (ccl->state)
	{
	case PICA_CCLINK_LOCAL_WAITREP:
		ccl->state = PICA_CCLINK_LOCAL_WAITCONNCLE;
		//
		break;
	case PICA_CCLINK_N2NCLR_WAITREP:
		ccl->state = PICA_CCLINK_N2NCLR_WAITCONNCLR;
		break;
	case PICA_CCLINK_N2NCLE_WAITREP:
		ccl->state = PICA_CCLINK_N2NCLE_WAITCONNCLE;
		break;
	default:
		;
	}
}

int cclink_allocbufs(struct cclink *ccl)
{
	ccl->buf_p1p2 = malloc(DEFAULT_BUF_SIZE);
	ccl->buf_p2p1 = malloc(DEFAULT_BUF_SIZE);
	if (!ccl->buf_p1p2 || !ccl->buf_p2p1)
	{
		if (ccl->buf_p1p2)
			free(ccl->buf_p1p2);
		if (ccl->buf_p2p1)
			free(ccl->buf_p2p1);
		return 0;
	}
	ccl->buflen_p1p2 = DEFAULT_BUF_SIZE;
	ccl->buflen_p2p1 = DEFAULT_BUF_SIZE;
	return 1;
}

void cclink_setwaitanontlsshutdown(struct cclink *link, struct newconn *nc)
{
	nc->iconn.cc = link;
	nc->state = PICA_NEWCONN_WAITANONTLSSHUTDOWN;

	switch(link->state)
	{
	case PICA_CCLINK_LOCAL_WAITCONNCLE:
		link->sck_p2 = nc->sck;
		link->state = PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN;
		break;

	case PICA_CCLINK_LOCAL_WAITCONNCLR:
		link->sck_p1 = nc->sck;

		if (!cclink_allocbufs(link))
		{
			cclink_list_delete(link);
			return;
		}
		link->state = PICA_CCLINK_LOCAL_WAITCONNCLRANONTLSSHUTDOWN;
		break;

	case PICA_CCLINK_N2NCLR_WAITCONNCLR:
		link->sck_p1 = nc->sck;

		if (!cclink_allocbufs(link))
		{
			cclink_list_delete(link);
			return;
		}
		link->state = PICA_CCLINK_N2NCLR_WAITCONNCLRANONTLSSHUTDOWN;
		break;

	case PICA_CCLINK_N2NCLE_WAITCONNCLE:
		link->sck_p2 = nc->sck;

		if (!cclink_allocbufs(link))
		{
			cclink_list_delete(link);
			return;
		}

		link->state = PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN;
		break;
	}
}

void cclink_activate(struct cclink *ccl)
{
	switch(ccl->state)
	{
	case PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN:
		ccl->state = PICA_CCLINK_LOCAL_WAITCONNCLR;
		break;
	case PICA_CCLINK_LOCAL_WAITCONNCLRANONTLSSHUTDOWN:
		ccl->state = PICA_CCLINK_LOCAL_ACTIVE;
		ccl->ev_read_p1 = event_new(pica_event_base, ccl->sck_p1, EV_READ | EV_PERSIST, process_c2c_read, ccl);
		ccl->ev_read_p2 = event_new(pica_event_base, ccl->sck_p2, EV_READ | EV_PERSIST, process_c2c_read, ccl);
		event_add(ccl->ev_read_p1, NULL);
		event_add(ccl->ev_read_p2, NULL);
		ccl->ev_write_p1 = event_new(pica_event_base, ccl->sck_p1, EV_WRITE, process_c2c_write, ccl);
		ccl->ev_write_p2 = event_new(pica_event_base, ccl->sck_p2, EV_WRITE, process_c2c_write, ccl);
		break;
	case PICA_CCLINK_N2NCLR_WAITCONNCLRANONTLSSHUTDOWN:
		ccl->state = PICA_CCLINK_N2NCLR_ACTIVE;
		ccl->ev_read_p1 = event_new(pica_event_base, ccl->sck_p1, EV_READ | EV_PERSIST, process_c2c_read, ccl);
		event_add(ccl->ev_read_p1, NULL);
		ccl->ev_write_p1 = event_new(pica_event_base, ccl->sck_p1, EV_WRITE, process_c2c_write, ccl);
		break;
	case PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN:
		ccl->state = PICA_CCLINK_N2NCLE_ACTIVE;
		ccl->ev_read_p2 = event_new(pica_event_base, ccl->sck_p2, EV_READ | EV_PERSIST, process_c2c_read, ccl);
		event_add(ccl->ev_read_p2, NULL);
		ccl->ev_write_p2 = event_new(pica_event_base, ccl->sck_p2, EV_WRITE, process_c2c_write, ccl);
		break;
	}
}

static void libevent_log_cb(int severity, const char *msg)
{
	switch(severity)
	{
	case EVENT_LOG_ERR:
		PICA_error("libevent error: %s", msg);
		break;

	case EVENT_LOG_WARN:
	case EVENT_LOG_MSG:
		PICA_info("libevent: %s", msg);
		break;

	case EVENT_LOG_DEBUG:
		PICA_debug1("libevent: %s", msg);
		break;
	}
}

int PICA_node_init()
{
	struct sockaddr_in sd;
	int ret;
	int flag;
	FILE *dh_file;
	DH *dh = NULL;
	struct event *listener_event;
	struct event *timeout_event;
	struct timeval tv;

	SSL_load_error_strings();
	SSL_library_init();


#ifdef WIN32
	WSADATA wsd;
	WSAStartup(MAKEWORD(2, 2), &wsd);
#endif

#ifndef WIN32
	signal(SIGPIPE, SIG_IGN);
#endif

	event_set_log_callback(libevent_log_cb);

	pica_event_base = event_base_new();

	if (!pica_event_base)
		PICA_fatal("failed to initialize libevent");

	PICA_debug1("libevent method: %s", event_base_get_method(pica_event_base));

//CONF -AF_INET6 ????
	listen_comm_sck = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

	if (listen_comm_sck == -1)
		PICA_fatal("unable to get socket - %s", strerror(errno));
#ifndef WIN32
	flag = 1; //enable SO_REUSEADDR
	if (0 != setsockopt(listen_comm_sck, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag)))
	{
		PICA_error("failed to set SO_REUSEADDR on socket: %s", strerror(errno));
	}
#endif
	memset(&sd, 0, sizeof(sd));

	sd.sin_family = AF_INET;
	sd.sin_addr.s_addr = INADDR_ANY; //CONF
	sd.sin_port = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) :  htons(PICA_COMM_PORT));//CONF

	ret = bind(listen_comm_sck, (struct sockaddr*)&sd, sizeof(sd));

	if (ret == -1)
		PICA_fatal("unable to bind socket - %s", strerror(errno));


	ret = listen(listen_comm_sck, 20);
	if (ret == -1)
		PICA_fatal("unable to listen socket - %s", strerror(errno));

	{
		int ret_comm;
		unsigned long arg = 1;
#ifdef WIN32
		ret_comm = ioctlsocket(listen_comm_sck, FIONBIO, &arg);
// ret_data=ioctlsocket(listen_data_sck,FIONBIO,&arg);
// ret_data=ioctlsocket(listen_node_sck,FIONBIO,&arg);
#else
		ret_comm = ioctl(listen_comm_sck, FIONBIO, (int*)&arg);
// ret_data=ioctl(listen_data_sck,FIONBIO,(int*)&arg);
// ret_data=ioctl(listen_node_sck,FIONBIO,(int*)&arg);
#endif
		if (ret_comm == -1)
			PICA_fatal("unable to set non-blocking mode on socket - %s", strerror(errno));
	}

//#ifdef NO_RAND_DEV
//PICA_rand_seed();
//#endif

	listener_event = event_new(pica_event_base, listen_comm_sck, EV_READ|EV_PERSIST, process_listen_read, NULL);

	if (!listener_event)
		PICA_fatal("failed to allocate listener event");

	tv.tv_sec = 1;
	tv.tv_usec = 0;

	event_add(listener_event, /*&tv*/ NULL);

	timeout_event = event_new(pica_event_base, -1, EV_TIMEOUT|EV_PERSIST, process_timeout_event, NULL);
	event_add(timeout_event, &tv);



	ctx = SSL_CTX_new(TLSv1_2_method());
	anon_ctx = SSL_CTX_new(TLSv1_2_method());

	if (!ctx || !anon_ctx)
		PICA_fatal("unable to create SSL_CTX object");

	ret = SSL_CTX_set_cipher_list(ctx, PICA_TLS_CIPHERLIST);

	if (!ret)
		PICA_fatal("failed to set cipher list " PICA_TLS_CIPHERLIST);

	SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER, verify_callback);

	dh_file = fopen(nodecfg.dh_param_file, "r");

	if (dh_file)
	{
		dh = PEM_read_DHparams(dh_file, NULL, NULL, NULL);
		fclose(dh_file);
	}
	else
		PICA_fatal("Unable to open DH parameters file %", nodecfg.dh_param_file);

	ret = SSL_CTX_set_tmp_dh(anon_ctx, dh);

	if (ret != 1)
		PICA_fatal("Failed to set DH parameters");

#if OPENSSL_VERSION_NUMBER < 0x10100000L
	ret = SSL_CTX_set_cipher_list(anon_ctx, PICA_TLS_ANONDHCIPHERLIST);
#else
	ret = SSL_CTX_set_cipher_list(anon_ctx, PICA_TLS_ANONDHCIPHERLIST "@SECLEVEL=0");
#endif

	if (!ret)
		PICA_fatal("failed to set cipher list " PICA_TLS_ANONDHCIPHERLIST);

	newconns_init();

	return 1;
}


void newconn_close(struct newconn* nc)
{
	if (nc->ev)
	{
		event_del(nc->ev);
		event_free(nc->ev);
	}
	nc->ev = NULL;

	CLOSE(nc->sck);
	PICA_debug3("newconn_close(): nc = %p closed socket %i", nc, nc->sck);

	nc->sck = -1;
	nc->pos = 0;

	if (nc->iconn.cl && nc->type == NEWCONN_C2N)
		client_list_delete(nc->iconn.cl);

	if (nc->iconn.cc && nc->type == NEWCONN_C2C)
		cclink_list_delete(nc->iconn.cc);

	nc->iconn.cl = NULL;

	if (nc->anonssl)
	{
		SSL_free(nc->anonssl);
		nc->anonssl = NULL;
	}

	nc->type = NEWCONN_UNKNOWN;
}

void newconns_init()
{
	int i;

	memset(newconns, 0, sizeof(struct newconn) * MAX_NEWCONNS);

	for (i = 0; i < MAX_NEWCONNS; i++)
	{
		newconns[i].sck = -1;
		newconns[i].type = NEWCONN_UNKNOWN;
	}
}

//функция ищет свободный элемент в массиве структур newconn и возвращает на него указатель
//В случае, если свободный элемент не найден, возвращается элемент с наиболее поздней отметкой времени,
//при этом соединение закрывается
struct newconn* newconn_add(struct newconn *ncs, int *pos)
{
	time_t min;
	int min_pos = 0;
	if (*pos < MAX_NEWCONNS && ncs[*pos].sck == -1)
	{
		goto ret;
	}
	else
	{
		*pos = 0;
		min = ncs->tmst;
		while(*pos < MAX_NEWCONNS && ncs[*pos].sck != -1)
		{
			if (ncs[*pos].tmst < min)
			{
				min = ncs[*pos].tmst;
				min_pos = *pos;
			}
			(*pos)++;
		}

		if ( (*pos) == MAX_NEWCONNS)
		{
			PICA_error("Too many new connections, closing newconn socket %i", ncs[min_pos].sck);
			newconn_close(&ncs[min_pos]);
			*pos = min_pos;
		}
	}

ret:
	ncs[*pos].sck = -1;
	ncs[*pos].pos = 0;
	ncs[*pos].tmst = time(0);
	return &ncs[(*pos)++];
}

void newconn_free(struct newconn* nc)
{
	nc->sck = -1;
	nc->iconn.cl = NULL;
	nc->anonssl = NULL;
	event_del(nc->ev);
	event_free(nc->ev);
	nc->ev = NULL;
//nc->pos=0; - pos value is being used in processmsgdatastream
}

int client_tree_add(struct client *ci)
{
	struct client *i;
	if (client_tree_root)
	{
		i = client_tree_root;

		do
		{
			int cmp;

			cmp = memcmp(ci->id, i->id, PICA_ID_SIZE);

			if (cmp == 0)
				return 0;

			if (cmp < 0) // ci->id < i->id
			{
				if (i->left)
					i = i->left;
				else
				{
					i->left = ci;
					ci->up = i;
					return 1;
				}
			}
			else
			{
				if (i->right)
					i = i->right;
				else
				{
					i->right = ci;
					ci->up = i;
					return 1;
				}
			}
		}
		while(1);
	}
	else
		client_tree_root = ci;

	return 1;
}

struct client* client_tree_search(const unsigned char *id)
{
	struct client* i_ptr;
	i_ptr = client_tree_root;

//PICA_debug1("client_tree_search: searching for %u...",id);

	while(i_ptr)
	{
		int cmp = memcmp(id, i_ptr->id, PICA_ID_SIZE);

		if (cmp == 0)
			return i_ptr;

		if (cmp < 0) //id < i_ptr->id
			i_ptr = i_ptr->left;
		else
			i_ptr = i_ptr->right;
	}

	PICA_debug1("client_tree_search: id %s  not found", PICA_id_to_base64(id, NULL));
	return 0;
}

void client_tree_print(struct client *c)
{
	char id_buf[2 * PICA_ID_SIZE];
	if (!c)
	{
		PICA_debug2("client_tree: NULL");
		return;
	}

	PICA_debug2("client_tree %p: %s state: %u\nUP: %p\nnext_multi: %p\nLEFT: %p, RIGHT: %p\n\n",
			(void*)c, PICA_id_to_base64(c->id, id_buf), c->state,
			(void*)c->up, (void*)c->next_multi,
			(void*)c->left, (void*)c->right);

	if (c->next_multi)
		client_tree_print(c->next_multi);

	if (c->left)
		client_tree_print(c->left);

	if (c->right)
		client_tree_print(c->right);
}


void client_attach_multi_secondary(struct client *primary, struct client *secondary)
{
	secondary->next_multi = primary->next_multi;
	primary->next_multi = secondary;
}

void client_tree_remove(struct client* ci)
{
	struct client** p_link;//указывает на left или right в родительском узле

	if (!ci->up)
	{
		p_link = &client_tree_root;
	}
	else
	{
		if (ci == ci->up->left)
			p_link = &(ci->up->left);
		else
			p_link = &(ci->up->right);
	}


	if (!ci->left && !ci->right)
	{
		*p_link = 0;
		return;
	}

	if (ci->left && !ci->right)
	{
		ci->left->up = ci->up;
		*p_link = ci->left;
		return;
	}

	if (!ci->left && ci->right)
	{
		ci->right->up = ci->up;
		*p_link = ci->right;
		return;
	}


	{
		struct client *lm;
		lm = ci->right;
		while(lm->left) lm = lm->left; //поиск самого левого узла правого поддерева

		if (lm != ci->right)
		{
			lm->up->left = lm->right;
			lm->right = ci->right;
		}
		lm->left = ci->left;
		lm->up = ci->up;
		*p_link = lm;
	}
}

struct client *client_list_addnew(struct newconn *nc)
{
	struct client* ci;

	ci = (struct client *)calloc(1, sizeof(struct client));

	if (!ci)	return 0;

	ci->ssl_comm = SSL_new (ctx);

	if (!ci->ssl_comm)
	{
		free(ci);
		return 0;
	}

	if (client_list_end)
	{
		client_list_end->next = ci;
		ci->prev = client_list_end;
		client_list_end = ci;
	}
	else
	{
		client_list_head = ci;
		client_list_end = ci;
	}

	ci->sck_comm = nc->sck;
	ci->addr = nc->addr;


	SSL_set_fd(ci->ssl_comm, ci->sck_comm);

	ci->r_buf = calloc(1, DEFAULT_BUF_SIZE);
	if (!ci->r_buf)
	{
		free(ci);
		return 0;
	}
	ci->buflen_r = DEFAULT_BUF_SIZE;

	ci->tmst = time(0);

	ci->ev_read = event_new(pica_event_base, ci->sck_comm, EV_READ | EV_PERSIST, process_c2n_read, ci);
	ci->ev_write = event_new(pica_event_base, ci->sck_comm, EV_WRITE, process_c2n_write, ci);

	return ci;
}

void client_tree_replace_multi(struct client *primary)
{
	struct client** p_link;
	struct client *secondary;

	PICA_debug2("client_tree_replace_multi(%p)", (void*)primary);

	if (!primary->up)
	{
		p_link = &client_tree_root;
	}
	else
	{
		if (primary == primary->up->left)
			p_link = &(primary->up->left);
		else
			p_link = &(primary->up->right);
	}

	*p_link = primary->next_multi;

	secondary = primary->next_multi;

	secondary->state = primary->state;
	secondary->up = primary->up;
	secondary->left = primary->left;
	secondary->right = primary->right;

	if (primary->left)
		primary->left->up = secondary;
	if (primary->right)
		primary->right->up = secondary;
}

void client_list_delete(struct client* ci)
{
	struct client *s;

	PICA_debug1("client_list_delete(%p)", (void*)ci);
	SSL_free(ci->ssl_comm);
//удаление из списка
	if (ci->prev)
		ci->prev->next = ci->next;
	else
		client_list_head = ci->next;

	if (ci->next)
		ci->next->prev = ci->prev;
	else
		client_list_end = ci->prev;

//удаление из дерева
	s = client_tree_search(ci->id);

	if (s == ci)
	{
		if (ci->next_multi)
			client_tree_replace_multi(ci);
		else
			client_tree_remove(ci);
		client_tree_print(client_tree_root);
	}
	else if (s)
	{
		while(s)
		{
			if (s->next_multi == ci)
			{
				s->next_multi = ci->next_multi;
				break;
			}

			s = s->next_multi;
		}
	}

	cclink_list_delete_by_client(ci);

	if (ci->r_buf)
		free(ci->r_buf);

	if (ci->w_buf)
		free(ci->w_buf);

	if (ci->ev_read)
	{
		event_del(ci->ev_read);
		event_free(ci->ev_read);
	}
	if (ci->ev_write)
	{
		event_del(ci->ev_write);
		event_free(ci->ev_write);
	}

	CLOSE(ci->sck_comm);
	PICA_debug2("closed c2n socket %i", ci->sck_comm);

	free(ci);
}

struct PICA_proto_msg* client_wbuf_push
(struct client *c, unsigned int msgid, unsigned int size)
{
	if (!c->w_buf)
	{
		c->w_buf = calloc(1, DEFAULT_BUF_SIZE + (size / DEFAULT_BUF_SIZE) * DEFAULT_BUF_SIZE );
		if (!c->w_buf)
		{
			PICA_error("calloc() failed. Out of memory");
			return 0;
		}
		c->buflen_w = DEFAULT_BUF_SIZE + (size / DEFAULT_BUF_SIZE) * DEFAULT_BUF_SIZE;
	}

	if ((c->buflen_w - c->w_pos) < size)
	{
		unsigned char *p;
		p = realloc(c->w_buf, size + c->w_pos);

		if (p)
		{
			c->w_buf = p;
			c->buflen_w = size + c->w_pos;
		}
		else
			return 0;
	}

	event_add(c->ev_write, NULL);

	struct PICA_proto_msg *mp = (struct PICA_proto_msg *)(c->w_buf + c->w_pos);
	mp->head[0] = mp->head[1] = msgid;
	c->w_pos += size;
	return mp;
}

int client_rbuf_grow(struct client *c)
{
	unsigned char *p;

	p = realloc(c->r_buf, c->buflen_r + DEFAULT_BUF_SIZE);

	if (!p)
	{
		PICA_error("realloc() failed");
		return 0;
	}
	c->r_buf = p;
	c->buflen_r += DEFAULT_BUF_SIZE;

	return 1;
}

struct nodelink *nodelink_list_addnew(struct newconn *nc)
{
	struct nodelink *nl;

	nl = (struct nodelink*)calloc(1, sizeof(struct nodelink));

	if (!nl)
		return 0;

	if (nodelink_list_end)
	{
		nodelink_list_end->next = nl;
		nl->prev = nodelink_list_end;
		nodelink_list_end = nl;
	}
	else
	{
		nodelink_list_head = nl;
		nodelink_list_end = nl;
	}

	nl->sck = nc->sck;
	nl->addr = nc->addr;
	nl->anonssl = nc->anonssl;
	nc->anonssl = NULL;

	nl->r_buf = calloc(1, DEFAULT_BUF_SIZE);
	if (!nl->r_buf)
	{
		free(nl);
		return 0;
	}
	nl->buflen_r = DEFAULT_BUF_SIZE;

	nl->tmst = time(0);

	nl->ev_read = event_new(pica_event_base, nl->sck, EV_READ | EV_PERSIST, process_n2n_read, nl);
	nl->ev_write = event_new(pica_event_base, nl->sck, EV_WRITE, process_n2n_write, nl);

	event_add(nl->ev_read, NULL);

	nodelink_list_count++;
	return nl;
}

void nodelink_list_delete(struct nodelink *n)
{
	PICA_debug1("nodelink_list_delete(%p)", (void*)n);

	if (n->r_buf)
		free(n->r_buf);

	if (n->w_buf)
		free(n->w_buf);

	if (n->node_addr)
		free(n->node_addr);

	SSL_free(n->anonssl);

	event_del(n->ev_read);
	event_free(n->ev_read);
	event_del(n->ev_write);
	event_free(n->ev_write);

	CLOSE(n->sck);
	PICA_debug2("closed n2n socket %i", n->sck);

	if (n->prev)
		n->prev->next = n->next;
	else
		nodelink_list_head = n->next;

	if (n->next)
		n->next->prev = n->prev;
	else
		nodelink_list_end = n->prev;

	cclink_list_delete_by_nodelink(n);

	nodelink_list_count--;
	free(n);
}

int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size)
{
	if (nl && nodeaddr)
	{
		if (nl->node_addr)
			free(nl->node_addr);

		nl->node_addr = malloc(nodeaddr_size);
		if (!nl->node_addr)
			return 0;//ERR_CHECK
		memcpy(nl->node_addr, nodeaddr, nodeaddr_size);

		nl->addr_type = addr_type;
		return 1;
	}
	return 0;
}

int nodelink_rbuf_grow(struct nodelink *nl)
{
	unsigned char *p;

	p = realloc(nl->r_buf, nl->buflen_r + DEFAULT_BUF_SIZE);

	if (!p)
	{
		PICA_error("realloc() failed");
		return 0;
	}
	nl->r_buf = p;
	nl->buflen_r += DEFAULT_BUF_SIZE;

	return 1;
}

struct PICA_proto_msg* nodelink_wbuf_push
(struct nodelink *nl, unsigned int msgid, unsigned int size)
{
	struct PICA_proto_msg *mp;

	if (!nl->w_buf)
	{
		nl->w_buf = calloc(1, DEFAULT_BUF_SIZE );
		if (!nl->w_buf)
		{
			PICA_error("calloc() failed. Out of memory");
			return 0;
		}
		nl->buflen_w = DEFAULT_BUF_SIZE;
	}

	if ((nl->buflen_w - nl->w_pos) < size)
	{
		unsigned char *p;
		p = realloc(nl->w_buf, size + nl->w_pos);

		if (p)
		{
			nl->w_buf = p;
			nl->buflen_w = size + nl->w_pos;
		}
		else
			return 0;
	}

	event_add(nl->ev_write, NULL);

	mp = (struct PICA_proto_msg *)(nl->w_buf + nl->w_pos);
	mp->head[0] = mp->head[1] = msgid;
	nl->w_pos += size;
	return mp;
}

struct nodelink *nodelink_search_by_ipv4addr(in_addr_t ip, in_port_t port)
{
	struct nodelink *nl;

	nl = nodelink_list_head;

	while(nl && !(
	            (nl->addr.sin_addr.s_addr == ip && nl->addr.sin_port == port) ||
	            (nl->node_addr && nl->addr_type == PICA_PROTO_NEWNODE_IPV4 &&
	             ((struct PICA_nodeaddr_ipv4*)nl->node_addr) ->addr == ip &&
	             ((struct PICA_nodeaddr_ipv4*)nl->node_addr) ->port == port
	            )
	        )
	     )
	{
		nl = nl->next;
	}

	return nl;
}


void process_timeouts_newconn()
{
	clock_t t;
	int i;

	t = time(0);

	for (i = 0; i < MAX_NEWCONNS; i++)
	{
		if (newconns[i].sck != -1)
		{
			if (newconns[i].tmst < t)
			{
				if (t - newconns[i].tmst > NEWCONN_TIMEOUT)
				{
					newconn_close(newconns + i);
				}
			}
			else if ( newconns[i].tmst - t > NEWCONN_TIMEOUT)
			{
				newconn_close(newconns + i);
			}
		}
	}
}

void process_timeouts_c2c()
{
	struct cclink *il;

	il = cclink_list_head;

	while(il)
	{
		switch(il->state)
		{
		case PICA_CCLINK_LOCAL_ACTIVE:
		case PICA_CCLINK_N2NCLR_ACTIVE:
		case PICA_CCLINK_N2NCLE_ACTIVE:

			//idle timeout (>=60sec), update timestamp at process_c2c

			break;
		default:
			if ((time(0) - il->tmst) > TMO_CCLINK_WAITACTIVE)
			{
				struct cclink *dl;
				dl = il;
				il = il->next;
				cclink_list_delete(dl);
				continue;
			}
		}
		il = il->next;
	}
}

void process_timeouts_c2n()
{
	struct client *i_ptr, *kill_ptr;
	int ret;

	i_ptr = client_list_head;
	kill_ptr = 0;

	clock_t cur_time = time(0);

	while(i_ptr)
	{
		if (i_ptr->disconnect_ticking)
		{
			if (cur_time >= i_ptr->tmst)
			{
				if ((cur_time - i_ptr->tmst) > NOREPLY_TIMEOUT)
					kill_ptr = i_ptr;
			}
			else
			{
				//2038 year has come, and this code is still working somewhere with 32bit Unix timestamps :)
				if ((i_ptr->tmst - cur_time ) > NOREPLY_TIMEOUT)
					kill_ptr = i_ptr;
			}
		}
		else
		{
			if (cur_time >= i_ptr->tmst)
			{
				if ((cur_time - i_ptr->tmst) > KEEPALIVE_TIMEOUT)
				{
					struct PICA_proto_msg *mp;

					if ((mp = client_wbuf_push(i_ptr, PICA_PROTO_PINGREQ, PICA_PROTO_PINGREQ_SIZE)))
					{
						PICA_debug3("sending PINGREQ to %p", i_ptr);
						RAND_pseudo_bytes(mp->tail, 2);

						i_ptr->tmst = time(0);
						i_ptr->disconnect_ticking = 1;
					}
					else
						kill_ptr = i_ptr;
				}
			}
		}

		i_ptr = i_ptr->next;

		if (kill_ptr)
		{
			PICA_info("Disconnecting user %s due to noreply timeout", PICA_id_to_base64(kill_ptr -> id, NULL));
			client_list_delete(kill_ptr);
			kill_ptr = 0;
		}
	}
}

void process_timeouts_n2n()
{
	struct nodelink *i_ptr, *kill_ptr;
	int ret;

	i_ptr = nodelink_list_head;
	kill_ptr = 0;

	clock_t cur_time = time(0);

	while(i_ptr)
	{
		if (i_ptr->disconnect_ticking)
		{
			if (cur_time >= i_ptr->tmst)
			{
				if ((cur_time - i_ptr->tmst) > NOREPLY_TIMEOUT)
					kill_ptr = i_ptr;
			}
			else
			{
				//2038 year has come, and this code is still working somewhere with 32bit Unix timestamps :)
				if ((i_ptr->tmst - cur_time ) > NOREPLY_TIMEOUT)
					kill_ptr = i_ptr;
			}
		}
		else
		{
			if (cur_time >= i_ptr->tmst)
			{
				if ((cur_time - i_ptr->tmst) > KEEPALIVE_TIMEOUT)
				{
					struct PICA_proto_msg *mp;

					if ((mp = nodelink_wbuf_push(i_ptr, PICA_PROTO_PINGREQ, PICA_PROTO_PINGREQ_SIZE)))
					{
						PICA_debug3("sending PINGREQ to node %p", i_ptr);
						RAND_pseudo_bytes(mp->tail, 2);

						i_ptr->tmst = time(0);
						i_ptr->disconnect_ticking = 1;
					}
					else
						kill_ptr = i_ptr;
				}
			}
		}

		i_ptr = i_ptr->next;

		if (kill_ptr)
		{
			PICA_info("Disconnecting node %p due to noreply timeout", kill_ptr);
			nodelink_list_delete(kill_ptr);
			kill_ptr = 0;
		}
	}
}

int process_timeouts()
{
	process_timeouts_newconn();
	process_timeouts_c2n();
	process_timeouts_c2c();
	process_timeouts_n2n();

	if (time(0) - skynet_refresh_tmst > SKYNET_REFRESH_TIMEOUT && !nodewait_list)
	{
		int clean_treshold = 100 - nodelink_list_count - cclink_list_count;

		if (clean_treshold <= 0)
			clean_treshold = 3;

		PICA_debug3("refreshing nodelist database");

		if (nodelink_list_count > 0)
		{
			PICA_debug3("trying to clean nodes unreachable after more than %i attempts", clean_treshold);
			PICA_nodeaddr_cleanold(nodecfg.nodes_db_file, clean_treshold);
		}

		PICA_node_joinskynet(nodecfg.nodes_db_file, nodecfg.announced_addr);
	}
	return 1;
}


void process_timeout_event(evutil_socket_t listener, short event, void *arg)
{
	process_timeouts();
	process_nodewait();
}

void process_listen_read(evutil_socket_t listener, short event, void *arg)
{
	struct newconn *nc;
	SOCKET s;
	struct sockaddr_in addr;
	int addrsize = sizeof(struct sockaddr_in);

	if (event & EV_READ)
	{
		int i = 0;

		while (i++ < MAX_NEWCONNS / 4 && (s = accept(listen_comm_sck, (struct sockaddr*)&addr, &addrsize)) != -1)
		{
			struct event *nc_read, *nc_write;

			nc = newconn_add(newconns, &newconns_pos);

			nc->anonssl = SSL_new(anon_ctx);

			if (!nc->anonssl)
			{
				PICA_error("failed to allocate SSL struct");
				return;
			}

			if (!SSL_set_fd(nc->anonssl, s))
			{
				SSL_free(nc->anonssl);
				nc->anonssl = NULL;
				return;
			}


			nc->sck = s;
			nc->addr = addr;
			nc->state = PICA_NEWCONN_WAITANONTLSACCEPT;

			IOCTLSETNONBLOCKINGSOCKET(s, 1);

			PICA_debug1("accepted connection from %.16s, new socket %i", inet_ntoa(addr.sin_addr), s);//IPv6

			nc->ev = event_new(pica_event_base, s, EV_READ, process_newconn_read, nc);
			event_add(nc->ev, NULL);
		}

		if (s == -1)
		{
#ifndef WIN32
			if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
				return;

			PICA_error("unable to accept connection: %s", strerror(errno));
#else
			DWORD err = WSAGetLastError();

			if (err != WSAEWOULDBLOCK)
				PICA_error("unable to accept connection: %u", err);
#endif
		}
	}

//IPv6
}

void process_newconn_write(struct newconn *nc)
{
	int ret;

	switch(nc->state)
	{
	case PICA_NEWCONN_SENDINGOK:
		PICA_debug3("sending OK response to newconn on socket %i...\n", nc->sck);
		ret = SSL_write(nc->anonssl, msg_INITRESPOK, sizeof msg_INITRESPOK);

		if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret))
			break;

		if (ret <= 0)
		{
			newconn_close(nc);
			break;
		}

		if (ret == sizeof msg_INITRESPOK)
		{
			if (nc->type == NEWCONN_N2N)
			{
				if (!nodelink_list_addnew(nc))
				{
					newconn_close(nc);
				}

				newconn_free(nc);
			}
			else
				nc->state = PICA_NEWCONN_WAITANONTLSSHUTDOWN;
		}
		break;

	case PICA_NEWCONN_SENDINGREJECT:
		PICA_debug3("sending reject response to newconn on socket %i...\n", nc->sck);
		ret = SSL_write(nc->anonssl, msg_VERDIFFER, sizeof msg_VERDIFFER);

		if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret))
			break;

		if (ret <= 0)
		{
			newconn_close(nc);
			break;
		}

		SSL_shutdown(nc->anonssl);
		SHUTDOWN(nc->sck);
		newconn_close(nc);
	}
}

void process_newconn_read(evutil_socket_t s, short event, void *arg)
{
	int ret;
	struct newconn *nc = arg;

	if (event & EV_READ)
	{

		switch(nc->state)
		{
		case PICA_NEWCONN_WAITANONTLSACCEPT:
			ret = SSL_accept(nc->anonssl);

			if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret))
				break;

			if (ret <= 0)
			{
				newconn_close(nc);
				break;
			}
			else if (ret == 1)
			{
				nc->state = PICA_NEWCONN_WAITREQUEST;
			}

		case PICA_NEWCONN_WAITREQUEST:
			ret = SSL_read(nc->anonssl, nc->buf + nc->pos, NEWCONN_BUFSIZE - nc->pos);

			if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret))
				break;

			if (ret <= 0)
			{
				newconn_close(nc);
				break;
			}

			nc->pos += ret;

			if(!PICA_processdatastream(nc->buf, &(nc->pos), nc, _msginfo_newconn, MSGINFO_MSGSNUM(_msginfo_newconn) ))
			{
				newconn_close(nc);
			}

			break;

		case PICA_NEWCONN_WAITANONTLSSHUTDOWN:
			PICA_debug3("waiting for anon TLS shutdown on socket %i...\n", nc->sck);
			ret = SSL_shutdown(nc->anonssl);

			if (ret == 0)
				break;

			if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret))
				break;

			if (ret < 0)
			{
				newconn_close(nc);
				break;
			}

			//switch state of client or cclink on successful shutdown
			if (ret == 1)
			{
				SSL_free(nc->anonssl);

				if (nc->iconn.cl && nc->type == NEWCONN_C2N)
				{
					nc->iconn.cl->state = PICA_CLSTATE_TLSNEGOTIATION;
					event_add(nc->iconn.cl->ev_read, NULL);
					event_add(nc->iconn.cl->ev_write, NULL);
					newconn_free(nc);
					break;
				}

				if (nc->iconn.cc && nc->type == NEWCONN_C2C)
				{
					cclink_sendcleconnected(nc->iconn.cc);
					cclink_activate(nc->iconn.cc);

					newconn_free(nc);
					break;
				}
			}
			break;

		}

		if (nc->sck != -1)
			process_newconn_write(nc);

		if (nc->sck != -1)
			event_add(nc->ev, NULL);
	}
}



void process_c2n_read(evutil_socket_t s, short event, void *arg)
{
	struct client *i_ptr = (struct client*)arg;
	struct client *kill_ptr = 0;
	int ret;

	switch(i_ptr->state)
	{
	case PICA_CLSTATE_CONNECTED:
	case PICA_CLSTATE_MULTILOGIN_SECONDARY:
		ret = SSL_read(i_ptr->ssl_comm, i_ptr->r_buf + i_ptr->r_pos, i_ptr->buflen_r - i_ptr->r_pos);

		if (!ret)
			kill_ptr = i_ptr;

		if (ret < 0)
			switch(SSL_get_error(i_ptr->ssl_comm, ret))
			{
			case SSL_ERROR_WANT_WRITE:
			case SSL_ERROR_WANT_READ:
				break;
			default:
				kill_ptr = i_ptr;
			}

		if (ret > 0)
		{
			const struct PICA_msginfo *msgs = _msginfo_comm;
			unsigned int nmsgs = MSGINFO_MSGSNUM(_msginfo_comm);

			if (i_ptr->state == PICA_CLSTATE_MULTILOGIN_SECONDARY)
			{
				msgs = _msginfo_multi;
				nmsgs = MSGINFO_MSGSNUM(_msginfo_multi);
			}

			i_ptr->r_pos += ret;
			if(!PICA_processdatastream(i_ptr->r_buf, &(i_ptr->r_pos), i_ptr, msgs, nmsgs))
				kill_ptr = i_ptr;

			if (i_ptr->buflen_r == i_ptr->r_pos)
				if (!client_rbuf_grow(i_ptr))
					kill_ptr = i_ptr;

		}
		break;
	case PICA_CLSTATE_TLSNEGOTIATION:
		i_ptr->w_pos = 1; //????
		//-------------- <<<<<<<<<<<<<<<<<<<<<<<<----------------------------------------
		break;
	};

	if (kill_ptr)
	{
		client_list_delete(kill_ptr);
		kill_ptr = 0;
	}
	else if (i_ptr->w_pos)
	{
		process_c2n_write(s, event, arg);
	}
}

void process_c2c_read(evutil_socket_t s, short event, void *arg)
{
	int ret;
	struct cclink *cc = (struct cclink*)arg;
	struct cclink *kill_ptr = 0;


	if ((cc->state == PICA_CCLINK_LOCAL_ACTIVE || cc->state == PICA_CCLINK_N2NCLR_ACTIVE)
		&& cc->sck_p1 == s && !cc->jam_p1p2 )
	{
		ret = recv(cc->sck_p1, cc->buf_p1p2 + cc->bufpos_p1p2, cc->buflen_p1p2 - cc->bufpos_p1p2, 0);
		PICA_debug3("process_c2c_read: recv  p1p2: ret=%i", ret);
		if (ret > 0)
		{
			cc->bufpos_p1p2 += ret;

			if (cc->buflen_p1p2 == cc->bufpos_p1p2)
			{
				event_del(cc->ev_read_p1);
				cc->jam_p1p2 = 1;
			}

			if (cc->state == PICA_CCLINK_LOCAL_ACTIVE)
				event_add(cc->ev_write_p2, NULL);
			else
				process_c2c_write(-1, event, cc);
		}
		else
			kill_ptr = cc;
	}

	if ((cc->state == PICA_CCLINK_LOCAL_ACTIVE || cc->state == PICA_CCLINK_N2NCLE_ACTIVE)
		&& cc->sck_p2 == s  && !cc->jam_p2p1 )
	{
		ret = recv(cc->sck_p2, cc->buf_p2p1 + cc->bufpos_p2p1, cc->buflen_p2p1 - cc->bufpos_p2p1, 0);
		PICA_debug3("process_c2c_read: recv  p2p1: ret=%i\n", ret);
		if (ret > 0)
		{
			cc->bufpos_p2p1 += ret;

			if (cc->buflen_p2p1 == cc->bufpos_p2p1)
			{
				event_del(cc->ev_read_p2);
				cc->jam_p2p1 = 1;
			}

			if (cc->state == PICA_CCLINK_LOCAL_ACTIVE)
				event_add(cc->ev_write_p1, NULL);
			else
				process_c2c_write(-1, event, cc);
		}
		else
			kill_ptr = cc;
	}


	if (kill_ptr)
	{
		cclink_list_delete(kill_ptr);
		kill_ptr = 0;
	}

}

void process_n2n_read(evutil_socket_t s, short event, void *arg)
{
	int ret;
	struct nodelink *nl = (struct nodelink*)arg;
	struct nodelink *kill_ptr = 0;

	ret = SSL_read(nl->anonssl, nl->r_buf + nl->r_pos, nl->buflen_r - nl->r_pos);

	if (ret > 0)
	{
		nl->r_pos += ret;

		if(!PICA_processdatastream(nl->r_buf, &(nl->r_pos), nl, _msginfo_node, MSGINFO_MSGSNUM(_msginfo_node) ))
			kill_ptr = nl;

		if (nl->buflen_r == nl->r_pos)
			if (!nodelink_rbuf_grow(nl))
				kill_ptr = nl;
	}
	else
	{
		if (ret == 0 || process_async_ssl_readwrite(nl->anonssl, ret) == 0)
		{
			PICA_debug3("SSL_read() failed in process_n2n_read()");
			kill_ptr = nl;
		}
	}


	if (kill_ptr)
	{
		nodelink_list_delete(kill_ptr);
		kill_ptr = 0;
	}
}

void process_c2n_write(evutil_socket_t s, short event, void *arg)
{
	struct client *i_ptr = (struct client*)arg;
	struct client *kill_ptr = 0;
	int ret;

	switch(i_ptr->state)
	{
	case PICA_CLSTATE_CONNECTED:
	case PICA_CLSTATE_MULTILOGIN_SECONDARY:
		if (!i_ptr->w_pos)
			break;

		if (!i_ptr->btw_ssl)
			i_ptr->btw_ssl = i_ptr->w_pos;

		ret = SSL_write(i_ptr->ssl_comm, i_ptr->w_buf, i_ptr->btw_ssl);

		PICA_debug3("process_c2n_write:  ret of SSL_write()=%i", ret);

		if (ret == i_ptr->btw_ssl)
		{
			if (ret < i_ptr->w_pos)
				memmove(i_ptr->w_buf, i_ptr->w_buf + ret, i_ptr->w_pos - ret); //may be it's better to implement ring buffer here

			i_ptr->w_pos -= i_ptr->btw_ssl;
			i_ptr->btw_ssl = 0;
		}

		if (!ret)
			kill_ptr = i_ptr;

		if (ret < 0)
		{
			ret = SSL_get_error(i_ptr->ssl_comm, ret);
			if (ret != SSL_ERROR_WANT_WRITE && ret != SSL_ERROR_WANT_READ)
				kill_ptr = i_ptr;
		}
		break;

	case PICA_CLSTATE_WAITANONTLSSHUTDOWN:
		// do nothing here, state will be switched by newconn
		break;

	case PICA_CLSTATE_TLSNEGOTIATION:
		i_ptr->w_pos = 0;
		ret = SSL_connect(i_ptr->ssl_comm);

		PICA_debug3("process_c2n_write: SSL_connect()=%i", ret);

		if (!ret)
			kill_ptr = i_ptr;

		if (ret < 0)
		{
			int eret;
			eret = SSL_get_error(i_ptr->ssl_comm, ret);
			switch(eret)
			{
			case SSL_ERROR_WANT_WRITE:
				i_ptr->w_pos = 1;
				PICA_debug3("SSL_ERROR_WANT_WRITE");
				break;
			case SSL_ERROR_WANT_READ:
				PICA_debug3("SSL_ERROR_WANT_READ");
				break;
			default:
				PICA_debug3("SSL_get_error() = %i", eret);
				kill_ptr = i_ptr;
			}
		}
		if (ret == 1)
			//проверить сертификат пользователя
		{
			X509* client_cert;
			struct client *primary;

			PICA_info("SSL c2n connection established using %s cipher", SSL_get_cipher(i_ptr->ssl_comm));

			client_cert = SSL_get_peer_certificate (i_ptr->ssl_comm);
			if (!client_cert)
			{
				//ERR_CHECK //нет сертификата
				kill_ptr = i_ptr;
			}

			ret = PICA_id_from_X509(client_cert, i_ptr->id);

			if (ret == 0)
			{
				PICA_error("Unable to get client id from certificate");
				kill_ptr = i_ptr;
			}

			X509_free(client_cert);

			//reject all-zeroes ID
			{
				static const char zeroID[PICA_ID_SIZE] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
				if (memcmp(zeroID, i_ptr->id, PICA_ID_SIZE) == 0)
				{
					PICA_error("Rejecting all-zero Pica ID");
					kill_ptr = i_ptr;
				}
			}

			if (!kill_ptr && (primary = client_tree_search(i_ptr->id)))
			{
				//PICA_info("Disconnecting client %p because other user with same ID = %u is already connected", i_ptr, i_ptr->id);
				//memset(i_ptr->id, 0, PICA_ID_SIZE);//reset id to all zeros to prevent removing existing ID from previous connection from client_tree
				//ret = 0;//id already exists in tree
				i_ptr->state = PICA_CLSTATE_MULTILOGIN_SECONDARY;
				client_attach_multi_secondary(primary, i_ptr);
				client_tree_print(client_tree_root);
			}
			else
			{
				ret = client_tree_add(i_ptr);
				i_ptr->state = PICA_CLSTATE_CONNECTED;
				client_tree_print(client_tree_root);
			}
			if (!ret)
			{
				//ERR_CHECK  //не уникальный id
				kill_ptr = i_ptr;
			}
		}
		break;
	}


	if (kill_ptr)
	{
		client_list_delete(kill_ptr);
		kill_ptr = 0;
	}
	else if(i_ptr->w_pos)
	{
		event_add(i_ptr->ev_write, NULL);
	}
}

void process_c2c_write(evutil_socket_t s, short event, void *arg)
{
	int ret;
	struct cclink *cc = (struct cclink*)arg;
	struct cclink *kill_ptr = 0;


	if (cc->bufpos_p1p2)
	{
		if (cc->state == PICA_CCLINK_N2NCLR_ACTIVE) //receiver is on remote node
		{
			int sendlen;
			struct PICA_proto_msg *mp;

			sendlen = (cc->bufpos_p1p2 > PICA_PROTO_N2NMSG_MAXDATASIZE) ? PICA_PROTO_N2NMSG_MAXDATASIZE : cc->bufpos_p1p2;

			if(mp = nodelink_wbuf_push(cc->callee_node, PICA_PROTO_N2NMSG, sendlen + 4 + 2 * PICA_ID_SIZE))
			{
				*((unsigned short*)mp->tail) = 2 * PICA_ID_SIZE + sendlen;
				memcpy(mp->tail + 2, cc->caller_id, PICA_ID_SIZE);
				memcpy(mp->tail + 2 + PICA_ID_SIZE, cc->callee_id, PICA_ID_SIZE);
				memcpy(mp->tail + 2 + 2 * PICA_ID_SIZE, cc->buf_p1p2, sendlen);
				ret = sendlen;
			}
			else
				ret = 0;
		}
		else if (s == cc->sck_p2)
			ret = send(cc->sck_p2, cc->buf_p1p2, cc->bufpos_p1p2, MSG_NOSIGNAL);

		PICA_debug3("process_c2c_write: send  p1p2: ret=%i", ret);
		if (ret > 0)
		{
			if (ret < cc->bufpos_p1p2)
				memmove(cc->buf_p1p2, cc->buf_p1p2 + ret, cc->bufpos_p1p2 - ret); //may be it's better to implement ring buffer here
			cc->bufpos_p1p2 -= ret;

			if (cc->bufpos_p1p2 == 0 && cc->sck_p1 != -1)
				event_add(cc->ev_read_p1, NULL);

			cc->jam_p1p2 = 0;
		}

		if (ret == 0)
		{
			kill_ptr = cc;
		}
		if (ret == -1)
		{
#ifdef WIN32
			ret = WSAGetLastError();
			if (ret == WSAEWOULDBLOCK || ret == WSAENOBUFS)
				cc->jam_p1p2 = 1;
			else
			{
				kill_ptr = cc;
			}
#else
			ret = errno;
			if (ret == EAGAIN || ret == ENOBUFS)
				cc->jam_p1p2 = 1;
			else
			{
				kill_ptr = cc;
			}
#endif
		}

		if (!kill_ptr && cc->bufpos_p1p2 && s != -1)
			event_add(cc->ev_write_p2, NULL);

	}
	if (cc->bufpos_p2p1 /*&& (cc->state==PICA_CCLINK_LOCAL_ACTIVE || cc->state==PICA_CCLINK_N2NCLR_ACTIVE)*/)
	{
		if (cc->state == PICA_CCLINK_N2NCLE_ACTIVE) //receiver is on remote node
		{
			int sendlen;
			struct PICA_proto_msg *mp;

			sendlen = (cc->bufpos_p2p1 > PICA_PROTO_N2NMSG_MAXDATASIZE) ? PICA_PROTO_N2NMSG_MAXDATASIZE : cc->bufpos_p2p1;

			if(mp = nodelink_wbuf_push(cc->caller_node, PICA_PROTO_N2NMSG, sendlen + 4 + 2 * PICA_ID_SIZE))
			{
				*((unsigned short*)mp->tail) = 2 * PICA_ID_SIZE + sendlen;
				memcpy(mp->tail + 2, cc->callee_id, PICA_ID_SIZE);
				memcpy(mp->tail + 2 + PICA_ID_SIZE, cc->caller_id, PICA_ID_SIZE);
				memcpy(mp->tail + 2 + 2 * PICA_ID_SIZE, cc->buf_p2p1, sendlen);
				ret = sendlen;
			}
			else
				ret = 0;
		}
		else if (s == cc->sck_p1)
			ret = send(cc->sck_p1, cc->buf_p2p1, cc->bufpos_p2p1, MSG_NOSIGNAL);

		PICA_debug3("process_c2c_write: send  p2p1: ret=%i", ret);
		if (ret > 0)
		{
			if (ret < cc->bufpos_p2p1)
				memmove(cc->buf_p2p1, cc->buf_p2p1 + ret, cc->bufpos_p2p1 - ret); //may be it's better to implement ring buffer here
			cc->bufpos_p2p1 -= ret;

			if (cc->bufpos_p2p1 == 0 && cc->sck_p2 != -1)
				event_add(cc->ev_read_p2, NULL);

			cc->jam_p2p1 = 0;
		}

		if (ret == 0)
		{
			kill_ptr = cc;
		}
		if (ret == -1)
		{
#ifdef WIN32
			ret = WSAGetLastError();
			if (ret == WSAEWOULDBLOCK || ret == WSAENOBUFS)
				cc->jam_p2p1 = 1;
			else
			{
				kill_ptr = cc;
			}
#else
			ret = errno;
			if (ret == EAGAIN || ret == ENOBUFS)
				cc->jam_p2p1 = 1;
			else
			{
				kill_ptr = cc;
			}
#endif

		}

		if (!kill_ptr && cc->bufpos_p2p1 && s != -1)
			event_add(cc->ev_write_p1, NULL);

	}

	if (kill_ptr)
	{
		cclink_list_delete(kill_ptr);
		kill_ptr = 0;
	}
}

void process_n2n_write(evutil_socket_t s, short event, void *arg)
{
	int ret;
	struct nodelink *nl = (struct nodelink*)arg;

	struct nodelink *kill_ptr = 0;

	if (nl->w_pos > 0)
	{
		if (!nl->btw_ssl)
			nl->btw_ssl = nl->w_pos;

		ret = SSL_write(nl->anonssl, nl->w_buf, nl->btw_ssl);

		if (ret == nl->btw_ssl)
		{
			if (ret < nl->w_pos)
				memmove(nl->w_buf, nl->w_buf + ret, nl->w_pos - ret); //may be it's better to implement ring buffer here
			nl->w_pos -= nl->btw_ssl;
			nl->btw_ssl = 0;
		}

		if (ret == 0 || ret < 0 && process_async_ssl_readwrite(nl->anonssl, ret) == 0)
		{
			kill_ptr = nl;
		}

	}

	if (kill_ptr)
	{
		nodelink_list_delete(kill_ptr);
		kill_ptr = 0;
	}
	else if (nl->w_pos > 0)
	{
		event_add(nl->ev_write, NULL);
	}
}

void process_nodewait()
{
	struct nodewait *kill_ptr = 0, *nw = nodewait_list;

	while(nw)
	{
		//PICA_debug3("checking nodewait pointer %p", nw);
		switch(nw->state)
		{
		case PICA_NODEWAIT_RESOLVED:
		{
			int skip = 0;
			struct nodelink *nl;

			nl = nodelink_list_head;

			while(nl && !skip)//check existing connections
			{
				struct addrinfo *p_ai = nw->ai;

				while(p_ai) //IPv6
				{
					struct sockaddr_in in = *(struct sockaddr_in*)p_ai->ai_addr;

					if (nl->addr.sin_family == in.sin_family && nl->addr.sin_addr.s_addr == in.sin_addr.s_addr && nl->addr.sin_port == in.sin_port)
					{
						skip = 1;
						break;
					}

					p_ai = p_ai->ai_next;
				}
				nl = nl->next;
			}

			if (!skip)
				nodewait_start_connect(nw);
			else
			{
				PICA_debug3("skipping node %.255s %u , connection already exist", nw->addr.addr, nw->addr.port);
				kill_ptr = nw;
			}
		}
		break;

		case PICA_NODEWAIT_CONNECTED:
		{
			struct nodelink *nlp;

			PICA_debug1("connected to %.255s %u, sending request ", nw->addr.addr, nw->addr.port);
			PICA_debug2("connection socket is %i", nw->nc.sck);
			nlp = nodelink_list_addnew(&nw->nc);
			if (nlp)
			{
				struct PICA_proto_msg *mp;
				mp = nodelink_wbuf_push(nlp, PICA_PROTO_NODECONNREQ, PICA_PROTO_NODECONNREQ_SIZE);
				if (mp)
				{
					mp->tail[0] = PICA_PROTO_VER_HIGH;
					mp->tail[1] = PICA_PROTO_VER_LOW;
				}
			}
			PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 1);
			kill_ptr = nw;
		}
		break;

		case PICA_NODEWAIT_RESOLVING_FAILED:
		{
			PICA_debug1("resolving %.255s %u failed: %s", nw->addr.addr, nw->addr.port, gai_strerror(nw->ai_errorcode));
			PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 0);
			kill_ptr = nw;
		}
		break;

		case PICA_NODEWAIT_CONNECT_FAILED:
		{
			PICA_debug1("connection to %.255s %u failed", nw->addr.addr, nw->addr.port);
			PICA_debug2("socket %i is closed", nw->nc.sck);
			PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 0);
			kill_ptr = nw;
		}
		break;
		}

		nw = nw->next;

		if (kill_ptr)
		{
			nodewait_list_delete(kill_ptr);
			kill_ptr = 0;
		}
	}

}

int node_loop()
{
	int ret;

	do
	{
		ret = event_base_loop(pica_event_base, EVLOOP_ONCE);
	}
	while(stop_node_loop == 0);////<<<<<<<<<<
	return ret;
}

void print_usage()
{
	puts("Usage: pica-node [-f config-file]\n\
\n\
Options:\n\
	-f 	configuration file\n\
	-h	show this text\n\
	-v	verbose\n\
	-V	print version info\n");
}

void print_version()
{
	puts("Pica Pica Node v"PACKAGE_VERSION" \nCopyright  (c) 2012 - 2018 Anton Sviridenko\n");
	puts("protocol version "PICA_PROTO_VER_STRING"\n");
	puts("License GPLv3: GNU GPL version 3 <http://gnu.org/licenses/gpl.html>\n\
This is free software: you are free to change and redistribute it.\n\
There is NO WARRANTY, to the extent permitted by law.\n");

////----include this only if OpenSSL is statically linked
	puts("This product includes software developed by the OpenSSL Project\n\
for use in the OpenSSL Toolkit (http://www.openssl.org/)\n");
	puts("This product includes cryptographic software written by\n\
Eric Young (eay@cryptsoft.com)\n");
	puts(SSLeay_version(SSLEAY_VERSION));
////---


	puts("\nThis product includes iniParser\n\
Copyright (c) 2000-2012 by Nicolas Devillard.\n\
MIT License");

}

void process_cmdline(int argc, char** argv)
{
	int opt;
	int verbosity = 0;

	while((opt = getopt(argc, argv, "f:hvV")) != -1)
	{
		switch(opt)
		{
		case 'f':
			nodecfg.config_file = strdup(optarg);
			break;

		case 'h':
			print_usage();
			exit(0);
			break;

		case 'v':
			verbosity++;

			break;

		case 'V':
			print_version();
			exit(0);
			break;

		default:
			PICA_error("unknown option - %c", optopt);
			print_usage();
			exit(-1);
		}
	}

	if (verbosity > 3)
		verbosity = 3;

	PICA_set_loglevel(PICA_LOG_INFO + verbosity);
}

void PICA_node_joinskynet(const char* addrlistfilename, const char *my_addr)
{
	struct PICA_nodeaddr *nap, *addrlist_h = 0;
	struct nodelink *nl;
	int ret;

	skynet_refresh_tmst = time(0);

	ret = PICA_nodeaddr_list_load(addrlistfilename, &addrlist_h);//MEM
	if (ret <= 0)
		return;

	nap = addrlist_h;

	while(nap)
	{
		int skip = 0;

		if (0 == strncmp(nap->addr, my_addr, 256) && nap->port == atoi(nodecfg.listen_port))
		{
			PICA_debug3("Skipping self address %.255s port %u", nap->addr, nap->port);
			skip = 1;
		}

		if (nodecfg.disable_reserved_addrs && INADDR_NONE != inet_addr(nap->addr) && PICA_is_reserved_addr_ipv4(inet_addr(nap->addr)))
		{
			PICA_debug3("Skipping IP from private range %.255s port %u", nap->addr, nap->port);
			skip = 1;
		}

		nl = nodelink_list_head;
		//check if connection to node with current address is already established
		while(nl && !skip)
		{
			if (nl->node_addr)
			{
				switch(nl->addr_type)
				{
				case PICA_PROTO_NEWNODE_IPV4:
					if (0 == strncmp(nap->addr, inet_ntoa(*(struct in_addr*) & (((struct PICA_nodeaddr_ipv4*)nl->node_addr)->addr)), 16)
					        &&
					        nap->port == ntohs(((struct PICA_nodeaddr_ipv4*)nl->node_addr)->port))
						skip = 1;
					break;
					//IPv6
				}
			}

			if (0 == strncmp(nap->addr, inet_ntoa(nl->addr.sin_addr), 16) && nap->port == ntohs(nl->addr.sin_port))
				skip = 1;


			nl = nl->next;
		}

		if (!skip)
			nodewait_start_resolve(nap);

		nap = nap->next;
	}

	PICA_nodeaddr_list_free(addrlist_h);
}

#ifdef WIN32
static int is_service = 0;
static SERVICE_STATUS_HANDLE service_status_handle;
static SERVICE_STATUS service_status;
static int service_argc;
static char **service_argv;

void ReportSvcStatus( DWORD dwCurrentState,
                      DWORD dwWin32ExitCode,
                      DWORD dwWaitHint)
{
	static DWORD dwCheckPoint = 1;

	// Fill in the SERVICE_STATUS structure.

	service_status.dwCurrentState = dwCurrentState;
	service_status.dwWin32ExitCode = dwWin32ExitCode;
	service_status.dwWaitHint = dwWaitHint;

	if (dwCurrentState == SERVICE_START_PENDING)
		service_status.dwControlsAccepted = 0;
	else
		service_status.dwControlsAccepted = SERVICE_ACCEPT_STOP;

	if ( (dwCurrentState == SERVICE_RUNNING) ||
	   (dwCurrentState == SERVICE_STOPPED) )
		service_status.dwCheckPoint = 0;
	else
		service_status.dwCheckPoint = dwCheckPoint++;

	// Report the status of the service to the SCM.
	SetServiceStatus( service_status_handle, &service_status);
}
#endif

int pica_node_main(int argc, char** argv)
{
	uint32_t pid;

	process_cmdline(argc, argv);

	nodecfg.config_file = (nodecfg.config_file ? nodecfg.config_file : PICA_NODECONFIG_DEF_CONFIG_FILE);

#ifdef WIN32
	pid = GetCurrentProcessId();
#else
	pid = getpid();
#endif
	PICA_info("pica-node v%s started. PID: %u, protocol version: %s", PACKAGE_VERSION, pid, PICA_PROTO_VER_STRING);

	if (!PICA_nodeconfig_load(nodecfg.config_file))
	{
		PICA_fatal("Unable to load configuration. Please check if config file %s exists and has correct permissions", nodecfg.config_file);
	}

	PICA_debug1("nodecfg.config_file = %s", nodecfg.config_file);
	PICA_debug1("nodecfg.announced_addr = %s", nodecfg.announced_addr);
	PICA_debug1("nodecfg.listen_port = %s", nodecfg.listen_port);
	PICA_debug1("nodecfg.nodes_db_file = %s", nodecfg.nodes_db_file);
	PICA_debug1("nodecfg.disable_reserved_addrs = %s", nodecfg.disable_reserved_addrs ? "yes" : "no");

	if (!PICA_node_init())
		return -1;

#ifdef WIN32
	if (is_service)
	{
		ReportSvcStatus(SERVICE_RUNNING, NO_ERROR, 0);
		PICA_info("pica-node started as a Windows service");
	}
	else
	{
		PICA_info("pica-node started as a normal Windows process.");
	}
#endif

	if (strcmp("autoconfigure", nodecfg.announced_addr) == 0)
	{
		in_addr_t guess;
		struct in_addr in;

		guess = PICA_guess_listening_addr_ipv4();

		in.s_addr = guess;

		free(nodecfg.announced_addr);
		nodecfg.announced_addr = strdup(inet_ntoa(in));

		PICA_debug1("guessed interface address to be announced: %s", nodecfg.announced_addr);

#ifdef HAVE_LIBMINIUPNPC
		if (PICA_is_reserved_addr_ipv4(guess) && nodecfg.disable_reserved_addrs)
		{
			char public_ip[64];
			int ret;

			PICA_info("trying to get global IP with UPnP...");

			ret = PICA_upnp_autoconfigure_ipv4(atoi(nodecfg.listen_port), atoi(nodecfg.listen_port), public_ip);
			if (ret)
			{
				free(nodecfg.announced_addr);
				nodecfg.announced_addr = strdup(public_ip);
			}
		}
#endif
		PICA_info("autoconfigured announced address %s port %s", nodecfg.announced_addr, nodecfg.listen_port);
	}

	if (INADDR_NONE == inet_addr(nodecfg.announced_addr) || INADDR_ANY == inet_addr(nodecfg.announced_addr))
	{
		PICA_fatal("announced_addr  (%.16s) is invalid or not configured. Please set correct public IP address of your pica-node instance in config file.",
		           nodecfg.announced_addr
		          );
	}

	if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(inet_addr(nodecfg.announced_addr)))
	{
		PICA_fatal("announced_addr  (%.16s) is in private range and is unacceptable for global Internet. Private ranges are disabled by node configuration.",
		           nodecfg.announced_addr
		          );
	}

	PICA_node_joinskynet(nodecfg.nodes_db_file, nodecfg.announced_addr);//CONF-CONF имя файла с адресами узлов, свой адрес

	my_addr = nodecfg.announced_addr;//TEMP FIXME

	return node_loop();
}

#ifdef WIN32
void WINAPI SvcHandler(DWORD ctl)
{
	switch(ctl)
	{
	case SERVICE_CONTROL_STOP:
		ReportSvcStatus(SERVICE_STOP_PENDING, NO_ERROR, 0);

		stop_node_loop = 1;

		ReportSvcStatus(service_status.dwCurrentState, NO_ERROR, 0);

		return;

	case SERVICE_CONTROL_INTERROGATE:
		break;

	default:
		break;
	}
}

void WINAPI winservice_main(DWORD argc, LPTSTR *argv)
{
	char filepath[MAX_PATH];
	int ret;
	HANDLE logfile;

	service_status_handle = RegisterServiceCtrlHandler("pica-node service", SvcHandler);

	if (!service_status_handle)
		return;

	service_status.dwServiceType = SERVICE_WIN32_OWN_PROCESS;
	service_status.dwServiceSpecificExitCode = 0;

	if (S_OK != SHGetFolderPath(NULL, CSIDL_COMMON_APPDATA, NULL, SHGFP_TYPE_CURRENT, filepath))
		return;

	strncat_s(filepath, MAX_PATH, "\\pica-node.log", MAX_PATH);

	freopen(filepath, "w", stdout);
	freopen(filepath, "w", stderr);

	//set current directory to executable location directory
	GetModuleFileName(GetModuleHandle(NULL), filepath, MAX_PATH);
	//strip exe filename part
	*(strrchr(filepath, '\\') + 1) = 0;
	SetCurrentDirectory(filepath);

	ReportSvcStatus(SERVICE_START_PENDING, NO_ERROR, 3000);

	is_service = 1;

	ret = pica_node_main(service_argc, service_argv);
	ReportSvcStatus(SERVICE_STOPPED, ret, 0);
}
#endif

int main(int argc, char** argv)
{
#ifndef WIN32
	return pica_node_main(argc, argv);
#else
	SERVICE_TABLE_ENTRY DispatchTable[] =
	{
		{"pica-node service", winservice_main},
		{NULL, NULL}
	};

	service_argc = argc;
	service_argv = argv;
	//if process is started as a service,
	//function will return only after service is stopped
	if (StartServiceCtrlDispatcher(DispatchTable))
		return 0;
	else if (GetLastError() == ERROR_FAILED_SERVICE_CONTROLLER_CONNECT)// not a service
		return pica_node_main(argc, argv);

	//failed to start service
	return -1;
#endif
}