Added support for session commands to readwrite split router.

Added support for multi-statement packets.

This is an intermediate commit to save work. Code is not cleaned and there are debug prints and prototypes to be removed.

server/core/buffer.c

@@ -128,10 +128,37 @@ GWBUF	*rval;
 	rval->start = buf->start;
 	rval->end = buf->end;
 	rval->next = NULL;
-	rval->command = buf->command;
+// 	rval->command = buf->command;
         CHK_GWBUF(rval);
 	return rval;
 }
+
+
+GWBUF *gwbuf_clone_portion(
+        GWBUF *buf,
+        size_t start_offset,
+        size_t length)
+{
+        GWBUF* clonebuf;
+        
+        CHK_GWBUF(buf);
+        ss_dassert(start_offset+length <= GWBUF_LENGTH(buf));
+        
+        if ((clonebuf = (GWBUF *)malloc(sizeof(GWBUF))) == NULL)
+        {
+                return NULL;
+        }
+        atomic_add(&buf->sbuf->refcount, 1);
+        clonebuf->sbuf = buf->sbuf;
+        clonebuf->start = (void *)((char*)buf->start)+start_offset;
+        clonebuf->end = (void *)((char *)clonebuf->start)+length;
+        clonebuf->next = NULL;
+        CHK_GWBUF(clonebuf);
+        return clonebuf;
+        
+}
+
+
 /**
  * Append a buffer onto a linked list of buffer structures.
  *

server/core/dcb.c

@@ -302,6 +302,8 @@ dcb_final_free(DCB *dcb)
 	if (dcb->remote)
 		free(dcb->remote);
 	bitmask_free(&dcb->memdata.bitmask);
+        simple_mutex_done(&dcb->dcb_read_lock);
+        simple_mutex_done(&dcb->dcb_write_lock);
 	free(dcb);
 }
 
@@ -520,6 +522,8 @@ int             rc;
          * Successfully connected to backend. Assign file descriptor to dcb
          */
         dcb->fd = fd;
+        /** Copy status field to DCB */
+        dcb->dcb_server_status = server->status;
 
 	/*<
 	 * backend_dcb is connected to backend server, and once backend_dcb
@@ -683,6 +687,15 @@ dcb_write(DCB *dcb, GWBUF *queue)
              dcb->state != DCB_STATE_LISTENING &&
              dcb->state != DCB_STATE_NOPOLLING))
         {
+                LOGIF(LD, (skygw_log_write(
+                        LOGFILE_DEBUG,
+                        "%lu [dcb_write] Write aborted to dcb %p because "
+                        "it is in state %s",
+                        pthread_self(),
+                        dcb->stats.n_buffered,
+                        dcb,
+                        STRDCBSTATE(dcb->state),
+                        dcb->fd)));
                 return 0;
         }
         
@@ -743,7 +756,11 @@ dcb_write(DCB *dcb, GWBUF *queue)
 #endif /* SS_DEBUG */
 			len = GWBUF_LENGTH(queue);
 			GW_NOINTR_CALL(
-                                w = gw_write(dcb->fd, GWBUF_DATA(queue), len);
+                                w = gw_write(
+#if defined(SS_DEBUG)
+                                        dcb,
+#endif
+                                        dcb->fd, GWBUF_DATA(queue), len);
                                 dcb->stats.n_writes++;
                                 );
                         
@@ -855,9 +872,13 @@ int saved_errno = 0;
 		while (dcb->writeq != NULL)
 		{
 			len = GWBUF_LENGTH(dcb->writeq);
-			GW_NOINTR_CALL(w = gw_write(dcb->fd,
+			GW_NOINTR_CALL(w = gw_write(
-                                                    GWBUF_DATA(dcb->writeq),
+#if defined(SS_DEBUG)
-                                                    len););
+                               dcb,
+#endif
+                               dcb->fd,
+                               GWBUF_DATA(dcb->writeq),
+                               len););
 			saved_errno = errno;
                         errno = 0;
                         
@@ -1312,12 +1333,15 @@ static bool dcb_set_state_nomutex(
 }
 
 int gw_write(
+#if defined(SS_DEBUG)
+        DCB* dcb,
+#endif
         int fd,
         const void* buf,
         size_t nbytes)
 {
         int w;
-#if defined(SS_DEBUG)
+#if defined(SS_DEBUG)                
         if (dcb_fake_write_errno[fd] != 0) {
                 ss_dassert(dcb_fake_write_ev[fd] != 0);
                 w = write(fd, buf, nbytes/2); /*< leave peer to read missing bytes */
@@ -1332,6 +1356,56 @@ int gw_write(
 #else
         w = write(fd, buf, nbytes);           
 #endif /* SS_DEBUG && SS_TEST */
+
+#if defined(SS_DEBUG)
+        {
+                size_t         len;
+                unsigned char* packet = (unsigned char *)buf;
+                char*          str;
+                
+                /** Print only MySQL packets */
+                if (w > 5)
+                {
+                        str = (char *)&packet[5];
+                        len      = packet[0];
+                        len     += 255*packet[1];
+                        len     += 255*255*packet[2];
+                                                
+                        if (strncmp(str, "insert", 6) == 0 ||
+                                strncmp(str, "create", 6) == 0 ||
+                                strncmp(str, "drop", 4) == 0)
+                        {
+                                ss_dassert((dcb->dcb_server_status & (SERVER_RUNNING|SERVER_MASTER|SERVER_SLAVE))==(SERVER_RUNNING|SERVER_MASTER));
+                        }
+                        
+                        if (strncmp(str, "set autocommit", 14) == 0 && nbytes > 17)
+                        {
+                                char* s = (char *)calloc(1, nbytes+1);
+                                
+                                if (nbytes-5 > len)
+                                {
+                                        size_t len2 = packet[4+len];
+                                        len2 += 255*packet[4+len+1];
+                                        len2 += 255*255*packet[4+len+2];
+                                        
+                                        char* str2 = (char *)&packet[4+len+5];
+                                        snprintf(s, 5+len+len2, "long %s %s", (char *)str, (char *)str2);
+                                }
+                                else
+                                {
+                                        snprintf(s, len, "%s", (char *)str);
+                                }
+                                LOGIF(LT, (skygw_log_write(
+                                        LOGFILE_TRACE,
+                                        "%lu [gw_write] Wrote %d bytes : %s ",
+                                        pthread_self(),
+                                        w,
+                                        s)));
+                                free(s);
+                        }
+                }
+        }
+#endif
         return w;
 }
 

server/include/buffer.h

@@ -92,6 +92,6 @@ extern GWBUF		*gwbuf_clone(GWBUF *buf);
 extern GWBUF		*gwbuf_append(GWBUF *head, GWBUF *tail);
 extern GWBUF		*gwbuf_consume(GWBUF *head, unsigned int length);
 extern unsigned int	gwbuf_length(GWBUF *head);
-
+extern GWBUF            *gwbuf_clone_portion(GWBUF *head, size_t offset, size_t len);
 
 #endif

server/include/dcb.h

@@ -87,6 +87,7 @@ typedef struct gw_protocol {
 	int		(*listen)(struct dcb *, char *);
 	int		(*auth)(struct dcb *, struct server *, struct session *, GWBUF *);
 	int		(*session)(struct dcb *, void *);
+        void*           (*getstmt)(void* buf);
 } GWPROTOCOL;
 
 /**
@@ -176,7 +177,7 @@ typedef struct dcb {
 	SPINLOCK	authlock;	/**< Generic Authorization spinlock */
 
 	DCBSTATS	stats;		/**< DCB related statistics */
-
+        unsigned int    dcb_server_status; /*< the server role indicator from SERVER */
 	struct dcb	*next;		/**< Next DCB in the chain of allocated DCB's */
 	struct service	*service;	/**< The related service */
 	void		*data;		/**< Specific client data */
@@ -202,7 +203,13 @@ int           fail_accept_errno;
 #define	DCB_ISZOMBIE(x)			((x)->state == DCB_STATE_ZOMBIE)
 
 DCB             *dcb_get_zombies(void);
-int             gw_write(int fd, const void* buf, size_t nbytes);
+int             gw_write(
+#if defined(SS_DEBUG)
+        DCB*        dcb,
+#endif
+        int         fd, 
+        const void* buf, 
+        size_t      nbytes);
 int             dcb_write(DCB *, GWBUF *);
 DCB             *dcb_alloc(dcb_role_t);
 void            dcb_free(DCB *);

server/include/gw.h

@@ -56,6 +56,12 @@ int  MySQLWrite(DCB *dcb, GWBUF *queue);
 int  gw_write_backend_event(DCB *dcb);
 int  gw_read_backend_event(DCB *dcb);
 int  setnonblocking(int fd);
-int  gw_write(int fd, const void* buf, size_t nbytes);
+int  gw_write(
+#if defined(SS_DEBUG)
+        DCB*        dcb,
+#endif
+        int         fd, 
+        const void* buf, 
+        size_t      nbytes);
 int  gw_getsockerrno(int fd);
 int  parse_bindconfig(char *, unsigned short, struct sockaddr_in *);

server/modules/include/readwritesplit.h

@@ -46,12 +46,19 @@ typedef struct router_client_session ROUTER_CLIENT_SES;
 
 typedef enum rses_property_type_t {
         RSES_PROP_TYPE_UNDEFINED=0,
-	RSES_PROP_TYPE_FIRST,
+        RSES_PROP_TYPE_SESCMD,
-	RSES_PROP_TYPE_SESCMD=RSES_PROP_TYPE_FIRST,
+        RSES_PROP_TYPE_FIRST = RSES_PROP_TYPE_SESCMD,
 	RSES_PROP_TYPE_LAST=RSES_PROP_TYPE_SESCMD,
 	RSES_PROP_TYPE_COUNT=RSES_PROP_TYPE_LAST+1
 } rses_property_type_t;
 
+typedef enum backend_type_t {
+        BE_UNDEFINED=-1, 
+        BE_MASTER, 
+        BE_SLAVE, 
+        BE_COUNT
+} backend_type_t;
+
 /**
  * Session variable command
  */
@@ -59,9 +66,9 @@ typedef struct mysql_sescmd_st {
 #if defined(SS_DEBUG)
         skygw_chk_t        my_sescmd_chk_top;
 #endif
-	ROUTER_CLIENT_SES* my_sescmd_rsession;   /*< parent router session */
+// 	ROUTER_CLIENT_SES* my_sescmd_rsession;   /*< parent router session */
 	rses_property_t*   my_sescmd_prop;       /*< parent property */
-        GWBUF*             my_sescmd_buf;        /*< client query reference */
+        GWBUF*             my_sescmd_buf;        /*< query buffer */
 	bool               my_sescmd_is_replied; /*< is cmd replied to client */
 #if defined(SS_DEBUG)
         skygw_chk_t        my_sescmd_chk_tail;
@@ -76,7 +83,8 @@ struct rses_property_st {
 #if defined(SS_DEBUG)
         skygw_chk_t          rses_prop_chk_top;
 #endif
-        SPINLOCK             rses_prop_lock; /*< protect property content */
+        ROUTER_CLIENT_SES*   rses_prop_rsession; /*< parent router session */
+//         SPINLOCK             rses_prop_lock;     /*< protect property content */
         int                  rses_prop_refcount;
         rses_property_type_t rses_prop_type;
         union rses_prop_data {
@@ -90,18 +98,13 @@ struct rses_property_st {
 };
 
 typedef struct sescmd_cursor_st {
-	rses_property_t** scmd_cur_ptr_property; /*< address of pointer to owner property */
+        ROUTER_CLIENT_SES* scmd_cur_rses;         /*< pointer to owning router session */
-	mysql_sescmd_t*   scmd_cur_cmd;          /*< pointer to current session command */
+	rses_property_t**  scmd_cur_ptr_property; /*< address of pointer to owner property */
-	bool              scmd_cur_active;       /*< true if command is being executed */
+	mysql_sescmd_t*    scmd_cur_cmd;          /*< pointer to current session command */
+	bool               scmd_cur_active;       /*< true if command is being executed */
+	backend_type_t     scmd_cur_be_type;      /*< BE_MASTER or BE_SLAVE */
 } sescmd_cursor_t;
 
-typedef enum backend_type_t {
-	BE_UNDEFINED=-1, 
-	BE_MASTER, 
-	BE_SLAVE, 
-	BE_COUNT
-} backend_type_t;
-
 /**
  * The client session structure used within this router.
  */

server/modules/protocol/mysql_backend.c

@@ -521,7 +521,7 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
         /*<
          * Don't write to backend if backend_dcb is not in poll set anymore.
          */
-	spinlock_acquire(&dcb->authlock);
+	spinlock_acquire(&dcb->dcb_initlock);
         if (dcb->state != DCB_STATE_POLLING) {
                 /*< vraa : errorHandle */
                 /*< Free buffer memory */
@@ -536,10 +536,12 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
                         dcb->fd,
                         STRDCBSTATE(dcb->state))));
 
-		spinlock_release(&dcb->authlock);
+		spinlock_release(&dcb->dcb_initlock);
                 return 0;
         }
-
+        spinlock_release(&dcb->dcb_initlock);
+        
+        spinlock_acquire(&dcb->authlock);
 	/*<
 	 * Now put the incoming data to the delay queue unless backend is
          * connected with auth ok
@@ -553,7 +555,6 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
                         dcb,
                         dcb->fd,
                         STRPROTOCOLSTATE(backend_protocol->state))));
-                
 		backend_set_delayqueue(dcb, queue);
 		spinlock_release(&dcb->authlock);
 		return 1;
@@ -562,9 +563,9 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
 	/*<
 	 * Now we set the last command received, from the current queue
 	 */
-        memcpy(&dcb->command, &queue->command, sizeof(dcb->command));
+//        memcpy(&dcb->command, &queue->command, sizeof(dcb->command));
-
 	spinlock_release(&dcb->authlock);
+//         LOGIF(LD, debuglog_statements(dcb, gwbuf_clone(queue)));
         rc = dcb_write(dcb, queue);
 	return rc;
 }
@@ -805,7 +806,7 @@ static int backend_write_delayqueue(DCB *dcb)
 	 * Now we set the last command received, from the delayed queue
 	 */
 
-        memcpy(&dcb->command, &localq->command, sizeof(dcb->command));
+//        memcpy(&dcb->command, &localq->command, sizeof(dcb->command));
 
 	spinlock_release(&dcb->delayqlock);
         rc = dcb_write(dcb, localq);
@@ -911,7 +912,6 @@ static int gw_change_user(DCB *backend, SERVER *server, SESSION *in_session, GWB
 		strcpy(current_session->user, username);
 		strcpy(current_session->db, database);
 		memcpy(current_session->client_sha1, client_sha1, sizeof(current_session->client_sha1));
-
 	}
 	
 	// consume all the data received from client

server/modules/protocol/mysql_client.c

@@ -48,6 +48,7 @@ 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 void* gw_MySQL_get_next_stmt(void* buffer);
 
 int mysql_send_ok(DCB *dcb, int packet_number, int in_affected_rows, const char* mysql_message);
 int MySQLSendHandshake(DCB* dcb);
@@ -67,8 +68,9 @@ static GWPROTOCOL MyObject = {
 	gw_client_close,			/* Close			 */
 	gw_MySQLListener,			/* Listen			 */
 	NULL,					/* Authentication		 */
-	NULL					/* Session			 */
+	NULL,					/* Session			 */
-	};
+        gw_MySQL_get_next_stmt                  /* get single stmt from read buf */
+};
 
 /**
  * Implementation of the mandatory version entry point
@@ -607,8 +609,7 @@ int gw_read_client_event(DCB* dcb) {
                  */
         {
                 int      len = -1;
-                GWBUF   *queue = NULL;
+                GWBUF   *read_buffer = NULL;
-                GWBUF   *gw_buffer = NULL;
                 uint8_t *ptr_buff = NULL;
                 int      mysql_command = -1;
                 
@@ -626,16 +627,15 @@ int gw_read_client_event(DCB* dcb) {
                 //////////////////////////////////////////////////////
                 // read and handle errors & close, or return if busy
                 //////////////////////////////////////////////////////
-                rc = gw_read_gwbuff(dcb, &gw_buffer, b); 
+                rc = gw_read_gwbuff(dcb, &read_buffer, b); 
                 
                 if (rc != 0) {
                         goto return_rc;
                 }
                 /* Now, we are assuming in the first buffer there is
                  * the information form mysql command */
-                queue = gw_buffer;
+                len = GWBUF_LENGTH(read_buffer);
-                len = GWBUF_LENGTH(queue);
+                ptr_buff = GWBUF_DATA(read_buffer);
-                ptr_buff = GWBUF_DATA(queue);
                 
                 /* get mysql commang at fifth byte */
                 if (ptr_buff) {
@@ -669,12 +669,12 @@ int gw_read_client_event(DCB* dcb) {
                         }
                         rc = 1;
                         /** Free buffer */
-                        queue = gwbuf_consume(queue, len);                
+                        read_buffer = gwbuf_consume(read_buffer, len);                
                         goto return_rc;
                 }
                 /** Route COM_QUIT to backend */
                 if (mysql_command == '\x01') {
-                        router->routeQuery(router_instance, rsession, queue);
+                        router->routeQuery(router_instance, rsession, read_buffer);
                         LOGIF(LD, (skygw_log_write_flush(
                                 LOGFILE_DEBUG,
                                 "%lu [gw_read_client_event] Routed COM_QUIT to "
@@ -693,7 +693,7 @@ int gw_read_client_event(DCB* dcb) {
                         /** Route other commands to backend */
                         rc = router->routeQuery(router_instance,
                                                 rsession,
-                                                queue);
+                                                read_buffer);
                         /** succeed */
                         if (rc == 1) {
                                 rc = 0; /**< here '0' means success */
@@ -1203,3 +1203,47 @@ gw_client_hangup_event(DCB *dcb)
 return_rc:
 	return rc;
 }
+
+/**
+ * Remove the first mysql statement from buffer. Return pointer to the removed
+ * statement or NULL if buffer is empty.
+ * 
+ * Clone buf, calculate the length of included mysql stmt, and point the 
+ * statement with cloned buffer. Move the start pointer of buf accordingly
+ * so that it only cover the remaining buffer.
+ * 
+ */
+static void* gw_MySQL_get_next_stmt(
+        void* buffer)
+{
+        GWBUF*         readbuf = (GWBUF *)buffer;
+        GWBUF*         stmtbuf;
+        unsigned char* packet;
+        size_t         len;
+        
+        CHK_GWBUF(readbuf);
+        
+        if (GWBUF_EMPTY(readbuf))
+        {
+                stmtbuf = NULL;
+                goto return_stmtbuf;
+        }        
+        packet = GWBUF_DATA(readbuf);
+        len      = packet[0];
+        len     += 255*packet[1];
+        len     += 255*255*packet[2];
+        
+        /** vraa :Multi-packet stmt is not supported as of 7.3.14 */
+        if (len+4 > GWBUF_LENGTH(readbuf))
+        {
+                stmtbuf = NULL;
+                goto return_stmtbuf;
+        }
+        stmtbuf = gwbuf_clone_portion(readbuf, 0, 4+len);
+        gwbuf_consume(readbuf, 4+len);
+        
+return_stmtbuf:
+        return (void *)stmtbuf;
+}
+
+

utils/skygw_debug.h

@@ -216,7 +216,11 @@ typedef enum skygw_chk_t {
                        ((r) == DCB_ROLE_REQUEST_HANDLER ? "DCB_ROLE_REQUEST_HANDLER" : \
                         "UNKNOWN DCB ROLE"))
 
-
+#define STRBETYPE(t) ((t) == BE_MASTER ? "BE_MASTER" : \
+                        ((t) == BE_SLAVE ? "BE_SLAVE" : \
+                        ((t) == BE_UNDEFINED ? "BE_UNDEFINED" : \
+                        "Unknown backend tpe")))
+                        
 #define CHK_MLIST(l) {                                                  \
             ss_info_dassert((l->mlist_chk_top ==  CHK_NUM_MLIST &&      \
                              l->mlist_chk_tail == CHK_NUM_MLIST),       \