mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-07 19:35:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

gh-133454: Mark test_queue tests with many threads as bigmem (gh-134575)

Browse source 
50 producer and 50 consumer threads need more than 5GB of memory.
This commit is contained in:
Serhiy Storchaka 2025-06-02 23:25:32 +03:00 • committed by GitHub
parent c7051a3669
commit 7a79f52d83
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

@ -6,7 +6,7 @@ import threading
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
@ -963,33 +963,33 @@ class BaseSimpleQueueTest:
# 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)