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