ga_server.c

一个不错的GA-NN的神经网络模型的示范代码。适合入门学习

{
	if ( (listenfd[k] = socket(PF_LOCAL, SOCK_STREAM, 0)) < 0)
	{
		syslog(LOG_CRIT,"Error creating listen socket");
		return(1);
	}
	bzero(&serv_un_addr,sizeof(serv_un_addr));
	serv_un_addr.sun_family = PF_LOCAL;
	sprintf(char_buffer,"unix%d.str",k);
	unlink(char_buffer);
	strcpy(serv_un_addr.sun_path,char_buffer);
	
	if ( (bind(listenfd[k], (struct sockaddr *) &serv_un_addr, sizeof(serv_un_addr)) ) < 0)
	{
		syslog(LOG_CRIT,"Error Binding socket: %d %s",k,strerror(errno));
		return(1);
	}
	if ( (listen(listenfd[k],LISTENQ)) <0 )
	{
		syslog(LOG_CRIT,"Error listenning %d %s",k,strerror(errno));
		return(1);
	}
}

for (k=max_local; k < poblation; k++)
{
	if ( (listenfd[k] = socket(PF_INET, SOCK_STREAM, 0)) < 0)
	{
		syslog(LOG_CRIT,"Error creating listen socket");
		return(1);
	}
	bzero(&serv_in_addr,sizeof(serv_in_addr));
	serv_in_addr.sin_family = PF_INET;
	
	if (inet_pton(AF_INET,ga_conf.ip,&serv_in_addr.sin_addr) < 1)
	{
		syslog(LOG_CRIT,"Error inet_pton() %d %s",k,strerror(errno));
		return(1);
	}
	serv_in_addr.sin_port = htons((uint16_t)(local_server_port + k));

	if ( (bind(listenfd[k], (struct sockaddr *) &serv_in_addr, sizeof(serv_in_addr)) ) < 0)
	{
		syslog(LOG_CRIT,"Error Binding socket: %d %s",k,strerror(errno));
		return(1);
	}
	if ( (listen(listenfd[k],LISTENQ)) <0 )
	{
		syslog(LOG_CRIT,"Error listenning %d %s",k,strerror(errno));
		return(1);
	}
}


/*** Code for common SLAVE ELEMENTS ***/
/**************************************/
if ( (pid = fork()) < 0)
{
	syslog(LOG_CRIT,"Error fork client activation: %s",strerror(errno));
	return(1);
}
else if (pid==0)
{
	for (k=0;(k < max_local) && (k < poblation); k++)
	{
		/*************** Client caller ****************/	
		/***************Re design pending *************/	
		if ((pid=fork()) < 0)
		{
			syslog(LOG_CRIT,"Error fork client: %d %s",k,strerror(errno));
			return(1);
		}
		else if (pid==0)
		{
			setsid();
			if ( !(char_vector = (char **)malloc(11*sizeof(char *)) ))
			{
				syslog(LOG_CRIT,"Error malloc char_vector");
				return(1);
			}
			for (l=0; l< 11; l++)
			{
				if ( !(char_vector[l] = (char *)malloc(BUFFSIZE*sizeof(char)) ))
				{
					syslog(LOG_CRIT,"Error malloc char_vector[%d]",l);
					return(1);
				}
			}
			char_vector[10]=NULL;
			sprintf(char_vector[0],"ga_client");
			sprintf(char_vector[1],"%d",k);
			sprintf(char_vector[2],"%d",ga_conf.max_num_layer);
			sprintf(char_vector[3],"%d",ga_conf.max_num_neuron);
			sprintf(char_vector[4],"%d",generations);
			sprintf(char_vector[5],"127.0.0.1");
			sprintf(char_vector[6],"%d",local_server_port+k);
			sprintf(char_vector[7],"%d",agr);
			sprintf(char_vector[8],"%s",ga_conf.init_preffix);
			sprintf(char_vector[9],"%s",ga_conf.result_preffix);

			execvp("ga_client",char_vector);
		}
	}

	/*** Starting remote clients ***/
	
	memcpy(&ga_conf_temp, &ga_conf, sizeof(struct ga_server_config));
	
	for (k=1; k< (poblation_f / max_local_f); k++)
	{
		syslog(LOG_INFO,"Using remote server %d",k);
		if ( (connfd_remote = socket(PF_INET, SOCK_STREAM, 0)) < 0)
		{
			syslog(LOG_CRIT,"Error creating socket: %d %s",k,strerror(errno));
			return(1);
		}
			
		bzero(&serv_in_addr_remote, sizeof(serv_in_addr_remote));
		serv_in_addr_remote.sin_family = PF_INET;
		serv_in_addr_remote.sin_port = htons((uint16_t)remote_server_port); 
		
		syslog(LOG_INFO,"remote host: %s",ga_conf.hosts[k-1]);
		
		if (inet_pton(AF_INET,ga_conf.hosts[k-1],&serv_in_addr_remote.sin_addr) < 1)
		{
			syslog(LOG_CRIT,"Error inet_pton() %d %s",k,strerror(errno));
			return(1);
		}
		if( (connect(connfd_remote, (struct sockaddr *) &serv_in_addr_remote, 
								sizeof(serv_in_addr_remote))) <0)
		{
			syslog(LOG_CRIT,"Couldn't connect to remote server: %d %s",k,strerror(errno));
			return(1);
		}
		sprintf(ga_conf_temp.message,"SPAWN");
	
		ga_conf_temp.first_local = k*max_local;
		ga_conf_temp.last_local = ((k+1)*max_local > poblation) ? 
															(poblation - 1) : ((k+1)*max_local - 1);
		
		write_ga_server_remote_config(&ga_conf_temp, connfd_remote);
		close(connfd_remote);
	}

	/*** Start remote niches ***/
	
	for (k=0; (k< num_niche) && (ga_conf.master > 0) ; k++)
	{
		syslog(LOG_INFO,"Init remote niche %d",k);
		if ( (connfd_remote = socket(PF_INET, SOCK_STREAM, 0)) < 0)
		{
			syslog(LOG_CRIT,"Error creating socket: %d %s",k,strerror(errno));
			return(1);
		}
			
		bzero(&serv_in_addr_remote, sizeof(serv_in_addr_remote));
		serv_in_addr_remote.sin_family = PF_INET;
		serv_in_addr_remote.sin_port = htons((uint16_t)remote_server_port); 
		
		syslog(LOG_INFO,"remote niche: %s",ga_conf.niches[k]);
		
		if (inet_pton(AF_INET,ga_conf.niches[k],&serv_in_addr_remote.sin_addr) < 1)
		{
			syslog(LOG_CRIT,"Error inet_pton() %d %s",k,strerror(errno));
			return(1);
		}
		if( (connect(connfd_remote, (struct sockaddr *) &serv_in_addr_remote, 
								sizeof(serv_in_addr_remote))) <0)
		{
			syslog(LOG_CRIT,"Couldn't connect to remote server: %d %s",k,strerror(errno));
			return(1);
		}

		sprintf(ga_conf_temp.message,"NICHE");
		
		write_ga_server_remote_config(&ga_conf_temp, connfd_remote);
		close(connfd_remote);
	}
	exit(0);
}


/************* Code for MASTER SERVER BEGINS HERE **************/

/*** Local niche fifo ***/
if (ga_conf.num_niche > 0)
{
	if( (connfd_niche=open("niche_fifo",O_WRONLY, 0)) <0)
	{
		syslog(LOG_CRIT,"Error openning niche_fifo for writting: %s",strerror(errno));
		return(1);
	}
}

/*** Accepting clients connections ***/
syslog(LOG_INFO,"Start waiting for client connections");

l=0;
while(1)
{
	if (l==poblation)
	{	
		break;
	}	

	errno = 0;
	for (k=0; (k< max_local) && (k< poblation); k++)
	{
		if (!( *(connected+k) ) )
		{
			clilen = sizeof(cli_un_addr);
			if ( (connfd[k] = accept(listenfd[k], (struct sockaddr *) &cli_un_addr, &clilen)) <0 )
			{
				if (errno == EINTR)
				{
					continue;
				}
				else
				{
					syslog(LOG_CRIT,"Accept error (PF_LOCAL): %s",strerror(errno));
					return(1);
				}
			}
			else
			{
				*(connected+k)=1;
				l++;
				syslog(LOG_INFO,"Connected client: %d",k);
			}
		}
	}

	for (k=max_local; k<poblation; k++)
	{
		if (!( *(connected+k) ) )
		{
			clilen = sizeof(cli_in_addr);
			if ( (connfd[k] = accept(listenfd[k], (struct sockaddr *) &cli_in_addr, &clilen)) <0 )
			{
				if (errno == EINTR)
				{
					continue;
				}
				else
				{
					syslog(LOG_CRIT,"Accept error: %s",strerror(errno));
					return (1);
				}
			}
			else
			{
				*(connected+k)=1;
				l++;
				syslog(LOG_INFO,"Connected client: %d",k);
			}
		}
	}
}

if ( !(fitness = (float *)calloc(poblation,sizeof(float)) ))
{
	syslog(LOG_CRIT,"Error calloc fitness: %s",strerror(errno));
	return(1);
}
if ( !(error = (double *)calloc(poblation,sizeof(double)) ))
{
	syslog(LOG_CRIT,"Error calloc error: %s",strerror(errno));
	return(1);
}
if ( !(num_neuron = (int *)calloc(poblation,sizeof(float)) ))
{
	syslog(LOG_CRIT,"Error calloc num_neuron: %s",strerror(errno));
	return(1);
}
if ( !(new_num_neuron = (int *)calloc(poblation,sizeof(float)) ))
{
	syslog(LOG_CRIT,"Error calloc new_num_neuron: %s",strerror(errno));
	return(1);
}
if ( !(net_array = (struct individual **)calloc(poblation,sizeof(struct individual *)) ))
{
	syslog(LOG_CRIT,"Error calloc net_array: %s",strerror(errno));
	return(1);
}
if ( !(vertice = (float **)calloc(2,2*sizeof(float *)) ))
{
	syslog(LOG_CRIT,"Error calloc vertice: %s",strerror(errno));
	return(1);
}
for (k=0; k<2; k++)
{
	if ( !(*(vertice+k) = (float *)calloc(2,sizeof(float)) ))
	{
		syslog(LOG_CRIT,"Error calloc vertice[%d]: %s",k,strerror(errno));
		return(1);
	}
}

if ( !(inner_elem1 = malloc(sizeof(int))) || !(inner_elem2 = malloc(sizeof(int))) )
{
	syslog(LOG_CRIT,"Error malloc inner_elem1 / 2");
	return(1);
}

if ( !(parents = (int **)calloc(poblation,2*sizeof(int *)) ))
{
	syslog(LOG_CRIT,"Error calloc parents: %s",strerror(errno));
	return(1);
}
for(k=0; k< poblation; k++)
{
	if( !(*(parents+k) = (int *)calloc(2, sizeof(int)) ))
	{
		syslog(LOG_CRIT,"Error calloc parents[%d]: %s",k,strerror(errno));
		return(1);
	}

	if ( !(net_array[k] = (struct individual *)calloc(1,sizeof(struct individual)) ))
	{
		syslog(LOG_CRIT,"Error calloc net_array[%d] in MASTER: %s",k,strerror(errno));
		return(1);
	}
}


/****************** LOOP FOR GENERATIONS ****************/
/********************************************************/
for (gen=0; gen< generations; gen ++)
{

max_fd=0;
FD_ZERO(&read_set);
for (k=0; k<poblation; k++)
{
	FD_SET(connfd[k], &read_set);
	max_fd = (connfd[k] > max_fd) ? connfd[k] : max_fd;
	connected[k]=1;
}

FD_ZERO(&write_set);
FD_ZERO(&ex_set);
max_fd += 1; //number of descriptors;

syslog(LOG_INFO,"max_fd: %d",max_fd); //needed for bug

l=0;
counter=0;
while(1)
{
	if ( l < poblation )
	{
		FD_ZERO(&read_set);
		for (k=0; k<poblation; k++)
		{
			if (*(connected+k))
			{
				FD_SET(connfd[k], &read_set);
			}
		}
		if ( (ready_fd=pselect(max_fd, &read_set, &write_set, &ex_set, NULL, NULL)) <0 )
		{
			syslog(LOG_CRIT,"Error in pselect: %s",strerror(errno));
			return(1);
		}

		for (k=0; k<poblation; k++)
		{
			if (FD_ISSET(connfd[k], &read_set) )
			{
				if( (readn(connfd[k],char_buffer,BUFFSIZE)) < 0)
				{
					syslog(LOG_CRIT,"Error closed connection %d",k);
					return(1);
				}
				else
				{
					if ( !(strncmp("Error",char_buffer,5) ) )
					{
						error[k] = atof(char_buffer+6);
					}

					else if ( !(strncmp("neuron_array",char_buffer,12) ) )
					{
						if( readn(connfd[k],char_buffer,BUFFSIZE) < 0)
						{
							syslog(LOG_CRIT,"Error closed connection %d",k);
							return(1);
						}
						num_neuron[k] = atoi(char_buffer+11);
						syslog(LOG_INFO,"client=%d num_neuron=%d",k,num_neuron[k]);  //required for bug
						counter += num_neuron[k];

						if ( !(net_array[k]->neuron_array = (struct neuron **)calloc(
																		num_neuron[k],sizeof(struct neuron *)) ))
						{
							syslog(LOG_CRIT,"Error calloc net_array[%d]->neuron_array: %s",k,
											strerror(errno));
							return(1);
						}

						for (m=0; m< num_neuron[k]; m++)
						{
							if ( !(net_array[k]->neuron_array[m] = (struct neuron *)calloc(
														1,sizeof(struct neuron)) ))
							{
								syslog(LOG_CRIT,"Error calloc net_array[%d]->neuron_array[%d]: %s",k,l,
														strerror(errno));
								return(1);
							}
						}
						read_neuron_array(num_neuron[k], net_array[k]->neuron_array, connfd[k]);
					}
					
					else if (!strncmp("All work done",char_buffer,13) )
					{
						l+=1;
						connected[k]=0;
					}
				}
			}
		}
	}
	else
	{
		syslog(LOG_INFO,"All work done with clients");
		break;
	}
}