Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
lib64
›
python2.7
›
test
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_threading.pyo
← Back
� z fc @ sL d d l Z d d l m Z m Z d d l m Z d d l Z d d l Z d d l Z e j j d � Z e j j d � Z d d l Z d d l Z d d l Z d d l Z d d l Z y d d l Z Wn e k r� e Z n Xd d l m Z d e f d � � YZ d e j f d � � YZ d e j f d � � YZ d e f d � � YZ d e f d � � YZ d e f d � � YZ d e j f d � � YZ d e j f d � � YZ d e j! f d � � YZ! d e j f d � � YZ" d e j# f d � � YZ# d e j$ f d � � YZ$ d e j% f d � � YZ% d! � Z& e' d" k rHe& � n d S(# i����N( t verboset cpython_only( t assert_python_okt threadt threading( t lock_testst Counterc B s, e Z d � Z d � Z d � Z d � Z RS( c C s d | _ d S( Ni ( t value( t self( ( s+ /usr/lib64/python2.7/test/test_threading.pyt __init__ s c C s | j d 7_ d S( Ni ( R ( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt inc s c C s | j d 8_ d S( Ni ( R ( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt dec s c C s | j S( N( R ( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt get s ( t __name__t __module__R R R R ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR s t TestThreadc B s e Z d � Z d � Z RS( c C s> t j j | d | �| | _ | | _ | | _ | | _ d S( Nt name( R t ThreadR t testcaset semat mutext nrunning( R R R R R R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR $ s c C s t j � d } t r/ d | j | d f GHn | j �� | j �I | j j � t rk | j j � Gd GHn | j j | j j � d � Wd QXt j | � t r� d G| j Gd GHn | j �R | j j � | j j | j j � d � t r d | j | j j � f GHn Wd QXWd QXd S( Ng ��@s task %s will run for %.1f usecg ��.As tasks are runningi t taskt donei s$ %s is finished. %d tasks are running( t randomR R R R R R R R t assertLessEqualt timet sleepR t assertGreaterEqual( R t delay( ( s+ /usr/lib64/python2.7/test/test_threading.pyt run+ s& " ( R R R R ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR # s t BaseTestCasec B s e Z d � Z d � Z RS( c C s t j j � | _ d S( N( t testt test_supportt threading_setupt _threads( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt setUpD s c C s$ t j j | j � t j j � d S( N( R R! t threading_cleanupR# t reap_children( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt tearDownG s ( R R R$ R' ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR C s t ThreadTestsc B s� e Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z e j e e d � d � d � � Z e j e e d � d � d � � Z d � Z RS( c C sC d } t j d d � } t j � } t � } g } xm t | � D]_ } t d | | | | | � } | j | � | j | j � | j t | � d � | j � q@ Wt r� d GHn xg | D]_ } | j | � | j | j � � | j | j d � | j | j � | j t | � d � q� Wt r)d GHn | j | j � d � d S( Ni R i s <thread %d>s ^<TestThread\(.*, initial\)>$s! waiting for all tasks to completei s ^<TestThread\(.*, \w+ -?\d+\)>$s all tasks done( R t BoundedSemaphoret RLockR t rangeR t appendt assertIsNonet identt assertRegexpMatchest reprt startR t joint assertFalset is_alivet assertNotEqualt assertIsNotNonet assertEqualR ( R t NUMTASKSR R t numrunningt threadst it t( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_various_opsP s, c sw | j t j � j � � � f d � } t j � � g � t j | d � � j � | j � d � t j � d =d S( Nc s$ � j t j � j � � j � d S( N( R, R t currentThreadR. t set( ( R R. ( s+ /usr/lib64/python2.7/test/test_threading.pyt fr s i ( ( R6 R R> R. t EventR t start_new_threadt waitt _active( R R@ ( ( R R. s+ /usr/lib64/python2.7/test/test_threading.pyt" test_ident_of_no_threading_threadso s c C s^ t r d GHn y t j d � Wn! t j k rB | j d � n X| j � t j d � d S( Ns with 256kB thread stack size...i s4 platform does not support changing thread stack sizei ( R R t stack_sizeR t errort skipTestR= ( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_various_ops_small_stack~ s c C s^ t r d GHn y t j d � Wn! t j k rB | j d � n X| j � t j d � d S( Ns with 1MB thread stack size...i s4 platform does not support changing thread stack sizei ( R R RF R RG RH R= ( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_various_ops_large_stack� s c C sy d � } t j � } | j � t j | | f � } | j � | j | t j � | j t j | t j � t j | =d S( Nc S s t j � | j � d S( N( R t current_threadt release( R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR@ � s ( R t Lockt acquireR RB t assertInRD t assertIsInstancet _DummyThread( R R@ R t tid( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_foreign_thread� s c sQ y d d l } Wn t k r0 | j d � n X| j j } d t f d � � Y� | j � � } t j � } y) | | j | � | � } x t r� q� WWn � k r� n X| j d � y | j | d � Wn t k r� n Xt j � � t j � � d t j f � � � f d � � Y} | � } t | _ | j � t rGd GHn t rUd GHn | | j d � | � } | j | d � t r�d GHn � j � } | j | � t r�d GHn | j | j � t r�d GHn | | j | j � | � } | j | d � t r d GHn � j d d � | j | j � t r7d GHn | j rM| j � n d S( Ni����s requires ctypest AsyncExcc B s e Z RS( ( R R ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyRT � s s AsyncExc not raisedi t Workerc s e Z � � � f d � Z RS( c sk t j � | _ t | _ y( x! t r> � j � t j d � q WWn$ � k rf t | _ � j � n Xd S( Ng�������?( R t get_identt idt Falset finishedt TrueR? R R ( R ( RT t worker_saw_exceptiont worker_started( s+ /usr/lib64/python2.7/test/test_threading.pyR � s ( R R R ( ( RT R[ R\ ( s+ /usr/lib64/python2.7/test/test_threading.pyRU � s s started worker threads trying nonsensical thread idi s, waiting for worker thread to get starteds" verifying worker hasn't exiteds2 attempting to raise asynch exception in workers5 waiting for worker to say it caught the exceptiont timeouti s all OK -- joining worker( t ctypest ImportErrorRH t pythonapit PyThreadState_SetAsyncExct Exceptiont py_objectR RV t c_longRZ t failR7 t UnboundLocalErrorR RA R t daemonR1 R RC t assertTrueR3 RY RW R2 ( R R^ t set_async_exct exceptionRR t resultRU R< t ret( ( RT R[ R\ s+ /usr/lib64/python2.7/test/test_threading.pyt test_PyThreadState_SetAsyncExc� sd % c C st d � } t j } | t _ zH t j d d � � } | j t j | j � | j | t j k d � Wd | t _ Xd S( Nc W s t j � � d S( N( R RG ( t args( ( s+ /usr/lib64/python2.7/test/test_threading.pyt fail_new_thread s t targetc S s d S( N( t None( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt <lambda> t s: Failed to cleanup _limbo map on failure of Thread.start().( R t _start_new_threadR t assertRaisesR RG R1 R3 t _limbo( R Ro Rt R< ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_limbo_cleanup� s c C s` y d d l } Wn t k r0 | j d � n Xt j t j d d g � } | j | d � d S( Ni����s requires ctypess -csK if 1: import ctypes, sys, time, thread # This lock is used as a simple event variable. ready = thread.allocate_lock() ready.acquire() # Module globals are cleared before __del__ is run # So we save the functions in class dict class C: ensure = ctypes.pythonapi.PyGILState_Ensure release = ctypes.pythonapi.PyGILState_Release def __del__(self): state = self.ensure() self.release(state) def waitingThread(): x = C() ready.release() time.sleep(100) thread.start_new_thread(waitingThread, ()) ready.acquire() # Be sure the other thread is waiting. sys.exit(42) i* ( R^ R_ RH t subprocesst callt syst executableR7 ( R R^ t rc( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_finalize_runnning_thread s c C s� t j t j d d g d t j d t j �} | j | j j � | j | j j � | j � \ } } | j } | j | d k d � | j | d k d t | � � d S( Ns -csO if 1: import sys, threading # A deadlock-killer, to prevent the # testsuite to hang forever def killer(): import os, time time.sleep(2) print 'program blocked; aborting' os._exit(2) t = threading.Thread(target=killer) t.daemon = True t.start() # This is the trace function def func(frame, event, arg): threading.current_thread() return func sys.settrace(func) t stdoutt stderri s interpreted was blockedi s Unexpected error: ( Rx t PopenRz R{ t PIPEt addCleanupR~ t closeR t communicatet returncodeR3 Rh R0 ( R t pR~ R R| ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_finalize_with_trace1 s c C s� t j t j d d g d t j d t j �} | j | j j � | j | j j � | j � \ } } | j | j � d � t j d d | t j � j � } | j | d � d S( Ns -cs� if 1: import threading from time import sleep def child(): sleep(1) # As a non-daemon thread we SHOULD wake up and nothing # should be torn down yet print "Woke up, sleep function is:", sleep threading.Thread(target=child).start() raise SystemExit R~ R s5 Woke up, sleep function is: <built-in function sleep>s ^\[\d+ refs\]Rs ( Rx R� Rz R{ R� R� R~ R� R R� R7 t stript ret subt MULTILINE( R R� R~ R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_join_nondaemon_on_shutdownS s !c C s� t j } t j � } z~ xw t d d � D]f } t j | d � t j d d � � } | j � | j � | � } | j | | d | | f � q( WWd t j | � Xd S( Ni id i Rp c S s d S( N( Rq ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyRr w Rs s& #1703448 triggered after %d trials: %s( R t enumerateRz t getcheckintervalt xranget setcheckintervalR R1 R2 t assertNotIn( R t enumt old_intervalR; R< t l( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_enumerate_after_joinm s c C s� d t f d � � Y} | d t � } t j | � } | j j � ~ | j d | � d d t j | � � �| d t � } t j | � } | j j � ~ | j d | � d d t j | � � �d S( Nt RunSelfFunctionc B s e Z d � Z d � Z RS( c S sH | | _ t j d | j d | f d i | d 6� | _ | j j � d S( NRp Rn t kwargst yet_another( t should_raiseR R t _runR R1 ( R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR � s c S s | j r t � n d S( N( R� t SystemExit( R t other_refR� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s ( R R R R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s R� t msgs %d references still around( t objectRX t weakreft refR R2 R7 Rq Rz t getrefcountRZ ( R R� t cyclic_objectt weak_cyclic_objectt raising_cyclic_objectt weak_raising_cyclic_object( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_no_refcycle_through_target� s t forks test needs fork()c C sB d } t d | � \ } } } | j | d � | j | d � d S( Ns� if 1: import thread, threading, os, time def background_thread(evt): # Creates and registers the _DummyThread instance threading.current_thread() evt.set() time.sleep(10) evt = threading.Event() thread.start_new_thread(background_thread, (evt,)) evt.wait() assert threading.active_count() == 2, threading.active_count() if os.fork() == 0: assert threading.active_count() == 1, threading.active_count() os._exit(0) else: os.wait() s -cRs ( R R7 ( R t codet _t outt err( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_dummy_thread_after_fork� s s needs os.fork()c C s� t j � } t j d � z� x� t d � D]� } t j d d � � } | j � t j � } | d k r� t j | j � r~ d n d � q) | j � t j | d � \ } } | j d | � q) WWd t j | � Xd S( Ni i Rp c S s d S( N( Rq ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyRr � Rs i i ( Rz R� R� R+ R R R1 t osR� t _exitR4 R2 t waitpidR7 ( R R� R; R< t pidt status( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_is_alive_after_fork� s " c C s xt d d � D]� } t j | � } g t | � D] } t j d | j � ^ q2 } x | D] } | j � qZ Wx | D] } | j � qu Wg t | � D] } t j d | j � ^ q� } x | D] } | j � q� Wx | D] } | j � q� W| j t | j � q Wd S( Ni i Rp ( R+ R R) R RN R1 R2 RL Ru t ValueError( R t limitt bsR� R: R< ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_BoundedSemaphore_limit� s + + ( R R R= RE RI RJ RS Rm Rw R} R� R� R� R� t unittestt skipUnlesst hasattrR� R� R� R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR( L s X $ " $$t ThreadJoinOnShutdownc B s� e Z d Z d � Z d � Z e j e e d � d � e j e j e k d � d � � � Z e j e e d � d � e j e j e k d � d � � � Z d � Z e j e e d � d � e j e j e k d � d � � � Z e j e e d � d � e j e j e k d � d � � � Z e j e e d � d � e j e j e k d � d � � � Z e e j e d k d � d � � � Z RS( t freebsd4t freebsd5t freebsd6t netbsd5t os2emxc C s� d | } t j t j d | g d t j �} | j � } | j j � j d d � } | j j � | j | d � | j | d k d � | j | d k d � d S( Ns� if 1: import sys, os, time, threading # a thread, which waits for the main program to terminate def joiningfunc(mainthread): mainthread.join() print 'end of thread' s -cR~ s Rs s end of main end of thread i s interpreter was blockedi s Unexpected error( Rx R� Rz R{ R� RC R~ t readt replaceR� R7 R3 Rh ( R t scriptR� R| t data( ( s+ /usr/lib64/python2.7/test/test_threading.pyt _run_and_join� s $ c C s d } | j | � d S( Ns� if 1: import os t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() time.sleep(0.1) print 'end of main' ( R� ( R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_1_join_on_shutdown� s R� s needs os.fork()s due to known OS bugc C s d } | j | � d S( NsF if 1: childpid = os.fork() if childpid != 0: os.waitpid(childpid, 0) sys.exit(0) t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() print 'end of main' ( R� ( R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_2_join_in_forked_process s c C s d } | j | � d S( Ns9 if 1: main_thread = threading.current_thread() def worker(): childpid = os.fork() if childpid != 0: os.waitpid(childpid, 0) sys.exit(0) t = threading.Thread(target=joiningfunc, args=(main_thread,)) print 'end of main' t.start() t.join() # Should not block: main_thread is already stopped w = threading.Thread(target=worker) w.start() ( R� ( R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt! test_3_join_in_forked_from_thread s c C sx t j t j d | g d t j �} | j � } | j j � j � j d d � } | j | d d � | j | | � d S( Ns -cR~ s Rs i s Unexpected error( Rx R� Rz R{ R� RC R~ R� t decodeR� R7 ( R R� t expected_outputR� R| R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt assertScriptHasOutput4 s !c C s d } | j | d � d S( NsD if 1: import os, time, threading finish_join = False start_fork = False def worker(): # Wait until this thread's lock is acquired before forking to # create the deadlock. global finish_join while not start_fork: time.sleep(0.01) # LOCK HELD: Main thread holds lock across this call. childpid = os.fork() finish_join = True if childpid != 0: # Parent process just waits for child. os.waitpid(childpid, 0) # Child process should just return. w = threading.Thread(target=worker) # Stub out the private condition variable's lock acquire method. # This acquires the lock and then waits until the child has forked # before returning, which will release the lock soon after. If # someone else tries to fix this test case by acquiring this lock # before forking instead of resetting it, the test case will # deadlock when it shouldn't. condition = w._block orig_acquire = condition.acquire call_count_lock = threading.Lock() call_count = 0 def my_acquire(): global call_count global start_fork orig_acquire() # LOCK ACQUIRED HERE start_fork = True if call_count == 0: while not finish_join: time.sleep(0.01) # WORKER THREAD FORKS HERE with call_count_lock: call_count += 1 condition.acquire = my_acquire w.start() w.join() print('end of main') s end of main ( R� ( R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt+ test_4_joining_across_fork_in_worker_thread< s Fc C s d } d } | j | | � d S( Ns if True: import os, time, threading start_fork = False def worker(): # Wait until the main thread has attempted to join this thread # before continuing. while not start_fork: time.sleep(0.01) childpid = os.fork() if childpid != 0: # Parent process just waits for child. (cpid, rc) = os.waitpid(childpid, 0) assert cpid == childpid assert rc == 0 print('end of worker thread') else: # Child process should just return. pass w = threading.Thread(target=worker) # Stub out the private condition variable's _release_save method. # This releases the condition's lock and flips the global that # causes the worker to fork. At this point, the problematic waiter # lock has been acquired once by the waiter and has been put onto # the waiters list. condition = w._block orig_release_save = condition._release_save def my_release_save(): global start_fork orig_release_save() # Waiter lock held here, condition lock released. start_fork = True condition._release_save = my_release_save w.start() w.join() print('end of main thread') s( end of worker thread end of main thread ( R� ( R R� t output( ( s+ /usr/lib64/python2.7/test/test_threading.pyt( test_5_clear_waiter_locks_to_avoid_crash� s 6c C sn d � } g } x= t d � D]/ } t j d | � } | j | � | j � q Wx | D] } | j � qV Wd S( Nc S s< t j � } | d k r+ t j | d � n t j d � d S( Ni ( R� R� R� R� ( R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt do_fork_and_wait� s i Rp ( R+ R R R, R1 R2 ( R R� R: R; R< ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_reinit_tls_after_fork� s s need _testcapi modulec s� � f d � � d � � � � f d � � d � _ t j � } t j � � z< t j � � t j � � x t d � D] } � � qv WWd t j | � Xd S( Nc s � S( N( ( t framet eventt arg( t noop_trace( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s c s s x d Vq Wd S( Nt generator( ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s c s+ � j d k r � � � _ n t � j � S( N( t genRq t next( ( t callbackR� ( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s i ( Rq R� Rz t gettracet settraceR t _testcapit call_in_temporary_c_threadR+ ( R t old_traceR ( ( R� R� R� s+ /usr/lib64/python2.7/test/test_threading.pyt test_frame_tstate_tracing� s ( R� R� R� R� R� N( R R t platforms_to_skipR� R� R� R� R� R� t skipIfRz t platformR� R� R� R� R� R� R R� Rq R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� � s" '' 'H'9't ThreadingExceptionTestsc B sG e Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( c C s- t j � } | j � | j t | j � d S( N( R R R1 Ru t RuntimeError( R R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_start_thread_again s c C s# t j � } | j t | j � d S( N( R RK Ru R� R2 ( R RK ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_joining_current_thread s c C s# t j � } | j t | j � d S( N( R R Ru R� R2 ( R R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_joining_inactive_thread s c C s3 t j � } | j � | j t t | d t � d S( NRg ( R R R1 Ru R� t setattrRZ ( R R ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_daemonize_active_thread s c C sr d } t d | � \ } } } | j | d � | j d | � | j d | � | j d | � | j d | � d S( Ns� if 1: import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1.0/0.0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() s -cRs s Exception in threads" Traceback (most recent call last):t ZeroDivisionErrors Unhandled exception( R R7 RO R� ( R R� R| R� R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_print_exception s c C sr d } t d | � \ } } } | j | d � | j d | � | j d | � | j d | � | j d | � d S( Ns� if 1: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1.0/0.0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) sys.stderr = None running = False t.join() s -cRs s Exception in threads" Traceback (most recent call last):R� s Unhandled exception( R R7 RO R� ( R R� R| R� R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt% test_print_exception_stderr_is_none_1/ s c C sB d } t d | � \ } } } | j | d � | j d | � d S( Ns� if 1: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1.0/0.0 sys.stderr = None t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() s -cRs s Unhandled exception( R R7 R� ( R R� R| R� R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt% test_print_exception_stderr_is_none_2K s ( R R R� R� R� R� R� R� R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� s t LockTestsc B s e Z e e j � Z RS( ( R R t staticmethodR RM t locktype( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� e s t RLockTestsc B s e Z e e j � Z RS( ( R R R� R R* R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� h s t EventTestsc B s e Z e e j � Z RS( ( R R R� R RA t eventtype( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� k s t ConditionAsRLockTestsc B s e Z e e j � Z RS( ( R R R� R t ConditionR� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� n s t ConditionTestsc B s e Z e e j � Z RS( ( R R R� R R� t condtype( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� r s t SemaphoreTestsc B s e Z e e j � Z RS( ( R R R� R t Semaphoret semtype( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� u s t BoundedSemaphoreTestsc B s; e Z e e j � Z e j e j d k d � d � � Z RS( t darwins test macosx problemc C s� d } d } t j t j d | g d t j �} | j � \ } } | j � j d d � } | j | j d d � | j | | � d S( Ns} if True: import threading def recurse(): return recurse() def outer(): try: recurse() except RuntimeError: pass w = threading.Thread(target=outer) w.start() w.join() print('end of main thread') s end of main thread s -cR~ s Rs i s Unexpected error( Rx R� Rz R{ R� R� R� R� R7 R� ( R R� R� R� R~ R R� ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_recursion_limit{ s ( R R R� R R) R� R� R� Rz R� R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyR� x s c C s/ t j j t t t t t t t t t t � d S( N( R R! t run_unittestR� R� R� R� R� R� R� R( R� R� ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt test_main� s t __main__(( t test.test_supportR R R t test.script_helperR R R� Rz R! t import_moduleR R R R� R� R� Rx R� R_ Rq R R� R R R t TestCaseR R( R� R� R� R� R� R� R� R� R� R� R ( ( ( s+ /usr/lib64/python2.7/test/test_threading.pyt <module> sF � �� e#
💻
System Info
Current path
/lib64/python2.7/test
Contents
10 folders, 1437 files
Disk free
100.01 GB
PHP version
8.3.30
🚀
Quick Actions
📍 Script location
🌐 Web root
🗑️ Temp
📁 /lib64/python2.7/test
⚡ Nexus File Manager • 1447 items