I hope that I don't break anything. Did some changes to Makefiles and sources so that liblog_manager.so can be linked-in and used.

core/Makefile

@@ -63,9 +63,7 @@ clean:
 tags:
 	ctags $(SRCS) $(HDRS)
 
-depend:	depend.mk
+depend:	
-
-depend.mk: $(SRCS)
 	@rm -f depend.mk
 	cc -M $(CFLAGS) $(SRCS) > depend.mk
 

core/poll.c

@@ -25,7 +25,7 @@
 #include <atomic.h>
 
 /**
- * @file poll.c  - Abraction of the epoll functionality
+ * @file poll.c  - Abstraction of the epoll functionality
  *
  * @verbatim
  * Revision History
@@ -104,11 +104,28 @@ struct	epoll_event	ev;
 	return epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dcb->fd, &ev);
 }
 
+#define	BLOCKINGPOLL	0	/* Set BLOCKING POLL to 1 if using a single thread and to make
+				 * debugging easier.
+				 */
 /**
  * The main polling loop
  *
  * This routine does the polling and despatches of IO events
  * to the DCB's
+ *
+ * The routine will loop as long as the variable "shutdown" is set to zero,
+ * setting this to a non-zero value will cause the polling loop to return.
+ *
+ * There are two options for the polling, a debug option that is only useful if
+ * you have a single thread. This blocks in epoll_wait until an event occurs.
+ *
+ * The non-debug option does an epoll with a time out. This allows the checking of
+ * shutdown value to be checked in all threads. The algorithm for polling in this
+ * mode is to do a poll with no-wait, if no events are detected then the poll is
+ * repeated with a time out. This allows for a quick check before making the call 
+ * with timeout. The call with the timeout differs in that the Linux scheduler may
+ * deschedule a process if a timeout is included, but will not do this if a 0 timeout
+ * value is given. this improves performance when the gateway is under heavy load.
  */
 void
 poll_waitevents()
@@ -118,6 +135,11 @@ int			i, nfds;
 
 	while (1)
 	{
+#if BLOCKINGPOLL
+		if ((nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1)) == -1)
+		{
+		}
+#else
 		if ((nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, 0)) == -1)
 		{
 		}
@@ -127,6 +149,7 @@ int			i, nfds;
 			{
 			}
 		}
+#endif
 		if (nfds > 0)
 		{
 			atomic_add(&pollStats.n_polls, 1);

modules/protocol/Makefile

@@ -63,9 +63,7 @@ tags:
 install: $(MODULES)
 	install -D $< $(DEST)/gateway/modules
 
-depend: depend.mk
+depend: 
-
-depend.mk: $(SRCS)
 	rm -f depend.mk
 	cc -M $(CFLAGS) $(SRCS) > depend.mk
 

modules/protocol/mysql_backend.c

@@ -382,6 +382,9 @@ int gw_mysql_connect(char *host, int port, char *dbname, char *user, uint8_t *pa
 	conn->state = MYSQL_ALLOC;
 	conn->fd = -1;
 
+	memset(&server_capabilities, '\0', sizeof(server_capabilities));
+	memset(&final_capabilities, '\0', sizeof(final_capabilities));
+
 #ifdef MYSQL_CONN_DEBUG
 	//fprintf(stderr, ")))) Connect to MySQL: user[%s], SHA1(passwd)[%s], db [%s]\n", user, passwd, dbname);
 #endif

modules/protocol/mysql_client.c

@@ -40,6 +40,7 @@ static int gw_write_client_event(DCB *dcb);
 static int gw_MySQLWrite_client(DCB *dcb, GWBUF *queue);
 static int gw_error_client_event(DCB *dcb);
 static int gw_client_close(DCB *dcb);
+static int gw_client_hangup_event(DCB *dcb);
 
 static int gw_check_mysql_scramble_data(DCB *dcb, uint8_t *token, unsigned int token_len, uint8_t *scramble, unsigned int scramble_len, char *username, uint8_t *stage1_hash);
 static int gw_find_mysql_user_password_sha1(char *username, uint8_t *gateway_password, void *repository);
@@ -56,7 +57,7 @@ static GWPROTOCOL MyObject = {
 	gw_MySQLWrite_client,			/* Write - data from gateway	 */
 	gw_write_client_event,			/* WriteReady - EPOLLOUT handler */
 	gw_error_client_event,			/* Error - EPOLLERR handler	 */
-	NULL,					/* HangUp - EPOLLHUP handler	 */
+	gw_client_hangup_event,			/* HangUp - EPOLLHUP handler	 */
 	gw_MySQLAccept,				/* Accept			 */
 	NULL,					/* Connect			 */
 	gw_client_close,			/* Close			 */
@@ -773,14 +774,12 @@ int gw_read_client_event(DCB* dcb) {
 	ROUTER          *router_instance = NULL;
 	void            *rsession = NULL;
 	MySQLProtocol *protocol = NULL;
-	//uint8_t		buffer[MAX_BUFFER_SIZE] = "";
 	int b = -1;
 
 	if (dcb) {
 		protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
 	}
 
-
 	if (ioctl(dcb->fd, FIONREAD, &b)) {
 		fprintf(stderr, "Client Ioctl FIONREAD error %i, %s\n", errno , strerror(errno));
 		return 1;
@@ -823,11 +822,33 @@ int gw_read_client_event(DCB* dcb) {
 				queue = gwbuf_consume(queue, len);
 
 				if (auth_val == 0)
+				{
+					SESSION *session = NULL;
+
 					protocol->state = MYSQL_AUTH_RECV;
+
+					//write to client mysql AUTH_OK packet, packet n. is 2
+					mysql_send_ok(dcb, 2, 0, NULL);
+
+					// start a new session, and connect to backends
+					session = session_alloc(dcb->service, dcb);
+
+					protocol->state = MYSQL_IDLE;
+
+					session->data = (MYSQL_session *)dcb->data;
+				}
 				else 
+				{
 					protocol->state = MYSQL_AUTH_FAILED;
+
+					// still to implement
+					mysql_send_auth_error(dcb, 2, 0, "Authorization failed");
+
+					dcb->func.close(dcb);
+				}
 			}
 
+
 			break;
 
 		case MYSQL_IDLE:
@@ -952,31 +973,6 @@ int gw_write_client_event(DCB *dcb) {
 		return 1;
 	}
 
-	if(protocol->state == MYSQL_AUTH_RECV) {
-		SESSION *session = NULL;
-
-		//write to client mysql AUTH_OK packet, packet n. is 2
-		mysql_send_ok(dcb, 2, 0, NULL);
-
-		// start a new session, and connect to backends
-		session = session_alloc(dcb->service, dcb);
-
-        	protocol->state = MYSQL_IDLE;
-
-		session->data = (MYSQL_session *)dcb->data;
-
-		return 0;
-	}
-
-	if (protocol->state == MYSQL_AUTH_FAILED) {
-		// still to implement
-		mysql_send_auth_error(dcb, 2, 0, "Authorization failed");
-
-		dcb->func.close(dcb);
-
-		return 0;
-	}
-
 	if ((protocol->state == MYSQL_IDLE) || (protocol->state == MYSQL_WAITING_RESULT)) {
 		int w;
 
@@ -1162,3 +1158,18 @@ gw_client_close(DCB *dcb)
         dcb_close(dcb);
 	return 1;
 }
+
+/**
+ * Handle a hangup event on the client side descriptor.
+ *
+ * We simply close the DCB, this will propogate the closure to any
+ * backend descriptors and perform the session cleanup.
+ *
+ * @param dcb		The DCB of the connection
+ */
+static int
+gw_client_hangup_event(DCB *dcb)
+{
+        dcb_close(dcb);
+	return 1;
+}

modules/routing/Makefile

@@ -20,6 +20,8 @@
 # 27/06/13      Vilho Raatikka          Added logmanager-related libs and 
 #                                       headers so that liblog_manager.so can 
 #                                       be linked in.
+# 27/06/13	Mark Riddoch		Addition of read write splitter
+
 include ../../../build_gateway.inc
 LOGPATH := $(ROOT_PATH)/log_manager
 UTILSPATH := $(ROOT_PATH)/utils
@@ -37,7 +39,9 @@ DEBUGCLIOBJ=$(DEBUGCLISRCS:.c=.o)
 SRCS=$(TESTSRCS) $(READCONSRCS) $(DEBUGCLISRCS)
 OBJ=$(SRCS:.c=.o)
 LIBS=$(UTILSPATH)/skygw_utils.o -lssl -llog_manager
-MODULES=libtestroute.so libreadconnroute.so libdebugcli.so
+MODULES=libtestroute.so libreadconnroute.so libdebugcli.so \
+	libreadwritesplit.so
+
 
 all:	$(MODULES)
 
@@ -50,20 +54,24 @@ libreadconnroute.so: $(READCONOBJ)
 libdebugcli.so: $(DEBUGCLIOBJ)
 	$(CC) $(LDFLAGS) $(DEBUGCLIOBJ) $(LIBS) -o $@
 
+libreadwritesplit.so:
+	(cd readwritesplit; make; cp $@ ..)
+
 .c.o:
 	$(CC) $(CFLAGS) $< -o $@
 
 clean:
 	rm -f $(OBJ) $(MODULES)
+	(cd readwritesplit; make clean)
 
 tags:
 	ctags $(SRCS) $(HDRS)
+	(cd readwritesplit; make tags)
 
-depend:	depend.mk
+depend:
-
-depend.mk: $(SRCS)
 	@rm -f depend.mk
 	cc -M $(CFLAGS) $(SRCS) > depend.mk
+	(cd readwritesplit; make depend)
 
 install: $(MODULES)
 	install -D $< $(DEST)/gateway/modules

modules/routing/debugcli.c

@@ -101,6 +101,7 @@ GetModuleObject()
  * within the gateway.
  * 
  * @param service	The service this router is being create for
+ * @param options	Any array of options for the query router
  *
  * @return The instance data for this new instance
  */

modules/routing/debugcmd.c

@@ -166,7 +166,8 @@ static struct {
  * Convert a string argument to a numeric, observing prefixes
  * for number bases, e.g. 0x for hex, 0 for octal
  *
- * @param arg	The string representation of the argument
+ * @param arg		The string representation of the argument
+ * @param arg_type	The target type for the argument
  * @return The argument as a long integer
  */
 static unsigned long

modules/routing/readconnroute.c

@@ -128,6 +128,7 @@ GetModuleObject()
  * within the gateway.
  * 
  * @param service	The service this router is being create for
+ * @param options	An array of options for this query router
  *
  * @return The instance data for this new instance
  */

modules/routing/readwritesplit/Makefile

@@ -0,0 +1,50 @@
+# This file is distributed as part of the SkySQL Gateway.  It is free
+# software: you can redistribute it and/or modify it under the terms of the
+# GNU General Public License as published by the Free Software Foundation,
+# version 2.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
+# details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program; if not, write to the Free Software Foundation, Inc., 51
+# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+#
+# Copyright SkySQL Ab 2013
+#
+# Revision History
+# Date		Who			Description
+# 27/06/13	Mark Riddoch		Initial framework put in place
+
+CC=cc
+CFLAGS=-c -fPIC -I/usr/include -I../../include -I../../../include -Wall -g
+LDFLAGS=-shared
+SRCS=router.c
+OBJ=$(SRCS:.c=.o)
+LIBS=-lssl
+MODULES=libreadwritesplit.so
+
+all:	$(MODULES)
+
+libreadwritesplit.so: $(OBJ)
+	$(CC) $(LDFLAGS) $(OBJ) $(LIBS) -o $@
+
+.c.o:
+	$(CC) $(CFLAGS) $< -o $@
+
+clean:
+	rm -f $(OBJ) $(MODULES)
+
+tags:
+	ctags $(SRCS) $(HDRS)
+
+depend:
+	@rm -f depend.mk
+	cc -M $(CFLAGS) $(SRCS) > depend.mk
+
+install: $(MODULES)
+	install -D $< $(DEST)/gateway/modules
+
+include depend.mk

modules/routing/readwritesplit/router.c

@@ -0,0 +1,160 @@
+/*
+ * This file is distributed as part of the SkySQL Gateway.  It is free
+ * software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation,
+ * version 2.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
+ * details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * this program; if not, write to the Free Software Foundation, Inc., 51
+ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Copyright SkySQL Ab 2013
+ */
+#include <stdio.h>
+#include <router.h>
+
+/**
+ * @file router.c	The entry points for the read/write query splitting
+ * router module.
+ *
+ * This file contains the entry points that comprise the API to the read write
+ * query splitting router.
+ *
+ */
+static char *version_str = "V1.0.0";
+
+static	ROUTER	*createInstance(SERVICE *service, char **options);
+static	void	*newSession(ROUTER *instance, SESSION *session);
+static	void 	closeSession(ROUTER *instance, void *session);
+static	int	routeQuery(ROUTER *instance, void *session, GWBUF *queue);
+static	void	diagnostic(ROUTER *instance, DCB *dcb);
+
+static ROUTER_OBJECT MyObject = { createInstance, newSession, closeSession, routeQuery, diagnostic };
+
+/**
+ * Implementation of the mandatory version entry point
+ *
+ * @return version string of the module
+ */
+char *
+version()
+{
+	return version_str;
+}
+
+/**
+ * The module initialisation routine, called when the module
+ * is first loaded.
+ */
+void
+ModuleInit()
+{
+	fprintf(stderr, "Initialse read/writer splitting query router module.\n");
+}
+
+/**
+ * The module entry point routine. It is this routine that
+ * must populate the structure that is referred to as the
+ * "module object", this is a structure with the set of
+ * external entry points for this module.
+ *
+ * @return The module object
+ */
+ROUTER_OBJECT *
+GetModuleObject()
+{
+	fprintf(stderr, "Returing test router module object.\n");
+	return &MyObject;
+}
+
+/**
+ * Create an instance of the router for a particular service
+ * within the gateway.
+ *
+ * The job of ths entry point is to create the service wide data needed
+ * for the query router. This is information needed to route queries that
+ * is not related to any individual client session, exmaples of data that
+ * might be stored in the ROUTER object for a particular query router are
+ * connections counts, last used connection etc so that balancing may
+ * take place.
+ * 
+ * @param service	The service this router is being create for
+ * @param options	The options for this query router
+ *
+ * @return The instance data for this new instance
+ */
+static	ROUTER	*
+createInstance(SERVICE *service, char **options)
+{
+	return NULL;
+}
+
+/**
+ * Associate a new session with this instance of the router.
+ *
+ * The session is used to store all the data required for a particular
+ * client connection.
+ *
+ * @param instance	The router instance data
+ * @param session	The session itself
+ * @return Session specific data for this session
+ */
+static	void	*
+newSession(ROUTER *instance, SESSION *session)
+{
+	return NULL;
+}
+
+/**
+ * Close a session with the router, this is the mechanism
+ * by which a router may cleanup data structure etc.
+ *
+ * @param instance	The router instance data
+ * @param session	The session being closed
+ */
+static	void 	
+closeSession(ROUTER *instance, void *session)
+{
+}
+
+
+/**
+ * The main routing entry, this is called with every packet that is
+ * received and has to be forwarded to the backend database.
+ *
+ * The routeQuery will make the routing decision based on the contents
+ * of the instance, session and the query itself in the queue. The
+ * data in the queue may not represent a complete query, it represents
+ * the data that has been received. The query router itself is responsible
+ * for buffering the partial query, a later call to the query router will
+ * contain the remainder, or part thereof of the query.
+ *
+ * @param instance	The query router instance
+ * @param session	The session assoicated with the client
+ * @param queue		Gateway buffer queue with the packets received
+ *
+ * @return The number of queries forwarded
+ */
+static	int	
+routeQuery(ROUTER *instance, void *session, GWBUF *queue)
+{
+	return 0;
+}
+
+/**
+ * Diagnostics routine
+ *
+ * Print query router statistics to the DCB passed in
+ *
+ * @param	instance	The router instance
+ * @param	dcb		The DCB for diagnostic output
+ */
+static	void
+diagnostic(ROUTER *instance, DCB *dcb)
+{
+}

modules/routing/testroute.c

@@ -69,6 +69,7 @@ GetModuleObject()
  * within the gateway.
  * 
  * @param service	The service this router is being create for
+ * @param options	The options for this query router
  *
  * @return The instance data for this new instance
  */