faubackup2/src/network network_receiver.cpp,1.5,1.6 network_receiver.hpp,1.4,1.5 network_sender.cpp,1.8,1.9 network_sender.hpp,1.5,1.6

Erik Merkel <[email protected]> Thu, 22 Sep 2005 15:33:19 +0000
Newsgroups gmane.comp.sysutils.backup.faubackup.cvs
Message-ID <[email protected]>
Update of /cvsroot/faubackup/faubackup2/src/network
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv10999

Modified Files:
	network_receiver.cpp network_receiver.hpp network_sender.cpp 
	network_sender.hpp 
Log Message:

now using streams instead of simple read / write calls to improve 
performance when processing small sized packages


Index: network_sender.cpp
===================================================================
RCS file: /cvsroot/faubackup/faubackup2/src/network/network_sender.cpp,v
retrieving revision 1.8
retrieving revision 1.9
diff -C2 -d -r1.8 -r1.9
*** network_sender.cpp	16 Sep 2005 19:11:18 -0000	1.8
--- network_sender.cpp	22 Sep 2005 15:33:17 -0000	1.9
***************
*** 15,34 ****
  {
  
! NetworkSender::NetworkSender(int newfd)
  {
! 	if(pthread_cond_init(&sender_client1_cond, NULL)) 
  		output.critical("error initializing sender_client1_cond", errno);
! 	if(pthread_cond_init(&sender_client2_cond, NULL)) 
  		output.critical("error initializing sender_client2_cond", errno);
! 	if(pthread_cond_init(&sender_network_cond, NULL)) 
  		output.critical("error initializing sender_network_cond", errno);
! 	if(pthread_mutex_init(&sender_client1_mutex, NULL)) 
  		output.critical("error initializing sender_client1_mutex", errno);
! 	if(pthread_mutex_init(&sender_client2_mutex, NULL)) 
  		output.critical("error initializing sender_client2_mutex", errno);
! 	if(pthread_mutex_init(&sender_network_mutex, NULL)) 
  		output.critical("error initializing sender_network_mutex", errno);
! 	
! 	this->fd = newfd;
  	die_on_end = false;
  	active = true;
--- 15,39 ----
  {
  
! NetworkSender::NetworkSender(FILE *newfd)
  {
! 	this->fd = newfd;
! 	if(!newfd)
! 		output.critical("bad filestream", errno);
! 	fflush(fd);
! 	setvbuf(fd, buffer,_IOFBF, 524288);
! 
! 	if(pthread_cond_init(&sender_client1_cond, NULL))
  		output.critical("error initializing sender_client1_cond", errno);
! 	if(pthread_cond_init(&sender_client2_cond, NULL))
  		output.critical("error initializing sender_client2_cond", errno);
! 	if(pthread_cond_init(&sender_network_cond, NULL))
  		output.critical("error initializing sender_network_cond", errno);
! 	if(pthread_mutex_init(&sender_client1_mutex, NULL))
  		output.critical("error initializing sender_client1_mutex", errno);
! 	if(pthread_mutex_init(&sender_client2_mutex, NULL))
  		output.critical("error initializing sender_client2_mutex", errno);
! 	if(pthread_mutex_init(&sender_network_mutex, NULL))
  		output.critical("error initializing sender_network_mutex", errno);
! 
  	die_on_end = false;
  	active = true;
***************
*** 44,47 ****
--- 49,53 ----
  		while (queue_client2.empty() && queue_client1.empty() && !die_on_end) {
  			//			output.information("inside lock" + int2string(queue_client1.size()) + " " + int2string(queue_client2.size()), EXCESSIV_DEBUG_OUTPUT);
+ 			fflush(fd);
  			pthread_cond_wait( &sender_network_cond, &sender_network_mutex);
  		}
***************
*** 68,77 ****
  			}
  		}
! 
  
  		if (die_on_end && queue_client1.empty() && queue_client2.empty()) {
  			uint8_t dummy = 0xff; //special sign
  			sendData(&dummy, 1);
! 			close(fd);
  			thread.setDead();
  			return ;
--- 74,83 ----
  			}
  		}
! 		//fflush(fd);
  
  		if (die_on_end && queue_client1.empty() && queue_client2.empty()) {
  			uint8_t dummy = 0xff; //special sign
  			sendData(&dummy, 1);
! 			fflush(fd);
  			thread.setDead();
  			return ;
***************
*** 87,92 ****
  	if (clientNo == 1) {
  		pthread_mutex_lock( &sender_client1_mutex);
! 		if (queue_client1.size() > 2000)
! 			while (queue_client1.size() > 1500)
  				pthread_cond_wait( &sender_client1_cond, &sender_client1_mutex);
  		pthread_mutex_unlock( &sender_client1_mutex);
--- 93,98 ----
  	if (clientNo == 1) {
  		pthread_mutex_lock( &sender_client1_mutex);
! 		if (queue_client1.size() > 100)
! 			while (queue_client1.size() > 50)
  				pthread_cond_wait( &sender_client1_cond, &sender_client1_mutex);
  		pthread_mutex_unlock( &sender_client1_mutex);
***************
*** 100,105 ****
  		if (clientNo == 2) {
  			pthread_mutex_lock( &sender_client2_mutex);
! 			if (queue_client2.size() > 2000)
! 				while (queue_client2.size() > 1500)
  					pthread_cond_wait( &sender_client2_cond, &sender_client2_mutex);
  			pthread_mutex_unlock( &sender_client2_mutex);
--- 106,111 ----
  		if (clientNo == 2) {
  			pthread_mutex_lock( &sender_client2_mutex);
! 			if (queue_client2.size() > 300)
! 				while (queue_client2.size() > 250)
  					pthread_cond_wait( &sender_client2_cond, &sender_client2_mutex);
  			pthread_mutex_unlock( &sender_client2_mutex);
***************
*** 163,173 ****
  	ssize_t count = 0, ret;
  	while(count < length){
! 	    if((ret = write (fd, (char*)data + count, length - count)) < 0 )
  		if((errno == EINTR) || (errno == EAGAIN))
  		    continue;
! 		else	
  		    output.critical("sendData: error sending Data", errno);
  	    count +=ret;
  	}
  	total += length;
  	//	output.information("senddata end", EXCESSIV_DEBUG_OUTPUT);
--- 169,180 ----
  	ssize_t count = 0, ret;
  	while(count < length){
! 	    if((ret = fwrite ((const char*)data + count, 1, length - count, fd)) <= 0 )
  		if((errno == EINTR) || (errno == EAGAIN))
  		    continue;
! 		else
  		    output.critical("sendData: error sending Data", errno);
  	    count +=ret;
  	}
+ 
  	total += length;
  	//	output.information("senddata end", EXCESSIV_DEBUG_OUTPUT);

Index: network_sender.hpp
===================================================================
RCS file: /cvsroot/faubackup/faubackup2/src/network/network_sender.hpp,v
retrieving revision 1.5
retrieving revision 1.6
diff -C2 -d -r1.5 -r1.6
*** network_sender.hpp	16 Sep 2005 19:11:18 -0000	1.5
--- network_sender.hpp	22 Sep 2005 15:33:17 -0000	1.6
***************
*** 28,41 ****
  	void sendPackage(NetworkItem *item);
  	uint8_t makeHeaderByte(uint8_t type, uint8_t dest);
! 	int fd;
  	bool die_on_end;
  	bool active;
  	uint64_t total;
! 	char buffer[32768];
  public:
! 	NetworkSender(int fd);
  	~NetworkSender()
  	{
! 		close(fd);
  	}
  	void work();
--- 28,42 ----
  	void sendPackage(NetworkItem *item);
  	uint8_t makeHeaderByte(uint8_t type, uint8_t dest);
! 	FILE *fd;
  	bool die_on_end;
  	bool active;
  	uint64_t total;
! 	char buffer[524288];
  public:
! 	NetworkSender(FILE *fd);
  	~NetworkSender()
  	{
! 		fflush(fd);
! 		fclose(fd);
  	}
  	void work();

Index: network_receiver.hpp
===================================================================
RCS file: /cvsroot/faubackup/faubackup2/src/network/network_receiver.hpp,v
retrieving revision 1.4
retrieving revision 1.5
diff -C2 -d -r1.4 -r1.5
*** network_receiver.hpp	16 Sep 2005 19:11:18 -0000	1.4
--- network_receiver.hpp	22 Sep 2005 15:33:17 -0000	1.5
***************
*** 3,6 ****
--- 3,7 ----
  
  #include <stdint.h>
+ #include <cstdio>
  #include <queue>
  #include "Threads/XThread.hpp"
***************
*** 21,25 ****
  	pthread_mutex_t receiver_network_mutex;
  	pthread_cond_t receiver_network_cond;
! 	
  	XThread thread;
  	queue<NetworkItem *> queue_client1;
--- 22,27 ----
  	pthread_mutex_t receiver_network_mutex;
  	pthread_cond_t receiver_network_cond;
! 	char buffer[524288];
! 
  	XThread thread;
  	queue<NetworkItem *> queue_client1;
***************
*** 27,31 ****
  	void  * readData_exactly(void * location, size_t length);
  	uint8_t extractDestination(uint8_t header);
! 	int fd;
  	uint64_t total;
  	uint8_t extractType(uint8_t header);
--- 29,33 ----
  	void  * readData_exactly(void * location, size_t length);
  	uint8_t extractDestination(uint8_t header);
! 	FILE *fd;
  	uint64_t total;
  	uint8_t extractType(uint8_t header);
***************
*** 38,45 ****
  	void readByte(int dest);
  public:
! 	NetworkReceiver(int fd);
  	~NetworkReceiver()
  	{
! 		close(fd);
  	}
  
--- 40,48 ----
  	void readByte(int dest);
  public:
! 	NetworkReceiver(FILE *fd);
  	~NetworkReceiver()
  	{
! 		//fpurge(fd);
! 		//fclose(fd);
  	}
  

Index: network_receiver.cpp
===================================================================
RCS file: /cvsroot/faubackup/faubackup2/src/network/network_receiver.cpp,v
retrieving revision 1.5
retrieving revision 1.6
diff -C2 -d -r1.5 -r1.6
*** network_receiver.cpp	16 Sep 2005 19:11:17 -0000	1.5
--- network_receiver.cpp	22 Sep 2005 15:33:17 -0000	1.6
***************
*** 12,31 ****
  namespace faubackup
  {
! NetworkReceiver::NetworkReceiver(int newfd)
  {
! 	if(pthread_cond_init(&receiver_client1_cond, NULL)) 
  		output.critical("error initializing receiver_client1_cond", errno);
! 	if(pthread_cond_init(&receiver_client2_cond, NULL)) 
  		output.critical("error initializing receiver_client2_cond", errno);
! 	if(pthread_cond_init(&receiver_network_cond, NULL)) 
  		output.critical("error initializing receiver_network_cond", errno);
! 	if(pthread_mutex_init(&receiver_client1_mutex, NULL)) 
  		output.critical("error initializing receiver_client1_mutex", errno);
! 	if(pthread_mutex_init(&receiver_client2_mutex, NULL)) 
  		output.critical("error initializing sender_client2_mutex", errno);
! 	if(pthread_mutex_init(&receiver_network_mutex, NULL)) 
  		output.critical("error initializing sender_network_mutex", errno);
- 	
- 	fd = newfd;
  	total=0;
  	thread.start(*this, &NetworkReceiver::work);
--- 12,35 ----
  namespace faubackup
  {
! NetworkReceiver::NetworkReceiver(FILE *newfd)
  {
! 	this->fd = newfd;
! 	if(!newfd)
! 		output.critical("bad filestream", errno);
! 	fflush(fd);
! 	setvbuf(fd, buffer,_IOFBF, 524288);
! 
! 	if(pthread_cond_init(&receiver_client1_cond, NULL))
  		output.critical("error initializing receiver_client1_cond", errno);
! 	if(pthread_cond_init(&receiver_client2_cond, NULL))
  		output.critical("error initializing receiver_client2_cond", errno);
! 	if(pthread_cond_init(&receiver_network_cond, NULL))
  		output.critical("error initializing receiver_network_cond", errno);
! 	if(pthread_mutex_init(&receiver_client1_mutex, NULL))
  		output.critical("error initializing receiver_client1_mutex", errno);
! 	if(pthread_mutex_init(&receiver_client2_mutex, NULL))
  		output.critical("error initializing sender_client2_mutex", errno);
! 	if(pthread_mutex_init(&receiver_network_mutex, NULL))
  		output.critical("error initializing sender_network_mutex", errno);
  	total=0;
  	thread.start(*this, &NetworkReceiver::work);
***************
*** 36,41 ****
  		uint8_t header;
  		pthread_mutex_lock( &receiver_network_mutex );
! 		if((queue_client1.size() > 1000) && (queue_client2.size() > 1000))
! 			while((queue_client1.size() > 500) && (queue_client2.size() > 500))
  				pthread_cond_wait( &receiver_network_cond , &receiver_network_mutex );
  		pthread_mutex_unlock( &receiver_network_mutex);
--- 40,45 ----
  		uint8_t header;
  		pthread_mutex_lock( &receiver_network_mutex );
! 		if((queue_client1.size() > 100) && (queue_client2.size() > 300))
! 			while((queue_client1.size() > 50) && (queue_client2.size() > 250))
  				pthread_cond_wait( &receiver_network_cond , &receiver_network_mutex );
  		pthread_mutex_unlock( &receiver_network_mutex);
***************
*** 44,48 ****
  			thread.setDead();
  			output.critical("lost connection",errno);
! 			close(fd);
  			return;
  		}
--- 48,52 ----
  			thread.setDead();
  			output.critical("lost connection",errno);
! 			fclose(fd);
  			return;
  		}
***************
*** 50,54 ****
  			thread.setDead();
  			output.information("work fertig", EXCESSIV_DEBUG_OUTPUT);
! 			close(fd);
  			return;
  		}
--- 54,58 ----
  			thread.setDead();
  			output.information("work fertig", EXCESSIV_DEBUG_OUTPUT);
! 			fclose(fd);
  			return;
  		}
***************
*** 244,248 ****
  void  *NetworkReceiver::readData_exactly(void * location, size_t length)
  {
! 	//output.information("readdata exact begin", EXCESSIV_DEBUG_OUTPUT);
  
  	ssize_t count, to_read = length;
--- 248,252 ----
  void  *NetworkReceiver::readData_exactly(void * location, size_t length)
  {
! 	//output.information("readdata exact begin" + int2string(length), EXCESSIV_DEBUG_OUTPUT);
  
  	ssize_t count, to_read = length;
***************
*** 250,254 ****
  
  	while(to_read > 0) {
! 		if((count = read(fd, data, to_read))<0) {
  			if((errno == EINTR) || (errno == EAGAIN))
  				continue;
--- 254,258 ----
  
  	while(to_read > 0) {
! 		if((count = fread(data, 1, to_read, fd))<=0) {
  			if((errno == EINTR) || (errno == EAGAIN))
  				continue;
***************
*** 257,262 ****
  		}
  		if(count == 0) {
- 
- 
  			output.information("readdata exact: connection closed",0);
  			return NULL;
--- 261,264 ----



-------------------------------------------------------
SF.Net email is sponsored by:
Tame your development challenges with Apache's Geronimo App Server. 
Download it for free - -and be entered to win a 42" plasma tv or your very
own Sony(tm)PSP.  Click here to play: http://sourceforge.net/geronimo.php