o
    ±ó®g¬(  ã                   @   sÞ   d Z ddlZddlZddlZddlZddlmZ ddlmZ ddl	Z
ddlZejdd… dkr7e dej ¡‚e ¡ Ze
 ¡ ZejjdkoJejjd	kZd
d„ Zdd„ Zdd„ ZG dd„ dejƒZdd„ Zedkrmeƒ  dS dS )z6PyUnit testing that threads honor our signal semanticsé    N)Úsupport)Úthreading_helperé   ÚwinzCan't test signal on %sZpthreadz
mutex+condc                 C   s4   t   t j| ¡}t   t j|¡}t   t j|¡}|||fS ©N)ÚsignalÚSIGUSR1ÚSIGUSR2ÚSIGALRM)Zfor_usr1Zfor_usr2Zfor_alrmZusr1Zusr2Zalrm© r   úM/Users/admin/.pyenv/versions/3.10.0/lib/python3.10/test/test_threadsignals.pyÚregisterSignals   s   
r   c                 C   s(   t |  d  d7  < 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  ttj¡ t  ttj¡ t ¡  d S r   )ÚosÚkillÚprocess_pidr   r   r	   Úsignalled_allÚreleaser   r   r   r   Úsend_signals#   s   r   c                   @   sÐ   e Zd Zdd„ Zdd„ Zdd„ Ze ed¡e e	j
 d¡o!e	jj d	¡e e	j
 d
¡d¡dd„ ƒƒƒZe ed¡e e	j
 d¡oDe	jj d	¡e e	j
 d
¡d¡dd„ ƒƒƒZdd„ Zdd„ Zdd„ Zdd„ ZdS )ÚThreadSignalsc              	   C   sþ   t  ¡  t ¡  |  ¡  t ¡  W d   ƒ n1 sw   Y  ttj d dks2ttj d dkrIzt 	d¡ t 
¡  W t 	d¡ nt 	d¡ w |  ttj d d¡ |  ttj d t ¡ ¡ |  ttj d d¡ |  ttj 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*   

ùÿ

ÿÿzThreadSignals.test_signalsc                 C   s   t  td¡ d S )Nr   )r   Ústart_new_threadr   r#   r   r   r   r    J   s   z#ThreadSignals.spawnSignallingThreadc                 C   s   t ‚r   )ÚKeyboardInterrupt)r$   r   r   r   r   r   Úalarm_interruptM   s   zThreadSignals.alarm_interruptz/POSIX condition variables cannot be interruptedÚlinuxzBIssue 34004: musl does not allow interruption of locks by signals.Zopenbsdz%lock cannot be interrupted on OpenBSDc              
   C   s–   t   t j| j¡}z5t ¡ }| ¡  t  d¡ t ¡ }| j	t
|jdd t ¡ | }|  |d¡ W t  d¡ t   t j|¡ d S t  d¡ t   t j|¡ w )Nr   é   ©Útimeoutç      @r   )r   r
   r(   r   Úallocate_lockr   r!   ÚtimeÚ	monotonicÚassertRaisesr'   Ú
assertLess)r$   ÚoldalrmÚlockÚt1Údtr   r   r   Útest_lock_acquire_interruptionP   s   


ÿz,ThreadSignals.test_lock_acquire_interruptionc              
      s  t   t j| j¡}zut ¡ ‰ ‡ fdd„}t ¡ A t |d¡ ˆ jddr3ˆ  	¡  t
 d¡ ˆ jdds$t  d¡ t
 ¡ }| jtˆ jdd	 t
 ¡ | }|  |d
¡ W d   ƒ n1 s[w   Y  W t  d¡ t   t j|¡ d S W t  d¡ t   t j|¡ d S t  d¡ t   t j|¡ w )Nc                      s   ˆ   ¡  d S r   )r   r   ©Zrlockr   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   r1   r'   r2   )r$   r3   r9   r5   r6   r   r8   r   Útest_rlock_acquire_interruptiono   s.   

þ
õ
ò

ÿz-ThreadSignals.test_rlock_acquire_interruptionc              
      sâ   dˆ_ ‡fdd„}t tj|¡}zX‡ fdd„}t ¡ 2 t |d¡ ˆ jddr7ˆ  ¡  t	 
d¡ ˆ jdds(ˆ  ¡ }ˆ ˆj ¡ ˆ |¡ W d   ƒ n1 sPw   Y  W t tj|¡ d S W t tj|¡ d S t tj|¡ w )	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 ttj¡ t d¡ ˆ  ¡  d S )Nç      à?)	r   r/   r>   r   r   r   r   r   r   r   )r4   r   r   r9   œ   s
   

z;ThreadSignals.acquire_retries_on_intr.<locals>.other_threadr   r:   r<   )r@   r   r   r   r   r   r&   r   r   r/   r>   Z
assertTrue)r$   r4   rA   Úold_handlerr9   Úresultr   ©r4   r$   r   Úacquire_retries_on_intr•   s$   

þø
ö"
z%ThreadSignals.acquire_retries_on_intrc                 C   ó   |   t ¡ ¡ d S r   )rF   r   r.   r#   r   r   r   Ú!test_lock_acquire_retries_on_intrµ   ó   z/ThreadSignals.test_lock_acquire_retries_on_intrc                 C   rG   r   )rF   r   r=   r#   r   r   r   Ú"test_rlock_acquire_retries_on_intr¸   rI   z0ThreadSignals.test_rlock_acquire_retries_on_intrc              
      s  d ˆ_ d ˆ_dˆ_t ¡ ‰ ˆ  ¡  t ¡ ‰ˆ ¡  ‡fdd„}t tj|¡}z]‡‡fdd„}‡ fdd„}t 	¡ 0 t 
|d¡ |ƒ  ˆ  ¡  ˆ ˆjˆj  d	¡ ˆ ˆjˆj  d
¡ ˆ ˆjd¡ W d   ƒ n1 skw   Y  W t tj|¡ d S W t tj|¡ d S t tj|¡ w )Nr   c                    s   ˆ  j d7  _ d S )Nr   )Ú
sigs_recvd)Úsignumr   r#   r   r   rA   Ê   rI   z@ThreadSignals.test_interrupted_timed_acquire.<locals>.my_handlerc                      s$   t  ¡ ˆ_ˆ jdd t  ¡ ˆ_d S )NrB   r+   )r/   r0   Ústartr   Úendr   rE   r   r   Útimed_acquireÎ   s   
zCThreadSignals.test_interrupted_timed_acquire.<locals>.timed_acquirec                     s2   t dƒD ]} t d¡ t ttj¡ 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Ó?)rM   rN   rK   r   r.   r   r   r   r   r   r&   r2   ZassertGreater)r$   rA   rC   rO   r   r   )rS   r4   r$   r   Útest_interrupted_timed_acquire»   s.   
óñ"z,ThreadSignals.test_interrupted_timed_acquireN)Ú__name__Ú
__module__Ú__qualname__r%   r    r(   ÚunittestZskipIfÚUSING_PTHREAD_CONDÚsysÚplatformÚ
startswithÚthread_infoÚversionr7   r?   rF   rH   rJ   rT   r   r   r   r   r   (   s<     ÿþÿÿþÿ r   c                  C   sX   t jdddœt jdddœt jdddœiattttƒ} zt t	¡ W t| Ž  d S t| Ž  w )Nr   )r   r   )
r   r   r	   r
   r   r   r   r   Zrun_unittestr   )Zoldsigsr   r   r   Ú	test_mainê   s   þr_   Ú__main__)Ú__doc__rX   r   r   rZ   Útestr   Ztest.supportr   Ú_threadr   r/   r[   ZSkipTestÚgetpidr   r.   r   r]   Únamer4   rY   r   r   r   ZTestCaser   r_   rU   r   r   r   r   Ú<module>   s2    
ÿ	 C
ÿ