WL#13701 DevAPI: Improve session pool logic

Author: Bogdan Degtyariov

cdk/core/session.cc

@@ -56,6 +56,7 @@ struct Session_builder
   bool m_throw_errors = false;
   scoped_ptr<Error>     m_error;
   unsigned              m_attempts = 0;
+  size_t                m_id = 0;
 
   Session_builder(bool throw_errors = false)
     : m_throw_errors(throw_errors)
@@ -76,11 +77,15 @@ struct Session_builder
     3. If a bail-out error was detected, throws that error.
   */
 
-  bool operator() (const ds::TCPIP &ds, const ds::TCPIP::Options &options);
+  bool operator() (size_t, const ds::TCPIP &ds,
+                   const ds::TCPIP::Options &options);
+
 #ifndef WIN32
-  bool operator() (const ds::Unix_socket&ds, const ds::Unix_socket::Options &options);
+  bool operator() (size_t, const ds::Unix_socket&ds,
+                   const ds::Unix_socket::Options &options);
 #endif
-  bool operator() (const ds::TCPIP_old &ds, const ds::TCPIP_old::Options &options);
+  bool operator() (size_t, const ds::TCPIP_old &ds,
+                   const ds::TCPIP_old::Options &options);
 
   /*
     Make a connection attempt using the given connection object. Returns true if
@@ -152,6 +157,7 @@ bool Session_builder::connect(Conn &connection)
 
 bool
 Session_builder::operator() (
+  size_t id,
   const ds::TCPIP &ds,
   const ds::TCPIP::Options &options
   )
@@ -213,12 +219,14 @@ Session_builder::operator() (
   }
 
   m_database = options.database();
+  m_id = id;
   return true;
 }
 
 #ifndef WIN32
 bool
 Session_builder::operator() (
+  size_t id,
   const ds::Unix_socket &ds,
   const ds::Unix_socket::Options &options
   )
@@ -235,7 +243,7 @@ Session_builder::operator() (
   m_conn.reset(connection.release());
 
   m_database = options.database();
-
+  m_id = id;
   return true;
 }
 
@@ -244,6 +252,7 @@ Session_builder::operator() (
 
 bool
 Session_builder::operator() (
+  size_t,
   const ds::TCPIP_old&,
   const ds::TCPIP_old::Options&
 )
@@ -252,7 +261,6 @@ Session_builder::operator() (
   //return false;
 }
 
-
 #ifdef WITH_SSL
 
 Session_builder::TLS*
@@ -366,30 +374,33 @@ Session::Session(ds::TCPIP &ds, const ds::TCPIP::Options &options)
 {
   Session_builder sb(true);  // throw errors if detected
 
-  sb(ds, options);
+  sb(0, ds, options);
 
   assert(sb.m_sess);
 
   m_session = sb.m_sess;
   m_connection = sb.m_conn.release();
+  m_id = sb.m_id;
 }
 
 
 struct ds::Multi_source::Access
 {
   template <class Visitor>
-  static void visit(Multi_source &ds, Visitor &visitor)
-  { ds.visit(visitor); }
+  static void visit(Multi_source &ds, Visitor &visitor,
+                    Multi_source::ep_filter_t ep_filter)
+  { ds.visit(visitor, ep_filter); }
 };
 
 
-Session::Session(ds::Multi_source &ds)
+
+Session::Session(ds::Multi_source &ds, ds::Multi_source::ep_filter_t ep_filter)
   : m_session(NULL)
   , m_connection(NULL)
 {
   Session_builder sb;
 
-  ds::Multi_source::Access::visit(ds, sb);
+  ds::Multi_source::Access::visit(ds, sb, ep_filter);
 
   if (!sb.m_sess)
   {
@@ -406,6 +417,7 @@ Session::Session(ds::Multi_source &ds)
   m_session = sb.m_sess;
   m_database = sb.m_database;
   m_connection = sb.m_conn.release();
+  m_id = sb.m_id;
 }
 
 
@@ -416,12 +428,13 @@ Session::Session(ds::Unix_socket &ds, const ds::Unix_socket::Options &options)
 {
   Session_builder sb(true);  // throw errors if detected
 
-  sb(ds, options);
+  sb(0, ds, options);
 
   assert(sb.m_sess);
 
   m_session = sb.m_sess;
   m_connection = sb.m_conn.release();
+  m_id = sb.m_id;
 }
 #endif //#ifndef WIN32
 

cdk/include/mysql/cdk/data_source.h

@@ -36,6 +36,7 @@
 PUSH_SYS_WARNINGS_CDK
 #include <functional>
 #include <algorithm>
+#include <iterator>
 #include <set>
 #include <random>
 #include "api/expression.h"
@@ -451,8 +452,10 @@ namespace ds {
 
     struct Prio
     {
+      size_t id;
       unsigned short prio;
       uint16_t weight;
+
       operator unsigned short() const
       {
         return prio;
@@ -482,7 +485,10 @@ namespace ds {
         );
       }
 
-      m_ds_list.emplace(Prio{ m_counter++, weight }, DS_pair<DS_t, DS_opt>{ ds, opt });
+      m_ds_list.emplace(
+        Prio{ m_ds_list.size() + 1, m_counter++, weight },
+        DS_pair<DS_t, DS_opt>{ ds, opt }
+      );
     }
 
     // Add data source with priority.
@@ -500,44 +506,54 @@ namespace ds {
         );
       }
 
-      m_ds_list.emplace(Prio{ prio, weight }, DS_pair<DS_t, DS_opt>{ ds, opt });
+      m_ds_list.emplace(Prio{ m_ds_list.size() + 1, prio, weight },
+         DS_pair<DS_t, DS_opt>{ ds, opt });
     }
 
   private:
 
     template <typename Visitor>
     struct Variant_visitor
     {
+
       Visitor *vis = nullptr;
+      size_t id = 0;
       bool stop_processing = false;
 
       template <class DS_t, class DS_opt>
-      void operator () (const DS_pair<DS_t, DS_opt> &ds_pair)
+      void operator () (
+        const DS_pair<DS_t, DS_opt> &ds_pair
+      )
       {
         assert(vis);
-        stop_processing = (bool)(*vis)(ds_pair.first, ds_pair.second);
+        stop_processing = (bool)(*vis)(id, ds_pair.first, ds_pair.second);
       }
     };
 
   public:
 
+    using ep_filter_t = std::function<bool(size_t)>;
+
     /*
-      Call visitor(ds,opts) for each data source ds with options
-      opts in the list. Do it in decreasing priority order, choosing
-      randomly among data sources with the same priority.
+      Call visitor(id, ds,opts) for each data source ds with options
+      opts in the list, where id is the data source identifier.
+      Do it in decreasing priority order, choosing
+      randomly among data sources with the same priority. If `ep_filter`
+      is given, ignore data sources for which this function returns true.
+
       If visitor(...) call returns true, stop the process.
     */
 
     template <class Visitor>
-    void visit(Visitor &visitor)
+    void visit(Visitor &visitor, ep_filter_t ep_filter = nullptr)
     {
       Variant_visitor<Visitor> variant_visitor;
       variant_visitor.vis = &visitor;
 
       std::random_device rnd;
       bool stop_processing = false;
       std::vector<uint16_t> weights;
-      std::set<DS_variant*> same_prio;
+      std::set<std::pair<size_t,DS_variant&>> same_prio;
 
       for (auto it = m_ds_list.begin(); !stop_processing;)
       {
@@ -555,7 +571,10 @@ namespace ds {
 
           for (auto it1 = same_range.first; it1 != same_range.second; ++it1)
           {
-            same_prio.insert(&(it1->second));
+            if (ep_filter && ep_filter(it1->first.id))
+              continue;
+
+            same_prio.emplace(it1->first.id, it1->second);
             weights.push_back(it1->first.weight);
             total_weight += it1->first.weight;
           }
@@ -601,7 +620,8 @@ namespace ds {
             std::advance(el, pos);
           }
 
-          (*el)->visit(variant_visitor);
+          variant_visitor.id = el->first;
+          el->second.visit(variant_visitor);
           stop_processing = variant_visitor.stop_processing;
 
           if (stop_processing)

cdk/include/mysql/cdk/session.h

@@ -56,6 +56,7 @@ class Session
   mysqlx::Session      *m_session;
   const mysqlx::string *m_database;
   api::Connection      *m_connection;
+  size_t                m_id = 0;
 
   typedef Reply::Initializer Reply_init;
 
@@ -68,7 +69,7 @@ class Session
   Session(ds::TCPIP &ds,
           const ds::TCPIP::Options &options = ds::TCPIP::Options());
 
-  Session(ds::Multi_source&);
+  Session(ds::Multi_source&, ds::Multi_source::ep_filter_t = nullptr);
 
 #ifndef _WIN32
   Session(ds::Unix_socket &ds,
@@ -99,6 +100,16 @@ class Session
     m_connection->close();
   }
 
+  /*
+    If this session was created from a ds::Multi_source, returns id of the single
+    data source inside the multi-source that was used to create this session.
+    Otherwise returns 0.
+  */
+
+  size_t id() {
+    return m_id;
+  }
+
   /*
     Transactions
     ------------