1102368308

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/gevent/gevent/commit/<a class=hub.com/gevent/gevent/commit/<a class="double-link" href="https://git"><a class=hub.com/gevent/gevent/commit/<a class="double-link" href="https://git"><a class=hub.com/gevent/gevent/commit/<a class="double-link" href="https://git"><a class=hub.com/gevent/gevent/commit/5bb9ed7469b0f877347fe80f897cdf133e4b6a9c">5bb9ed746">&lt;a href=&quot;https://github.com/gevent/gevent/commit/</a><a class="double-link" href="https://github.com/gevent/gevent/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/gevent/gevent/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/gevent/gevent/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/gevent/gevent/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/gevent/gevent/commit/5bb9ed7469b0f877347fe80f897cdf133e4b6a9c">5bb9ed746</a><a href="https://github.com/gevent/gevent/commit/5bb9ed7469b0f877347fe80f897cdf133e4b6a9c">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/gevent/gevent/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/gevent/gevent/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/gevent/gevent/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/gevent/gevent/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/gevent/gevent/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/gevent/gevent/commit/5bb9ed7469b0f877347fe80f897cdf133e4b6a9c&quot;&gt;5bb9ed746&lt;/a&gt;&lt;a href=&quot;https://github.com/gevent/gevent/commit/5bb9ed7469b0f877347fe80f897cdf133e4b6a9c&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/gevent/gevent/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/gevent/gevent/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;htt... (continued)

Pull Request Pull Request #1809: Support 3.10 now that the ABI is officially stable with 3.10rc1

Run Details

37 of 62 new or added lines in 15 files covered. (59.68%)

23780 of 28022 relevant lines covered (84.86%)

4.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.61

/src/gevent/tests/test__hub.py

# Copyright (c) 2009 AG Projects
# Author: Denis Bilenko
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import re
import time
import unittest

import gevent.testing as greentest
import gevent.testing.timing

import gevent
from gevent import socket
from gevent.hub import Waiter, get_hub
from gevent._compat import NativeStrIO
from gevent._compat import get_this_psutil_process

DELAY = 0.1


class TestCloseSocketWhilePolling(greentest.TestCase):

    def test(self):
        sock = socket.socket()
        self._close_on_teardown(sock)
        t = get_hub().loop.timer(0)
        t.start(sock.close)
        with self.assertRaises(socket.error):
            try:
                sock.connect(('python.org', 81))
            finally:
                t.close()

        gevent.sleep(0)


class TestExceptionInMainloop(greentest.TestCase):

    def test_sleep(self):
        # even if there was an error in the mainloop, the hub should continue to work
        start = time.time()
        gevent.sleep(DELAY)
        delay = time.time() - start

        delay_range = DELAY * 0.9
        self.assertTimeWithinRange(delay, DELAY - delay_range, DELAY + delay_range)

        error = greentest.ExpectedException('TestExceptionInMainloop.test_sleep/fail')

        def fail():
            raise error

        with get_hub().loop.timer(0.001) as t:
            t.start(fail)

            self.expect_one_error()

            start = time.time()
            gevent.sleep(DELAY)
            delay = time.time() - start

            self.assert_error(value=error)
            self.assertTimeWithinRange(delay, DELAY - delay_range, DELAY + delay_range)



class TestSleep(gevent.testing.timing.AbstractGenericWaitTestCase):

    def wait(self, timeout):
        gevent.sleep(timeout)

    def test_simple(self):
        gevent.sleep(0)


class TestWaiterGet(gevent.testing.timing.AbstractGenericWaitTestCase):

    def setUp(self):
        super(TestWaiterGet, self).setUp()
        self.waiter = Waiter()

    def wait(self, timeout):
        with get_hub().loop.timer(timeout) as evt:
            evt.start(self.waiter.switch, None)
            return self.waiter.get()


class TestWaiter(greentest.TestCase):

    def test(self):
        waiter = Waiter()
        self.assertEqual(str(waiter), '<Waiter greenlet=None>')
        waiter.switch(25)
        self.assertEqual(str(waiter), '<Waiter greenlet=None value=25>')
        self.assertEqual(waiter.get(), 25)

        waiter = Waiter()
        waiter.throw(ZeroDivisionError)
        assert re.match('^<Waiter greenlet=None exc_info=.*ZeroDivisionError.*$', str(waiter)), str(waiter)
        self.assertRaises(ZeroDivisionError, waiter.get)

        waiter = Waiter()
        g = gevent.spawn(waiter.get)
        g.name = 'AName'
        gevent.sleep(0)
        str_waiter = str(waiter)
        self.assertTrue(str_waiter.startswith('<Waiter greenlet=<Greenlet "AName'),
                        str_waiter)

        g.kill()


@greentest.skipOnCI("Racy on CI")
class TestPeriodicMonitoringThread(greentest.TestCase):

    def _reset_hub(self):
        hub = get_hub()
        try:
            del hub.exception_stream
        except AttributeError:
            pass
        if hub._threadpool is not None:
            hub.threadpool.join()
            hub.threadpool.kill()
            del hub.threadpool


    def setUp(self):
        super(TestPeriodicMonitoringThread, self).setUp()
        self.monitor_thread = gevent.config.monitor_thread
        gevent.config.monitor_thread = True
        from gevent.monkey import get_original
        self.lock = get_original('threading', 'Lock')()
        self.monitor_fired = 0
        self.monitored_hubs = set()
        self._reset_hub()

    def tearDown(self):
        hub = get_hub()
        if not self.monitor_thread and hub.periodic_monitoring_thread:
            # If it was true, nothing to do. If it was false, tear things down.
            hub.periodic_monitoring_thread.kill()
            hub.periodic_monitoring_thread = None
        gevent.config.monitor_thread = self.monitor_thread
        self.monitored_hubs = None
        self._reset_hub()
        super(TestPeriodicMonitoringThread, self).tearDown()

    def _monitor(self, hub):
        with self.lock:
            self.monitor_fired += 1
            if self.monitored_hubs is not None:
                self.monitored_hubs.add(hub)

    def test_config(self):
        self.assertEqual(0.1, gevent.config.max_blocking_time)

    def _run_monitoring_threads(self, monitor, kill=True):
        self.assertTrue(monitor.should_run)
        from threading import Condition
        cond = Condition()
        cond.acquire()

        def monitor_cond(_hub):
            cond.acquire()
            cond.notify_all()
            cond.release()
            if kill:
                # Only run once. Especially helpful on PyPy, where
                # formatting stacks is expensive.
                monitor.kill()

        monitor.add_monitoring_function(monitor_cond, 0.01)

        cond.wait()
        cond.release()
        monitor.add_monitoring_function(monitor_cond, None)

    @greentest.ignores_leakcheck
    def test_kill_removes_trace(self):
        from greenlet import gettrace
        hub = get_hub()
        hub.start_periodic_monitoring_thread()
        self.assertIsNotNone(gettrace())
        hub.periodic_monitoring_thread.kill()
        self.assertIsNone(gettrace())

    @greentest.ignores_leakcheck
    def test_blocking_this_thread(self):
        hub = get_hub()
        stream = hub.exception_stream = NativeStrIO()
        monitor = hub.start_periodic_monitoring_thread()
        self.assertIsNotNone(monitor)

        basic_monitor_func_count = 1
        if get_this_psutil_process() is not None:
            # psutil is installed
            basic_monitor_func_count += 1

        self.assertEqual(basic_monitor_func_count,
                         len(monitor.monitoring_functions()))
        monitor.add_monitoring_function(self._monitor, 0.1)
        self.assertEqual(basic_monitor_func_count + 1,
                         len(monitor.monitoring_functions()))
        self.assertEqual(self._monitor, monitor.monitoring_functions()[-1].function)
        self.assertEqual(0.1, monitor.monitoring_functions()[-1].period)

        # We must make sure we have switched greenlets at least once,
        # otherwise we can't detect a failure.
        gevent.sleep(hub.loop.approx_timer_resolution)
        assert hub.exception_stream is stream
        try:
            time.sleep(0.3) # Thrice the default
            self._run_monitoring_threads(monitor)
        finally:
            monitor.add_monitoring_function(self._monitor, None)
            self.assertEqual(basic_monitor_func_count,
                             len(monitor._monitoring_functions))
            assert hub.exception_stream is stream
            monitor.kill()
            del hub.exception_stream


        self.assertGreaterEqual(self.monitor_fired, 1)
        data = stream.getvalue()
        self.assertIn('appears to be blocked', data)
        self.assertIn('PeriodicMonitoringThread', data)

    def _prep_worker_thread(self):
        hub = get_hub()
        threadpool = hub.threadpool

        worker_hub = threadpool.apply(get_hub)
        assert hub is not worker_hub
        stream = NativeStrIO()

        # It does not have a monitoring thread yet
        self.assertIsNone(worker_hub.periodic_monitoring_thread)
        # So switch to it and give it one by letting it run.
        # XXX: Python 3.10 appears to have made some changes in the memory model.
        # Specifically, reading values from the background that are set in the
        # background hub *from this thread* is flaky. It takes them awhile to show up.
        # Really, that's correct and expected from a standard C point of view, as we
        # don't insert any memory barriers or things like that. It just always used to
        # work in the past. So now, rather than read them directly, we need to read them
        # from the background thread itself. The same, apparently, goes for
        # writing.
        # Need to figure out what exactly the change was.
        def task():
            get_hub().exception_stream = stream
            gevent.sleep(0.01)
            mon = get_hub().periodic_monitoring_thread
            mon.add_monitoring_function(self._monitor, 0.1)
            return mon
        worker_monitor = threadpool.apply(task)
        self.assertIsNotNone(worker_monitor)

        return worker_hub, stream, worker_monitor

    @greentest.ignores_leakcheck
    def test_blocking_threadpool_thread_task_queue(self):
        # A threadpool thread spends much of its time
        # blocked on the native Lock object. Unless we take
        # care, if that thread had created a hub, it will constantly
        # be reported as blocked.

        worker_hub, stream, worker_monitor = self._prep_worker_thread()

        # Now wait until the monitoring threads have run.
        self._run_monitoring_threads(worker_monitor)
        worker_monitor.kill()

        # We did run the monitor in the worker thread, but it
        # did NOT report itself blocked by the worker thread sitting there.
        with self.lock:
            self.assertIn(worker_hub, self.monitored_hubs)
            self.assertEqual(stream.getvalue(), '')

    @greentest.ignores_leakcheck
    def test_blocking_threadpool_thread_one_greenlet(self):
        # If the background threadpool thread has no other greenlets to run
        # and never switches, then even if it has a hub
        # we don't report it blocking. The threadpool is *meant* to run
        # tasks that block.

        hub = get_hub()
        threadpool = hub.threadpool
        worker_hub, stream, worker_monitor = self._prep_worker_thread()

        task = threadpool.spawn(time.sleep, 0.3)
        # Now wait until the monitoring threads have run.
        self._run_monitoring_threads(worker_monitor)
        # and be sure the task ran
        task.get()
        worker_monitor.kill()

        # We did run the monitor in the worker thread, but it
        # did NOT report itself blocked by the worker thread
        with self.lock:
            self.assertIn(worker_hub, self.monitored_hubs)
            self.assertEqual(stream.getvalue(), '')


    @greentest.ignores_leakcheck
    def test_blocking_threadpool_thread_multi_greenlet(self):
        # If the background threadpool thread ever switches
        # greenlets, monitoring goes into affect.

        hub = get_hub()
        threadpool = hub.threadpool
        worker_hub, stream, worker_monitor = self._prep_worker_thread()

        def task():
            g = gevent.spawn(time.sleep, 0.7)
            g.join()

        task = threadpool.spawn(task)
        # Now wait until the monitoring threads have run.
        self._run_monitoring_threads(worker_monitor, kill=False)
        # and be sure the task ran
        task.get()
        worker_monitor.kill()

        # We did run the monitor in the worker thread, and it
        # DID report itself blocked by the worker thread
        self.assertIn(worker_hub, self.monitored_hubs)
        data = stream.getvalue()
        self.assertIn('appears to be blocked', data)
        self.assertIn('PeriodicMonitoringThread', data)


class TestLoopInterface(unittest.TestCase):

    def test_implemensts_ILoop(self):
        from gevent.testing import verify
        from gevent._interfaces import ILoop

        loop = get_hub().loop

        verify.verifyObject(ILoop, loop)

    def test_callback_implements_ICallback(self):
        from gevent.testing import verify
        from gevent._interfaces import ICallback

        loop = get_hub().loop

        cb = loop.run_callback(lambda: None)
        verify.verifyObject(ICallback, cb)

    def test_callback_ts_implements_ICallback(self):
        from gevent.testing import verify
        from gevent._interfaces import ICallback

        loop = get_hub().loop

        cb = loop.run_callback_threadsafe(lambda: None)
        verify.verifyObject(ICallback, cb)


class TestHandleError(unittest.TestCase):

    def tearDown(self):
        try:
            del get_hub().handle_error
        except AttributeError:
            pass

    def test_exception_in_custom_handle_error_does_not_crash(self):

        def bad_handle_error(*args):
            raise AttributeError

        get_hub().handle_error = bad_handle_error

        class MyException(Exception):
            pass

        def raises():
            raise MyException

        with self.assertRaises(MyException):
            gevent.spawn(raises).get()


if __name__ == '__main__':
    greentest.main()

1	# Copyright (c) 2009 AG Projects
2	# Author: Denis Bilenko
3	#
4	# Permission is hereby granted, free of charge, to any person obtaining a copy
5	# of this software and associated documentation files (the "Software"), to deal
6	# in the Software without restriction, including without limitation the rights
7	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8	# copies of the Software, and to permit persons to whom the Software is
9	# furnished to do so, subject to the following conditions:
10	#
11	# The above copyright notice and this permission notice shall be included in
12	# all copies or substantial portions of the Software.
13	#
14	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15	# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16	# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17	# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18	# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19	# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
20	# THE SOFTWARE.
21
22	import re	6✔
23	import time	6✔
24	import unittest	6✔
25
26	import gevent.testing as greentest	6✔
27	import gevent.testing.timing	6✔
28
29	import gevent	6✔
30	from gevent import socket	6✔
31	from gevent.hub import Waiter, get_hub	6✔
32	from gevent._compat import NativeStrIO	6✔
33	from gevent._compat import get_this_psutil_process	6✔
34
35	DELAY = 0.1	6✔
36
37
38	class TestCloseSocketWhilePolling(greentest.TestCase):	6✔
39
40	def test(self):	6✔
41	sock = socket.socket()	6✔
42	self._close_on_teardown(sock)	6✔
43	t = get_hub().loop.timer(0)	6✔
44	t.start(sock.close)	6✔
45	with self.assertRaises(socket.error):	6✔
46	try:	6✔
47	sock.connect(('python.org', 81))	6✔
48	finally:
49	t.close()	6✔
50
51	gevent.sleep(0)	6✔
52
53
54	class TestExceptionInMainloop(greentest.TestCase):	6✔
55
56	def test_sleep(self):	6✔
57	# even if there was an error in the mainloop, the hub should continue to work
58	start = time.time()	6✔
59	gevent.sleep(DELAY)	6✔
60	delay = time.time() - start	6✔
61
62	delay_range = DELAY * 0.9	6✔
63	self.assertTimeWithinRange(delay, DELAY - delay_range, DELAY + delay_range)	6✔
64
65	error = greentest.ExpectedException('TestExceptionInMainloop.test_sleep/fail')	6✔
66
67	def fail():	6✔
68	raise error	6✔
69
70	with get_hub().loop.timer(0.001) as t:	6✔
71	t.start(fail)	6✔
72
73	self.expect_one_error()	6✔
74
75	start = time.time()	6✔
76	gevent.sleep(DELAY)	6✔
77	delay = time.time() - start	6✔
78
79	self.assert_error(value=error)	6✔
80	self.assertTimeWithinRange(delay, DELAY - delay_range, DELAY + delay_range)	6✔
81
82
83
84	class TestSleep(gevent.testing.timing.AbstractGenericWaitTestCase):	6✔
85
86	def wait(self, timeout):	6✔
87	gevent.sleep(timeout)	6✔
88
89	def test_simple(self):	6✔
90	gevent.sleep(0)	6✔
91
92
93	class TestWaiterGet(gevent.testing.timing.AbstractGenericWaitTestCase):	6✔
94
95	def setUp(self):	6✔
96	super(TestWaiterGet, self).setUp()	6✔
97	self.waiter = Waiter()	6✔
98
99	def wait(self, timeout):	6✔
100	with get_hub().loop.timer(timeout) as evt:	6✔
101	evt.start(self.waiter.switch, None)	6✔
102	return self.waiter.get()	6✔
103
104
105	class TestWaiter(greentest.TestCase):	6✔
106
107	def test(self):	6✔
108	waiter = Waiter()	6✔
109	self.assertEqual(str(waiter), '<Waiter greenlet=None>')	6✔
110	waiter.switch(25)	6✔
111	self.assertEqual(str(waiter), '<Waiter greenlet=None value=25>')	6✔
112	self.assertEqual(waiter.get(), 25)	6✔
113
114	waiter = Waiter()	6✔
115	waiter.throw(ZeroDivisionError)	6✔
116	assert re.match('^<Waiter greenlet=None exc_info=.ZeroDivisionError.$', str(waiter)), str(waiter)	6✔
117	self.assertRaises(ZeroDivisionError, waiter.get)	6✔
118
119	waiter = Waiter()	6✔
120	g = gevent.spawn(waiter.get)	6✔
121	g.name = 'AName'	6✔
122	gevent.sleep(0)	6✔
123	str_waiter = str(waiter)	6✔
124	self.assertTrue(str_waiter.startswith('<Waiter greenlet=<Greenlet "AName'),	6✔
125	str_waiter)
126
127	g.kill()	6✔
128
129
130	@greentest.skipOnCI("Racy on CI")	6✔
131	class TestPeriodicMonitoringThread(greentest.TestCase):	6✔
132
133	def _reset_hub(self):	6✔
134	hub = get_hub()	×
135	try:	×
136	del hub.exception_stream	×
137	except AttributeError:	×
138	pass	×
139	if hub._threadpool is not None:	×
140	hub.threadpool.join()	×
141	hub.threadpool.kill()	×
142	del hub.threadpool	×
143
144
145	def setUp(self):	6✔
146	super(TestPeriodicMonitoringThread, self).setUp()	×
147	self.monitor_thread = gevent.config.monitor_thread	×
148	gevent.config.monitor_thread = True	×
NEW 149	from gevent.monkey import get_original	×
NEW 150	self.lock = get_original('threading', 'Lock')()	×
151	self.monitor_fired = 0	×
152	self.monitored_hubs = set()	×
153	self._reset_hub()	×
154
155	def tearDown(self):	6✔
156	hub = get_hub()	×
157	if not self.monitor_thread and hub.periodic_monitoring_thread:	×
158	# If it was true, nothing to do. If it was false, tear things down.
159	hub.periodic_monitoring_thread.kill()	×
160	hub.periodic_monitoring_thread = None	×
161	gevent.config.monitor_thread = self.monitor_thread	×
162	self.monitored_hubs = None	×
163	self._reset_hub()	×
164	super(TestPeriodicMonitoringThread, self).tearDown()	×
165
166	def _monitor(self, hub):	6✔
NEW 167	with self.lock:	×
NEW 168	self.monitor_fired += 1	×
NEW 169	if self.monitored_hubs is not None:	×
NEW 170	self.monitored_hubs.add(hub)	×
171
172	def test_config(self):	6✔
173	self.assertEqual(0.1, gevent.config.max_blocking_time)	×
174
175	def _run_monitoring_threads(self, monitor, kill=True):	6✔
176	self.assertTrue(monitor.should_run)	×
177	from threading import Condition	×
178	cond = Condition()	×
179	cond.acquire()	×
180
181	def monitor_cond(_hub):	×
182	cond.acquire()	2✔
NEW 183	cond.notify_all()	×
184	cond.release()	×
185	if kill:	×
186	# Only run once. Especially helpful on PyPy, where
187	# formatting stacks is expensive.
188	monitor.kill()	×
189
190	monitor.add_monitoring_function(monitor_cond, 0.01)	×
191
192	cond.wait()	×
193	cond.release()	×
194	monitor.add_monitoring_function(monitor_cond, None)	2✔
195
196	@greentest.ignores_leakcheck	6✔
197	def test_kill_removes_trace(self):	3✔
198	from greenlet import gettrace	×
199	hub = get_hub()	×
200	hub.start_periodic_monitoring_thread()	×
201	self.assertIsNotNone(gettrace())	×
202	hub.periodic_monitoring_thread.kill()	×
203	self.assertIsNone(gettrace())	×
204
205	@greentest.ignores_leakcheck	6✔
206	def test_blocking_this_thread(self):	3✔
207	hub = get_hub()	×
208	stream = hub.exception_stream = NativeStrIO()	×
209	monitor = hub.start_periodic_monitoring_thread()	×
210	self.assertIsNotNone(monitor)	×
211
212	basic_monitor_func_count = 1	×
213	if get_this_psutil_process() is not None:	×
214	# psutil is installed
215	basic_monitor_func_count += 1	×
216
217	self.assertEqual(basic_monitor_func_count,	×
218	len(monitor.monitoring_functions()))
219	monitor.add_monitoring_function(self._monitor, 0.1)	×
220	self.assertEqual(basic_monitor_func_count + 1,	×
221	len(monitor.monitoring_functions()))
222	self.assertEqual(self._monitor, monitor.monitoring_functions()[-1].function)	×
223	self.assertEqual(0.1, monitor.monitoring_functions()[-1].period)	×
224
225	# We must make sure we have switched greenlets at least once,
226	# otherwise we can't detect a failure.
227	gevent.sleep(hub.loop.approx_timer_resolution)	×
228	assert hub.exception_stream is stream	×
229	try:	×
230	time.sleep(0.3) # Thrice the default	×
231	self._run_monitoring_threads(monitor)	×
232	finally:
233	monitor.add_monitoring_function(self._monitor, None)	×
234	self.assertEqual(basic_monitor_func_count,	×
235	len(monitor._monitoring_functions))
236	assert hub.exception_stream is stream	×
237	monitor.kill()	×
238	del hub.exception_stream	×
239
240
241	self.assertGreaterEqual(self.monitor_fired, 1)	×
242	data = stream.getvalue()	×
243	self.assertIn('appears to be blocked', data)	×
244	self.assertIn('PeriodicMonitoringThread', data)	×
245
246	def _prep_worker_thread(self):	6✔
247	hub = get_hub()	×
248	threadpool = hub.threadpool	×
249
250	worker_hub = threadpool.apply(get_hub)	×
NEW 251	assert hub is not worker_hub	×
NEW 252	stream = NativeStrIO()	×
253
254	# It does not have a monitoring thread yet
255	self.assertIsNone(worker_hub.periodic_monitoring_thread)	×
256	# So switch to it and give it one by letting it run.
257	# XXX: Python 3.10 appears to have made some changes in the memory model.
258	# Specifically, reading values from the background that are set in the
259	# background hub from this thread is flaky. It takes them awhile to show up.
260	# Really, that's correct and expected from a standard C point of view, as we
261	# don't insert any memory barriers or things like that. It just always used to
262	# work in the past. So now, rather than read them directly, we need to read them
263	# from the background thread itself. The same, apparently, goes for
264	# writing.
265	# Need to figure out what exactly the change was.
NEW 266	def task():	×
NEW 267	get_hub().exception_stream = stream	×
NEW 268	gevent.sleep(0.01)	×
NEW 269	mon = get_hub().periodic_monitoring_thread	×
NEW 270	mon.add_monitoring_function(self._monitor, 0.1)	×
NEW 271	return mon	×
NEW 272	worker_monitor = threadpool.apply(task)	×
NEW 273	self.assertIsNotNone(worker_monitor)	×
274
275	return worker_hub, stream, worker_monitor	×
276
277	@greentest.ignores_leakcheck	6✔
278	def test_blocking_threadpool_thread_task_queue(self):	3✔
279	# A threadpool thread spends much of its time
280	# blocked on the native Lock object. Unless we take
281	# care, if that thread had created a hub, it will constantly
282	# be reported as blocked.
283
284	worker_hub, stream, worker_monitor = self._prep_worker_thread()	×
285
286	# Now wait until the monitoring threads have run.
287	self._run_monitoring_threads(worker_monitor)	×
288	worker_monitor.kill()	×
289
290	# We did run the monitor in the worker thread, but it
291	# did NOT report itself blocked by the worker thread sitting there.
NEW 292	with self.lock:	×
NEW 293	self.assertIn(worker_hub, self.monitored_hubs)	×
NEW 294	self.assertEqual(stream.getvalue(), '')	×
295
296	@greentest.ignores_leakcheck	6✔
297	def test_blocking_threadpool_thread_one_greenlet(self):	3✔
298	# If the background threadpool thread has no other greenlets to run
299	# and never switches, then even if it has a hub
300	# we don't report it blocking. The threadpool is meant to run
301	# tasks that block.
302
303	hub = get_hub()	×
304	threadpool = hub.threadpool	×
305	worker_hub, stream, worker_monitor = self._prep_worker_thread()	×
306
307	task = threadpool.spawn(time.sleep, 0.3)	×
308	# Now wait until the monitoring threads have run.
309	self._run_monitoring_threads(worker_monitor)	×
310	# and be sure the task ran
311	task.get()	×
312	worker_monitor.kill()	×
313
314	# We did run the monitor in the worker thread, but it
315	# did NOT report itself blocked by the worker thread
NEW 316	with self.lock:	×
NEW 317	self.assertIn(worker_hub, self.monitored_hubs)	×
NEW 318	self.assertEqual(stream.getvalue(), '')	×
319
320
321	@greentest.ignores_leakcheck	6✔
322	def test_blocking_threadpool_thread_multi_greenlet(self):	3✔
323	# If the background threadpool thread ever switches
324	# greenlets, monitoring goes into affect.
325
326	hub = get_hub()	×
327	threadpool = hub.threadpool	×
328	worker_hub, stream, worker_monitor = self._prep_worker_thread()	×
329
330	def task():	×
331	g = gevent.spawn(time.sleep, 0.7)	×
332	g.join()	×
333
334	task = threadpool.spawn(task)	×
335	# Now wait until the monitoring threads have run.
336	self._run_monitoring_threads(worker_monitor, kill=False)	×
337	# and be sure the task ran
338	task.get()	×
339	worker_monitor.kill()	×
340
341	# We did run the monitor in the worker thread, and it
342	# DID report itself blocked by the worker thread
343	self.assertIn(worker_hub, self.monitored_hubs)	×
344	data = stream.getvalue()	×
345	self.assertIn('appears to be blocked', data)	×
346	self.assertIn('PeriodicMonitoringThread', data)	×
347
348
349	class TestLoopInterface(unittest.TestCase):	6✔
350
351	def test_implemensts_ILoop(self):	6✔
352	from gevent.testing import verify	6✔
353	from gevent._interfaces import ILoop	6✔
354
355	loop = get_hub().loop	6✔
356
357	verify.verifyObject(ILoop, loop)	6✔
358
359	def test_callback_implements_ICallback(self):	6✔
360	from gevent.testing import verify	6✔
361	from gevent._interfaces import ICallback	6✔
362
363	loop = get_hub().loop	6✔
364
365	cb = loop.run_callback(lambda: None)	6✔
366	verify.verifyObject(ICallback, cb)	6✔
367
368	def test_callback_ts_implements_ICallback(self):	6✔
369	from gevent.testing import verify	6✔
370	from gevent._interfaces import ICallback	6✔
371
372	loop = get_hub().loop	6✔
373
374	cb = loop.run_callback_threadsafe(lambda: None)	6✔
375	verify.verifyObject(ICallback, cb)	6✔
376
377
378	class TestHandleError(unittest.TestCase):	6✔
379
380	def tearDown(self):	6✔
381	try:	6✔
382	del get_hub().handle_error	6✔
383	except AttributeError:	×
384	pass	×
385
386	def test_exception_in_custom_handle_error_does_not_crash(self):	6✔
387
388	def bad_handle_error(*args):	6✔
389	raise AttributeError	6✔
390
391	get_hub().handle_error = bad_handle_error	6✔
392
393	class MyException(Exception):	6✔
394	pass	6✔
395
396	def raises():	6✔
397	raise MyException	6✔
398
399	with self.assertRaises(MyException):	6✔
400	gevent.spawn(raises).get()	6✔
401
402
403	if __name__ == '__main__':
404	greentest.main()

gevent / gevent / 1102368308

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous