| Server IP : 118.27.122.248 / Your IP : 216.73.216.230 Web Server : Apache System : Linux web0264.sh.tyo1 4.18.0-553.79.1.lve.el7h.x86_64 #1 SMP Wed Oct 15 16:34:46 UTC 2025 x86_64 User : c9415830 ( 11735) PHP Version : 8.4.17 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /opt/alt/python36/lib64/python3.6/test/__pycache__/ |
Upload File : |
3
x2_Q( � @ s� d Z ddlZddlZddlZddlZddlmZ ejd�Zddl Z ej
dd� dkrdej dej
��ej � Z
ej� Zejjdko�ejjd kZd
d � Zd d
� Zdd� ZG dd� dej�Zdd� Zedkr�e� dS )z6PyUnit testing that threads honor our signal semantics� N)�support�_thread� �winzCan't test signal on %sZpthreadz
mutex+condc C s4 t j t j| �}t j t j|�}t j t j|�}|||fS )N)�signal�SIGUSR1�SIGUSR2�SIGALRM)Zfor_usr1Zfor_usr2Zfor_alrmZusr1Zusr2Zalrm� r
�</opt/alt/python36/lib64/python3.6/test/test_threadsignals.py�registerSignals s r c C s( t | d d7 < tj� t | d< d S )N�tripped� �
tripped_by)�signal_blackboard�thread� get_ident)�sig�framer
r
r �handle_signals s r c C s( t jttj� t jttj� tj� d S )N)�os�kill� process_pidr r r �
signalled_all�releaser
r
r
r � send_signals" s r c @ s� e Zd Zdd� Zdd� Zdd� Zejed�eje j
j d�oBe j j
d �eje j
j d
�d �d d
� ���Zejed�eje j
j d�o�e j j
d �eje j
j d
�d �dd� ���Zdd� Zdd� Zdd� Zdd� ZdS )�
ThreadSignalsc C s� t j� � tj� | j� tj� W d Q R X ttj d dksPttj d dkrtztj d� tj
� W d tj d� X | j ttj d d� | j ttj d t j
� � | j ttj d d� | j ttj d t j
� � tj� d S )Nr
r r r )r �wait_threads_exitr �acquire�spawnSignallingThreadr r r r �alarm�pauseZ assertEqualr r r )�selfr
r
r � test_signals) s"
zThreadSignals.test_signalsc C s t jtf � d S )N)r �start_new_threadr )r" r
r
r r I s z#ThreadSignals.spawnSignallingThreadc C s t �d S )N)�KeyboardInterrupt)r" r r r
r
r �alarm_interruptL s z ThreadSignals.alarm_interruptz/POSIX condition variables cannot be interrupted�linuxzBIssue 34004: musl does not allow interruption of locks by signals.Zopenbsdz%lock cannot be interrupted on OpenBSDc C s� t j t j| j�}zPtj� }|j� t jd� tj� }| jt |jdd� tj� | }| j
|d� W d t jd� t j t j|� X d S )Nr � )�timeoutg @r ) r r r& r �
allocate_lockr r �time� assertRaisesr% �
assertLess)r" �oldalrm�lock�t1�dtr
r
r � test_lock_acquire_interruptionO s
z,ThreadSignals.test_lock_acquire_interruptionc s� t j t j| j�}z�tj� � � fdd�}tj� �r tj|f � x"� jdd�r^� j � t
j d� q>W t j d� t
j
� }| j
t� jdd� t
j
� | }| j|d � W d Q R X W d t j d
� t j t j|� X d S ) Nc s � j � d S )N)r r
)�rlockr
r � other_thread� s zCThreadSignals.test_rlock_acquire_interruption.<locals>.other_threadF)�blockingg{�G�z�?r r( )r) g @r )r r r& r �RLockr r r$ r r r+ �sleepr r, r% r- )r" r. r4 r0 r1 r
)r3 r � test_rlock_acquire_interruptionn s
z-ThreadSignals.test_rlock_acquire_interruptionc s� d�_ �fdd�}tjtj|�}zr� fdd�}tj� �T tj|f � x"� jdd�rf� j� t j
d� qFW � j� }�j �j � �j |� W d Q R X W d tjtj|� X d S )NFc s
d� _ d S )NT)� sig_recvd)r r )r" r
r �
my_handler� s z9ThreadSignals.acquire_retries_on_intr.<locals>.my_handlerc s6 � j � tjd� tjttj� tjd� � j� d S )Ng �?) r r+ r7 r r r r r r r
)r/ r
r r4 � s
z;ThreadSignals.acquire_retries_on_intr.<locals>.other_thread)r5 g{�G�z�?) r9 r r r r r r$ r r r+ r7 Z
assertTrue)r" r/ r: � old_handlerr4 �resultr
)r/ r" r �acquire_retries_on_intr� s
z%ThreadSignals.acquire_retries_on_intrc C s | j tj� � d S )N)r= r r* )r" r
r
r �!test_lock_acquire_retries_on_intr� s z/ThreadSignals.test_lock_acquire_retries_on_intrc C s | j tj� � d S )N)r= r r6 )r" r
r
r �"test_rlock_acquire_retries_on_intr� s z0ThreadSignals.test_rlock_acquire_retries_on_intrc s� d �_ d �_d�_tj� � � j� tj� ��j� �fdd�}tjtj|�}z���fdd�}� fdd�}tj � �V tj
|f � |� � j� �j �j�j d� �j �j�j d � �j �jd� W d Q R X W d tjtj|� X d S )
Nr c s � j d7 _ d S )Nr )�
sigs_recvd)Zsignumr )r" r
r r: � s z@ThreadSignals.test_interrupted_timed_acquire.<locals>.my_handlerc s$ t j � �_� jdd� t j � �_d S )Ng �?)r) )r+ �startr �endr
)r/ r" r
r �
timed_acquire� s
zCThreadSignals.test_interrupted_timed_acquire.<locals>.timed_acquirec s6 x(t d�D ] } tjd� tjttj� q
W � j� d S )N�( g{�G�z�?) �ranger+ r7 r r r r r r )�_)�doner
r r � s
zBThreadSignals.test_interrupted_timed_acquire.<locals>.send_signalsg @g333333�?)
rA rB r@ r r* r r r r r r$ r- Z
assertGreater)r" r: r; rC r r
)rG r/ r" r � test_interrupted_timed_acquire� s(
z,ThreadSignals.test_interrupted_timed_acquireN)�__name__�
__module__� __qualname__r# r r& �unittestZskipIf�USING_PTHREAD_COND�sys�platform�
startswith� thread_info�versionr2 r8 r= r> r? rH r
r
r
r r ' s* r c C sR t jddd�t jddd�t jddd�iatttt�} ztjt � W d t| � X d S )Nr )r
r )
r r r r r r r r Z run_unittestr )Zoldsigsr
r
r � test_main� s rS �__main__)�__doc__rL r r rN Ztestr �
import_moduler r+ rO ZSkipTest�getpidr r* r rQ �namer/ rM r r r ZTestCaser rS rI r
r
r
r �<module> s*
C