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

client/Makefile

@@ -33,7 +33,7 @@ endif
 
 CC=cc
 
-CFLAGS=-c -Wall -g $(HISTFLAG)
+CFLAGS=-c -Wall -g $(HISTFLAG) -I ../server/include
 
 SRCS= maxadmin.c
 

client/maxadmin.c

@@ -47,6 +47,9 @@
 #include <dirent.h>
 #include <locale.h>
 #include <errno.h>
+#include <getopt.h>
+
+#include <version.h>
 
 #ifdef HISTORY
 #include <histedit.h>
@@ -59,6 +62,7 @@ static int sendCommand(int so, char *cmd);
 static void DoSource(int so, char *cmd);
 static void DoUsage();
 static int isquit(char *buf);
+static void PrintVersion(const char *progname);
 
 #ifdef HISTORY
 static char *
@@ -70,6 +74,16 @@ prompt(EditLine *el __attribute__((__unused__)))
 }
 #endif
 
+static struct option long_options[] = {
+  {"host",     required_argument, 0, 'h'},
+  {"user",     required_argument, 0, 'u'},
+  {"password", required_argument, 0, 'p'},
+  {"port",     required_argument, 0, 'P'},
+  {"version",  no_argument,       0, 'v'},
+  {"help",     no_argument,       0, '?'},
+  {0, 0, 0, 0}
+};
+
 /**
  * The main for the maxadmin client
  *
@@ -79,7 +93,7 @@ prompt(EditLine *el __attribute__((__unused__)))
 int
 main(int argc, char **argv)
 {
-int		i, num, rv, fatal = 0;
+int		i, num, rv;
 #ifdef HISTORY
 char		*buf;
 EditLine	*el = NULL;
@@ -96,107 +110,39 @@ char		*user = "admin";
 char		*passwd = NULL;
 int		so, cmdlen;
 char		*cmd;
-int		argno = 0;
+int             option_index = 0;
+char            c;
 
 	cmd = malloc(1);
 	*cmd = 0;
 	cmdlen = 1;
 
-	for (i = 1; i < argc; i++)
+        while ((c = getopt_long(argc, argv, "h:p:P:u:v?", 
-	{
+				long_options, &option_index))
-		if (argv[i][0] == '-')
+	       >= 0)
-		{
+        {
-			switch (argv[i][1])
+	  switch (c) {
-			{
+	  case 'h':
-			case 'u':	/* User */
+	    hostname = strdup(optarg);
-				if (argv[i][2])
+	    break;
-					user = &(argv[i][2]);
+	  case 'p':
-				else if (i + 1 < argc)
+	    passwd = strdup(optarg);
-					user = argv[++i];
+	    break;
-				else
+	  case 'P':
-				{
+	    port = strdup(optarg);
-					fprintf(stderr, "Missing username"
+	    break;
-						"in -u option.\n");
+	  case 'u':
-					fatal = 1;
+	    user = strdup(optarg);
-				}
+	    break;
-				break;
+	  case 'v':
-			case 'p':	/* Password */
+	    PrintVersion(*argv);
-				if (argv[i][2])
+	    exit(EXIT_SUCCESS);
-					passwd = &(argv[i][2]);
+	  case '?':
-				else if (i + 1 < argc)
+	    DoUsage(*argv);
-					passwd = argv[++i];
+	    exit(optopt ? EXIT_FAILURE : EXIT_SUCCESS);
-				else
+	  }	  
-				{
-					fprintf(stderr, "Missing password "
-						"in -p option.\n");
-					fatal = 1;
-				}
-				break;
-			case 'h':	/* hostname */
-				if (argv[i][2])
-					hostname = &(argv[i][2]);
-				else if (i + 1 < argc)
-					hostname = argv[++i];
-				else
-				{
-					fprintf(stderr, "Missing hostname value "
-						"in -h option.\n");
-					fatal = 1;
-				}
-				break;
-			case 'P':	/* Port */
-				if (argv[i][2])
-					port = &(argv[i][2]);
-				else if (i + 1 < argc)
-					port = argv[++i];
-				else
-				{
-					fprintf(stderr, "Missing Port value "
-						"in -P option.\n");
-					fatal = 1;
-				}
-				break;
-			case '-':
-				{
-					char *word;
-
-					word = &argv[i][2];
-					if (strcmp(word, "help") == 0)
-					{
-						DoUsage();
-						exit(0);
-					}
-					break;
-				}
-			}
-		}
-		else
-		{
-			/* Arguments after the second argument are quoted
-			 * to allow for quoted names on the command line
-			 * to be passed on in quotes.
-			 */
-			if (argno++ > 1)
-			{
-				cmdlen += strlen(argv[i]) + 3;
-				cmd = realloc(cmd, cmdlen);
-				strcat(cmd, "\"");
-				strcat(cmd, argv[i]);
-				strcat(cmd, "\" ");
-			}
-			else
-			{
-				cmdlen += strlen(argv[i]) + 1;
-				cmd = realloc(cmd, cmdlen);
-				strcat(cmd, argv[i]);
-				strcat(cmd, " ");
-			}
-		}
 	}
 
-	if (fatal)
-		exit(1);
-
 	if (passwd == NULL)
 	{
 		struct termios tty_attr;
@@ -533,23 +479,34 @@ FILE		*fp;
 	return;
 }
 
+/**
+ * Print version information
+ */
+static void 
+PrintVersion(const char *progname) 
+{
+        printf("%s Version %s\n", progname, MAXSCALE_VERSION);
+}
+
 /**
  * Display the --help text.
  */
 static void
-DoUsage()
+DoUsage(const char *progname)
 {
-	printf("maxadmin: The MaxScale administrative and monitor client.\n\n");
+	PrintVersion(progname);
-	printf("Usage: maxadmin [-u user] [-p password] [-h hostname] [-P port] [<command file> | <command>]\n\n");
+	printf("The MaxScale administrative and monitor client.\n\n");
-	printf("	-u user		The user name to use for the connection, default\n");
+	printf("Usage: %s [-u user] [-p password] [-h hostname] [-P port] [<command file> | <command>]\n\n", progname);
+	printf("  -u|--user=...	        The user name to use for the connection, default\n");
 	printf("			is admin.\n");
-	printf("	-p password	The user password, if not given the password will\n");
+	printf("  -p|--password=...	The user password, if not given the password will\n");
 	printf("			be prompted for interactively\n");
-	printf("	-h hostname	The maxscale host to connecto to. The default is\n");
+	printf("  -h|--hostname=...	The maxscale host to connecto to. The default is\n");
 	printf("			localhost\n");
-	printf("	-P port		The port to use for the connection, the default\n");
+	printf("  -P|--port=...       	The port to use for the connection, the default\n");
 	printf("			port is 6603.\n");
-	printf("	--help		Print this help text.\n");
+	printf("  -v|--version          print version information and exit\n");
+	printf("  -?|--help		Print this help text.\n");
 	printf("Any remaining arguments are treated as MaxScale commands or a file\n");
 	printf("containing commands to execute.\n");
 }

log_manager/log_manager.cc

@@ -255,7 +255,7 @@ static int  logmanager_write_log(
         bool         use_valist,
         bool         spread_down,
         size_t       len,
-        char*        str,
+        const char*  str,
         va_list      valist);
 
 static blockbuf_t* blockbuf_init(logfile_id_t id);
@@ -609,7 +609,7 @@ static int logmanager_write_log(
         bool          use_valist,
         bool          spread_down,
         size_t        str_len,
-        char*         str,
+        const char*   str,
         va_list       valist)
 {
         logfile_t*   lf;
@@ -623,7 +623,7 @@ static int logmanager_write_log(
         CHK_LOGMANAGER(lm);
         
         if (id < LOGFILE_FIRST || id > LOGFILE_LAST) {
-                char* errstr = "Invalid logfile id argument.";
+                const char* errstr = "Invalid logfile id argument.";
                 /**
                  * invalid id, since we don't have logfile yet.
                  */
@@ -1181,7 +1181,7 @@ static bool logfile_set_enabled(
         CHK_LOGMANAGER(lm);
         
         if (id < LOGFILE_FIRST || id > LOGFILE_LAST) {
-            char* errstr = "Invalid logfile id argument.";
+            const char* errstr = "Invalid logfile id argument.";
             /**
              * invalid id, since we don't have logfile yet.
              */
@@ -1235,7 +1235,7 @@ return_succp:
 
 int skygw_log_write_flush(
         logfile_id_t  id,
-        char*         str,
+        const char*   str,
         ...)
 {
         int     err = 0;
@@ -1291,7 +1291,7 @@ return_err:
 
 int skygw_log_write(
         logfile_id_t  id,
-        char*         str,
+        const char*   str,
         ...)
 {
         int     err = 0;

log_manager/log_manager.h

@@ -72,9 +72,9 @@ void skygw_logmanager_exit(void);
  * free private write buffer list
  */
 void skygw_log_done(void);
-int  skygw_log_write(logfile_id_t id, char* format, ...);
+int  skygw_log_write(logfile_id_t id, const char* format, ...);
 int  skygw_log_flush(logfile_id_t id);
-int  skygw_log_write_flush(logfile_id_t id, char* format, ...);
+int  skygw_log_write_flush(logfile_id_t id, const char* format, ...);
 int  skygw_log_enable(logfile_id_t id);
 int  skygw_log_disable(logfile_id_t id);
 

server/core/config.c

@@ -1575,6 +1575,7 @@ static char *service_params[] =
 		"use_sql_variables_in",		/*< rwsplit only */
 		"version_string",
 		"filters",
+		"weightby",
                 NULL
         };
 

server/core/dcb.c

@@ -403,13 +403,16 @@ DCB_CALLBACK		*cb;
  * operation of clearing this bit means that no bits are set in
  * the memdata.bitmask then the DCB is no longer able to be 
  * referenced and it can be finally removed.
- * Thread won't clear its bit from bitmask of the DCB it is still using. 
+ *
+ * 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	dcb_in_use	The DCB the thread currently uses, NULL or valid DCB.
+ * @param	excluded		The DCB the thread currently uses, NULL or valid DCB.
  */
 DCB *
-dcb_process_zombies(int threadid, DCB *dcb_in_use)
+dcb_process_zombies(int threadid, DCB *excluded)
 {
 DCB	*ptr, *lptr;
 DCB*    dcb_list = NULL;
@@ -426,78 +429,100 @@ 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);
-		/** Don't clear the bit from DCB the user currently uses */
+
-		if (dcb_in_use == NULL || ptr != dcb_in_use)
+		/*
-		{
+		 * Skip processing of the excluded DCB
-			bitmask_clear(&ptr->memdata.bitmask, threadid);
+		 */
-		}
+		if (ptr == excluded)
-		if (ptr == dcb_in_use)
-			ss_dassert(!bitmask_isallclear(&ptr->memdata.bitmask));
-		
-		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
-			 * tptr is the DCB after the one we are processing
-			 * on the zombie queue
-			 */
-			DCB	*tptr = ptr->memdata.next;
-			if (lptr == NULL)
-				zombies = tptr;
-			else
-				lptr->memdata.next = tptr;
-                        LOGIF(LD, (skygw_log_write_flush(
-                                LOGFILE_DEBUG,
-                                "%lu [dcb_process_zombies] Remove dcb %p fd %d "
-                                "in state %s from zombies list.",
-                                pthread_self(),
-                                ptr,
-                                ptr->fd,
-                                STRDCBSTATE(ptr->state)))); 
-                        ss_info_dassert(ptr->state == DCB_STATE_ZOMBIE,
-                                        "dcb not in DCB_STATE_ZOMBIE state.");
-                        /*<
-                         * Move dcb to linked list of victim dcbs.
-                         */
-                        if (dcb_list == NULL) {
-                                dcb_list = ptr;
-                                dcb = dcb_list;
-                        } else {
-                                dcb->memdata.next = ptr;
-                                dcb = dcb->memdata.next;
-                        }
-                        dcb->memdata.next = NULL;
-			ptr = tptr;
-		}
-		else
 		{
 			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 tptr is the DCB after the one we are
+				 * processing on the zombie queue
+				 */
+				DCB	*tptr = ptr->memdata.next;
+				if (lptr == NULL)
+					zombies = tptr;
+				else
+					lptr->memdata.next = tptr;
+				LOGIF(LD, (skygw_log_write_flush(
+					LOGFILE_DEBUG,
+					"%lu [dcb_process_zombies] Remove dcb "
+					"%p fd %d " "in state %s from the "
+					"list of zombies.",
+					pthread_self(),
+					ptr,
+					ptr->fd,
+					STRDCBSTATE(ptr->state)))); 
+				ss_info_dassert(ptr->state == DCB_STATE_ZOMBIE,
+						"dcb not in DCB_STATE_ZOMBIE state.");
+				/*<
+				 * Move dcb to linked list of victim dcbs.
+				 */
+				if (dcb_list == NULL) {
+					dcb_list = ptr;
+					dcb = dcb_list;
+				} else {
+					dcb->memdata.next = ptr;
+					dcb = dcb->memdata.next;
+				}
+				dcb->memdata.next = NULL;
+				ptr = tptr;
+			}
+			else
+			{
+				lptr = ptr;
+				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(dcb_in_use != dcb);
+		ss_dassert(excluded != dcb);
                 rc = close(dcb->fd);
 
                 if (rc < 0) {
@@ -1119,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
@@ -1943,9 +1968,11 @@ int	rval = 0;
  * 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);
@@ -1956,7 +1983,8 @@ dcb_pollin(DCB *dcb, int thread_id)
 			if (dcb->readcheck)
 			{
 				dcb->stats.n_readrechecks++;
-				dcb_process_zombies(thread_id, dcb);
+				if (!nozombies)
+					dcb_process_zombies(thread_id, dcb);
 			}
 			dcb->readcheck = 0;
 			spinlock_release(&dcb->pollinlock);
@@ -1987,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);
@@ -1999,7 +2029,8 @@ dcb_pollout(DCB *dcb, int thread_id)
 		do {
 			if (dcb->writecheck)
 			{
-				dcb_process_zombies(thread_id, dcb);
+				if (!nozombies)
+					dcb_process_zombies(thread_id, dcb);
 				dcb->stats.n_writerechecks++;
 			}
 			dcb->writecheck = 0;

server/core/gateway.c

@@ -44,6 +44,7 @@
 #include <string.h>
 #include <gw.h>
 #include <unistd.h>
+#include <getopt.h>
 #include <service.h>
 #include <server.h>
 #include <dcb.h>
@@ -131,6 +132,17 @@ static bool libmysqld_started = FALSE;
  */
 static bool     daemon_mode = true;
 
+const char *progname = NULL;
+static struct option long_options[] = {
+  {"homedir",  required_argument, 0, 'c'},
+  {"config",   required_argument, 0, 'f'},
+  {"nodeamon", required_argument, 0, 'd'},
+  {"log",      required_argument, 0, 'l'},
+  {"version",  no_argument,       0, 'v'},
+  {"help",     no_argument,       0, '?'},
+  {0, 0, 0, 0}
+};
+
 static void log_flush_shutdown(void);
 static void log_flush_cb(void* arg);
 static int write_pid_file(char *); /* write MaxScale pidfile */
@@ -878,15 +890,19 @@ return_cnf_file_buf:
         return cnf_file_buf;
 }
 
-
 static void usage(void)
 {
         fprintf(stderr,
-                "*\n* Usage : maxscale [-h] | [-d] [-c <home "
+                "\nUsage : %s [-h] | [-d] [-c <home directory>] [-f <config file name>]\n\n"
-                "directory>] [-f <config file name>]\n* where:\n* "
+		"  -d|--nodaemon     enable running in terminal process (default:disabled)\n"
-                "-h help\n* -d enable running in terminal process (default:disabled)\n* "
+                "  -c|--homedir=...  relative|absolute MaxScale home directory\n"
-                "-c relative|absolute MaxScale home directory\n* "
+                "  -f|--config=...   relative|absolute pathname of MaxScale configuration file\n"
-                "-f relative|absolute pathname of MaxScale configuration file (default:MAXSCALE_HOME/etc/MaxScale.cnf)\n*\n");
+		"                    (default: $MAXSCALE_HOME/etc/MaxScale.cnf)\n"
+		"  -l|--log=...      log to file or shared memory\n"
+		"                    -lfile or -lshm - defaults to shared memory\n"
+		"  -v|--version      print version info and exit\n"
+                "  -?|--help         show this help\n"
+		, progname);
 }
 
 /** 
@@ -943,6 +959,8 @@ int main(int argc, char **argv)
         char*    cnf_file_path = NULL;        /*< conf file, to be freed */
         char*    cnf_file_arg = NULL;         /*< conf filename from cmd-line arg */
         void*    log_flush_thr = NULL;
+	int      option_index;
+	int	 logtofile = 0;	      	      /* Use shared memory or file */
         ssize_t  log_flush_timeout_ms = 0;
         sigset_t sigset;
         sigset_t sigpipe_mask;
@@ -954,6 +972,8 @@ int main(int argc, char **argv)
         sigemptyset(&sigpipe_mask);
         sigaddset(&sigpipe_mask, SIGPIPE);
 
+	progname = *argv;
+
 #if defined(SS_DEBUG)
         memset(conn_open, 0, sizeof(bool)*10240);
         memset(dcb_fake_write_errno, 0, sizeof(unsigned char)*10240);
@@ -980,7 +1000,8 @@ int main(int argc, char **argv)
                         goto return_main;
                 }
         }
-        while ((opt = getopt(argc, argv, "dc:f:h")) != -1)
+        while ((opt = getopt_long(argc, argv, "dc:f:l:v?",
+				 long_options, &option_index)) != -1)
         {
                 bool succp = true;
                 
@@ -1061,9 +1082,36 @@ int main(int argc, char **argv)
                                 succp = false;
                         }
                         break;  
+			
+		case 'v':
+		  rc = EXIT_SUCCESS;
+                  goto return_main;		  
+
+		case 'l':
+			if (strncasecmp(optarg, "file") == 0)
+				logtofile = 1;
+			else if (strncasecmp(optarg, "shm") == 0)
+				logtofile = 0;
+			else
+			{
+                                char* logerr = "Configuration file argument "
+                                        "identifier \'-l\' was specified but "
+                                        "the argument didn't specify\n  a valid "
+                                        "configuration file or the argument "
+                                        "was missing.";
+                                print_log_n_stderr(true, true, logerr, logerr, 0);
+                                usage();
+                                succp = false;
+			}
+			break;
+		  
+		case '?':
+		  usage();
+		  rc = EXIT_SUCCESS;
+                  goto return_main;		  
                         
                 default:
-                        usage();
+		  usage();
                         succp = false;
                         break;
                 }
@@ -1345,12 +1393,23 @@ int main(int argc, char **argv)
                 argv[0] = "MaxScale";
                 argv[1] = "-j";
                 argv[2] = buf;
-                argv[3] = "-s"; /*< store to shared memory */
+		if (logtofile)
-                argv[4] = "LOGFILE_DEBUG,LOGFILE_TRACE";   /*< ..these logs to shm */
+		{
-                argv[5] = "-l"; /*< write to syslog */
+			argv[3] = "-l"; /*< write to syslog */
-                argv[6] = "LOGFILE_MESSAGE,LOGFILE_ERROR"; /*< ..these logs to syslog */
+			argv[4] = "LOGFILE_MESSAGE,LOGFILE_ERROR"
-                argv[7] = NULL;
+				"LOGFILE_DEBUG,LOGFILE_TRACE"; 
-                skygw_logmanager_init(7, argv);
+			argv[5] = NULL;
+			skygw_logmanager_init(5, argv);
+		}
+		else
+		{
+			argv[3] = "-s"; /*< store to shared memory */
+			argv[4] = "LOGFILE_DEBUG,LOGFILE_TRACE";   /*< ..these logs to shm */
+			argv[5] = "-l"; /*< write to syslog */
+			argv[6] = "LOGFILE_MESSAGE,LOGFILE_ERROR"; /*< ..these logs to syslog */
+			argv[7] = NULL;
+			skygw_logmanager_init(7, argv);
+		}
         }
 
         /*<

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
@@ -225,4 +225,4 @@ char* modutil_get_query(
         } /*< switch */
 retblock:
         return query_str;
-}
+}

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 */
@@ -442,7 +453,7 @@ DCB                *zombies = NULL;
 
 			for (i = 0; i < nfds; i++)
 			{
-				DCB 		*dcb = (DCB *)events[i].data.ptr;
+				DCB 	*dcb = (DCB *)events[i].data.ptr;
 				__uint32_t	ev = events[i].events;
 
                                 CHK_DCB(dcb);
@@ -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,7 +620,15 @@ DCB                *zombies = NULL;
                                                 eno,
                                                 strerror(eno))));
                                         atomic_add(&pollStats.n_hup, 1);
-					dcb->func.hangup(dcb);
+					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,7 +647,15 @@ DCB                *zombies = NULL;
                                                 eno,
                                                 strerror(eno))));
                                         atomic_add(&pollStats.n_hup, 1);
-					dcb->func.hangup(dcb);
+					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;

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)
@@ -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 */