Issue #8865: Concurrent invocation of select.poll.poll() now raises a

RuntimeError exception.  Patch by Christian Schubert.
This commit is contained in:
Serhiy Storchaka 2013-08-20 20:38:21 +03:00
parent ec67d187ee
commit b1973c252c
4 changed files with 57 additions and 2 deletions

View file

@ -1,8 +1,16 @@
# Test case for the os.poll() function
import os, select, random, unittest
import os
import random
import select
import _testcapi
from test.support import TESTFN, run_unittest
try:
import threading
except ImportError:
threading = None
import time
import unittest
from test.support import TESTFN, run_unittest, reap_threads
try:
select.poll
@ -160,6 +168,36 @@ class PollTests(unittest.TestCase):
self.assertRaises(OverflowError, pollster.poll, _testcapi.INT_MAX + 1)
self.assertRaises(OverflowError, pollster.poll, _testcapi.UINT_MAX + 1)
@unittest.skipUnless(threading, 'Threading required for this test.')
@reap_threads
def test_threaded_poll(self):
r, w = os.pipe()
self.addCleanup(os.close, r)
self.addCleanup(os.close, w)
rfds = []
for i in range(10):
fd = os.dup(r)
self.addCleanup(os.close, fd)
rfds.append(fd)
pollster = select.poll()
for fd in rfds:
pollster.register(fd, select.POLLIN)
t = threading.Thread(target=pollster.poll)
t.start()
try:
time.sleep(0.5)
# trigger ufds array reallocation
for fd in rfds:
pollster.unregister(fd)
pollster.register(w, select.POLLOUT)
self.assertRaises(RuntimeError, pollster.poll)
finally:
# and make the call to poll() from the thread return
os.write(w, b'spam')
t.join()
def test_main():
run_unittest(PollTests)