GameThread slow

[decompile]

Hey there,

I'm facing an issue that I haven't been able to resolve on my own. The problem involves using GameThreads with the pymysql module. Specifically, I'm loading a plugin that fetches queries from a MariaDB database, and I've noticed that the results take significantly longer than running them in the main thread.

I've tested using following snippet:

Syntax: Select all

import pymysql
import time
import threading

connection = pymysql.connect(
    host="",
    user="",
    password="",
    db="",
    charset="",
    cursorclass=pymysql.cursors.DictCursor,
)

cursor = connection.cursor()


def do_thread():
    current_time = time.time()

    cursor.execute("...")
    result = cursor.fetchall()

    print(len(result))

    now = time.time()
    print(f"Time: {now - current_time}")


thread = threading.Thread(target=do_thread)
thread.start()

I've ran the code through:
- Windows Python3
- Windows CS:S Source.Python

Windows Python3:

Code: Select all

2445
Time: 0.18452095985412598

Windows CS:S Source.Python:

Code: Select all

2445
Time: 27.85248613357544

Additional Note - When running it outside a thread in Windows CS:S Source.Python, the output looks like this:

2445
Time: 0.09724783897399902

Version:

Code: Select all

sp info
[Source.Python]
IMPORTANT: Please copy the full output.
--------------------------------------------------------
Checksum      : fe51d1cec1f41c7e6aa1434392a82db3
Date          : 2023-07-19 18:09:17.378787
OS            : Windows-10-10.0.19041
Game          : css
SP version    : 720
Github commit : ad71981b892d4324cf49a07005a858c7682ffe92
Server plugins:
   00: Source.Python, (C) 2012-2021, Source.Python Team.
   01: Tickrate_Enabler 0.4, Didrole
SP plugins:
   00: mysql
--------------------------------------------------------

Not sure if this issue is windows specific. Any help would be appreciated.

[Articha]

Any help? All right. Do you sure that I/O operations have more priority in alternate threads? I/O operations in separate threads can have lower priority than main, which leads to block of the auxiliary thread

[decompile]

Any help? All right. Do you sure that I/O operations have more priority in alternate threads? I/O operations in separate threads can have lower priority than main, which leads to block of the auxiliary thread

Hey! I've tried to set srcds.exe to the highest priority but sadly without any success..

[L'In20Cible]

My first guess would be that the execution is not necessarily slow, but rather the GIL is more than likely deadlocked. For instance, does calling do_thread from the main thread resume the threaded call?

Syntax: Select all

thread = threading.Thread(target=do_thread)
thread.start()
do_thread()

EDIT: What about this?

Syntax: Select all

from os import system
from listeners import on_tick_listener_manager
from listeners.tick import GameThread

class FocusedGameThread(GameThread):
    refcount = 0

    @staticmethod
    def _():
        system('')

    @classmethod
    def incref(cls):
        if not cls.refcount:
            on_tick_listener_manager.register_listener(cls._)
        cls.refcount += 1

    @classmethod
    def decref(cls):
        if not cls.refcount:
            return
        cls.refcount -= 1
        if cls.refcount:
            return
        on_tick_listener_manager.unregister_listener(cls._)

    def start(self, *args, **kwargs):
        self.incref()
        try:
            super().start(*args, **kwargs)
        except:
            self.decref()
            raise

    def run(self, *args, **kwargs):
        try:
            super().run(*args, **kwargs)
        finally:
            self.decref()

[decompile]

Hey,

thank you for investing your time into my issue. I've got amazing results!

I've replaced threading.Thread with your FocusedGameThread class, and I've got the following results:

Code: Select all

2567
Time: 0.08876204490661621

Syntax: Select all

# Python
from os import system
import pymysql
import time

# Source.Python
from listeners import on_tick_listener_manager
from listeners.tick import GameThread


connection = pymysql.connect(
    host="",
    user="",
    password="",
    db="",
    charset="",
    cursorclass=pymysql.cursors.DictCursor,
)

cursor = connection.cursor()


def do_thread():
    current_time = time.time()

    cursor.execute("...")
    result = cursor.fetchall()

    print(len(result))

    now = time.time()
    print(f"Time: {now - current_time}")


class FocusedGameThread(GameThread):
    refcount = 0

    @staticmethod
    def _():
        system("")

    @classmethod
    def incref(cls):
        if not cls.refcount:
            on_tick_listener_manager.register_listener(cls._)
        cls.refcount += 1

    @classmethod
    def decref(cls):
        if not cls.refcount:
            return
        cls.refcount -= 1
        if cls.refcount:
            return
        on_tick_listener_manager.unregister_listener(cls._)

    def start(self, *args, **kwargs):
        self.incref()
        try:
            super().start(*args, **kwargs)
        except:
            self.decref()
            raise

    def run(self, *args, **kwargs):
        try:
            super().run(*args, **kwargs)
        finally:
            self.decref()


thread = FocusedGameThread(target=do_thread)
thread.start()

When running it like:

Syntax: Select all

thread = threading.Thread(target=do_thread)
thread.start()

I get following results:

Code: Select all

2567
Time: 21.3028781414032

This is huge!

[L'In20Cible]

This is huge!

os.system is quite heavy so I implemented that on the c++ side instead and patched GameThread directly.

Test build: css-d2fe149.zip

[decompile]

Hey,

I've tried it out now and the snippet works as expected with your newly updated GameThread class. Thank you.

Syntax: Select all

thread = GameThread(target=do_thread)
thread.start()

Code: Select all

2582
Time: 0.06981301307678223

However, when wrapping the GameThread class, the slow process is sadly back..

Syntax: Select all

# Python
import pymysql
import time
import queue

# Source.Python
from listeners.tick import GameThread


class ThreadedMySQL(GameThread):
    def __init__(self):
        self.queue = queue.Queue()

        self.connected = False
        self.connection = None
        self.cursor = None

        super().__init__()

    def connect(self, *args, **kwargs):
        self.connection = pymysql.connect(*args, **kwargs)
        self.cursor = self.connection.cursor()
        self.connected = True

    def execute(self, query, args=()):
        return self.cursor.execute(query, args)

    def fetchall(self):
        return self.cursor.fetchall()

    def run(self):
        while self.connected:
            while not self.queue.empty():
                callback, args, kwargs = self.queue.get_nowait()

                self.queue.task_done()

                callback(*args, **kwargs)

    def function_call(self, callback, args=(), kwargs=None):
        self.queue.put_nowait((callback, args, {} if kwargs is None else kwargs))


SQL = ThreadedMySQL()
SQL.connect(
    host="",
    user="",
    password="",
    db="",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)
SQL.start()


def do_thread():
    current_time = time.time()

    SQL.execute("...")

    result = SQL.fetchall()

    print(len(result))

    now = time.time()
    print(f"Time: {now - current_time}")


SQL.function_call(do_thread)

Code: Select all

2582
Time: 29.230831146240234

[L'In20Cible]

However, when wrapping the GameThread class, the slow process is sadly back..

You override GameThread.run but never super'call the original. Therefore, the bear is never poked. Just create your own poker: ../listeners/tick.py#L73

[decompile]

So I could just import:

Syntax: Select all

from _listeners._tick import ThreadPoker

and update the run function to: (??)

Syntax: Select all

def run(self):
    with ThreadPoker():
        while self.connected:
            while not self.queue.empty():
                callback, args, kwargs = self.queue.get_nowait()

                self.queue.task_done()

                callback(*args, **kwargs)

or how should it look like?

[L'In20Cible]

So I could just import:

Syntax: Select all

from _listeners._tick import ThreadPoker

You can import it from listeners.tick.

and update the run function to: (??)

Syntax: Select all

def run(self):
    with ThreadPoker():
        while self.connected:
            while not self.queue.empty():
                callback, args, kwargs = self.queue.get_nowait()

                self.queue.task_done()

                callback(*args, **kwargs)

or how should it look like?

If you don't super'call the original then that's correct.

[decompile]

It works! Thank you so much for your effort and investigating. <3

Code: Select all

2609
Time: 0.06781792640686035

[decompile]

Hey, looks like something is still not right. I cant get it to work consistent..

Syntax: Select all

# Python
import pymysql
import time
import queue

# Source.Python
from listeners.tick import GameThread
from listeners.tick import ThreadPoker


class ThreadedMySQL(GameThread):
    def __init__(self):
        self.queue = queue.Queue()

        self.connected = False
        self.connection = None
        self.cursor = None

        super().__init__()

    def connect(self, *args, **kwargs):
        self.connection = pymysql.connect(*args, **kwargs)
        self.cursor = self.connection.cursor()
        self.connected = True

    def execute(self, query, args=()):
        return self.cursor.execute(query, args)

    def fetchall(self):
        return self.cursor.fetchall()

    def run(self, *args, **kwargs):
        while ThreadPoker():
            while self.connected:
                while not self.queue.empty():
                    callback, args, kwargs = self.queue.get_nowait()

                    self.queue.task_done()

                    callback(*args, **kwargs)

    def function_call(self, callback, args=(), kwargs=None):
        self.queue.put_nowait((callback, args, {} if kwargs is None else kwargs))


SQL = ThreadedMySQL()
SQL.connect(
    host="",
    user="",
    password="",
    db="",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)
SQL.start()


def do_thread():
    current_time = time.time()

    SQL.execute("...")

    result = SQL.fetchall()

    print(len(result))

    now = time.time()
    print(f"Time: {now - current_time}")


SQL.function_call(do_thread)

2609
Time: 21.491878986358643

Code: Select all

sp info
[Source.Python]
IMPORTANT: Please copy the full output.
--------------------------------------------------------
Checksum      : be4d272097d8d270248b0800dad19ba4
Date          : 2023-07-24 22:05:04.179113
OS            : Windows-10-10.0.19041
Game          : css
SP version    : 721
Github commit : f129713a17014795de445ff8fa782e079c76ef4e
Server plugins:
   00: Source.Python, (C) 2012-2021, Source.Python Team.
   01: Tickrate_Enabler 0.4, Didrole
SP plugins:
   00: mysql
--------------------------------------------------------

[L'In20Cible]

99.9% sure you should remove the self.queue.task_done() call. IIRC, this join to the main thread.

EDIT: This also does not look quite right:

Syntax: Select all

while ThreadPoker():

In any case, I think you are making it overly hard to maintain with a single class that tries to be them all (an infinite thread, a queue, and a database...).

[decompile]

It's a class I use for everything database related, which works fine on my linux machines but is very slow on my windows server.

Removing self.queue.task_done() gives me the same results. How is while ThreadPoker(): properly used in my example?

[L'In20Cible]

How is while ThreadPoker(): properly used in my example?

It should be with, not while. I'm about to push something that should make your life easier.

[L'In20Cible]

With 21d3dff you could simply do something like this to handle queued calls:

Syntax: Select all

from functools import partial
from listeners.tick import InfiniteThread

class Queued(list):
    def __call__(self, func, *args, **kwargs):
        self.append(partial(func, *args, **kwargs))

    def process(self):
        try:
            self.pop(0)()
        except IndexError:
            return

queued = Queued()
thread = InfiniteThread(target=queued.process).start()

def foo(*args, **kwargs):
    print(f'foo called with {args} and {kwargs}')

queued(foo)
queued(foo, 1, 1, bar=1)
queued(foo, bar=2)
queued(foo, 3, 3, bar=3, baz=3)
queued(foo, 4, 4, bar=4)
queued(foo, 5, 5)
queued(foo, 6, 6, bar=6)

[decompile]

Yeah, youre right.. No idea how I haven't noticed using while instead of with, haha.

Changing it to:

Syntax: Select all

def run(self):
    with ThreadPoker():
        while self.connected:
            while not self.queue.empty():
                callback, args, kwargs = self.queue.get_nowait()
                callback(*args, **kwargs)
            
            # time.sleep(0.1)  # Reduce CPU Load

works, but it somehow changes the server tickrate to exactly 10.00.

Before:

Code: Select all

stats
CPU    In_(KB/s)  Out_(KB/s)  Uptime  Map_changes  FPS      Players  Connects
0.00   0.00       0.00        0       0            100.25   0        0

After:

Code: Select all

[SP] Successfully loaded plugin 'mysql'.
2609
Time: 0.061354875564575195
stats
CPU    In_(KB/s)  Out_(KB/s)  Uptime  Map_changes  FPS      Players  Connects
104.47 0.00       0.00        0       0            10.00    0        0

[L'In20Cible]

but it somehow changes the server tickrate to exactly 10.00.

Are you running multiple threads that execute CPU-bound routines at the same time (e.g. by reloading your plugin while the current one is still running)? If so, then that's likely your issue. For example, Thread would also suffer from this:

Syntax: Select all

from math import sqrt
from threading import Thread

for _ in range(10):
    Thread(target=lambda: [
        sum([sqrt(i) for i in range(10000)])
        for i in range(10000)]
    ).start()

Granted that it's likely more noticeable now with GameThread since poking the system is effectively yielding the processor but, if you have multiple parallel threads that try to take a bite then there won't be much left for the server beside crumbs.

[decompile]

I'm simply just starting srcds, wait until the map is loaded and load my snippet from here by typing sp plugin load mysql, which then caps the tickrate / fps to 10.00. If I would reload the plugin several times in the same session, then there would be indeed several threads running, but the output I sent in here are always after a clean start.

Code: Select all

sp info
[Source.Python]
IMPORTANT: Please copy the full output.
--------------------------------------------------------
Checksum      : be4d272097d8d270248b0800dad19ba4
Date          : 2023-07-24 22:05:04.179113
OS            : Windows-10-10.0.19041
Game          : css
SP version    : 721
Github commit : f129713a17014795de445ff8fa782e079c76ef4e
Server plugins:
   00: Source.Python, (C) 2012-2021, Source.Python Team.
   01: Tickrate_Enabler 0.4, Didrole
SP plugins:
   00: mysql
--------------------------------------------------------

[L'In20Cible]

I'm simply just starting srcds, wait until the map is loaded and load my snippet from here by typing sp plugin load mysql, which then caps the tickrate / fps to 10.00. If I would reload the plugin several times in the same session, then there would be indeed several threads running, but the output I sent in here are always after a clean start.

This:

Syntax: Select all

def run(self):
    with ThreadPoker():
        while self.connected:
            while not self.queue.empty():
                callback, args, kwargs = self.queue.get_nowait()
                callback(*args, **kwargs)

Should rather be:

Syntax: Select all

def run(self):
    while self.connected:
        while not self.queue.empty():
            with ThreadPoker():
                callback, args, kwargs = self.queue.get_nowait()
                callback(*args, **kwargs)

Because you only want to poke the system when processing a call. That said, I would recommend you use InfiniteThread that does extra safety. You can leave your code almost as is but override __call__ instead of run:

Syntax: Select all

class ThreadedMySQL(InfiniteThread):
    def __init__(self):
        self.queue = queue.Queue()

        self.connected = False
        self.connection = None
        self.cursor = None

        super().__init__()

    def connect(self, *args, **kwargs):
        self.connection = pymysql.connect(*args, **kwargs)
        self.cursor = self.connection.cursor()
        self.connected = True

    def execute(self, query, args=()):
        return self.cursor.execute(query, args)

    def fetchall(self):
        return self.cursor.fetchall()

    def __call__(self, *args, **kwargs):
        if self.queue.empty():
            return

        with self.poker:
            callback, args, kwargs = self.queue.get()
            callback(*args, **kwargs)

    def function_call(self, callback, args=(), kwargs=None):
        self.queue.put_nowait((callback, args, {} if kwargs is None else kwargs))

As for the fps, it's normal. There are no player, so the server is hibernating and yielding all available resources. Should get back to normal whenever a player joins.