mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 17:08:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

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

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:50:43 +02:00 • committed by GitHub
parent bf0d4127b0
commit 9bbb0c5df8
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)