File: //usr/local/lib/python3.7/test/__pycache__/lock_tests.cpython-37.pyc
B
��gq � @ s d Z ddlZddlZddlmZmZ ddlZddlZddlZddl m
Z
dd� ZG dd� de�Z
G d d
� d
ej�ZG dd� de�ZG d
d� de�ZG dd� de�ZG dd� de�ZG dd� de�ZG dd� de�ZG dd� de�ZG dd� de�ZG dd� de�ZdS )z/
Various tests for synchronization primitives.
� N)�start_new_thread�TIMEOUT_MAX)�supportc C s t �d� d S )Ng{�G�z�?)�time�sleep� r r �+/usr/local/lib/python3.7/test/lock_tests.py�_wait s r c @ s2 e Zd ZdZddd�Zdd� Zdd� Zd d
� ZdS )
�Bunchz
A bunch of threads.
Fc s| � �_ |�_g �_g �_| �_t�� �_�j�� � �fdd�}y xt |�D ]}t
|d� qNW W n d�_� Y nX dS )z�
Construct a bunch of `n` threads running the same function `f`.
If `wait_before_exit` is True, the threads won't terminate until
do_finish() is called.
c sD t �� } �j�| � z
� � W d �j�| � x�js<t� q.W X d S )N)� threading� get_ident�started�append�finished� _can_exitr )�tid)�f�selfr r �task% s
zBunch.__init__.<locals>.taskr TN)r �nr
r r r �wait_threads_exit�wait_thread� __enter__�ranger )r r r Zwait_before_exitr �ir )r r r �__init__ s
zBunch.__init__c C s xt | j�| jk rt� qW d S )N)�lenr
r r )r r r r �wait_for_started6 s zBunch.wait_for_startedc C s0 xt | j�| jk rt� qW | j�d d d � d S )N)r r r r r �__exit__)r r r r �wait_for_finished: s
zBunch.wait_for_finishedc C s
d| _ d S )NT)r )r r r r � do_finish@ s zBunch.do_finishN)F)�__name__�
__module__�__qualname__�__doc__r r r r r r r r r
s
r
c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�BaseTestCasec C s t �� | _d S )N)r Zthreading_setup�_threads)r r r r �setUpE s zBaseTestCase.setUpc C s t j| j� t �� d S )N)r Zthreading_cleanupr&