BUG27489937: Support c extension for connection pools

The reset connection command was missing from the c extension implementation
which was required to reuse a connection from the pool, this patch adds this
functionality in order of use the c extension in connection pools.

Author: isrgomez

lib/mysql/connector/connection_cext.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2014, 2020, Oracle and/or its affiliates. All rights reserved.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License, version 2.0, as
@@ -652,6 +652,23 @@ def cmd_change_user(self, username='', password='', database='',
         self._charset_id = charset
         self._post_connection()
 
+    def cmd_reset_connection(self):
+        """Resets the session state without re-authenticating
+
+        Works only for MySQL server 5.7.3 or later.
+        """
+        if self._server_version < (5, 7, 3):
+            raise errors.NotSupportedError("MySQL version 5.7.2 and "
+                                           "earlier does not support "
+                                           "COM_RESET_CONNECTION.")
+        try:
+            self._cmysql.reset_connection()
+        except MySQLInterfaceError as exc:
+            raise errors.get_mysql_exception(msg=exc.msg, errno=exc.errno,
+                                             sqlstate=exc.sqlstate)
+
+        self._post_connection()
+
     def cmd_refresh(self, options):
         """Send the Refresh command to the MySQL server"""
         try:

lib/mysql/connector/pooling.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2013, 2017, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2013, 2020, Oracle and/or its affiliates. All rights reserved.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License, version 2.0, as
@@ -41,7 +41,9 @@
 import threading
 
 from . import errors
+from . import Connect
 from .connection import MySQLConnection
+from .connection_cext import CMySQLConnection
 
 CONNECTION_POOL_LOCK = threading.RLock()
 CNX_POOL_MAXSIZE = 32
@@ -97,7 +99,7 @@ def __init__(self, pool, cnx):
         if not isinstance(pool, MySQLConnectionPool):
             raise AttributeError(
                 "pool should be a MySQLConnectionPool")
-        if not isinstance(cnx, MySQLConnection):
+        if not isinstance(cnx, (MySQLConnection, CMySQLConnection)):
             raise AttributeError(
                 "cnx should be a MySQLConnection")
         self._cnx_pool = pool
@@ -194,7 +196,7 @@ def set_config(self, **kwargs):
 
         with CONNECTION_POOL_LOCK:
             try:
-                test_cnx = MySQLConnection()
+                test_cnx = Connect()
                 if "use_pure" in kwargs:
                     del kwargs["use_pure"]
                 test_cnx.config(**kwargs)
@@ -243,7 +245,7 @@ def _queue_connection(self, cnx):
 
         Raises PoolError on errors.
         """
-        if not isinstance(cnx, MySQLConnection):
+        if not isinstance(cnx, (MySQLConnection, CMySQLConnection)):
             raise errors.PoolError(
                 "Connection instance not subclass of MySQLConnection.")
 
@@ -275,7 +277,7 @@ def add_connection(self, cnx=None):
                     "Failed adding connection; queue is full")
 
             if not cnx:
-                cnx = MySQLConnection(**self._cnx_config)
+                cnx = Connect(**self._cnx_config)
                 try:
                     if (self._reset_session and self._cnx_config['compress']
                             and cnx.get_server_version() < (5, 7, 3)):
@@ -291,7 +293,7 @@ def add_connection(self, cnx=None):
                 cnx._pool_config_version = self._config_version
                 # pylint: enable=W0201,W0212
             else:
-                if not isinstance(cnx, MySQLConnection):
+                if not isinstance(cnx, (MySQLConnection, CMySQLConnection)):
                     raise errors.PoolError(
                         "Connection instance not subclass of MySQLConnection.")
 

src/include/mysql_capi.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2014, 2020, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -208,6 +208,9 @@ MySQL_query(MySQL *self, PyObject *args, PyObject *kwds);
 PyObject*
 MySQL_refresh(MySQL *self, PyObject *args);
 
+PyObject*
+MySQL_reset_connection(MySQL *self);
+
 PyObject*
 MySQL_rollback(MySQL *self);
 

src/include/mysql_connector.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2014, 2020, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as

src/mysql_capi.c

@@ -2750,6 +2750,37 @@ MySQL_refresh(MySQL *self, PyObject *args)
     Py_RETURN_NONE;
 }
 
+/**
+  Resets current connection.
+
+  Resets this connection to MySQL.
+
+  @param    self    MySQL instance
+
+  @return   Boolean Object Py_True or Py_False
+    @retval Py_True for success
+    @retval Py_False if an error occurred
+*/
+PyObject*
+MySQL_reset_connection(MySQL *self)
+{
+    int res;
+
+    if (!self->connected)
+    {
+        Py_RETURN_FALSE;
+    }
+
+    res= mysql_reset_connection(&self->session);
+
+    if (!res)
+    {
+        Py_RETURN_TRUE;
+    }
+
+    Py_RETURN_FALSE;
+}
+
 /**
   Shut down the MySQL server.
 

src/mysql_connector.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2014, 2020, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -204,6 +204,9 @@ static PyMethodDef MySQL_methods[]=
     {"refresh", (PyCFunction)MySQL_refresh,
      METH_VARARGS,
      "Flush tables, caches or reset replication server info"},
+    {"reset_connection", (PyCFunction)MySQL_reset_connection,
+     METH_NOARGS,
+     "Resets current connection"},
     {"rollback", (PyCFunction)MySQL_rollback,
      METH_NOARGS,
 	 "Rolls back the current transaction"},

tests/cext/test_cext_connection.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-# Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2014, 2020, Oracle and/or its affiliates. All rights reserved.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License, version 2.0, as
@@ -32,7 +32,9 @@
 """
 
 import tests
+import unittest
 
+from .. import PY2
 from mysql.connector import errors
 from mysql.connector.constants import ClientFlag, flag_is_set
 from mysql.connector.connection import MySQLConnection
@@ -126,4 +128,19 @@ def test_cmd_query(self):
             'warning_count': 0, 'insert_id': 0, 'affected_rows': 0,
             'server_status': 0, 'field_count': 0
         }
-        self.assertEqual(exp, info)
+        self.assertEqual(exp, info)
+
+    @unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 3),
+                     "MySQL >= 5.7.3 is required for reset command")
+    def test_cmd_reset_connection(self):
+        """Resets session without re-authenticating"""
+        exp_session_id = self.cnx.connection_id
+        self.cnx.cmd_query("SET @ham = 2")
+        self.cnx.cmd_reset_connection()
+
+        self.cnx.cmd_query("SELECT @ham")
+        self.assertEqual(exp_session_id, self.cnx.connection_id)
+        if PY2:
+            self.assertNotEqual(('2',), self.cnx.get_rows()[0][0])
+        else:
+            self.assertNotEqual((b'2',), self.cnx.get_rows()[0][0])

tests/test_bugs.py

@@ -63,6 +63,7 @@
 from mysql.connector import (connection, cursor, conversion, protocol,
                              errors, constants, pooling)
 from mysql.connector.optionfiles import read_option_files
+from mysql.connector.pooling import PooledMySQLConnection
 from mysql.connector.catch23 import STRING_TYPES
 import mysql.connector
 import cpy_distutils
@@ -3385,22 +3386,32 @@ def test_binary_charset(self):
 class BugOra19549363(tests.MySQLConnectorTests):
     """BUG#19549363: Compression does not work with Change User
     """
-    def test_compress_reset_connection(self):
-        config = tests.get_mysql_config()
-        config['compress'] = True
+    def setUp(self):
+        self.config = tests.get_mysql_config()
+        self.config['compress'] = True
 
         mysql.connector._CONNECTION_POOLS = {}
-        config['pool_name'] = 'mypool'
-        config['pool_size'] = 3
-        config['pool_reset_session'] = True
-        cnx1 = mysql.connector.connect(**config)
+        self.config['pool_name'] = 'mypool'
+        self.config['pool_size'] = 3
+        self.config['pool_reset_session'] = True
 
-        try:
-            cnx1.close()
-        except:
-            self.fail("Reset session with compression test failed.")
-        finally:
-            mysql.connector._CONNECTION_POOLS = {}
+
+    def tearDown(self):
+        # Remove pools created by test
+        mysql.connector._CONNECTION_POOLS = {}
+
+    def test_compress_reset_connection(self):
+        for use_pure in [True, False]:
+            self.config['use_pure'] = use_pure
+
+            cnx1 = mysql.connector.connect(**self.config)
+
+            try:
+                cnx1.close()
+            except:
+                self.fail("Reset session with compression test failed.")
+            finally:
+                mysql.connector._CONNECTION_POOLS = {}
 
 
 class BugOra19803702(tests.MySQLConnectorTests):
@@ -4564,8 +4575,12 @@ def test_pool_exhaustion(self):
                 self.mysql_server.wait_down()
                 cur.execute(sql)
             except (mysql.connector.errors.OperationalError,
-                    mysql.connector.errors.ProgrammingError):
+                    mysql.connector.errors.ProgrammingError,
+                    mysql.connector.errors.DatabaseError) as err:
                 try:
+                    if isinstance(err, mysql.connector.errors.DatabaseError):
+                        if err.errno != 2006:
+                            raise
                     cur.close()
                     cnx.close()
                 except mysql.connector.errors.OperationalError:
@@ -5689,3 +5704,62 @@ def test_read_default_file_alias(self):
         self.assertEqual("my_pool", conn.pool_name)
         mysql.connector._CONNECTION_POOLS = {}
         conn.close()
+
+@unittest.skipIf(tests.MYSQL_VERSION < (5, 7, 3),
+                 "MySQL >= 5.7.3 is required for reset command")
+@unittest.skipIf(CMySQLConnection is None,
+                 "CMySQLConnection is required but is not available")
+class Bug27489937(tests.MySQLConnectorTests):
+    """BUG#17414258: SUPPORT C EXTENSION FOR CONNECTION POOLS
+    """
+
+    def setUp(self):
+        self.config = tests.get_mysql_config()
+        self.config['pool_name'] = 'Bug27489937'
+        self.config['pool_size'] = 3
+        self.config['use_pure'] = False
+        try:
+            del mysql.connector._CONNECTION_POOLS[self.config['pool_name']]
+        except:
+            pass
+
+    def tearDown(self):
+        # Remove pools created by test
+        del mysql.connector._CONNECTION_POOLS[self.config['pool_name']]
+
+    def test_cext_pool_support(self):
+        """Basic pool tests"""
+        cnx_list = []
+        session_ids = []
+        for _ in range(self.config['pool_size']):
+            cnx = mysql.connector.connect(**self.config)
+            self.assertIsInstance(cnx, PooledMySQLConnection,
+                                  "Expected a CMySQLConnection instance")
+            self.assertIsInstance(cnx._cnx, CMySQLConnection,
+                                  "Expected a CMySQLConnection instance")
+            cnx_list.append(cnx)
+            exp_session_id = cnx.connection_id
+            session_ids.append(exp_session_id)
+            cnx.cmd_query("SET @ham = 2")
+            cnx.cmd_reset_connection()
+
+            cnx.cmd_query("SELECT @ham")
+            self.assertEqual(exp_session_id, cnx.connection_id)
+            if PY2:
+                self.assertNotEqual(('2',), cnx.get_rows()[0][0])
+            else:
+                self.assertNotEqual((b'2',), cnx.get_rows()[0][0])
+        self.assertRaises(errors.PoolError, mysql.connector.connect,
+                          **self.config)
+
+        for cnx in cnx_list:
+            cnx.close()
+
+        cnx = mysql.connector.connect(**self.config)
+        cnx.cmd_query("SELECT @ham")
+        self.assertIn(cnx.connection_id, session_ids,
+                      "Pooled connection was not reused.")
+        if PY2:
+            self.assertNotEqual(('2',), cnx.get_rows()[0][0])
+        else:
+            self.assertNotEqual((b'2',), cnx.get_rows()[0][0])

tests/test_pooling.py

@@ -40,6 +40,7 @@
 import mysql.connector
 from mysql.connector import errors
 from mysql.connector.connection import MySQLConnection
+from mysql.connector.connection_cext import CMySQLConnection
 from mysql.connector import pooling
 from mysql.connector.constants import ClientFlag
 
@@ -237,14 +238,14 @@ def test_add_connection(self):
         pcnx = pooling.PooledMySQLConnection(
             cnxpool,
             cnxpool._cnx_queue.get(block=False))
-        self.assertTrue(isinstance(pcnx._cnx, MySQLConnection))
+        self.assertTrue(isinstance(pcnx._cnx, (CMySQLConnection, MySQLConnection)))
         self.assertEqual(cnxpool, pcnx._cnx_pool)
         self.assertEqual(cnxpool._config_version,
                          pcnx._cnx._pool_config_version)
 
         cnx = pcnx._cnx
         pcnx.close()
-        # We should get the same connectoin back
+        # We should get the same connection back
         self.assertEqual(cnx, cnxpool._cnx_queue.get(block=False))
         cnxpool.add_connection(cnx)
 
@@ -256,7 +257,7 @@ def test_add_connection(self):
         cnxpool._remove_connections()
         cnxpool._cnx_config['port'] = 9999999
         cnxpool._cnx_config['unix_socket'] = '/ham/spam/foobar.socket'
-        self.assertRaises(errors.InterfaceError, cnxpool.add_connection)
+        self.assertRaises(errors.Error, cnxpool.add_connection)
 
         self.assertRaises(errors.PoolError, cnxpool.add_connection, cnx=str)