MXS-1392 Manage DCB lifetime using refcounts

Now it is also possible to ensure that a DCB stays alive while
a task referring to it is posted from one worker to another.
That will be implemented in a subsequent commit.

include/maxscale/dcb.h

@@ -216,7 +216,6 @@ void dcb_global_init();
 int dcb_write(DCB *, GWBUF *);
 DCB *dcb_accept(DCB *listener);
 DCB *dcb_alloc(dcb_role_t, struct servlistener *);
-void dcb_free_all_memory(DCB *dcb);
 DCB *dcb_connect(struct server *, struct session *, const char *);
 int dcb_read(DCB *, GWBUF **, int);
 int dcb_drain_writeq(DCB *);

include/maxscale/poll_core.h

@@ -131,7 +131,8 @@ static inline void poll_inc_ref(MXS_POLL_DATA* data)
  * @param data  The poll data whose reference count should be decreased.
  *
  * @return The previous reference count. If the returned value is 1, then
- *         the caller is the last user of the data.
+ *         the caller should call @c data->free(data) to dispose of the
+ *         poll data.
  */
 static inline uint32_t poll_dec_ref(MXS_POLL_DATA* data)
 {

include/maxscale/poll_core.hh

@@ -23,12 +23,14 @@ struct MxsPollData : MXS_POLL_DATA
     MxsPollData()
     {
         handler = NULL;
+        free = NULL;
         thread.id = 0;
     }
 
     MxsPollData(mxs_poll_handler_t h)
     {
         handler = h;
+        free = NULL;
         thread.id = 0;
     }
 };

server/core/dcb.cc

@@ -20,7 +20,7 @@
  * the state data and pointers to other components that relate to the
  * use of a file descriptor.
  */
-#include <maxscale/dcb.h>
+#include "maxscale/dcb.h"
 
 #include <arpa/inet.h>
 #include <errno.h>
@@ -123,6 +123,12 @@ static uint32_t dcb_poll_handler(MXS_POLL_DATA *data, int thread_id, uint32_t ev
 static uint32_t dcb_process_poll_events(DCB *dcb, uint32_t ev);
 static bool dcb_session_check(DCB *dcb, const char *);
 
+static void poll_dcb_free(MXS_POLL_DATA* data)
+{
+    // MXS_POLL_DATA is the first member of DCB.
+    dcb_free_all_memory(reinterpret_cast<DCB*>(data));
+}
+
 void dcb_global_init()
 {
     this_unit.dcb_initialized.dcb_chk_top = CHK_NUM_DCB;
@@ -130,7 +136,8 @@ void dcb_global_init()
     this_unit.dcb_initialized.state = DCB_STATE_ALLOC;
     this_unit.dcb_initialized.ssl_state = SSL_HANDSHAKE_UNKNOWN;
     this_unit.dcb_initialized.poll.handler = dcb_poll_handler;
-    this_unit.dcb_initialized.poll.free = NULL;
+    this_unit.dcb_initialized.poll.free = poll_dcb_free;
+    this_unit.dcb_initialized.poll.refcount = 1;
     this_unit.dcb_initialized.dcb_chk_tail = CHK_NUM_DCB;
 
     int nthreads = config_threadcount();
@@ -170,18 +177,17 @@ dcb_initialize(DCB *dcb)
 }
 
 /**
- * @brief Allocate or recycle a new DCB.
+ * @brief Allocate a new DCB.
  *
  * This routine performs the generic initialisation on the DCB before returning
- * the newly allocated or recycled DCB.
+ * the newly allocated DCB.
- *
- * Most fields will be already initialized by the list manager, through the
- * call to list_find_free, passing the DCB initialization function.
  *
  * Remaining fields are set from the given parameters, and then the DCB is
  * flagged as ready for use.
  *
  * @param dcb_role_t    The role for the new DCB
+ * @param listener      The listener if applicable.
+ *
  * @return An available DCB or NULL if none could be allocated.
  */
 DCB *
@@ -240,7 +246,8 @@ dcb_final_free(DCB *dcb)
             }
         }
     }
-    dcb_free_all_memory(dcb);
+
+    dcb_dec_ref(dcb);
 }
 
 /**
@@ -400,7 +407,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
                                              MODULE_PROTOCOL)) == NULL)
     {
         dcb->state = DCB_STATE_DISCONNECTED;
-        dcb_free_all_memory(dcb);
+        dcb_dec_ref(dcb);
         MXS_ERROR("Failed to load protocol module '%s'", protocol);
         return NULL;
     }
@@ -416,7 +423,7 @@ dcb_connect(SERVER *server, MXS_SESSION *session, const char *protocol)
     if (authfuncs == NULL)
     {
         MXS_ERROR("Failed to load authenticator module '%s'", authenticator);
-        dcb_free_all_memory(dcb);
+        dcb_dec_ref(dcb);
         return NULL;
     }
 
@@ -436,7 +443,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_free_all_memory(dcb);
+        dcb_dec_ref(dcb);
         return NULL;
     }
     else
@@ -473,7 +480,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_free_all_memory(dcb);
+        dcb_dec_ref(dcb);
         return NULL;
     }
 
@@ -489,7 +496,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_free_all_memory(dcb);
+        dcb_dec_ref(dcb);
         return NULL;
     }
     /**

server/core/maxscale/dcb.h

@@ -0,0 +1,75 @@
+#pragma once
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include <maxscale/dcb.h>
+
+MXS_BEGIN_DECLS
+
+void dcb_free_all_memory(DCB *dcb);
+
+/**
+ * @brief Increase the reference count of the DCB
+ *
+ * @param dcb  The dcb whose reference count should be increased.
+ */
+static inline void dcb_inc_ref(DCB* dcb)
+{
+    // A DCB starts out with a refcount of 1.
+    ss_dassert(dcb->poll.refcount >= 1);
+
+    poll_inc_ref(&dcb->poll);
+}
+
+/**
+ * @brief Decrease the reference count of the DCB. If it reaches 0,
+ *        the DCB will be freed.
+ *
+ * @param dcb  The dcb whose reference count should be decreased.
+ *
+ * @return True, if the dcb is still usable after the call, otherwise false.
+ *
+ * @note If the function returns false, the caller cannot use the dcb
+ *       for anything.
+ */
+static inline bool dcb_dec_ref(DCB* dcb)
+{
+    // A DCB starts out with a refcount of 1.
+    ss_dassert(dcb->poll.refcount >= 1);
+
+    bool rv = true;
+    if (poll_dec_ref(&dcb->poll) == 1)
+    {
+        dcb_free_all_memory(dcb);
+        rv = false;
+    }
+    return rv;
+}
+
+/**
+ * @brief Increase the reference count of the DCB.
+ *
+ * Convenience function for the situation where a received dcb is stored
+ * and the reference count needs to be increased.
+ *
+ * @param dcb  The dcb whose reference count should be increased.
+ *
+ * @return The dcb provided as argument.
+ */
+static inline DCB* dcb_get_ref(DCB* dcb)
+{
+    dcb_inc_ref(dcb);
+    return dcb;
+}
+
+MXS_END_DECLS

server/core/session.cc

@@ -37,6 +37,7 @@
 #include <maxscale/utils.h>
 #include <maxscale/json_api.h>
 
+#include "maxscale/dcb.h"
 #include "maxscale/session.h"
 #include "maxscale/filter.h"
 #include "maxscale/worker.hh"
@@ -392,7 +393,8 @@ static void session_free(MXS_SESSION *session)
 
     if (session->client_dcb)
     {
-        dcb_free_all_memory(session->client_dcb);
+        dcb_dec_ref(session->client_dcb);
+        session->client_dcb = NULL;
     }
     /**
      * If session is not child of some other session, free router_session.