proxysql/lib/mysql_thread.cpp_

#include "proxysql.h"
#include "cpp.h"


#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */
void mypoll_add(proxy_poll_t *, uint32_t, int, MySQL_Data_Stream *);
void __mypoll_add(proxy_poll_t *, struct pollfd *, MySQL_Data_Stream *); 
int mypoll_del(proxy_poll_t *_myp, int i);
#ifdef __cplusplus
}
#endif /* __cplusplus */






void mypoll_add(proxy_poll_t *_myp, uint32_t _events, int _fd, MySQL_Data_Stream *_myds) {
	struct pollfd _pollfd;
	_pollfd.revents=0;
	_pollfd.events=_events;
	_pollfd.fd=_fd;
	__mypoll_add(_myp, &_pollfd, _myds);
}

void __mypoll_add(proxy_poll_t *_myp, struct pollfd *_fd, MySQL_Data_Stream *_myds) {
	proxy_debug(PROXY_DEBUG_NET,1,"Session=%p, DataStream=%p -- Adding MySQL Data Stream %p for FD %d\n", _myds->sess, _myds, _myds, _fd->fd);
	if (_myp->nfds==_myp->max_nfds) {  // no space left, expand
		_myp->max_nfds+=MIN_POLL_FDS_PER_THREAD;
		struct pollfd *fds_tmp=(struct pollfd *)malloc(sizeof(struct pollfd)*_myp->max_nfds);
		assert(fds_tmp);
		MySQL_Data_Stream **myds_tmp=(MySQL_Data_Stream **)malloc(sizeof(MySQL_Data_Stream *)*_myp->max_nfds);
		assert(myds_tmp);
		memcpy(fds_tmp,_myp->fds,sizeof(struct pollfd)*_myp->nfds);
		memcpy(myds_tmp,_myp->myds,sizeof(MySQL_Data_Stream *)*_myp->nfds);
		free(_myp->fds);
		free(_myp->myds);
		_myp->fds=fds_tmp;
		_myp->myds=myds_tmp;
	}

//	memcpy(&(_myp->fds[_myp->nfds]),&_fd,sizeof(struct pollfd));
	_myp->fds[_myp->nfds].events=_fd->events;
	_myp->fds[_myp->nfds].revents=_fd->revents;
	_myp->fds[_myp->nfds].fd=_fd->fd;

	_myp->myds[_myp->nfds]=_myds;
	_myds->poll_fds_idx=_myp->nfds;	// pointer back
	_myp->nfds+=1;
}

// return if data was moved or not
int mypoll_del(proxy_poll_t *_myp, int i) {
	if (i >= _myp->nfds) return 0;
	if (i < _myp->nfds-1) {
		_myp->nfds--;
		memcpy(&_myp->fds[i],&_myp->fds[_myp->nfds],sizeof(struct pollfd));
		_myp->myds[i]=_myp->myds[_myp->nfds];
		_myp->myds[i]->poll_fds_idx=i;
		return 1;
	}
	_myp->nfds--;
	return 0;
}

MySQL_Thread::MySQL_Thread() {
	//events=NULL;
	mypolls.fds=NULL;
	mypolls.myds=NULL;
	mysql_sessions=NULL;
}


MySQL_Thread::~MySQL_Thread() {
	if (mypolls.fds)
		free(mypolls.fds);
	if (mypolls.myds)
		free(mypolls.myds);
//	destroy_all_sessions(mysql_sessions);

        while(mysql_sessions->len) {
            MySQL_Session *sess=(MySQL_Session *)mysql_sessions->remove_index_fast(0);
            delete sess;
        }
        delete mysql_sessions;


}

void MySQL_Thread::init() {
	mysql_sessions = new PtrArray();
	assert(mysql_sessions);
	shutdown=0;
	init_poll();
}


void MySQL_Thread::init_poll() {
	mypolls.max_nfds = MIN_POLL_FDS_PER_THREAD;
	mypolls.fds=(struct pollfd *)malloc(sizeof(struct pollfd)*mypolls.max_nfds);
	assert(mypolls.fds);
	mypolls.myds=(MySQL_Data_Stream **)malloc(sizeof(MySQL_Data_Stream *)*mypolls.max_nfds);
	assert(mypolls.myds);
	mypolls.nfds=0;
}

void MySQL_Thread::poll_listener_add(int sock) {
	MySQL_Data_Stream *listener_DS = new MySQL_Data_Stream;	
	listener_DS->myds_type=MYDS_LISTENER;
	listener_DS->fd=sock;

	proxy_debug(PROXY_DEBUG_NET,1,"Created listener %p for socket %d\n", listener_DS, sock);
	mypoll_add(&mypolls, POLLIN, sock, listener_DS);	
}


void MySQL_Thread::register_session(MySQL_Session *_sess) {
	mysql_sessions->add(_sess);
	_sess->thread=this;
	proxy_debug(PROXY_DEBUG_NET,1,"Thread=%p, Session=%p -- Registered new session\n", _sess->thread, _sess);
}

void MySQL_Thread::unregister_session(int idx) {
	proxy_debug(PROXY_DEBUG_NET,1,"Thread=%p, Session=%p -- Unregistered session\n", this, mysql_sessions->index(idx));
	mysql_sessions->remove_index_fast(idx);
}


MySQL_Session * MySQL_Thread::create_new_session_and_client_data_stream(int _fd) {
	int arg_on=1;
	MySQL_Session *sess=new MySQL_Session;
	register_session(sess); // register session
	sess->client_fd=_fd;
	setsockopt(sess->client_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &arg_on, sizeof(int));
	sess->client_myds = new MySQL_Data_Stream();
	sess->client_myds->init(MYDS_FRONTEND, sess, sess->client_fd);
	proxy_debug(PROXY_DEBUG_NET,1,"Thread=%p, Session=%p, DataStream=%p -- Created new client Data Stream\n", sess->thread, sess, sess->client_myds);
	return sess;
}

// main loop
void MySQL_Thread::run() {
	int n, rc;
	//int arg_on=1;


	while (shutdown==0) {


		rc=poll(mypolls.fds,mypolls.nfds,10000);
			if (rc == -1 && errno == EINTR)
				// poll() timeout, try again
				continue;
			if (rc == -1) {
				// error , exit
				perror("poll()");
				exit(EXIT_FAILURE);
			}



		for (n = 0; n < mypolls.nfds; n++) {
			if (mypolls.fds[n].revents==0) continue;
			//MySQL_Data_Stream *myds=(MySQL_Data_Stream *)events[n].data.ptr;
			MySQL_Data_Stream *myds=mypolls.myds[n];
			if (myds->myds_type==MYDS_LISTENER) {
				// we got a new connection!
				int c=accept(myds->fd, NULL, NULL);
				if (c>-1) { // accept() succeeded

					// create a new client connection
					mypolls.fds[n].revents=0;
					MySQL_Session *sess=create_new_session_and_client_data_stream(c);

					// connect to the server (FIXME : hardcoded for now)
//					sess->server_fd=connect_socket((char *)"127.0.0.1", 3306);
					//setsockopt(sess->server_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &arg_on, sizeof(int));
					sess->server_myds = new MySQL_Data_Stream();
					sess->server_fd=sess->server_myds->myds_connect((char *)"127.0.0.1", 3306);
					//if (sess->server_fd==-1) {
					//	delete sess->server_myds;
					//	delete sess->client_myds;
					//	delete sess;
					//}
					sess->server_myds->init(MYDS_BACKEND, sess, sess->server_fd);
					proxy_debug(PROXY_DEBUG_NET,1,"Session=%p, DataStream=%p -- Created new server Data Stream\n", sess, sess->server_myds);

					ioctl_FIONBIO(sess->client_fd, 1);
					//ioctl_FIONBIO(sess->server_fd, 1);

					mypoll_add(&mypolls, POLLIN|POLLOUT, sess->client_fd, sess->client_myds);
					mypoll_add(&mypolls, POLLIN|POLLOUT, sess->server_fd, sess->server_myds);
					proxy_debug(PROXY_DEBUG_NET,1,"Session=%p -- Adding client FD %d and server FD %d\n", sess, sess->client_fd, sess->server_fd);

				}
			// if we arrive here, accept() failed
				continue;
			} else {
				// data on exiting connection
				MySQL_Session *sess=myds->sess;	
				myds->revents=mypolls.fds[n].revents;
				myds->read_from_net();
				myds->read_pkts();

				myds->check_data_flow();

				myds->write_to_net_poll();
				//sess->inactive_handler();

				myds->sess->to_process=1;


	      if (myds->active==FALSE) {
					rc=mypoll_del(&mypolls, n);
					if (rc==1) n--;
					proxy_debug(PROXY_DEBUG_NET,1, "Session=%p, DataStream=%p -- Deleting FD %d\n", myds->sess, myds, myds->fd);
					myds->shut_hard();
					if (sess->client_myds==myds) sess->client_myds=NULL;
					if (sess->server_myds==myds) sess->server_myds=NULL;
					delete myds;
					myds=NULL;
  	   		if (sess->client_myds==NULL && sess->server_myds==NULL) {
						mysql_sessions->remove_fast(sess);
						delete sess;
						continue;
					}
  	    }






			}
	      // always move pkts from queue to evbuffer
			//	sess->writeout();
			//if (myds) myds->write_to_net_poll();
		}

		// iterate through all sessions and process the session logic
		for (n=0; n<(int)mysql_sessions->len; n++) {
			MySQL_Session *sess=(MySQL_Session *)mysql_sessions->index(n);
			if (sess->to_process==1) {
				rc=sess->handler();
				if (rc==-1) {
					unregister_session(n);
					n--;
				}
			}
		}

	}
}