Performance considerations and test

[Safihre]

I have been using yappi and line_profiler a bit, and wanted a place where I can store some of my results for later.
And maybe if anyone has better ideas, then let me know :)

[Safihre]

ThreadPoolExecutor vs Queue.
I thought the overhead of all the stuff done by ThreadPoolExecutor would be higher, but it was not.
In case of 1 thread, we could still considering adding back the direct-function-call instead of threads which I had removed right now.

Threads=1
Funct: 0.2498975999187678
Queue: 0.26235809992067516
TPool: 0.2960000999737531

Threads=2
Funct: 0.23323020001407713
Queue: 0.12882659991737455
TPool: 0.1603575999615714

Code

import queue
import threading
import time
import timeit
from concurrent.futures import ThreadPoolExecutor
from threading import Thread
import concurrent.futures

def worker(inp):
    time.sleep(0.0001)


def q_worker(inp_q):
    while True:
        worker(inp_q.get())
        inp_q.task_done()


nr_threads = 1
news_wrapper = 'abc'

tq = queue.Queue()

tp = ThreadPoolExecutor(max_workers=nr_threads)
tp_tasks = set()

for _ in range(nr_threads):
    Thread(target=q_worker, args=(tq,), daemon=True).start()


print("Funct:", min(timeit.repeat("worker(news_wrapper); worker(news_wrapper); worker(news_wrapper); worker(news_wrapper);", setup="from __main__ import worker, news_wrapper", number=100, repeat=5)))
print("Queue:", min(timeit.repeat("tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.join()", setup="from __main__ import tq, news_wrapper", number=100, repeat=5)))
print("TPool:", min(timeit.repeat("tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); concurrent.futures.wait(tp_tasks); tp_tasks.clear()", setup="import concurrent.futures; from __main__ import worker, tp, tp_tasks, news_wrapper", number=100, repeat=5)))

[Safihre]

Seems SimpleQueue is also much faster then Queue. But we need the task_done in our case.
And of course, we are talking usec differences here 😆

(venv312) > python -m timeit --setup "import queue; q=queue.SimpleQueue(); inp='abc'*1000" "q.put(inp); q.get()"
2000000 loops, best of 5: 98.5 nsec per loop
(venv312) > python -m timeit --setup "import queue; q=queue.Queue(); inp='abc'*1000" "q.put(inp); q.get()"
100000 loops, best of 5: 1.73 usec per loop

[Safihre]

I made a wrapper-class that adds task_done and join to SimpleQueue.
It is faster, but if it's worth it?
I also saw in the cpython source that putting sleep(0) just triggers a GIL switch, so that's all that is needed for testing.

Funct: 0.0047617999371141195
Queue: 0.09013130003586411
TestQ: 0.05271199997514486
TPool: 0.1531970000360161

Code

import multiprocessing
import queue
import threading
import time
import timeit
from concurrent.futures import ThreadPoolExecutor
from threading import Thread
import concurrent.futures

def worker(inp):
    time.sleep(0)


def q_worker(inp_q):
    while True:
        worker(inp_q.get())
        inp_q.task_done()


class Test(queue.SimpleQueue):
    def __init__(self):
        self.all_tasks_done = threading.Condition(threading.Lock())
        self.unfinished_tasks = 0
        super().__init__()

    def task_done(self):
        with self.all_tasks_done:
            unfinished = self.unfinished_tasks - 1
            if unfinished <= 0:
                self.all_tasks_done.notify_all()
            self.unfinished_tasks = unfinished

    def join(self):
        with self.all_tasks_done:
            while self.unfinished_tasks:
                self.all_tasks_done.wait()

    def put(self, item):
        super().put(item)
        self.unfinished_tasks += 1

nr_threads = 2
news_wrapper = 'abc'

tq = queue.Queue()
tp_test = Test()

tp = ThreadPoolExecutor(max_workers=nr_threads)
tp_tasks = set()

for _ in range(nr_threads):
    Thread(target=q_worker, args=(tq,), daemon=True).start()
    Thread(target=q_worker, args=(tp_test,), daemon=True).start()


print("Funct:", min(timeit.repeat("worker(news_wrapper); worker(news_wrapper); worker(news_wrapper); worker(news_wrapper);", setup="from __main__ import worker, news_wrapper", number=1000, repeat=5)))
print("Queue:", min(timeit.repeat("tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.join()", setup="from __main__ import tq, news_wrapper", number=1000, repeat=5)))
print("TestQ:", min(timeit.repeat("tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.join()", setup="from __main__ import tp_test, news_wrapper", number=1000, repeat=5)))
print("TPool:", min(timeit.repeat("tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); concurrent.futures.wait(tp_tasks); tp_tasks.clear()", setup="import concurrent.futures; from __main__ import worker, tp, tp_tasks, news_wrapper", number=1000, repeat=5)))

[Safihre]

Seems we have a winner!
I noticed that queue.Queue uses a deque under the hood, which has an extend method to add multiple items at once.
It's even faster than the SimpleQueue

Funct: 0.003658300032839179
Queue: 0.08986920001916587
TestQ: 0.05976609996287152
Test2: 0.041304800019133836
TPool: 0.16489040001761168

Code

import collections
import multiprocessing
import queue
import threading
import time
import timeit
from concurrent.futures import ThreadPoolExecutor
from threading import Thread
import concurrent.futures

def worker(inp):
    time.sleep(0)


def q_worker(inp_q):
    while True:
        worker(inp_q.get())
        inp_q.task_done()


class Test(queue.SimpleQueue):
    def __init__(self):
        self.all_tasks_done = threading.Condition(threading.Lock())
        self.unfinished_tasks = 0
        super().__init__()

    def task_done(self):
        with self.all_tasks_done:
            unfinished = self.unfinished_tasks - 1
            if unfinished <= 0:
                self.all_tasks_done.notify_all()
            self.unfinished_tasks = unfinished

    def join(self):
        with self.all_tasks_done:
            while self.unfinished_tasks:
                self.all_tasks_done.wait()

    def put(self, item):
        super().put(item)
        self.unfinished_tasks += 1


class Test2(queue.Queue):
    def put_multiple(self, multiple):
        self.queue.extend(multiple)
        self.unfinished_tasks += len(multiple)


nr_threads = 2
news_wrapper = 'abc'
news_wrapper_list = [news_wrapper]*3

tq = queue.Queue()
tp_test = Test()

tq2 = queue.Queue()
tp_test2 = Test()

tp = ThreadPoolExecutor(max_workers=nr_threads)
tp_tasks = set()

for _ in range(nr_threads):
    Thread(target=q_worker, args=(tq,), daemon=True).start()
    Thread(target=q_worker, args=(tp_test,), daemon=True).start()
    Thread(target=q_worker, args=(tp_test2,), daemon=True).start()


print("Funct:", min(timeit.repeat("worker(news_wrapper); worker(news_wrapper); worker(news_wrapper); worker(news_wrapper);", setup="from __main__ import worker, news_wrapper", number=1000, repeat=5)))
print("Queue:", min(timeit.repeat("tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.put(news_wrapper); tq.join()", setup="from __main__ import tq, news_wrapper", number=1000, repeat=5)))
print("TestQ:", min(timeit.repeat("tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.put(news_wrapper); tp_test.join()", setup="from __main__ import tp_test, news_wrapper", number=1000, repeat=5)))
print("Test2:", min(timeit.repeat("tp_test2.put([news_wrapper,news_wrapper,news_wrapper]); tp_test2.join()", setup="from __main__ import tp_test2, news_wrapper, news_wrapper_list", number=1000, repeat=5)))
print("TPool:", min(timeit.repeat("tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); tp_tasks.add(tp.submit(worker, news_wrapper)); concurrent.futures.wait(tp_tasks); tp_tasks.clear()", setup="import concurrent.futures; from __main__ import worker, tp, tp_tasks, news_wrapper", number=1000, repeat=5)))

[thezoggy]

by any chance have you taken a look at py3.12 per interp gil?
https://docs.python.org/3.12/whatsnew/3.12.html#pep-684-a-per-interpreter-gil

[Safihre]

@thezoggy I did, but I couldn't think of a way that we could use it. The multi-interpreters would have their own state, so no shared objects between them. Almost like starting 2 python interpreters or using multiprocessing.
https://peps.python.org/pep-0554/#concerns

The big one for us will be nogil for 3.13!
https://peps.python.org/pep-0703/