Merge branch 'release-1.0beta-refresh' into blr

server/core/dcb.c

@@ -404,17 +404,22 @@ DCB_CALLBACK		*cb;
  * the memdata.bitmask then the DCB is no longer able to be 
  * referenced and it can be finally removed.
  *
+ * The excluded DCB allows a thread to exclude a DCB from zombie processing.
+ * It is used when a thread calls dcb_process_zombies when there is
+ * a DCB that the caller knows it will continue processing with.
+ *
  * @param	threadid	The thread ID of the caller
+ * @param	excluded		The DCB the thread currently uses, NULL or valid DCB.
  */
 DCB *
-dcb_process_zombies(int threadid)
+dcb_process_zombies(int threadid, DCB *excluded)
 {
 DCB	*ptr, *lptr;
 DCB*    dcb_list = NULL;
 DCB*    dcb = NULL;
 bool    succp = false;
 
-	/*<
+	/**
 	 * Perform a dirty read to see if there is anything in the queue.
 	 * This avoids threads hitting the queue spinlock when the queue 
 	 * is empty. This will really help when the only entry is being
@@ -424,24 +429,47 @@ bool    succp = false;
 	if (!zombies)
 		return NULL;
 
+	/*
+	 * Process the zombie queue and create a list of DCB's that can be
+	 * finally freed. This processing is down under a spinlock that
+	 * will prevent new entries being added to the zombie queue. Therefore
+	 * we do not want to do any expensive operations under this spinlock
+	 * as it will block other threads. The expensive operations will be
+	 * performed on the victim queue within holding the zombie queue
+	 * spinlock.
+	 */
 	spinlock_acquire(&zombiespin);
 	ptr = zombies;
 	lptr = NULL;
 	while (ptr)
 	{
+		CHK_DCB(ptr);
+
+		/*
+		 * Skip processing of the excluded DCB
+		 */
+		if (ptr == excluded)
+		{
+			lptr = ptr;
+			ptr = ptr->memdata.next;
+		}
+		else
+		{
+
 			bitmask_clear(&ptr->memdata.bitmask, threadid);
+			
 			if (bitmask_isallclear(&ptr->memdata.bitmask))
 			{
-			/*<
+				/**
 				 * Remove the DCB from the zombie queue
 				 * and call the final free routine for the
 				 * DCB
 				 *
 				 * ptr is the DCB we are processing
 				 * lptr is the previous DCB on the zombie queue
-			 * or NULL if the DCB is at the head of the queue
+				 * or NULL if the DCB is at the head of the
-			 * tptr is the DCB after the one we are processing
+				 * queue tptr is the DCB after the one we are
-			 * on the zombie queue
+				 * processing on the zombie queue
 				 */
 				DCB	*tptr = ptr->memdata.next;
 				if (lptr == NULL)
@@ -450,8 +478,9 @@ bool    succp = false;
 					lptr->memdata.next = tptr;
 				LOGIF(LD, (skygw_log_write_flush(
 					LOGFILE_DEBUG,
-                                "%lu [dcb_process_zombies] Remove dcb %p fd %d "
+					"%lu [dcb_process_zombies] Remove dcb "
-                                "in state %s from zombies list.",
+					"%p fd %d " "in state %s from the "
+					"list of zombies.",
 					pthread_self(),
 					ptr,
 					ptr->fd,
@@ -477,16 +506,23 @@ bool    succp = false;
 				ptr = ptr->memdata.next;
 			}
 		}
+	}
 	spinlock_release(&zombiespin);
 
+	/*
+	 * Process the victim queue. These are DCBs that are not in
+	 * use by any thread. 
+	 * The corresponding file descriptor is closed, the DCB marked
+	 * as disconnected and the DCB itself is fianlly freed.
+	 */
         dcb = dcb_list;
-        /*< Close, and set DISCONNECTED victims */
         while (dcb != NULL) {
 		DCB* dcb_next = NULL;
                 int  rc = 0;
                 /*<
                  * Close file descriptor and move to clean-up phase.
                  */
+		ss_dassert(excluded != dcb);
                 rc = close(dcb->fd);
 
                 if (rc < 0) {
@@ -1108,8 +1144,8 @@ int	above_water;
 /** 
  * Removes dcb from poll set, and adds it to zombies list. As a consequense,
  * dcb first moves to DCB_STATE_NOPOLLING, and then to DCB_STATE_ZOMBIE state.
- * At the end of the function state may not be DCB_STATE_ZOMBIE because once dcb_initlock
+ * At the end of the function state may not be DCB_STATE_ZOMBIE because once
- * is released parallel threads may change the state.
+ * dcb_initlock is released parallel threads may change the state.
  *
  * Parameters:
  * @param dcb The DCB to close
@@ -1928,13 +1964,15 @@ int	rval = 0;
  * and instead implements a queuing mechanism in which nested events are
  * queued on the DCB such that when the thread processing the first event
  * returns it will read the queued event and process it. This allows the
- * thread that woudl otherwise have to wait to process the nested event
+ * thread that would otherwise have to wait to process the nested event
  * to return immediately and and process other events.
  *
  * @param dcb		The DCB that has data available
+ * @param thread_id	The ID of the calling thread
+ * @param nozombies	If non-zero then do not do zombie processing
  */
 void
-dcb_pollin(DCB *dcb, int thread_id)
+dcb_pollin(DCB *dcb, int thread_id, int nozombies)
 {
 
 	spinlock_acquire(&dcb->pollinlock);
@@ -1945,7 +1983,8 @@ dcb_pollin(DCB *dcb, int thread_id)
 			if (dcb->readcheck)
 			{
 				dcb->stats.n_readrechecks++;
-				dcb_process_zombies(thread_id);
+				if (!nozombies)
+					dcb_process_zombies(thread_id, dcb);
 			}
 			dcb->readcheck = 0;
 			spinlock_release(&dcb->pollinlock);
@@ -1976,9 +2015,11 @@ dcb_pollin(DCB *dcb, int thread_id)
  * to return immediately and and process other events.
  *
  * @param dcb		The DCB thats available for writes 
+ * @param thread_id	The ID of the calling thread
+ * @param nozombies	If non-zero then do not do zombie processing
  */
 void
-dcb_pollout(DCB *dcb, int thread_id)
+dcb_pollout(DCB *dcb, int thread_id, int nozombies)
 {
 
 	spinlock_acquire(&dcb->polloutlock);
@@ -1988,7 +2029,8 @@ dcb_pollout(DCB *dcb, int thread_id)
 		do {
 			if (dcb->writecheck)
 			{
-				dcb_process_zombies(thread_id);
+				if (!nozombies)
+					dcb_process_zombies(thread_id, dcb);
 				dcb->stats.n_writerechecks++;
 			}
 			dcb->writecheck = 0;

server/core/housekeeper.c

@@ -16,6 +16,7 @@
  * Copyright SkySQL Ab 2014
  */
 #include <stdlib.h>
+#include <string.h>
 #include <housekeeper.h>
 #include <thread.h>
 #include <spinlock.h>

server/core/modutil.c

@@ -125,7 +125,7 @@ unsigned char	*ptr;
 /**
  * Replace the contents of a GWBUF with the new SQL statement passed as a text string.
  * The routine takes care of the modification needed to the MySQL packet,
- * returning a GWBUF chian that cna be used to send the data to a MySQL server
+ * returning a GWBUF chain that can be used to send the data to a MySQL server
  *
  * @param orig	The original request in a GWBUF
  * @param sql	The SQL text to replace in the packet

server/core/poll.c

@@ -49,10 +49,19 @@ extern int lm_enabled_logfiles_bitmask;
  * @endverbatim
  */
 
+/**
+ * Control the use of mutexes for the epoll_wait call. Setting to 1 will
+ * cause the epoll_wait calls to be moved under a mutex. This may be useful
+ * for debuggign purposes but should be avoided in general use.
+ */
+#define	MUTEX_EPOLL	0
+
 static	int		epoll_fd = -1;	  /*< The epoll file descriptor */
 static	int		do_shutdown = 0;  /*< Flag the shutdown of the poll subsystem */
 static	GWBITMASK	poll_mask;
+#if MUTEX_EPOLL
 static  simple_mutex_t  epoll_wait_mutex; /*< serializes calls to epoll_wait */
+#endif
 static	int		n_waiting = 0;	  /*< No. of threads in epoll_wait */
 
 /**
@@ -154,7 +163,9 @@ int	i;
 			thread_data[i].state = THREAD_STOPPED;
 		}
 	}
+#if MUTEX_EPOLL
         simple_mutex_init(&epoll_wait_mutex, "epoll_wait_mutex");        
+#endif
 
 	hktask_add("Load Average", poll_loadav, NULL, POLL_LOAD_FREQ);
 	n_avg_samples = 15 * 60 / POLL_LOAD_FREQ;
@@ -359,7 +370,7 @@ DCB                *zombies = NULL;
                                            thread_id)));                        
                         no_op = TRUE;
                 }
-#if 0
+#if MUTEX_EPOLL
                 simple_mutex_lock(&epoll_wait_mutex, TRUE);
 #endif
 		if (thread_data)
@@ -385,7 +396,7 @@ DCB                *zombies = NULL;
 		{
 			atomic_add(&n_waiting, -1);
                         if (process_zombies_only) {
-#if 0
+#if MUTEX_EPOLL
                                 simple_mutex_unlock(&epoll_wait_mutex);
 #endif
                                 goto process_zombies;
@@ -413,7 +424,7 @@ DCB                *zombies = NULL;
 
 		if (n_waiting == 0)
 			atomic_add(&pollStats.n_nothreads, 1);
-#if 0
+#if MUTEX_EPOLL
                 simple_mutex_unlock(&epoll_wait_mutex);
 #endif
 #endif /* BLOCKINGPOLL */
@@ -504,7 +515,7 @@ DCB                *zombies = NULL;
 #else
                                                 atomic_add(&pollStats.n_write,
                                                         		1);
-						dcb_pollout(dcb, thread_id);
+						dcb_pollout(dcb, thread_id, nfds);
 #endif
                                         } else {
                                                 LOGIF(LD, (skygw_log_write(
@@ -554,7 +565,7 @@ DCB                *zombies = NULL;
 #if MUTEX_BLOCK
 						dcb->func.read(dcb);
 #else
-						dcb_pollin(dcb, thread_id);
+						dcb_pollin(dcb, thread_id, nfds);
 #endif
 					}
 #if MUTEX_BLOCK
@@ -609,8 +620,16 @@ DCB                *zombies = NULL;
                                                 eno,
                                                 strerror(eno))));
                                         atomic_add(&pollStats.n_hup, 1);
+					spinlock_acquire(&dcb->dcb_initlock);
+					if ((dcb->flags & DCBF_HUNG) == 0)
+					{
+						dcb->flags |= DCBF_HUNG;
+						spinlock_release(&dcb->dcb_initlock);
 						dcb->func.hangup(dcb);
 					}
+					else
+						spinlock_release(&dcb->dcb_initlock);
+				}
 
 				if (ev & EPOLLRDHUP)
 				{
@@ -628,8 +647,16 @@ DCB                *zombies = NULL;
                                                 eno,
                                                 strerror(eno))));
                                         atomic_add(&pollStats.n_hup, 1);
+					spinlock_acquire(&dcb->dcb_initlock);
+					if ((dcb->flags & DCBF_HUNG) == 0)
+					{
+						dcb->flags |= DCBF_HUNG;
+						spinlock_release(&dcb->dcb_initlock);
 						dcb->func.hangup(dcb);
 					}
+					else
+						spinlock_release(&dcb->dcb_initlock);
+				}
 			} /*< for */
                         no_op = FALSE;
 		}
@@ -638,7 +665,7 @@ DCB                *zombies = NULL;
 		{
 			thread_data[thread_id].state = THREAD_ZPROCESSING;
 		}
-		zombies = dcb_process_zombies(thread_id);
+		zombies = dcb_process_zombies(thread_id, NULL);
                 
                 if (zombies == NULL) {
                         process_zombies_only = false;

server/include/dcb.h

@@ -271,8 +271,8 @@ int           fail_accept_errno;
 #define DCB_BELOW_LOW_WATER(x)		((x)->low_water && (x)->writeqlen < (x)->low_water)
 #define DCB_ABOVE_HIGH_WATER(x)		((x)->high_water && (x)->writeqlen > (x)->high_water)
 
-void		dcb_pollin(DCB *, int);
+void		dcb_pollin(DCB *, int, int);
-void		dcb_pollout(DCB *, int);
+void		dcb_pollout(DCB *, int, int);
 DCB             *dcb_get_zombies(void);
 int             gw_write(
 #if defined(SS_DEBUG)
@@ -289,7 +289,7 @@ DCB		*dcb_clone(DCB *);
 int             dcb_read(DCB *, GWBUF **);
 int             dcb_drain_writeq(DCB *);
 void            dcb_close(DCB *);
-DCB		*dcb_process_zombies(int);		/* Process Zombies */
+DCB		*dcb_process_zombies(int, DCB*);	/* Process Zombies except the one behind the pointer */
 void		printAllDCBs();				/* Debug to print all DCB in the system */
 void		printDCB(DCB *);			/* Debug print routine */
 void		dprintAllDCBs(DCB *);			/* Debug to print all DCB in the system */
@@ -317,6 +317,9 @@ void dcb_call_foreach (DCB_REASON reason);
 void dcb_call_foreach (
         DCB_REASON reason);
 
-/* DCB flags values */
+/**
-#define	DCBF_CLONE		0x0001	/* DCB is a clone */
+ * DCB flags values
+ */
+#define	DCBF_CLONE		0x0001	/*< DCB is a clone */
+#define DCBF_HUNG		0x0002	/*< Hangup has been dispatched */
 #endif /*  _DCB_H */

server/modules/routing/readwritesplit/readwritesplit.c

@@ -1674,7 +1674,6 @@ static int routeQuery(
         {
                 rses_is_closed = true;
         }
-        
         ss_dassert(!GWBUF_IS_TYPE_UNDEFINED(querybuf));
         
         packet = GWBUF_DATA(querybuf);
@@ -1702,7 +1701,6 @@ static int routeQuery(
                                 (rses_is_closed ? "Router was closed" :
                                 "Router has no backend servers where to "
                                 "route to"))));
-                        free(querybuf);
                 }
                 goto retblock;
         }
@@ -2196,6 +2194,16 @@ static void clientReply (
         }
         bref = get_bref_from_dcb(router_cli_ses, backend_dcb);
 
+#if !defined(FOR_BUG548_FIX_ONLY)
+	/** This makes the issue becoming visible in poll.c */
+	if (bref == NULL)
+	{
+		/** Unlock router session */
+		rses_end_locked_router_action(router_cli_ses);
+		goto lock_failed;
+	}
+#endif
+	
         CHK_BACKEND_REF(bref);
         scur = &bref->bref_sescmd_cur;
         /**