Implemented all benchmarks

Author: 4br3mm0rd

README.md

@@ -0,0 +1,19 @@
+# pysql-benchmark
+
+This repository contains the benchmark of the following solutions for mysql drivers in Python:
+
+- PyMySQL
+- mysqlclient within a threadpool
+- mysqlclient
+- mysqlclient within a threadpool
+- aiomysql
+- asyncmy
+
+# motivation
+
+Because the page of asyncmy offers a benchmark which is limited to only one query, we cannot really estimate the impact of asyncio on the performance.
+This is where this project comes into life.
+
+# results
+
+The benchmark has been run on a Macbook Pro M1 Pro 32GB

__pycache__/aiomysql_bench.cpython-311.pyc

@@ -0,0 +1,30 @@
+import random
+import time
+
+import aiomysql
+
+from benchmark_async import AsyncBenchmark
+
+
+class AIOMySQLBench(AsyncBenchmark):
+    async def run(self):
+        # Create a connection pool using the specified pool class
+        self.pool = await aiomysql.create_pool(
+            host=self.DB_HOST,
+            user=self.DB_USER,
+            password=self.DB_PASSWORD,
+            db=self.DB_NAME,
+            minsize=1,
+            maxsize=self.num_batches,
+        )
+
+        res = {
+            "insert": await self.insert(),
+            "select": await self.select(),
+            "update": await self.update(),
+            "delete": await self.delete(),
+        }
+
+        self.pool.close()
+        await self.pool.wait_closed()
+        return res

__pycache__/asyncmy_bench.cpython-311.pyc

@@ -0,0 +1,30 @@
+import asyncio
+import random
+import time
+from asyncio import Queue
+
+import asyncmy
+
+from benchmark_async import AsyncBenchmark
+
+
+class AsyncMyBench(AsyncBenchmark):
+    async def run(self):
+        # Create a connection pool manually
+        self.pool = await asyncmy.create_pool(
+            host=self.DB_HOST,
+            user=self.DB_USER,
+            password=self.DB_PASSWORD,
+            db=self.DB_NAME,
+            maxsize=self.num_batches,
+        )
+        res = {
+            "insert": await self.insert(),
+            "select": await self.select(),
+            "update": await self.update(),
+            "delete": await self.delete(),
+        }
+
+        self.pool.close()
+        await self.pool.wait_closed()
+        return res

__pycache__/benchmark.cpython-311.pyc

@@ -0,0 +1,88 @@
+import random
+import time
+
+from benchmark_base import BenchmarkBase
+
+
+class Benchmark(BenchmarkBase):
+    def __init__(self) -> None:
+        super().__init__()
+
+    def run(self):
+        return {
+            "insert": self.insert(),
+            "select": self.select(),
+            "update": self.update(),
+            "delete": self.delete(),
+        }
+
+    def insert(self):
+        connection, cursor = self._connect()
+
+        # Drop the table if it exists and create it again
+        cursor.execute(self.drop_table_query)
+        cursor.execute(self.create_table_query)
+
+        start_time = time.time()
+
+        # Insert 1000 rows using pymysql
+        for _ in range(self.NUM_QUERIES):
+            data = f"Sample data {random.randint(1, 1000)}"
+            cursor.execute(self.insert_query, (data,))
+
+        connection.commit()
+        end_time = time.time()
+
+        cursor.close()
+        connection.close()
+
+        return end_time - start_time
+
+    def select(self):
+        connection, cursor = self._connect()
+
+        start_time = time.time()
+
+        # Execute 1000 SELECT queries using pymysql
+        for i in range(1, self.NUM_QUERIES + 1):
+            cursor.execute(self.select_query, (i,))
+            cursor.fetchone()
+
+        end_time = time.time()
+
+        cursor.close()
+        connection.close()
+
+        return end_time - start_time
+
+    def update(self):
+        connection, cursor = self._connect()
+
+        start_time = time.time()
+
+        for i in range(1, self.NUM_QUERIES + 1):
+            new_data = f"Updated data {random.randint(1, 1000)}"
+            cursor.execute(self.update_query, (new_data, i))
+        connection.commit()
+        end_time = time.time()
+
+        cursor.close()
+        connection.close()
+
+        return end_time - start_time
+
+    def delete(self):
+        connection, cursor = self._connect()
+
+        start_time = time.time()
+
+        for i in range(1, self.NUM_QUERIES + 1):
+            cursor.execute(self.delete_query)
+
+        connection.commit()
+        end_time = time.time()
+
+        cursor.close()
+        connection.close()
+
+        return end_time - start_time

__pycache__/benchmark_async.cpython-311.pyc

@@ -0,0 +1,77 @@
+import asyncio
+import random
+import time
+
+import asyncmy
+
+from benchmark_base import BenchmarkBase
+
+
+class AsyncBenchmark(BenchmarkBase):
+    def __init__(self, num_batches=10):
+        super().__init__()
+        self.num_batches = num_batches  # Number of queries to run in parallel at once
+
+    async def insert(self):
+        start_time = time.time()
+
+        async def insert_query(_):
+            async with self.pool.acquire() as connection:
+                async with connection.cursor() as cursor:
+                    data = f"Sample data {random.randint(1, 1000)}"
+                    await cursor.execute(self.insert_query, (data,))
+                    await connection.commit()
+
+        await self._run_in_batches(insert_query)
+
+        end_time = time.time()
+        return end_time - start_time
+
+    async def select(self):
+        start_time = time.time()
+
+        async def select_query(i):
+            async with self.pool.acquire() as connection:
+                async with connection.cursor() as cursor:
+                    await cursor.execute(self.select_query, (i,))
+                    await cursor.fetchone()
+
+        await self._run_in_batches(select_query)
+
+        end_time = time.time()
+        return end_time - start_time
+
+    async def update(self):
+        start_time = time.time()
+
+        async def update_query(i):
+            async with self.pool.acquire() as connection:
+                async with connection.cursor() as cursor:
+                    new_data = f"Updated data {random.randint(1, 1000)}"
+                    await cursor.execute(self.update_query, (new_data, i))
+                    await connection.commit()
+
+        await self._run_in_batches(update_query)
+
+        end_time = time.time()
+        return end_time - start_time
+
+    async def delete(self):
+        start_time = time.time()
+
+        async def delete_query(_):
+            async with self.pool.acquire() as connection:
+                async with connection.cursor() as cursor:
+                    await cursor.execute(self.delete_query)
+                    await connection.commit()
+
+        await self._run_in_batches(delete_query)
+
+        end_time = time.time()
+        return end_time - start_time
+
+    async def _run_in_batches(self, query_func):
+        # Run queries in smaller batches
+        for i in range(0, self.NUM_QUERIES, self.num_batches):
+            batch = range(i + 1, min(i + self.num_batches + 1, self.NUM_QUERIES + 1))
+            await asyncio.gather(*(query_func(j) for j in batch))

__pycache__/benchmark_base.cpython-311.pyc

@@ -0,0 +1,23 @@
+class BenchmarkBase:
+    def __init__(self) -> None:
+        self.DB_HOST = "127.0.0.1"
+        self.DB_USER = "root"
+        self.DB_PASSWORD = ""
+        self.DB_NAME = "benchmark_test"
+        self.NUM_QUERIES = 10000
+
+        self.create_table_query = """
+        CREATE TABLE IF NOT EXISTS test_table (
+            id INT AUTO_INCREMENT PRIMARY KEY,
+            data VARCHAR(100)
+        )
+        """
+        self.drop_table_query = "DROP TABLE IF EXISTS test_table"
+
+        self.insert_query = "INSERT INTO test_table (data) VALUES (%s)"
+        self.select_query = "SELECT * FROM test_table WHERE id = %s"
+        self.delete_query = "DELETE FROM test_table WHERE 1 LIMIT 1"
+        self.update_query = "UPDATE test_table SET data = %s WHERE id = %s"
+
+    def _connect():
+        raise NotImplementedError()

__pycache__/benchmark_threaded.cpython-311.pyc

@@ -0,0 +1,117 @@
+import random
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from benchmark_base import BenchmarkBase
+
+
+class ThreadedBenchmark(BenchmarkBase):
+    def __init__(self):
+        super().__init__()
+        self.NUM_THREADS = 10
+        self.connection_pool = {}
+
+    def _disconnect(self):
+        for thread in self.connection_pool:
+            connection, cursor = self.connection_pool[thread]
+            cursor.close()
+            connection.close()
+
+    def _get_connection_from_pool(self):
+        thread_id = threading.get_ident()
+        if thread_id not in self.connection_pool:
+            self.connection_pool[thread_id] = self._connect()
+        return self.connection_pool[thread_id]
+
+    def run(self):
+        self._disconnect()
+        return {
+            "insert": self.insert(),
+            "select": self.select(),
+            "update": self.update(),
+            "delete": self.delete(),
+        }
+
+    def _run_queries_in_threads(self, query_func):
+        with ThreadPoolExecutor(max_workers=self.NUM_THREADS) as executor:
+            futures = [
+                executor.submit(query_func, i) for i in range(1, self.NUM_QUERIES + 1)
+            ]
+            for future in as_completed(futures):
+                future.result()
+
+    def insert(self):
+        connection, cursor = self._connect()
+
+        # Drop the table if it exists and create it again
+        cursor.execute(self.drop_table_query)
+        cursor.execute(self.create_table_query)
+        cursor.close()
+        connection.close()
+
+        start_time = time.time()
+
+        def insert_query(_):
+            connection, cursor = self._get_connection_from_pool()
+            data = f"Sample data {random.randint(1, 1000)}"
+            cursor.execute(self.insert_query, (data,))
+            connection.commit()
+
+        self._run_queries_in_threads(insert_query)
+        end_time = time.time()
+
+        return end_time - start_time
+
+    def select(self):
+        connection, cursor = self._connect()
+
+        cursor.close()
+        connection.close()
+        start_time = time.time()
+
+        def select_query(i):
+            connection, cursor = self._get_connection_from_pool()
+            cursor.execute(self.select_query, (i,))
+            cursor.fetchone()
+
+        self._run_queries_in_threads(select_query)
+
+        end_time = time.time()
+
+        return end_time - start_time
+
+    def update(self):
+        connection, cursor = self._connect()
+        cursor.close()
+        connection.close()
+        start_time = time.time()
+
+        def update_query(i):
+            connection, cursor = self._get_connection_from_pool()
+            new_data = f"Updated data {random.randint(1, 1000)}"
+            cursor.execute(self.update_query, (new_data, i))
+            connection.commit()
+
+        self._run_queries_in_threads(update_query)
+
+        end_time = time.time()
+
+        return end_time - start_time
+
+    def delete(self):
+        connection, cursor = self._connect()
+        cursor.close()
+        connection.close()
+
+        start_time = time.time()
+
+        def delete_query(_):
+            connection, cursor = self._get_connection_from_pool()
+            cursor.execute(self.delete_query)
+            connection.commit()
+
+        self._run_queries_in_threads(delete_query)
+        end_time = time.time()
+
+        return end_time - start_time