Archives
Trending
Support
Login
clear text
XML
Django
JavaScript
MATLAB
C
C++
C#
Python
SQL
Shell
Bash
Markdown
YAML
JSON
HTML
CSS
PHP
Java
Ruby
Go
Rust
Swift
Kotlin
Arduino
TypeScript
Perl
Autohotkey
Lua
SQF
R
Scala
Haskell
Groovy
Dart
Clojure
VB.NET
Objective-C
PowerShell
Bash
CoffeeScript
Verilog
from contextlib import contextmanager from threading import Condition, Thread import time class SharedLock: def __init__(self): self._cond = Condition() self._readers = 0 @contextmanager def __call__(self, lock_type: str): if lock_type == 'shared': with self._cond: self._readers += 1 yield self with self._cond: self._readers -= 1 if not self._readers: self._cond.notify() elif lock_type == 'exclusive': with self._cond: while self._readers: self._cond.wait() yield self else: raise ValueError(f'Invalid lock_type {lock_type!r}') if __name__ == '__main__': shared_lock = SharedLock() def reader(): with shared_lock('shared'): print('start reading', time.time()) time.sleep(1) print('end reading', time.time()) # Give writer thread a chance to acquire exclusive use: #time.sleep(.1) #with shared_lock('shared'): # print('start reading', time.time()) # time.sleep(1) # print('end reading', time.time()) def writer(): # Give reader threads a chance to acquire shared use: time.sleep(.1) with shared_lock('exclusive'): print('start writing', time.time()) time.sleep(1) print('end writing', time.time()) def test(): reader_threads = [ Thread(target=reader) for _ in range(1) ] writer_threads = [ Thread(target=writer) for _ in range(2) ] for t in reader_threads: t.start() for t in writer_threads: t.start() for t in reader_threads: t.join() for t in writer_threads: t.join() test()
Mark as private (unlisted)
for 30 minutes
for 6 hours
for 1 day
for 1 week
for 1 month
for 1 year
everlasting (like CCCP)