Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-01 06:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity

[3.13] gh-133454: Mark test_queue tests with many threads as bigmem (gh-134575) (GH-135047)

Browse source 
50 producer and 50 consumer threads need more than 5GB of memory.
(cherry picked from commit 7a79f52d83)

Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>
This commit is contained in:
Miss Islington (bot) 2025-06-02 22:49:32 +02:00 • committed by GitHub
parent 7e2ebdbe9c
commit f67dede253
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 10 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

20
Lib/test/test_queue.py
View file

@ -7,7 +7,7 @@
import time
import unittest
import weakref
from test.support import gc_collect
from test.support import gc_collect, bigmemtest
from test.support import import_helper
from test.support import threading_helper
@ -964,33 +964,33 @@ def test_order(self):
# One producer, one consumer => results appended in well-defined order
self.assertEqual(results, inputs)
def test_many_threads(self):
@bigmemtest(size=50, memuse=100*2**20, dry_run=False)
def test_many_threads(self, size):
# Test multiple concurrent put() and get()
N = 50
q = self.q
inputs = list(range(10000))
results = self.run_threads(N, q, inputs, self.feed, self.consume)
results = self.run_threads(size, q, inputs, self.feed, self.consume)
# Multiple consumers without synchronization append the
# results in random order
self.assertEqual(sorted(results), inputs)
def test_many_threads_nonblock(self):
@bigmemtest(size=50, memuse=100*2**20, dry_run=False)
def test_many_threads_nonblock(self, size):
# Test multiple concurrent put() and get(block=False)
N = 50
q = self.q
inputs = list(range(10000))
results = self.run_threads(N, q, inputs,
results = self.run_threads(size, q, inputs,
self.feed, self.consume_nonblock)
self.assertEqual(sorted(results), inputs)
def test_many_threads_timeout(self):
@bigmemtest(size=50, memuse=100*2**20, dry_run=False)
def test_many_threads_timeout(self, size):
# Test multiple concurrent put() and get(timeout=...)
N = 50
q = self.q
inputs = list(range(1000))
results = self.run_threads(N, q, inputs,
results = self.run_threads(size, q, inputs,
self.feed, self.consume_timeout)
self.assertEqual(sorted(results), inputs)