3. Eventbasierender TCP Server

Eventbasierender TCP Server

  • ?

    Hallo,

    Ich würde mich gerne in meinem C++ verbessern, derzeit habe Ich versucht einen Event basierenden Server zu Schreiben. Ihr seit so ziemlich die einzigen C++ Profis die Ich kenne und möchte euch deshalb darum bitten kurz über den Quellcode zu schauen und mir eure Ideen von Verbesserungswürdigen zu geben. Das ganze ist lauffähig und scheint fehlerfrei zu Funktionieren. Ich suche im Moment Möglichkeiten das ganze Effizienter und Eleganter zu Gestalten.

    Die std::cout .. print Anweisungen sind Debugger Anweisungen für meinen Logger und werden Exceptions weichen.

    SERVE.h

    #ifndef SERVEH
#define SERVEH

#include "../GLOBAL/GLOBAL.h"
#include "../NOTIFY/NOTIFIER.h"
#include "../LOG/LOG.h"
#include "SERVE_MESSAGE.h"
#include <vector>
#include <map>
#include <thread>
#include <sys/types.h>
#include <event2/event.h>
#include <event2/event-config.h>
#include <event2/thread.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include "SERVE_CLIENT.h"

#define DEFAULT_BUFFER_SIZE 2048
#define SIG_WRITE 0x0BABABAB

class SERVE : public NOTIFIER
{
public:
	SERVE(int);
	virtual ~SERVE();

	void start();
	void stop();
	void setListenPort(int port);
	int getListenPort();

	/* send to all clients */
	void broadcastData( unsigned char *buffer, unsigned int len );

	/* add send data to list */
	void addSendDataToList(SERVE_CLIENT *client, unsigned char *buffer, unsigned int len);
	void addSendDataToList(int sockfd, unsigned char *buffer, unsigned int len );

	void tcp_server_write(int fd, byte * buf, int buflen);
	void tcp_server_write(int fd, byte buf );
	void tcp_server_broadcast( byte * buf, int buflen );
private:

	void main_loop();

	/* Server Properties */
	int listenPort;

	/* Listen Socket */
	int listenSocket;

	/* List of clients */
	std::map<int,SERVE_CLIENT*> clientList;

	void deleteClient( SERVE_CLIENT *client );

	/* lib event Functions */
	struct event* addEvent(int fd, short flags );
	static void cb_func( evutil_socket_t fd, short what, void* arg);
	void intern_cb( evutil_socket_t fd, short what );

	/* Buffer */
	unsigned char *buffer;

	/* Client Events */
	void acceptClient();
	void clientRcv( SERVE_CLIENT *client);

	/* lib event vars */
	struct event_base * eb;
	struct event * lisent_event;

	std::thread* main_loop_thread;

	static SERVE_SEND * SENDER;
	static int ref_count;
	static std::mutex thread_lock;
};

#endif

    SERVE.cpp

    #include "SERVE.h"
#include "SERVE_CLIENT.h"

SERVE::SERVE( int lisentPort )
:listenPort(lisentPort)
,eb(NULL)
,main_loop_thread(NULL)
{
	thread_lock.lock();
	ref_count = ref_count + 1;
	thread_lock.unlock();
	buffer = new unsigned char[DEFAULT_BUFFER_SIZE];
	if( evthread_use_pthreads() < 0 ){
		std::cout << debug << "SERVE evthread_use_pthreads failed" << print;
	// error
	}
	clientList.get_allocator().allocate(100);
	std::cout << debug << "SERVE constructor finished" << print;
}
/*--------------------------------------------------------------------*/
void SERVE::setListenPort(int port)
{
	this->listenPort = port;
}
/*--------------------------------------------------------------------*/
SERVE::~SERVE()
{
	stop();

	if(buffer != NULL) 
		delete[] buffer;

	/* clean up clients */
	for ( auto it=clientList.begin(); it!=clientList.end(); ++it)
		deleteClient(it->second);
	clientList.clear();

	thread_lock.lock();
	ref_count = ref_count - 1;
	if( ref_count <= 0 ){
		delete SENDER;
		SENDER = NULL;
	}
	thread_lock.unlock();	
}
/*--------------------------------------------------------------------*/
void SERVE::start()
{
	if( SENDER == NULL ){
		SENDER = new SERVE_SEND();
	}	
	struct sockaddr_in localAddr;

	/* setup listen socket */
	if((listenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1){
		// error
		std::cout << debug << "SERVE Listen socket creation failed" << print;
	}

	/* set socket options */
	int yes = 1;
	if(setsockopt(listenSocket,SOL_SOCKET,SO_REUSEADDR,&yes,sizeof(int)) < 0) {
		// error
		std::cout << debug << "SERVE setsockopt failed" << print;
	}

	/* setup local addr */
	localAddr.sin_family = AF_INET;
	localAddr.sin_port = htons(this->listenPort);
	localAddr.sin_addr.s_addr = INADDR_ANY;

	/* bind to socket */
	if(bind(listenSocket, (struct sockaddr*)&localAddr, sizeof(localAddr)) == -1){
		// error
		std::cout << debug << "SERVE bind failed" << print;
	}

	/* let's listen */
	if(listen(listenSocket, 10) == -1){
		// error
		std::cout << debug << "SERVE listen failed" << print;
	}

	/* setup event base */
	if((eb = event_base_new()) == NULL){
		// error
		std::cout << debug << "SERVE event_base_new failed" << print;
	}	
	/* setup our initial event list */

	/* let's add an event for the listen socket */
	std::cout << debug << "SERVE creating events" << print;
	lisent_event = addEvent( listenSocket, EV_READ|EV_PERSIST );
	std::cout << debug << "SERVE created lisent_event event" << print;
	std::cout << debug << "SERVE entering event loop" << print;
	main_loop_thread = new std::thread( &SERVE::main_loop, this );
}
/*--------------------------------------------------------------------*/
void SERVE::main_loop(){
	std::cout << debug << "SERVE main_loop entered" << print;
	event_base_dispatch(eb);
	std::cout << debug << "SERVE main_loop returning" << print;
}
/*--------------------------------------------------------------------*/
struct event* SERVE::addEvent( int fd, short flags )
{
	/* set event */
	struct event * n = event_new( eb, fd, flags, SERVE::cb_func, reinterpret_cast<void*>(this) );
	if( n == NULL ){
		// error 
	}
	event_add( n, NULL );
	return n;
}
/*--------------------------------------------------------------------*/
void SERVE::cb_func( evutil_socket_t fd, short what, void* arg){
	reinterpret_cast<SERVE*>(arg)->intern_cb(fd,what);
}
/*--------------------------------------------------------------------*/
void SERVE::intern_cb( evutil_socket_t fd, short what ){
	std::cout << debug << "SERVE Event occured" << print;
	if( fd == listenSocket ){
		acceptClient();
	}else if( what == EV_READ ){
		auto it = clientList.find( fd );
		if( it == clientList.end() ){
			std::cout << debug << "SERVE Client not found in list" << print;
			// error
			// client not found in list
		}else{
			clientRcv(it->second);		
		}	
	}
}
/*--------------------------------------------------------------------*/
void SERVE::stop(){
	event_base_loopbreak(eb);
	event_base_loopexit(eb,NULL);
	event_del(lisent_event);
	main_loop_thread->detach();
	::close(listenSocket);
	event_base_free(eb);
	if( main_loop_thread != NULL ){
		delete main_loop_thread;
		main_loop_thread = NULL;
	}
}
/*--------------------------------------------------------------------*/
void SERVE::acceptClient()
{
	//std::cout << "acceptClient called\n";
	int remoteSocket, addrlen;
	struct sockaddr_in remoteAddr;
	SERVE_CLIENT *client;
	addrlen = sizeof(remoteAddr);

	/* accept client */
	remoteSocket = accept(listenSocket, (struct sockaddr*)&remoteAddr, (socklen_t*)&addrlen);
	if( remoteSocket < 0 ){
		std::cout << debug << "SERVE Accept failed" << print;
	}else{
		std::cout << debug << "SERVE Accepting client " << std::string(inet_ntoa(remoteAddr.sin_addr)) << print;
		/* setup new client */
		if( SIG_WRITE == remoteSocket ){
			std::cout << debug << "SERVE MUST CHOOSE A DIFFEREN SIG_WRITE!" << print;
		}
		client = new SERVE_CLIENT(remoteSocket, std::string(inet_ntoa(remoteAddr.sin_addr)), addEvent(remoteSocket, EV_READ|EV_PERSIST), SENDER );
		clientList.insert(std::pair<int,SERVE_CLIENT*>(remoteSocket,client));
		//std::cout << "Accepted client\n";
	}
//	client->add_send( (byte*)Buffer, sizeof(Buffer) );
}
/*--------------------------------------------------------------------*/
void SERVE::deleteClient( SERVE_CLIENT *client )
{
	std::cout << debug << "SERVE Disconnecting client" << print;
	::close(client->getFiledescriptor());
	event_del(client->getEvent());
	clientList.erase(client->getFiledescriptor());
	//std::cout << "clientList size: " << clientList.size() << std::endl;
	delete client;
}
/*--------------------------------------------------------------------*/
void SERVE::clientRcv( SERVE_CLIENT *client )
{
	int count = recv( client->getFiledescriptor(), this->buffer, DEFAULT_BUFFER_SIZE, 0);

	/* Client disconnected */
	if(count <= 0) {
		deleteClient( client );
		return;
	}
	std::cout << debug << "SERVE Received " << count << " bytes" << print;
	SERVE_MESSAGE * sm = new SERVE_MESSAGE( client->getFiledescriptor(), this->buffer, count );
	NOTIFIER::update( reinterpret_cast<void*>(sm) );
	delete sm;
}
/*--------------------------------------------------------------------*/
void SERVE::broadcastData( unsigned char *buffer, unsigned int len ){
	for (auto it=clientList.begin(); it!=clientList.end(); ++it){
			it->second->add_send( buffer, len );
	}	
}
/*--------------------------------------------------------------------*/
void SERVE::addSendDataToList(int sockfd, unsigned char *buffer, unsigned int len ){
	auto it = clientList.find( sockfd );
	if( it == clientList.end() ){
		// error
		std::cout << debug << "SERVOM can't send data because client not found in list" << print;
	}else{
		addSendDataToList( it->second, buffer, len );
	}
}
/*--------------------------------------------------------------------*/
void SERVE::addSendDataToList(SERVE_CLIENT *client, unsigned char *buffer, unsigned int len )
{
	client->add_send( buffer, len );
}
/*--------------------------------------------------------------------*/
int SERVE::getListenPort()
{
	return listenPort;
}
/*--------------------------------------------------------------------*/
void SERVE::tcp_server_write(int fd, byte * buf, int buflen){
	addSendDataToList(fd, buf, buflen);
}
/*--------------------------------------------------------------------*/
void SERVE::tcp_server_write(int fd, byte buf ){
	tcp_server_write(fd, &buf, 1);
}
/*--------------------------------------------------------------------*/
void SERVE::tcp_server_broadcast( byte * buf, int buflen ){
	broadcastData(buf, buflen);
}

SERVE_SEND * SERVE::SENDER = NULL;
int SERVE::ref_count = 0;
std::mutex SERVE::thread_lock;

    SERVE_SEND.h

    #ifndef SERVE_SENDH
#define SERVE_SENDH
#include <sys/types.h>
#include <sys/socket.h>
#include <thread>
#include <mutex>
#include <vector>
#include "../GLOBAL/GLOBAL.h"
#include "../LOG/LOG.h"

struct supply{
	byte* buffer;
	unsigned int len;
	int fd;
};

class SERVE_SEND{
public:
	SERVE_SEND();	
	~SERVE_SEND();
	void addSupplies( int fd, byte* buffer, unsigned int len );
	void addSupplies( int fd, byte* bufer, unsigned int len, int * clients, unsigned int clen );
	int getSupplyCount();

private:
	void sendSupplies();
	void clearSupplies();
	bool running;
	std::thread send_thread;
	std::mutex thread_lock;
	std::mutex supply_lock;
	std::vector<supply> supplies;
};
#endif

    SERVE_SEND.cpp

    #include "SERVE_SEND.h"

SERVE_SEND::SERVE_SEND()
:running(true){
	send_thread = std::thread( &SERVE_SEND::sendSupplies, this );
}

SERVE_SEND::~SERVE_SEND(){
	std::cout << debug << "SERVE_SEND deconstructor" << print;
	running = false;
	clearSupplies();
	thread_lock.unlock();
	send_thread.detach();
}

void SERVE_SEND::addSupplies( int fd, byte* buffer, unsigned int len ){
	std::cout << "SERVE_SEND Adding " << len << " bytes of supplies for client: " << fd << print;
	supply_lock.lock();
	struct supply s;
	s.fd = fd;
	s.buffer = new byte[len];
	std::copy( buffer, buffer+len, s.buffer );
	s.len = len;
	supplies.push_back(s);
	supply_lock.unlock();
	thread_lock.unlock();
}

void SERVE_SEND::addSupplies( int fd, byte* bufer, unsigned int len, int * clients, unsigned int clen ) {

}

int SERVE_SEND::getSupplyCount(){
	supply_lock.lock();
	int n = supplies.size();
	supply_lock.unlock();
	return n;
}

void SERVE_SEND::clearSupplies(){
	std::cout << debug << "SERVE_SEND clearing supplies" << print;
	supply_lock.lock();
	while( !supplies.empty() ){
		delete[] supplies.begin()->buffer;
		supplies.erase(supplies.begin());
	}
	supply_lock.unlock();
}

void SERVE_SEND::sendSupplies(){
	std::cout << debug << "SERVE_SEND sendSupplies thread running" << print;
	while( running ){
		thread_lock.lock();
		struct supply s;
		supply_lock.lock();
		if( !supplies.empty() ){
			thread_lock.unlock();
			s = *supplies.begin();
			supplies.erase(supplies.begin());
			supply_lock.unlock();
			std::cout << debug << "SERVE_SEND Sending " << s.len << " bytes to: " << s.fd << print;
			::send( s.fd, s.buffer, s.len, 0 );
			delete[] s.buffer;
		}else{
			supply_lock.unlock();
		}
	}
	std::cout << debug << "SERVE_SEND sendSupplies thread ending" << print;
}

    SERVE_CLIENT.h

    #ifndef SERVE_CLIENTH
#define SERVE_CLIENTH
#include <string>
#include <mutex>
#include <vector>
#include <thread>
#include <sys/types.h>
#include <sys/socket.h>
#include "../GLOBAL/GLOBAL.h"
#include "SERVE_SEND.h"

class SERVE_CLIENT 
{
public: 
	SERVE_CLIENT( int sockfd, std::string ip, struct event * e, SERVE_SEND * sender );
	virtual ~SERVE_CLIENT();

	int getFiledescriptor();
	struct event* getEvent();
	void add_send( byte* buffer, unsigned int len);
	std::string getIP();

private: 
	int sockfd;
	struct event * e;
	std::string ip;
	SERVE_SEND * sender;
};

#endif

    SERVE_CLIENT.cpp

    #ifndef SERVE_CLIENTH
#define SERVE_CLIENTH
#include <string>
#include <mutex>
#include <vector>
#include <thread>
#include <sys/types.h>
#include <sys/socket.h>
#include "../GLOBAL/GLOBAL.h"
#include "SERVE_SEND.h"

class SERVE_CLIENT 
{
public: 
	SERVE_CLIENT( int sockfd, std::string ip, struct event * e, SERVE_SEND * sender );
	virtual ~SERVE_CLIENT();

	int getFiledescriptor();
	struct event* getEvent();
	void add_send( byte* buffer, unsigned int len);
	std::string getIP();

private: 
	int sockfd;
	struct event * e;
	std::string ip;
	SERVE_SEND * sender;
};

#endif

    SERVE_MESSAGE.h

    #ifndef _SERVE_MESSAGEH_
#define _SERVE_MESSAGEH_
#include "../GLOBAL/GLOBAL.h"
#include <algorithm>
class SERVE_MESSAGE{
	public:
	SERVE_MESSAGE( int fd_sender, byte * msg_buf, unsigned int msg_len );
	~SERVE_MESSAGE();

	int getSender() const;
	byte* getMsgBuf() const;
	unsigned int getMsgLen() const;

	private:
	int fd_sender;
	byte* msg_buf;
	unsigned int msg_len;
};

#endif

    SERVE_MESSAGE.cpp

    #include "SERVE_MESSAGE.h"

SERVE_MESSAGE::SERVE_MESSAGE( int fd_sender, byte * msg_buf, unsigned int msg_len ){
	this->fd_sender = fd_sender;
	this->msg_buf = new byte[msg_len];
	this->msg_len = msg_len;
	std::copy( msg_buf, msg_buf + msg_len, this->msg_buf );
}

SERVE_MESSAGE::~SERVE_MESSAGE(){
	delete[] msg_buf;
}

int SERVE_MESSAGE::getSender() const {
	return fd_sender;
}

byte* SERVE_MESSAGE::getMsgBuf() const {
	return msg_buf;
}

unsigned int  SERVE_MESSAGE::getMsgLen() const {
	return msg_len;
}
    0
  • ?

    "eure Ideen von Verbesserungswürdigen zu geben"

    eure Ideen zur Verbessrung zu Geben :x

    0
  • ?

    deine klassennamen schreien alle

    0
  • ?

    Manuelles lock / unlock. Böööse: http://www.meetingcpp.com/index.php/br/items/anti-pattern-the-need-to-call-a-function.html

    0
Anmelden zum Antworten