|
Server : LiteSpeed System : Linux server51.dnsbootclub.com 4.18.0-553.62.1.lve.el8.x86_64 #1 SMP Mon Jul 21 17:50:35 UTC 2025 x86_64 User : nandedex ( 1060) PHP Version : 8.1.33 Disable Function : NONE Directory : /opt/cppython/lib/python3.8/test/__pycache__/ |
U
>��g^( � @ s� d dl Z d dlZd dlZd dlZd dlmZ d dlZd dlZej dd� dkr^e �
dej ��e� � Z e�
� Zejjdko�ejjdkZdd � Zd
d � Zd d
� ZG dd� de j�Zdd� Zedkr�e� dS )� N)�support� �winzCan't test signal on %sZpthreadz
mutex+condc C s4 t � t j| �}t � t j|�}t � t j|�}|||fS �N)�signal�SIGUSR1�SIGUSR2�SIGALRM)Zfor_usr1Zfor_usr2Zfor_alrmZusr1Zusr2Zalrm� r
�6/opt/cppython/lib/python3.8/test/test_threadsignals.py�registerSignals s r c C s( t | d d7 < t�� 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 �ttj� t �ttj� t�� d S r )�os�kill� process_pidr r r �
signalled_all�releaser
r
r
r � send_signals" s r c @ s� e Zd Zdd� Zdd� Zdd� Ze�ed�e�e j
� d�oBe j j
d �e�e j
� d
�d �d d
� ���Ze�ed�e�e j
� d�o�e j j
d �e�e j
� d
�d �dd� ���Zdd� Zdd� Zdd� Zdd� ZdS )�
ThreadSignalsc C s� t �� � t�� | �� t�� W 5 Q R X ttj d dksPttj d dkrtzt� d� t�
� W 5 t� d� X | � ttj d d� | � ttj d t �
� � | � ttj d d� | � ttj d t �
� � t�� 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 �td� d S )Nr
)r �start_new_threadr r" r
r
r r I s z#ThreadSignals.spawnSignallingThreadc C s t �d S r )�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 � t j| j�}zPt�� }|�� t �d� t�� }| j t
|jdd� t�� | }| � |d� W 5 t �d� t � t j|� X d S )Nr r � ��timeout� @) r r r' r r �
allocate_lockr �time� monotonic� assertRaisesr& �
assertLess)r# �oldalrm�lock�t1�dtr
r
r � test_lock_acquire_interruptionO s
z,ThreadSignals.test_lock_acquire_interruptionc
s� t � t j| j�}z�t�� � � fdd�}t�� �n t�|d� � j dd�r\� �
� t � d� q<t �d� t �
� }| jt� j d d
� t �
� | }| �|d � W 5 Q R X W 5 t �d� t � t j|� X d S ) Nr c s � � � d S r )r r
�Zrlockr
r � other_thread� s zCThreadSignals.test_rlock_acquire_interruption.<locals>.other_threadr
F��blocking�{�G�z�?r r) r* r, )r r r' r r �RLockr r r% r r r. �sleepr/ r0 r& r1 )r# r2 r8 r4 r5 r
r7 r � test_rlock_acquire_interruptionn s
z-ThreadSignals.test_rlock_acquire_interruptionc
s� d�_ �fdd�}t�tj|�}zn� fdd�}t�� �P t�|d� � jdd�rd� �� t �
d� qD� �� }�� �j � �� |� W 5 Q R X W 5 t�tj|� 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 � � � t�d� t�ttj� t�d� � �� d S )N� �?) r r. r= r r r r r r r
)r3 r
r r8 � s
z;ThreadSignals.acquire_retries_on_intr.<locals>.other_threadr
r9 r; ) r? r r r r r r% r r r. r= Z
assertTrue)r# r3 r@ � old_handlerr8 �resultr
�r3 r# r �acquire_retries_on_intr� s
z%ThreadSignals.acquire_retries_on_intrc C s | � t�� � d S r )rE r r- r" r
r
r �!test_lock_acquire_retries_on_intr� s z/ThreadSignals.test_lock_acquire_retries_on_intrc C s | � t�� � d S r )rE r r< r" r
r
r �"test_rlock_acquire_retries_on_intr� s z0ThreadSignals.test_rlock_acquire_retries_on_intrc
s� d �_ d �_d�_t�� � � �� t�� ���� �fdd�}t�tj|�}z���fdd�}� fdd�}t� � �V t�
|d� |� � �� �� �j�j d � �� �j�j d
� �� �jd� W 5 Q R X W 5 t�tj|� X d S ) Nr c s � j d7 _ d S )Nr )�
sigs_recvd)�signumr r" r
r r@ � s z@ThreadSignals.test_interrupted_timed_acquire.<locals>.my_handlerc s$ t �� �_� jdd� t �� �_d S )NrA r* )r. r/ �startr �endr
rD r
r �
timed_acquire� s
zCThreadSignals.test_interrupted_timed_acquire.<locals>.timed_acquirec s2 t d�D ] } t�d� t�ttj� q� �� d S )N�( g{�G�z�?) �ranger. r= r r r r r r )�_)�doner
r r � s
zBThreadSignals.test_interrupted_timed_acquire.<locals>.send_signalsr
g @g333333�?)
rJ rK rH r r- r r r r r r% r1 Z
assertGreater)r# r@ rB rL r r
)rP r3 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�versionr6 r> rE rF rG rQ r
r
r
r r ' s: ������ r c C sR t jddd�t jddd�t jddd�iatttt�} zt�t � W 5 t| � X d S )Nr )r
r )
r r r r r r r r Z run_unittestr )Zoldsigsr
r
r � test_main� s � r\ �__main__)rU r r rW �testr �_threadr r. rX ZSkipTest�getpidr r- r rZ �namer3 rV r r r ZTestCaser r\ rR r
r
r
r �<module> s*
� C