From a30e2ab29ba7ef6381a58ea7843fd6865ba5f6a8 Mon Sep 17 00:00:00 2001
From: vraatikka <vilho.raatikka@skysql.com>
Date: Wed, 20 Nov 2013 17:23:04 +0200
Subject: [PATCH] Added back read and write mutex to dcb. Locks are acquired in
 poll.c before calling call-back functions. Additional locks efficiently hide
 memory corruption issue, which appears when readconn router is used without
 using those mutexes, with multiple parallel maxscale threads and intensive
 load.

---
 server/core/dcb.c    |  6 ++++++
 server/core/poll.c   | 28 +++++++++++++++++++++++++++-
 server/include/dcb.h |  6 ++++++
 3 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/server/core/dcb.c b/server/core/dcb.c
index 4bee9a23f..6e2c32ea4 100644
--- a/server/core/dcb.c
+++ b/server/core/dcb.c
@@ -106,6 +106,12 @@ if ((rval = calloc(1, sizeof(DCB))) == NULL)
         rval->dcb_chk_tail = CHK_NUM_DCB;
 #endif
         rval->dcb_role = role;
+#if 1
+        simple_mutex_init(&rval->dcb_write_lock, "DCB write mutex");
+        simple_mutex_init(&rval->dcb_read_lock, "DCB read mutex");
+        rval->dcb_write_active = false;
+        rval->dcb_read_active = false;
+#endif
         spinlock_init(&rval->dcb_initlock);
 	spinlock_init(&rval->writeqlock);
 	spinlock_init(&rval->delayqlock);
diff --git a/server/core/poll.c b/server/core/poll.c
index fd706b060..fb1bf2a52 100644
--- a/server/core/poll.c
+++ b/server/core/poll.c
@@ -406,10 +406,24 @@ poll_waitevents(void *arg)
                                         eno = gw_getsockerrno(dcb->fd);
 
                                         if (eno == 0)  {
+#if 1
+                                                simple_mutex_lock(
+                                                        &dcb->dcb_write_lock,
+                                                        true);
+                                                ss_info_dassert(
+                                                        !dcb->dcb_write_active,
+                                                        "Write already active");
+                                                dcb->dcb_write_active = TRUE;
+#endif
                                                 atomic_add(
-                                                        &pollStats.n_write,
+                                                &pollStats.n_write,
                                                         1);
                                                 dcb->func.write_ready(dcb);
+#if 1
+                                                dcb->dcb_write_active = FALSE;
+                                                simple_mutex_unlock(
+                                                        &dcb->dcb_write_lock);
+#endif
                                         } else {
                                                 skygw_log_write(
                                                         LOGFILE_DEBUG,
@@ -425,6 +439,13 @@ poll_waitevents(void *arg)
                                 }
                                 if (ev & EPOLLIN)
                                 {
+#if 1
+                                        simple_mutex_lock(&dcb->dcb_read_lock,
+                                                          true);
+                                        ss_info_dassert(!dcb->dcb_read_active,
+                                                        "Read already active");
+                                        dcb->dcb_read_active = TRUE;
+#endif
 					if (dcb->state == DCB_STATE_LISTENING)
 					{
                                                 skygw_log_write(
@@ -449,6 +470,11 @@ poll_waitevents(void *arg)
 						atomic_add(&pollStats.n_read, 1);
 						dcb->func.read(dcb);
 					}
+#if 1
+                                        dcb->dcb_read_active = FALSE;
+                                        simple_mutex_unlock(
+                                                &dcb->dcb_read_lock);
+#endif
 				}
 			} /**< for */
                         no_op = FALSE;
diff --git a/server/include/dcb.h b/server/include/dcb.h
index 6a15b4509..9a7341c61 100644
--- a/server/include/dcb.h
+++ b/server/include/dcb.h
@@ -156,6 +156,12 @@ typedef struct dcb {
 #endif
         dcb_role_t      dcb_role;
         SPINLOCK        dcb_initlock;
+#if 1
+        simple_mutex_t  dcb_read_lock;
+        simple_mutex_t  dcb_write_lock;
+        bool            dcb_read_active;
+        bool            dcb_write_active;
+#endif
 	int	 	fd;		/**< The descriptor */
 	dcb_state_t	state;		/**< Current descriptor state */
 	char		*remote;	/**< Address of remote end */