nodewait - asynchronous DNS resolve and connect to nodes using threads

git-svn-id: svn://localhost/project_picapica/trunk@215 fb96df88-216d-11de-830c-f3023f12d8ac

8 files changed, 317 insertions, 223 deletions

diff --git a/Makefile.am b/Makefile.am
index 0167680..7dd6ab0 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -13,10 +13,10 @@ if PICA_NODE
 bin_PROGRAMS += pica-node
 
 pica_node_SOURCES = PICA_node.c PICA_msgproc.c PICA_nodeaddrlist.c \
-		PICA_nodejoinskynet.c PICA_nodeconfig.c PICA_log.c \
+		PICA_nodewait.c PICA_nodeconfig.c PICA_log.c \
 		iniparser/src/dictionary.c iniparser/src/iniparser.c \
 		PICA_node.h PICA_msgproc.h PICA_nodeaddrlist.h \
-		PICA_nodejoinskynet.h PICA_nodeconfig.h PICA_log.h \
+		PICA_nodewait.h PICA_nodeconfig.h PICA_log.h \
 		iniparser/src/dictionary.h iniparser/src/iniparser.h \
 		PICA_common.h PICA_proto.h
 		
diff --git a/PICA_node.c b/PICA_node.c
index c943adf..ad4307d 100755
--- a/PICA_node.c
+++ b/PICA_node.c
@@ -12,9 +12,9 @@
 #include "PICA_msgproc.h"
 #include "PICA_nodeaddrlist.h"
 #include "PICA_common.h"
-#include "PICA_nodejoinskynet.h"
 #include "PICA_nodeconfig.h"
 #include "PICA_log.h"
+#include "PICA_nodewait.h"
 
 #ifdef NO_RAND_DEV
 #include "PICA_rand_seed.h"
@@ -53,9 +53,6 @@ struct nodelink *nodelink_list_head;
 struct nodelink *nodelink_list_end;
 unsigned int nodelink_list_count;
 
-struct PICA_nodeaddr *nodeaddr_list_head;
-struct PICA_nodeaddr *nodeaddr_list_end;
-unsigned int nodeaddr_list_count;
 
 
 char msg_VERDIFFER[4]={PICA_PROTO_VERDIFFER,PICA_PROTO_VERDIFFER,PICA_PROTO_VER_HIGH,PICA_PROTO_VER_LOW};
@@ -114,6 +111,7 @@ void newconn_close(struct newconn* nc);
 void newconn_free(struct newconn* nc);
 struct newconn* newconn_add(struct newconn *ncs,int *pos);
 
+
 //описание формата сообщений, которые могут приходить _ОТ_ клиентов по управляющему защищенному соединению 
 #define MSGINFO_MSGSNUM(arr) (sizeof(arr)/sizeof(struct PICA_msginfo))
 
@@ -526,6 +524,8 @@ 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");
 
@@ -535,11 +535,35 @@ switch(buf[0])
 	if (buf[2]!='O' || buf[3]!='K')
 		return 0;
 
-	*((int*)ptr)=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);
+	*((uint16_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_bytes(mp->tail, 2);
+	}
+else 
+	return 0;
+	
 	break;
 	case PICA_PROTO_VERDIFFER:
-	((char*)ptr)[0]=buf[2];
-	((char*)ptr)[1]=buf[3];
+	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;
 	}
 
@@ -706,16 +730,19 @@ while(listleft)
 			na_ipv4.port=*(in_port_t*)(buf+size-listleft+5);
 
 			sprintf(na.addr,"%.16s",inet_ntoa(*(struct in_addr*)&na_ipv4.addr));
-			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));
-					}
-					
 			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));
+// 					}
+					
+			
 
 			listleft-=PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;
 			break;
@@ -1753,7 +1780,6 @@ return mp;
 }
 
 
-
 // обработка различных таймаутов. Отправка пингов
 
 void process_timeouts_newconn()
@@ -2541,6 +2567,48 @@ while(nl)
 	}
 }
 
+void process_nodewait()
+{
+struct nodewait *kill_ptr = 0, *nw = nodewait_list;
+
+while(nw)
+	{
+	if (PICA_NODEWAIT_FINISHED_OK == nw->state)
+		{
+		struct nodelink *nlp;
+		
+		PICA_debug1("connected to %.255s %u, sending request ", nw->addr.addr, nw->addr.port);
+		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[0] = PICA_PROTO_VER_LOW;
+				}
+			}
+		kill_ptr = nw;
+		}
+
+	if (PICA_NODEWAIT_FINISHED_ERR == nw->state)
+		{
+		PICA_debug1("connection to %.255s %u failed", nw->addr.addr, nw->addr.port);
+		kill_ptr = nw;
+		}
+
+	nw = nw->next;
+
+	if (kill_ptr)
+		{
+		nodewait_list_delete(nw);
+		kill_ptr = 0;
+		}
+	}
+
+}
+
 int node_loop()
 {
 fd_set fds;
@@ -2583,6 +2651,7 @@ do
 	process_n2n_write();
 	
 	process_timeouts();
+	process_nodewait();
 	}
 while(1);////<<<<<<<<<<
 return 0;
@@ -2659,6 +2728,26 @@ if (verbosity > 3)
 PICA_set_loglevel(PICA_LOG_INFO + verbosity);
 }
 
+void PICA_node_joinskynet(char* addrlistfilename,const char *my_addr)
+{
+struct PICA_nodeaddr *nap,*addrlist_h=0;
+int ret;
+
+ret = PICA_nodeaddr_list_load(addrlistfilename, &addrlist_h);//MEM
+if (ret <= 0)
+	return;
+
+nap = addrlist_h;
+
+while(nap)
+	{
+	nodewait_start_connection(nap);
+	nap=nap->next;
+	}
+
+PICA_nodeaddr_list_free(addrlist_h);
+}
+
 int main(int argc,char** argv)
 {
 
diff --git a/PICA_node.h b/PICA_node.h
index c387dc1..efedcb9 100755
--- a/PICA_node.h
+++ b/PICA_node.h
@@ -163,4 +163,5 @@ time_t tmst;
 int disconnect_ticking;
 };
 
+
 #endif
diff --git a/PICA_nodejoinskynet.c b/PICA_nodejoinskynet.c
deleted file mode 100644
index 740d7e4..0000000
--- a/PICA_nodejoinskynet.c
+++ /dev/null
@@ -1,186 +0,0 @@
-#define JOINSKYNET
-#include "PICA_node.h"
-#include "PICA_nodejoinskynet.h"
-#include "PICA_proto.h"
-#include "PICA_msgproc.h"
-#include "PICA_common.h"
-#include "PICA_nodeconfig.h"
-#include "PICA_log.h"
-#include <time.h>
-#include <sys/time.h>
-
-#warning "USE HEADER  VVV"
-struct nodelink *nodelink_list_addnew(struct newconn *nc);
-extern struct PICA_msginfo _msginfo_node[];
-struct PICA_proto_msg* nodelink_wbuf_push
-	(struct nodelink *nl,unsigned int msgid,unsigned int size);
-
-//Делает попытку подключения к указанному узлу, при удачном подключении отправляет узлу NODECONNREQ и возвращает 1. При неудаче возвращает 0
-//static int try_connect_to_node(struct PICA_nodeaddr *nap,struct newconn *nc)
-	
-/*static*/ int try_connect_to_node(char *addr, unsigned int port,struct newconn *nc)
-{
-struct addrinfo h,*r,*ap;
-char portbuf[8];
-char msg[PICA_PROTO_NODECONNREQ_SIZE]=
-{PICA_PROTO_NODECONNREQ,PICA_PROTO_NODECONNREQ,
-PICA_PROTO_VER_HIGH,PICA_PROTO_VER_LOW};
-int ret;
-
-memset(&h,0,sizeof(struct addrinfo));
-
-PICA_debug1("connecting to node %.255s %u ...", addr, port);
-
-#ifdef AI_IDN
-#define H_AI_FLAGS (AI_ADDRCONFIG | AI_IDN)
-#else
-#define H_AI_FLAGS AI_ADDRCONFIG
-#endif
-
-h.ai_flags=H_AI_FLAGS;
-h.ai_family=AF_INET;//CONF
-h.ai_socktype=SOCK_STREAM;
-h.ai_protocol=IPPROTO_TCP;
-
-sprintf(portbuf,"%u",port);
-
-if (0!=getaddrinfo(addr,portbuf,&h,&r))
-	return 0;
-
-ap=r;
-
-while(ap)
-	{
-	nc->sck=socket(ap->ai_family,ap->ai_socktype,ap->ai_protocol);
-	if 	(nc->sck==SOCKET_ERROR)
-		{
-		ap=ap->ai_next;
-		continue;
-		}
-
-	if (0!=connect(nc->sck,ap->ai_addr,ap->ai_addrlen))	
-		{
-		ap=ap->ai_next;
-
-		CLOSE(nc->sck);
-		continue;
-		}
-	
-	if (PICA_PROTO_NODECONNREQ_SIZE!=
-			send(nc->sck,msg,PICA_PROTO_NODECONNREQ_SIZE,0))
-		{
-		ap=ap->ai_next;
-		CLOSE(nc->sck);
-		}
-	else
-		break;
-	}
-
-if (ap)
-	{
-	nc->addr=*((struct sockaddr_in*)(ap->ai_addr));
-	freeaddrinfo(r);
-	return 1;
-	}
-
-freeaddrinfo(r);
-return 0;
-}
-
-/*static*/ int try_get_reply(struct newconn *nc)
-{
-int ret;
-fd_set fds;
-struct timeval tv={10,0};
-union _pv
-	{
-	int i;
-	char c[4];
-	} 
-proto_version;
-
-proto_version.c[0]=PICA_PROTO_VER_HIGH;
-proto_version.c[1]=PICA_PROTO_VER_LOW;
-FD_ZERO(&fds);		
-FD_SET(nc->sck,&fds);
-ret=select(nc->sck+1,&fds,0,0,&tv);
-if (0==ret)
-	return 0;
-if (-1==ret)
-	return 0;
-
-ret=recv(nc->sck,nc->buf,4,0);
-if (ret<=0)
-	return 0;
-
-nc->pos+=ret;
-if (!PICA_processdatastream(nc->buf,(unsigned int*)&nc->pos,&proto_version,_msginfo_node,2))
-	return 0;
-		
-if (!proto_version.i) //server reply was ok
-	return 1;
-else
-	{
-	if (proto_version.c[0]> PICA_PROTO_VER_HIGH ||  (proto_version.c[1]> PICA_PROTO_VER_LOW && proto_version.c[0]== 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;
-}
-
-/*static*/ void send_newnode(struct nodelink *nlp, const char *my_addr)
-{
-struct PICA_proto_msg *mp;
-mp=nodelink_wbuf_push(nlp,PICA_PROTO_NEWNODE_IPV4,PICA_PROTO_NEWNODE_IPV4_SIZE);
-*((in_addr_t*)(mp->tail))=inet_addr(my_addr);//CONF TEMP
-*((uint16_t*)(mp->tail+4))=(nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) :  htons(PICA_COMM_PORT));//CONF; 
-}
-
-static void send_nodelistreq(struct nodelink *nlp)
-{
-struct PICA_proto_msg *mp;
-mp=nodelink_wbuf_push(nlp,PICA_PROTO_NODELISTREQ,PICA_PROTO_NODELISTREQ_SIZE);
-RAND_bytes(mp->tail, 2); 
-}
-
-//Загружает из указанного файла addrlistfilename список адресов узлов, делает попытки  подключения ко всем узлам, после удачного подключения  к какому-либо из них запрашивает у узла актуальный список адресов узлов, после чего возвращает количество активных узлов в сети
-int PICA_node_joinskynet(char* addrlistfilename,const char *my_addr)
-{
-struct PICA_nodeaddr *nap,*addrlist_h=0;
-struct nodelink *nlp;
-struct newconn nc;
-int ret;
-
-
-ret=PICA_nodeaddr_list_load(addrlistfilename,&addrlist_h);//MEM
-if (ret<=0)
-	return ret;
-nap=addrlist_h;
-memset(&nc,0,sizeof(nc));
-while(nap)
-	{
-	if (try_connect_to_node(nap->addr,nap->port,&nc))
-		if (try_get_reply(&nc))
-			{
-			nlp=nodelink_list_addnew(&nc);
-			send_newnode(nlp,my_addr);
-			send_nodelistreq(nlp);
-			ret=1;
-			break;
-			}
-			
-		
-	CLOSE(nc.sck);
-	nc.pos=0;
-	nap=nap->next;
-	}
-
-PICA_nodeaddr_list_free(addrlist_h);
-return ret;
-}
-
diff --git a/PICA_nodejoinskynet.h b/PICA_nodejoinskynet.h
deleted file mode 100644
index 1ca2578..0000000
--- a/PICA_nodejoinskynet.h
+++ /dev/null
@@ -1,17 +0,0 @@
- 
-#ifndef PICA_NODEJOINSKYNET_H
-#define PICA_NODEJOINSKYNET_H
-
-int PICA_node_joinskynet(char* addrlistfilename,const char *my_addr);
-int try_connect_to_node(char *addr, unsigned int port,struct newconn *nc);
-int try_get_reply(struct newconn *nc);
-void send_newnode(struct nodelink *nlp, const char *my_addr);
-
-#ifdef JOINSKYNET //include from PICA_nodejoinskynet.c, private data types definition
-#include "PICA_nodeaddrlist.h"
-
-
-#endif
-
-
-#endif
diff --git a/PICA_nodewait.c b/PICA_nodewait.c
new file mode 100644
index 0000000..2716d18
--- /dev/null
+++ b/PICA_nodewait.c
@@ -0,0 +1,170 @@
+ 
+#include "PICA_nodewait.h"
+#include "PICA_log.h"
+#include "PICA_common.h"
+
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+
+#ifdef WIN32
+
+#include <windows.h>
+
+#else
+
+#include <pthread.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+
+#endif
+
+struct nodewait *nodewait_list;
+
+struct nodewait *nodewait_list_addnew()
+{
+struct nodewait *nw;
+    
+nw = (struct nodewait*)calloc(1, sizeof(struct nodewait));
+    
+if (!nw)
+	return 0;
+
+nw->next = nodewait_list;
+nodewait_list = nw;
+
+nw->state = PICA_NODEWAIT_NEW;
+return nw;
+}
+
+void nodewait_list_delete(struct nodewait *nw)
+{
+struct nodewait **pp = &nodewait_list;
+struct nodewait *p = nodewait_list;
+
+while(p) 
+	{
+	if (p == nw)
+		{
+		*pp = nw->next;
+		//
+		//free resources
+		//
+		free(nw);
+	
+		return;
+		}
+		
+	pp = &p->next;
+	p = p->next;
+	}
+}
+
+#ifdef WIN32
+static DWORD WINAPI nodewait_thread (void *arg)
+#else
+static void  *nodewait_thread (void *arg)
+#endif
+{
+struct nodewait *nw = (struct nodewait *)arg;
+struct addrinfo h,*r,*ap;
+char portbuf[8];
+
+nw->state = PICA_NODEWAIT_RUNNING;
+
+memset(&h,0,sizeof(struct addrinfo));
+
+#ifdef AI_IDN
+#define H_AI_FLAGS (AI_ADDRCONFIG | AI_IDN)
+#else
+#define H_AI_FLAGS AI_ADDRCONFIG
+#endif
+
+h.ai_flags=H_AI_FLAGS;
+h.ai_family=AF_INET;//CONF - IPv4, IPv6, IPv4 then IPv6, IPv6 then IPv4
+h.ai_socktype=SOCK_STREAM;
+h.ai_protocol=IPPROTO_TCP;
+
+sprintf(portbuf,"%u",nw->addr.port);
+
+if (0!=getaddrinfo(nw->addr.addr, portbuf, &h, &r))
+	return 0;
+
+ap = r;
+
+while(ap)
+	{
+	nw->nc.sck=socket(ap->ai_family,ap->ai_socktype,ap->ai_protocol);
+	if 	(nw->nc.sck==SOCKET_ERROR)
+		{
+		ap=ap->ai_next;
+		continue;
+		}
+
+	if (0 != connect(nw->nc.sck,ap->ai_addr,ap->ai_addrlen))	
+		{
+		ap=ap->ai_next;
+
+		CLOSE(nw->nc.sck);
+		continue;
+		}
+	else
+		{
+		break;
+		}
+	}
+
+if (ap)
+	{
+	nw->nc.addr=*((struct sockaddr_in*)(ap->ai_addr));
+	freeaddrinfo(r);
+
+	nw->state = PICA_NODEWAIT_FINISHED_OK;
+	return 1;
+	}
+
+freeaddrinfo(r);
+nw->state = PICA_NODEWAIT_FINISHED_ERR;
+return 0;
+}
+
+void nodewait_start_connection(struct PICA_nodeaddr *a)
+{
+struct nodewait *nw;
+#ifndef WIN32
+pthread_t thr;
+pthread_attr_t attr;
+#else
+DWORD thread_id;
+HANDLE thread_h;
+#endif
+
+nw = nodewait_list_addnew();
+
+if (!nw)
+	return;
+
+nw ->addr = *a;
+
+PICA_debug1("starting connection to node %.255s %u ...", nw->addr.addr, nw->addr.port);
+
+#ifndef WIN32
+if (0 != pthread_attr_init(&attr))
+	PICA_error("pthread_attr_init() call failed.");
+
+pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+
+if (0 != pthread_create(&thr, &attr, nodewait_thread, (void*)nw))
+	PICA_error("unable to create thread: %s", strerror(errno));
+
+pthread_attr_destroy(&attr);
+#else
+thread_h = CreateThread(NULL, 4096, nodewait_thread, (void*)nw, 0, &thread_id);
+
+if (NULL != thread_h)
+	CloseHandle(thread_h);
+else
+	PICA_error("unable to create thread: %u", GetLastError());
+#endif
+}
diff --git a/PICA_nodewait.h b/PICA_nodewait.h
new file mode 100644
index 0000000..614db09
--- /dev/null
+++ b/PICA_nodewait.h
@@ -0,0 +1,37 @@
+#ifndef PICA_NODEWAIT_H
+#define PICA_NODEWAIT_H
+
+#include "PICA_nodeaddrlist.h"
+#include "PICA_node.h"
+
+#define PICA_NODEWAIT_NEW 1
+#define PICA_NODEWAIT_RUNNING 2
+#define PICA_NODEWAIT_FINISHED_OK 3
+#define PICA_NODEWAIT_FINISHED_ERR 4
+
+struct nodewait
+{
+struct newconn nc;
+int state;
+time_t tmst;
+
+//unsigned int addr_type;
+// union address_ptr
+// 	{
+// 	struct PICA_nodeaddr_ipv4 *ipv4;
+// 	struct PICA_nodeaddr_ipv6 *ipv6;
+// 	struct PICA_nodeaddr_dns *dns;
+// 	} addrptr;
+
+struct PICA_nodeaddr addr;
+	
+struct nodewait *next;
+};
+
+extern struct nodewait *nodewait_list;
+
+struct nodewait *nodewait_list_addnew();
+void nodewait_list_delete(struct nodewait *nw);
+void nodewait_start_connection(struct PICA_nodeaddr *a);
+
+#endif
diff --git a/configure.ac b/configure.ac
index 7fb5f13..70ea2c0 100644
--- a/configure.ac
+++ b/configure.ac
@@ -14,7 +14,7 @@ AC_PROG_INSTALL
 # Checks for common libraries, required by both pica-node and pica-client
 AC_CHECK_LIB([crypto], [RAND_add],[],[AC_MSG_ERROR(["libcrypto from OpenSSL package not found"])],[ -ldl -lz ])
 AC_CHECK_LIB([ssl], [SSL_library_init],[],[AC_MSG_ERROR(["libssl from OpenSSL package not found"])], [ -lcrypto -ldl -lz ])
-
+AC_CHECK_LIB([pthread], [pthread_create])
 
 #checking if pica-node is disabled
 AC_ARG_ENABLE([node],