MXS-1392 Take new zombie mechanism into use

Next commit will remove the remnants of the reference counting
mechanism.

server/core/dcb.cc

@@ -93,7 +93,6 @@ static thread_local struct
 
 static void dcb_initialize(DCB *dcb);
 static void dcb_final_free(DCB *dcb);
-static void dcb_final_close(DCB *dcb);
 static void dcb_call_callback(DCB *dcb, DCB_REASON reason);
 static int  dcb_null_write(DCB *dcb, GWBUF *buf);
 static int  dcb_null_auth(DCB *dcb, SERVER *server, MXS_SESSION *session, GWBUF *buf);
@@ -247,7 +246,7 @@ dcb_final_free(DCB *dcb)
         }
     }
 
-    dcb_dec_ref(dcb);
+    dcb_free_all_memory(dcb);
 }
 
 /**
@@ -407,7 +406,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
                                              MODULE_PROTOCOL)) == NULL)
     {
         dcb->state = DCB_STATE_DISCONNECTED;
-        dcb_dec_ref(dcb);
+        dcb_free_all_memory(dcb);
         MXS_ERROR("Failed to load protocol module '%s'", protocol);
         return NULL;
     }
@@ -423,7 +422,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
     if (authfuncs == NULL)
     {
         MXS_ERROR("Failed to load authenticator module '%s'", authenticator);
-        dcb_dec_ref(dcb);
+        dcb_free_all_memory(dcb);
         return NULL;
     }
 
@@ -443,7 +442,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
         // Remove the inc ref that was done in session_link_backend_dcb().
         session_put_ref(dcb->session);
         dcb->session = NULL;
-        dcb_dec_ref(dcb);
+        dcb_free_all_memory(dcb);
         return NULL;
     }
     else
@@ -480,7 +479,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
         // Remove the inc ref that was done in session_link_backend_dcb().
         session_put_ref(dcb->session);
         dcb->session = NULL;
-        dcb_dec_ref(dcb);
+        dcb_free_all_memory(dcb);
         return NULL;
     }
 
@@ -496,7 +495,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
         // Remove the inc ref that was done in session_link_backend_dcb().
         session_put_ref(dcb->session);
         dcb->session = NULL;
-        dcb_dec_ref(dcb);
+        dcb_free_all_memory(dcb);
         return NULL;
     }
     /**
@@ -1094,14 +1093,10 @@ void dcb_close(DCB *dcb)
     {
         dcb->n_close = 1;
 
-        if (dcb != dcb_get_current())
+        Worker* worker = Worker::get(dcb->poll.thread.id);
-        {
+        ss_dassert(worker);
-            // If the dcb to be closed is *not* the dcb for which event callbacks are being
+
-            // called, we close it immediately. Otherwise it will be closed once all callbacks
+        worker->register_zombie(dcb);
-            // have been called, in the end of dcb_process_poll_events(), which currently is
-            // above us in the call stack.
-            dcb_final_close(dcb);
-        }
     }
     else
     {
@@ -1139,7 +1134,7 @@ void dcb_close_in_owning_thread(DCB* dcb)
     }
 }
 
-static void dcb_final_close(DCB* dcb)
+void dcb_final_close(DCB* dcb)
 {
 #if defined(SS_DEBUG)
     int wid = Worker::get_current_id();
@@ -3145,11 +3140,6 @@ static uint32_t dcb_handler(DCB* dcb, uint32_t events)
         rv |= dcb_process_poll_events(dcb, events);
     }
 
-    if (dcb->n_close != 0)
-    {
-        dcb_final_close(dcb);
-    }
-
     this_thread.current_dcb = NULL;
 
     return rv;

server/core/maxscale/dcb.h

@@ -17,6 +17,7 @@
 MXS_BEGIN_DECLS
 
 void dcb_free_all_memory(DCB *dcb);
+void dcb_final_close(DCB *dcb);
 
 /**
  * @brief Increase the reference count of the DCB

server/core/session.cc

@@ -393,7 +393,7 @@ static void session_free(MXS_SESSION *session)
 
     if (session->client_dcb)
     {
-        dcb_dec_ref(session->client_dcb);
+        dcb_free_all_memory(session->client_dcb);
         session->client_dcb = NULL;
     }
     /**

server/core/worker.cc

@@ -31,6 +31,7 @@
 #include <maxscale/json_api.h>
 #include <maxscale/utils.hh>
 
+#include "maxscale/dcb.h"
 #include "maxscale/modules.h"
 #include "maxscale/poll.h"
 #include "maxscale/statistics.h"
@@ -845,8 +846,15 @@ void Worker::register_zombie(DCB* pDcb)
 
 void Worker::delete_zombies()
 {
-    // TODO: for_each(m_zombies.begin(), m_zombies.end(), dcb_free_all_memory);
+    // An algorithm cannot be used, as the final closing of a DCB may cause
-    m_zombies.resize(0);
+    // other DCBs to be registered in the zombie queue.
+
+    while (!m_zombies.empty())
+    {
+        DCB* pDcb = m_zombies.back();
+        m_zombies.resize(m_zombies.size() - 1);
+        dcb_final_close(pDcb);
+    }
 }
 
 void Worker::run()
@@ -1154,15 +1162,6 @@ void Worker::poll_waitevents()
 
         uint64_t cycle_start = hkheartbeat;
 
-        for (int i = 0; i < nfds; i++)
-        {
-            MXS_POLL_DATA *data = (MXS_POLL_DATA*)events[i].data.ptr;
-            if (data->free)
-            {
-                poll_inc_ref(data);
-            }
-        }
-
         for (int i = 0; i < nfds; i++)
         {
             /** Calculate event queue statistics */
@@ -1224,18 +1223,6 @@ void Worker::poll_waitevents()
             m_statistics.maxexectime = MXS_MAX(m_statistics.maxexectime, qtime);
         }
 
-        for (int i = 0; i < nfds; i++)
-        {
-            MXS_POLL_DATA *data = (MXS_POLL_DATA*)events[i].data.ptr;
-            if (data->free)
-            {
-                if (poll_dec_ref(data) == 1)
-                {
-                    data->free(data);
-                }
-            }
-        }
-
         dcb_process_idle_sessions(m_id);
 
         m_state = ZPROCESSING;