<s> import os <EOL> import unittest <EOL> import random <EOL> from test import support <EOL> thread = support . import_module ( '<STR_LIT>' ) <EOL> import time <EOL> import sys <EOL> import weakref <EOL> from test import lock_tests <EOL> NUMTASKS = <NUM_LIT:10> <EOL> NUMTRIPS = <NUM_LIT:3> <EOL> _print_mutex = thread . allocate_lock ( ) <EOL> def verbose_print ( arg ) : <EOL> """<STR_LIT>""" <EOL> if support . verbose : <EOL> with _print_mutex : <EOL> print ( arg ) <EOL> class BasicThreadTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . done_mutex = thread . allocate_lock ( ) <EOL> self . done_mutex . acquire ( ) <EOL> self . running_mutex = thread . allocate_lock ( ) <EOL> self . random_mutex = thread . allocate_lock ( ) <EOL> self . created = <NUM_LIT:0> <EOL> self . running = <NUM_LIT:0> <EOL> self . next_ident = <NUM_LIT:0> <EOL> class ThreadRunningTests ( BasicThreadTest ) : <EOL> def newtask ( self ) : <EOL> with self . running_mutex : <EOL> self . next_ident += <NUM_LIT:1> <EOL> verbose_print ( "<STR_LIT>" % self . next_ident ) <EOL> thread . start_new_thread ( self . task , ( self . next_ident , ) ) <EOL> self . created += <NUM_LIT:1> <EOL> self . running += <NUM_LIT:1> <EOL> def task ( self , ident ) : <EOL> with self . random_mutex : <EOL> delay = random . random ( ) / <NUM_LIT> <EOL> verbose_print ( "<STR_LIT>" % ( ident , round ( delay * <NUM_LIT> ) ) ) <EOL> time . sleep ( delay ) <EOL> verbose_print ( "<STR_LIT>" % ident ) <EOL> with self . running_mutex : <EOL> self . running -= <NUM_LIT:1> <EOL> if self . created == NUMTASKS and self . running == <NUM_LIT:0> : <EOL> self . done_mutex . release ( ) <EOL> def test_starting_threads ( self ) : <EOL> for i in range ( NUMTASKS ) : <EOL> self . newtask ( ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> self . done_mutex . acquire ( ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> def test_stack_size ( self ) : <EOL> self . assertEqual ( thread . stack_size ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> thread . stack_size ( <NUM_LIT:0> ) <EOL> self . assertEqual ( thread . stack_size ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> if os . name not in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> return <EOL> tss_supported = True <EOL> try : <EOL> thread . stack_size ( <NUM_LIT> ) <EOL> except ValueError : <EOL> verbose_print ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> except thread . error : <EOL> tss_supported = False <EOL> verbose_print ( "<STR_LIT>" <EOL> "<STR_LIT:size>" ) <EOL> if tss_supported : <EOL> fail_msg = "<STR_LIT>" <EOL> for tss in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0> ) : <EOL> thread . stack_size ( tss ) <EOL> self . assertEqual ( thread . stack_size ( ) , tss , fail_msg % tss ) <EOL> verbose_print ( "<STR_LIT>" % tss ) <EOL> for tss in ( <NUM_LIT> , <NUM_LIT> ) : <EOL> verbose_print ( "<STR_LIT>" % tss ) <EOL> self . next_ident = <NUM_LIT:0> <EOL> self . created = <NUM_LIT:0> <EOL> for i in range ( NUMTASKS ) : <EOL> self . newtask ( ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> self . done_mutex . acquire ( ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> thread . stack_size ( <NUM_LIT:0> ) <EOL> def test__count ( self ) : <EOL> orig = thread . _count ( ) <EOL> mut = thread . allocate_lock ( ) <EOL> mut . acquire ( ) <EOL> started = [ ] <EOL> def task ( ) : <EOL> started . append ( None ) <EOL> mut . acquire ( ) <EOL> mut . release ( ) <EOL> thread . start_new_thread ( task , ( ) ) <EOL> while not started : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( thread . _count ( ) , orig + <NUM_LIT:1> ) <EOL> mut . release ( ) <EOL> done = [ ] <EOL> wr = weakref . ref ( task , lambda _ : done . append ( None ) ) <EOL> del task <EOL> while not done : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( thread . _count ( ) , orig ) <EOL> def test_save_exception_state_on_error ( self ) : <EOL> def task ( ) : <EOL> started . release ( ) <EOL> raise SyntaxError <EOL> def mywrite ( self , * args ) : <EOL> try : <EOL> raise ValueError <EOL> except ValueError : <EOL> pass <EOL> real_write ( self , * args ) <EOL> c = thread . _count ( ) <EOL> started = thread . allocate_lock ( ) <EOL> with support . captured_output ( "<STR_LIT>" ) as stderr : <EOL> real_write = stderr . write <EOL> stderr . write = mywrite <EOL> started . acquire ( ) <EOL> thread . start_new_thread ( task , ( ) ) <EOL> started . acquire ( ) <EOL> while thread . _count ( ) > c : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertIn ( "<STR_LIT>" , stderr . getvalue ( ) ) <EOL> class Barrier : <EOL> def __init__ ( self , num_threads ) : <EOL> self . num_threads = num_threads <EOL> self . waiting = <NUM_LIT:0> <EOL> self . checkin_mutex = thread . allocate_lock ( ) <EOL> self . checkout_mutex = thread . allocate_lock ( ) <EOL> self . checkout_mutex . acquire ( ) <EOL> def enter ( self ) : <EOL> self . checkin_mutex . acquire ( ) <EOL> self . waiting = self . waiting + <NUM_LIT:1> <EOL> if self . waiting == self . num_threads : <EOL> self . waiting = self . num_threads - <NUM_LIT:1> <EOL> self . checkout_mutex . release ( ) <EOL> return <EOL> self . checkin_mutex . release ( ) <EOL> self . checkout_mutex . acquire ( ) <EOL> self . waiting = self . waiting - <NUM_LIT:1> <EOL> if self . waiting == <NUM_LIT:0> : <EOL> self . checkin_mutex . release ( ) <EOL> return <EOL> self . checkout_mutex . release ( ) <EOL> class BarrierTest ( BasicThreadTest ) : <EOL> def test_barrier ( self ) : <EOL> self . bar = Barrier ( NUMTASKS ) <EOL> self . running = NUMTASKS <EOL> for i in range ( NUMTASKS ) : <EOL> thread . start_new_thread ( self . task2 , ( i , ) ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> self . done_mutex . acquire ( ) <EOL> verbose_print ( "<STR_LIT>" ) <EOL> def task2 ( self , ident ) : <EOL> for i in range ( NUMTRIPS ) : <EOL> if ident == <NUM_LIT:0> : <EOL> delay = <NUM_LIT:0> <EOL> else : <EOL> with self . random_mutex : <EOL> delay = random . random ( ) / <NUM_LIT> <EOL> verbose_print ( "<STR_LIT>" % <EOL> ( ident , round ( delay * <NUM_LIT> ) ) ) <EOL> time . sleep ( delay ) <EOL> verbose_print ( "<STR_LIT>" % ( ident , i ) ) <EOL> self . bar . enter ( ) <EOL> verbose_print ( "<STR_LIT>" % ident ) <EOL> with self . running_mutex : <EOL> self . running -= <NUM_LIT:1> <EOL> finished = self . running == <NUM_LIT:0> <EOL> if finished : <EOL> self . done_mutex . release ( ) <EOL> class LockTests ( lock_tests . LockTests ) : <EOL> locktype = thread . allocate_lock <EOL> class TestForkInThread ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . read_fd , self . write_fd = os . pipe ( ) <EOL> @ unittest . skipIf ( sys . platform . startswith ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> @ support . reap_threads <EOL> def test_forkinthread ( self ) : <EOL> def thread1 ( ) : <EOL> try : <EOL> pid = os . fork ( ) <EOL> except RuntimeError : <EOL> os . _exit ( <NUM_LIT:1> ) <EOL> if pid == <NUM_LIT:0> : <EOL> try : <EOL> os . close ( self . read_fd ) <EOL> os . write ( self . write_fd , b"<STR_LIT:OK>" ) <EOL> finally : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> else : <EOL> os . close ( self . write_fd ) <EOL> thread . start_new_thread ( thread1 , ( ) ) <EOL> self . assertEqual ( os . read ( self . read_fd , <NUM_LIT:2> ) , b"<STR_LIT:OK>" , <EOL> "<STR_LIT>" ) <EOL> def tearDown ( self ) : <EOL> try : <EOL> os . close ( self . read_fd ) <EOL> except OSError : <EOL> pass <EOL> try : <EOL> os . close ( self . write_fd ) <EOL> except OSError : <EOL> pass <EOL> def test_main ( ) : <EOL> support . run_unittest ( ThreadRunningTests , BarrierTest , LockTests , <EOL> TestForkInThread ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
