锁

Lock

>>> help(type(threading.Lock()))
Help on class lock in module _thread:
class lock(builtins.object)
    锁对象是一个同步原语。
    A lock object is a synchronization primitive.  To create a lock,
    call threading.Lock().  Methods are:
    
    acquire() -- 上锁，在被获取之前可能会一直阻塞
    release() -- 释放锁
    locked() -- 测试锁是否被上锁
    acquire() -- lock the lock, possibly blocking until it can be obtained
    release() -- unlock of the lock
    locked() -- test whether the lock is currently locked
    
    锁不会被对它上锁的线程占用；其他线程可以释放锁。如果线程尝试对已经上锁的锁再次上锁，会被阻塞。
    可能会出现死锁。
    A lock is not owned by the thread that locked it; another thread may
    unlock it.  A thread attempting to lock a lock that it has already locked
    will block until another thread unlocks it.  Deadlocks may ensue.
    
    Methods defined here:
    
    __enter__(...)
        acquire(blocking=True, timeout=-1) -> bool
        (acquire_lock() is an obsolete synonym)
        
        上锁。没有参数时，如果锁已经被上锁（甚至被同一个线程上锁），则会被阻塞，等待其他线程释放锁，
        当锁被成功上锁后会返回 True。
        带有参数时，只会在 blocking=True 时阻塞，然后返回值来反映是否上锁成功。
        Lock the lock.  Without argument, this blocks if the lock is already
        locked (even by the same thread), waiting for another thread to release
        the lock, and return True once the lock is acquired.
        With an argument, this will only block if the argument is true,
        and the return value reflects whether the lock is acquired.
        The blocking operation is interruptible.
    
    __exit__(...)
        release()
        (release_lock() is an obsolete synonym)
        
        释放锁，允许其他被阻塞的、正在等待锁的线程去上锁。锁必须是上锁的状态，但是它不必被同一个上锁它的线程解锁。
        （__enter__ 和 __exit__ 实现了锁的上下文管理器，所以可以使用 with 语句执行锁）
        Release the lock, allowing another thread that is blocked waiting for
        the lock to acquire the lock.  The lock must be in the locked state,
        but it needn't be locked by the same thread that unlocks it.
    
    __repr__(self, /)
        Return repr(self).
    
    acquire(...)
        acquire(blocking=True, timeout=-1) -> bool
        (acquire_lock() is an obsolete synonym)
        
        上锁。无参数时如果锁已经被上锁，会阻塞线程，然后等待其他线程释放锁，当锁被成功上锁后会返回 True。
        当参数 blocking=True 时，会阻塞线程，否则不会阻塞
        Lock the lock.  Without argument, this blocks if the lock is already
        locked (even by the same thread), waiting for another thread to release
        the lock, and return True once the lock is acquired.
        With an argument, this will only block if the argument is true,
        and the return value reflects whether the lock is acquired.
        The blocking operation is interruptible.
    
    acquire_lock(...)
        acquire(blocking=True, timeout=-1) -> bool
        (acquire_lock() is an obsolete synonym)
        
        Lock the lock.  Without argument, this blocks if the lock is already
        locked (even by the same thread), waiting for another thread to release
        the lock, and return True once the lock is acquired.
        With an argument, this will only block if the argument is true,
        and the return value reflects whether the lock is acquired.
        The blocking operation is interruptible.
    
    locked(...)
        locked() -> bool
        (locked_lock() is an obsolete synonym)
        
        Return whether the lock is in the locked state.
    
    locked_lock(...)
        locked() -> bool
        (locked_lock() is an obsolete synonym)
        
        Return whether the lock is in the locked state.
    
    release(...)
        release()
        (release_lock() is an obsolete synonym)
        
        Release the lock, allowing another thread that is blocked waiting for
        the lock to acquire the lock.  The lock must be in the locked state,
        but it needn't be locked by the same thread that unlocks it.
    
    release_lock(...)
        release()
        (release_lock() is an obsolete synonym)
        
        Release the lock, allowing another thread that is blocked waiting for
        the lock to acquire the lock.  The lock must be in the locked state,
        but it needn't be locked by the same thread that unlocks it.

RLock

class _RLock:
    """This class implements reentrant lock objects. 
    这个类实现了可重入锁。

    一个可重入锁必须被获取它的线程释放。一旦一个线程获取了一个可重入锁，这个线程可以再次不阻塞的获取它；
    这个线程每次 acquire 该锁后，必须 release 它。
    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it
    again without blocking; the thread must release it once for each time it
    has acquired it.
    """

    def __init__(self):
        self._block = _allocate_lock()  # 创建锁
        self._owner = None
        self._count = 0

    # 不重要，不用看
    def __repr__(self):
        owner = self._owner
        try:
            owner = _active[owner].name
        except KeyError:
            pass
        return "<%s %s.%s object owner=%r count=%d at %s>" % (
            "locked" if self._block.locked() else "unlocked",
            self.__class__.__module__,
            self.__class__.__qualname__,
            owner,
            self._count,
            hex(id(self))
        )
    
    # 不重要，不用看
    def _at_fork_reinit(self):
        self._block._at_fork_reinit()
        self._owner = None
        self._count = 0

    def acquire(self, blocking=True, timeout=-1):
        """Acquire a lock, blocking or non-blocking. 以阻塞或非阻塞方式获取一个锁。

        当不使用参数调用时：如果线程已经拥有该锁，就将该锁的递归层级+1，并立即返回。否则，
        如果其他线程拥有该锁，在该锁被释放之前会一直阻塞。一旦锁被释放（没有被任何线程占用），
        就会获取该锁的所有权，将递归层级设置为 1 并返回。如果多个线程都被阻塞，并且等待锁被释放，
        一次只会有一个线程能占有该锁。这种情况下没有返回值。
        When invoked without arguments: if this thread already owns the lock,
        increment the recursion level by one, and return immediately. Otherwise,
        if another thread owns the lock, block until the lock is unlocked. Once
        the lock is unlocked (not owned by any thread), then grab ownership, set
        the recursion level to one, and return. If more than one thread is
        blocked waiting until the lock is unlocked, only one at a time will be
        able to grab ownership of the lock. There is no return value in this
        case.

        当以阻塞的方式调用此函数，行为和无参数调用一样，并且返回 True
        When invoked with the blocking argument set to true, do the same thing
        as when called without arguments, and return true.

        当用 blocking=False 调用此函数时，不会阻塞。假设无参数调用时也会阻塞，那么使用这个参数
        就会立即返回 False；否则就会像无参数调用时一样，并返回 True。（也就是说，blocking 这个参数
        指的是这个函数是否会阻塞：False 代表不阻塞，如果没能立刻获取到锁，就立即返回 False；True 
        代表阻塞，如果没能立即获取到锁，就会阻塞在这里等待，不会立刻返回）
        When invoked with the blocking argument set to false, do not block. If a
        call without an argument would block, return false immediately;
        otherwise, do the same thing as when called without arguments, and
        return true.

        如果将 timeout 设置成一个正浮点数，如果没能获取到锁，就会阻塞至少 timeout 秒。
        如果在 timeout 秒内能获取到锁，就返回 True，否则超时返回 False。
        When invoked with the floating-point timeout argument set to a positive
        value, block for at most the number of seconds specified by timeout
        and as long as the lock cannot be acquired.  Return true if the lock has
        been acquired, false if the timeout has elapsed.

        """
        me = get_ident()  # 返回一个整数代表当前线程
        
        # 情况1：当前线程重复获取锁
        if self._owner == me:
            self._count += 1  # 当前线程自身 acquire，递归层级+1
            return 1
        
        # 情况2：其他线程获取锁
        rc = self._block.acquire(blocking, timeout)
        if rc:
            self._owner = me
            self._count = 1
        return rc

    __enter__ = acquire

    def release(self):
        """Release a lock, decrementing the recursion level. 释放锁，减少递归层级。

        如果减少层级后它的值变成 0，重置锁为 unlocked 状态（不被任何线程占有），并且如果有其他
        被阻塞的、正在等待锁变成 unlocked 状态的线程，就会只允许这些线程中的一个能继续执行。
        如果减少层级后它的值还是非零值，这个锁会保持 locked 状态并继续被调用此函数的线程占用。
        If after the decrement it is zero, reset the lock to unlocked (not owned
        by any thread), and if any other threads are blocked waiting for the
        lock to become unlocked, allow exactly one of them to proceed. If after
        the decrement the recursion level is still nonzero, the lock remains
        locked and owned by the calling thread.

        只有当线程占有锁时才能调用这个方法。否则会报错。
        Only call this method when the calling thread owns the lock. A
        RuntimeError is raised if this method is called when the lock is
        unlocked.
        
        此方法没有返回值。
        There is no return value.
        """
        
        if self._owner != get_ident():
            raise RuntimeError("cannot release un-acquired lock")
        
        self._count = count = self._count - 1
        if not count:
            self._owner = None
            self._block.release()

    def __exit__(self, t, v, tb):
        self.release()

    # Internal methods used by condition variables。接下来的三个方法，是给 Condition 变量使用的。

    def _acquire_restore(self, state):
        # 获取锁，并将锁的递归层级、所有者还原(state 是个元组，保存了锁的递归层级、所有者)
        self._block.acquire()
        self._count, self._owner = state

    def _release_save(self):
        # 保存锁的递归层级、所有者，然后将锁的信息清空，并释放锁。
        if self._count == 0:
            raise RuntimeError("cannot release un-acquired lock")
            
        count = self._count
        self._count = 0  # 清除递归层级
        owner = self._owner
        self._owner = None  # 清除所有者
        self._block.release()  # 释放锁
        return (count, owner)  # 返回锁的原始信息

    def _is_owned(self):
        return self._owner == get_ident()

Condition

class Condition:
    """Class that implements a condition variable. 实现 condition 变量的类。

    条件变量允许一个或多个线程 wait，直到它们被别的线程 notify 为止。
    A condition variable allows one or more threads to wait until they are
    notified by another thread.

    如果 lock 这个参数是一个非 None 的值，它必须是一个 Lock 或 RLock 对象，并且它会被用作底层锁。
    否则，一个新的 RLock 对象会被自动创建并用作底层锁。（注意，为了实现条件变量，它用到了两个锁，这个是底层锁）
    If the lock argument is given and not None, it must be a Lock or RLock
    object, and it is used as the underlying lock. Otherwise, a new RLock object
    is created and used as the underlying lock.
    """

    def __init__(self, lock=None):
        if lock is None:
            lock = RLock()
        self._lock = lock
        # 导出锁的 acquire 和 release 方法
        self.acquire = lock.acquire
        self.release = lock.release
        # 如果 lock 对象定义了 _release_save(), _acquire_restore()等，这些方法会覆盖此类中的同名方法
        # 譬如上文中的 RLock 类就实现了这些方法
        try:
            self._release_save = lock._release_save
        except AttributeError:
            pass
        try:
            self._acquire_restore = lock._acquire_restore
        except AttributeError:
            pass
        try:
            self._is_owned = lock._is_owned
        except AttributeError:
            pass
        self._waiters = _deque()

    def _at_fork_reinit(self):
        self._lock._at_fork_reinit()
        self._waiters.clear()

    def __enter__(self):
        return self._lock.__enter__()

    def __exit__(self, *args):
        return self._lock.__exit__(*args)

    def __repr__(self):
        return "<Condition(%s, %d)>" % (self._lock, len(self._waiters))
    
    # 这些方法会被覆盖掉，上文已经提到了。
    def _release_save(self):
        self._lock.release()           # No state to save

    def _acquire_restore(self, x):
        self._lock.acquire()           # Ignore saved state

    def _is_owned(self):
        # Return True if lock is owned by current_thread.
        # This method is called only if _lock doesn't have _is_owned().
        if self._lock.acquire(False):
            self._lock.release()
            return False
        else:
            return True

    def wait(self, timeout=None):
        """Wait until notified or until a timeout occurs. 在被 notify 或 timeout 之前会一直 wait。

        如果线程调用此方法时没有获取锁，就会报出 RuntimeError 错误。
        If the calling thread has not acquired the lock when this method is
        called, a RuntimeError is raised.

        这个方法会释放底层锁，并且在被别的线程中的条件变量调用 notify() 或 notify_all() 之前，又
        或者超时之前会一直阻塞。当被唤醒或超时后，它会重新获取锁，并返回。
        This method releases the underlying lock, and then blocks until it is
        awakened by a notify() or notify_all() call for the same condition
        variable in another thread, or until the optional timeout occurs. Once
        awakened or timed out, it re-acquires the lock and returns.
        
        timeout 参数应该是个浮点型的值，代表了超时的秒数。
        When the timeout argument is present and not None, it should be a
        floating point number specifying a timeout for the operation in seconds
        (or fractions thereof).

        当底层锁是个 RLock 时，它不会被它的 release() 方法释放，因为它之前可能多次被递归的 acquire，所以
        不一定能被释放掉。相反，一个内部的 RLock 接口会被使用，它会真正的释放锁。当锁被重新 acquire 时，
        另一个内部的 RLock 接口被用来还原递归层级。
        When the underlying lock is an RLock, it is not released using its
        release() method, since this may not actually unlock the lock when it
        was acquired multiple times recursively. Instead, an internal interface
        of the RLock class is used, which really unlocks it even when it has
        been recursively acquired several times. Another internal interface is
        then used to restore the recursion level when the lock is reacquired.

        """
        if not self._is_owned():
            raise RuntimeError("cannot wait on un-acquired lock")
            
        waiter = _allocate_lock()      # 创建一个新 Lock 对象
        waiter.acquire()               # 上锁，第一次上锁，因此不会阻塞
        self._waiters.append(waiter)   # 将锁添加到等候列表中
        saved_state = self._release_save()  # 释放底层锁，并保存该锁的状态信息
        gotit = False
        try:    # restore state no matter what (e.g., KeyboardInterrupt)
            if timeout is None:
                waiter.acquire()  # 再次上锁，会阻塞在这里，因为上面 acquire 一次了
                gotit = True
            else:
                if timeout > 0:
                    gotit = waiter.acquire(True, timeout)
                else:
                    gotit = waiter.acquire(False)
            return gotit
        finally:
            self._acquire_restore(saved_state)  # 还原底层锁
            if not gotit:
                try:
                    self._waiters.remove(waiter)  # 如果新锁没有上锁成功，就会从列表删除它
                except ValueError:
                    pass

    # 略
    def wait_for(self, predicate, timeout=None):
        """Wait until a condition evaluates to True.

        predicate should be a callable which result will be interpreted as a
        boolean value.  A timeout may be provided giving the maximum time to
        wait.

        """
        endtime = None
        waittime = timeout
        result = predicate()
        while not result:
            if waittime is not None:
                if endtime is None:
                    endtime = _time() + waittime
                else:
                    waittime = endtime - _time()
                    if waittime <= 0:
                        break
            self.wait(waittime)
            result = predicate()
        return result

    def notify(self, n=1):
        """Wake up one or more threads waiting on this condition, if any.
           唤醒一个或多个等待这个条件变量的线程。
        
        如果调用者没有上锁就执行此方法，会报错。
        If the calling thread has not acquired the lock when this method is
        called, a RuntimeError is raised.
        
        这个方法会唤醒 n 个等待条件变量的线程；如果没有线程在等待，这会是一个空指令，什么都不做。
        This method wakes up at most n of the threads waiting for the condition
        variable; it is a no-op if no threads are waiting.

        """
        if not self._is_owned():
            raise RuntimeError("cannot notify on un-acquired lock")
        
        # waiters 里面存放的是锁
        waiters = self._waiters
        while waiters and n > 0:
            waiter = waiters[0]
            try:
                waiter.release()  # 释放锁，此时 wait() 中被阻塞的锁会不再阻塞，因此调用 wait() 的线程会被唤醒
            except RuntimeError:
                # gh-92530: The previous call of notify() released the lock,
                # but was interrupted before removing it from the queue.
                # It can happen if a signal handler raises an exception,
                # like CTRL+C which raises KeyboardInterrupt.
                pass
            else:
                n -= 1
            try:
                waiters.remove(waiter)
            except ValueError:
                pass

    def notify_all(self):
        """Wake up all threads waiting on this condition.

        If the calling thread has not acquired the lock when this method
        is called, a RuntimeError is raised.

        """
        self.notify(len(self._waiters))

    def notifyAll(self):
        """Wake up all threads waiting on this condition.

        This method is deprecated, use notify_all() instead.

        """
        import warnings
        warnings.warn('notifyAll() is deprecated, use notify_all() instead',
                      DeprecationWarning, stacklevel=2)
        self.notify_all()

Semaphore

class Semaphore:
    """This class implements semaphore objects. 这个类实现了 semaphore 即信号量。

    信号量管理了一个计数器，它代表了 "初始值 + release() 调用次数 - acquire() 调用次数"，初始值默认为 1.
    acquire 方法会阻塞，直到它可以返回并且不让计数器变成负数。
    Semaphores manage a counter representing the number of release() calls minus
    the number of acquire() calls, plus an initial value. The acquire() method
    blocks if necessary until it can return without making the counter
    negative. If not given, value defaults to 1.

    """

    # After Tim Peters' semaphore class, but not quite the same (no maximum)

    def __init__(self, value=1):
        if value < 0:
            raise ValueError("semaphore initial value must be >= 0")
        self._cond = Condition(Lock())  # 使用 Condition 和 Lock 对象
        self._value = value

    def __repr__(self):
        cls = self.__class__
        return (f"<{cls.__module__}.{cls.__qualname__} at {id(self):#x}:"
                f" value={self._value}>")

    def acquire(self, blocking=True, timeout=None):
        """Acquire a semaphore, decrementing the internal counter by one. 获取信号量，内部计数器-1

        无参数调用：如果内部计数器>0，计数器会-1并立即返回。如果计数器==0，在其他线程通过调用 release() 来
        将计数器增大到比 0 大之前都会保持阻塞。如果多个 acquire() 都被阻塞，release() 会随机唤醒其中一个。
        这种情况下没有返回值。
        When invoked without arguments: if the internal counter is larger than
        zero on entry, decrement it by one and return immediately. If it is zero
        on entry, block, waiting until some other thread has called release() to
        make it larger than zero. This is done with proper interlocking so that
        if multiple acquire() calls are blocked, release() will wake exactly one
        of them up. The implementation may pick one at random, so the order in
        which blocked threads are awakened should not be relied on. There is no
        return value in this case.

        当调用者使用 blocking=True 参数来调用，和无参数调用一样，并返回 True
        When invoked with blocking set to true, do the same thing as when called
        without arguments, and return true.

        当调用者使用 blocking=False 参数调用，将不会阻塞。如果无参数情况下会阻塞，那么使用这个参数时会
        立刻返回 False；否则会和无参数一样。（设成 False 不会阻塞调用者，即不能立刻 acquire 就返回 False）
        When invoked with blocking set to false, do not block. If a call without
        an argument would block, return false immediately; otherwise, do the
        same thing as when called without arguments, and return true.
        
        如果 timeout 设置了非 None 值，会最多阻塞 timeout 秒
        When invoked with a timeout other than None, it will block for at
        most timeout seconds.  If acquire does not complete successfully in
        that interval, return false.  Return true otherwise.

        """
        if not blocking and timeout is not None:
            raise ValueError("can't specify timeout for non-blocking acquire")
        rc = False
        endtime = None
        with self._cond:  # condition 对象实现了上下文管理器
            while self._value == 0:  # == 0 时，等待别的线程调用 release() 来将计数器增加到 >0 状态 
                if not blocking:
                    break
                if timeout is not None:
                    if endtime is None:
                        endtime = _time() + timeout
                    else:
                        timeout = endtime - _time()
                        if timeout <= 0:
                            break
                self._cond.wait(timeout)  # 等待别的线程调用 release() 来增加计数器到 >0 的状态
            else:
                self._value -= 1
                rc = True
        return rc

    __enter__ = acquire

    def release(self, n=1):
        """Release a semaphore, incrementing the internal counter by one or more.
           释放信号量，给计数器的值 +n
        
        当计数器为 0 且别的线程正在等待它变得大于 0 时，会唤醒等待它的线程。
        When the counter is zero on entry and another thread is waiting for it
        to become larger than zero again, wake up that thread.

        """
        if n < 1:
            raise ValueError('n must be one or more')
        with self._cond:
            self._value += n
            for i in range(n):
                self._cond.notify()  # 通知别的线程

    def __exit__(self, t, v, tb):
        self.release()

本栏目推荐文章
• Linux的信号管理 [补档-2023-07-30]
• 【LeetCode 2701. 连续递增交易】MySQL用户变量编程得到严格递增连续子序列的开始位置，结束位置，长度等统计信息
• 【Leetcode 2474. 购买量严格增加的客户】MySQL用户变量编程解决严格递增连续子序列问题
• 【LeetCode 2494. 合并在同一个大厅重叠的活动】MySQL用户变量编程解决区间合并问题
• Ansible - 定义变量
• 【LeetCode1747. 应该被禁止的 Leetflex 账户】MySQL用户变量编程；尝试维护一个multiset
• 深入浅出 C 语言：学变量、掌控流程、玩指针，全方位掌握 C 编程技能
• js中的对象，如果赋值给多个变量，那么会有性能问题吗
• 【LeetCode 2142. 每辆车的乘客人数 I】乘客到站等车，车到站载客。MySQL用户变量编程完成
• 采用递归的方法简化函数 在允许创建临时变量的前提下实现求字符串长度

变量 Condition Semaphore 信号 条件变量condition semaphore信号 变量condition条件pyhton 线程 变量 信号 条件 semaphore信号 内核semaphore信号linux 底层semaphore信号 原理 condition_variable变量condition variable 内核semaphore源码 信号 多线 变量 条件 变量lock_guard条件guard