Python 3.10 optimisation¶
In 3.10 an optimisation (introduced by commit https://github.com/python/cpython/commit/4958f5d69dd2bf86866c43491caf72f774ddec97 ) made it instead release and acquire the GIL only at specific bytecode instructions, rather than at any of them. In your example, your code is a loop of [LOAD_GLOBAL counter, LOAD_CONST 1, INPLACE_ADD, STORE_GLOBAL counter], importantly None of these are magic "eval breaking" instructions that check the GIL (the final JUMP_ABSOLUTE however is, and so each iteration of the loop is a potential GIL release+acquire point). That means that the load of counter, adding 1, and storing it back, all happen atomically because there were no bytecode instructions in this sequence that caused the eval breaker to do its periodic GIL release+acquire cycle, meaning the GIL is held the entire time. This explains the behaviour you see.
In [2]:
import dis
In [3]:
def race_condition(count):
global shared #global variable
for _ in range(count):
shared += 1 #shared = shared+1
dis.dis(race_condition)
1 0 RESUME 0
3 2 LOAD_GLOBAL 1 (NULL + range)
14 LOAD_FAST 0 (count)
16 PRECALL 1
20 CALL 1
30 GET_ITER
>> 32 FOR_ITER 12 (to 58)
34 STORE_FAST 1 (_)
4 36 LOAD_GLOBAL 2 (shared)
48 LOAD_CONST 1 (1)
50 BINARY_OP 13 (+=)
54 STORE_GLOBAL 1 (shared)
56 JUMP_BACKWARD 13 (to 32)
3 >> 58 LOAD_CONST 0 (None)
60 RETURN_VALUE
In [4]:
def race_condition(count):
global shared #global variable
for _ in range(count):
shared += int(1) #shared = shared+1
dis.dis(race_condition)
1 0 RESUME 0
3 2 LOAD_GLOBAL 1 (NULL + range)
14 LOAD_FAST 0 (count)
16 PRECALL 1
20 CALL 1
30 GET_ITER
>> 32 FOR_ITER 25 (to 84)
34 STORE_FAST 1 (_)
4 36 LOAD_GLOBAL 2 (shared)
48 LOAD_GLOBAL 5 (NULL + int)
60 LOAD_CONST 1 (1)
62 PRECALL 1
66 CALL 1
76 BINARY_OP 13 (+=)
80 STORE_GLOBAL 1 (shared)
82 JUMP_BACKWARD 26 (to 32)
3 >> 84 LOAD_CONST 0 (None)
86 RETURN_VALUE
Lock (synchronise)¶
https://docs.python.org/3/library/threading.html#lock-objects
In [ ]:
def race_condition_lock(lock, count):
global shared #global variable
for _ in range(count):
lock.acquire() #start of critical section
shared += int(1) #shared = shared+1
lock.release() #end of critical section
lock = threading.Lock()
In [ ]:
def race_condition_lock(lock, count):
global shared #global variable
for _ in range(count):
with lock: #critical section
shared += int(1) #shared = shared+1
In [ ]:
Generate random values at random time¶
In [3]:
import threading
from random import randrange
import time
#number of values as parameter
def thread_gen(count):
name = threading.current_thread().name
print(f'Thread {name} started.')
for _ in range(count):
time.sleep(0.3*randrange(1, 7))
print(f'Thread {name} generated: {randrange(1000, 10000)}')
print(f'Thread {name} stopped.')
In [4]:
thread_gen(5)
Thread MainThread started. Thread MainThread generated: 2141 Thread MainThread generated: 2554 Thread MainThread generated: 1492 Thread MainThread generated: 9828 Thread MainThread generated: 5988 Thread MainThread stopped.
In [5]:
# run in parallel 7 times
In [6]:
threads = [threading.Thread(target=thread_gen, args=(5,), name=f'PDS thread {1+i}') for i in range(7)]
#start
for thread in threads:
thread.start()
#wait for finish
for thread in threads:
thread.join()
Thread PDS thread 1 started. Thread PDS thread 2 started. Thread PDS thread 3 started. Thread PDS thread 4 started. Thread PDS thread 5 started. Thread PDS thread 6 started. Thread PDS thread 7 started. Thread PDS thread 3 generated: 6322 Thread PDS thread 2 generated: 7570 Thread PDS thread 4 generated: 6367 Thread PDS thread 1 generated: 6578 Thread PDS thread 6 generated: 4325 Thread PDS thread 7 generated: 9785 Thread PDS thread 5 generated: 5306 Thread PDS thread 1 generated: 6789 Thread PDS thread 3 generated: 1455 Thread PDS thread 6 generated: 5322 Thread PDS thread 2 generated: 5641 Thread PDS thread 4 generated: 2096 Thread PDS thread 4 generated: 3897 Thread PDS thread 7 generated: 6034 Thread PDS thread 4 generated: 3611 Thread PDS thread 5 generated: 1223 Thread PDS thread 3 generated: 2307 Thread PDS thread 1 generated: 5624 Thread PDS thread 4 generated: 5533 Thread PDS thread 4 stopped. Thread PDS thread 2 generated: 5524Thread PDS thread 6 generated: 2655 Thread PDS thread 1 generated: 2044 Thread PDS thread 7 generated: 2391 Thread PDS thread 3 generated: 7475Thread PDS thread 6 generated: 1814 Thread PDS thread 7 generated: 4051 Thread PDS thread 5 generated: 5083 Thread PDS thread 6 generated: 9591 Thread PDS thread 6 stopped. Thread PDS thread 1 generated: 5192 Thread PDS thread 1 stopped. Thread PDS thread 2 generated: 7029 Thread PDS thread 7 generated: 9348 Thread PDS thread 7 stopped. Thread PDS thread 5 generated: 5611 Thread PDS thread 3 generated: 8748 Thread PDS thread 3 stopped. Thread PDS thread 5 generated: 8871 Thread PDS thread 5 stopped. Thread PDS thread 2 generated: 8734 Thread PDS thread 2 stopped.
In [ ]:
import threading
from random import randrange
import time
def thread_gen(count):
#number of values to generate as parameter
name = threading.current_thread().name
print(f'Thread {name} started.')
for _ in range(count):
time.sleep(0.3*randrange(1, 7))
print(f'Thread {name} generated: {randrange(1000, 10000)}')
print(f'Thread {name} stopped.')
threads = [threading.Thread(target=thread_gen, args=(5,), name=f'PDS thread {1+i}') for i in range(7)]
#start
for thread in threads:
thread.start()
#wait for finish
for thread in threads:
thread.join()
In [7]:
semaphore = threading.BoundedSemaphore(3)
import threading
from random import randrange
import time
def thread_gen(sem, count):
#number of values to generate as parameter
with sem:
name = threading.current_thread().name
print(f'Thread {name} started.')
for _ in range(count):
time.sleep(0.3*randrange(1, 7))
print(f'Thread {name} generated: {randrange(1000, 10000)}')
print(f'Thread {name} stopped.')
threads = [threading.Thread(target=thread_gen, args=(semaphore, 5,), name=f'PDS thread {1+i}') for i in range(7)]
#start
for thread in threads:
thread.start()
#wait for finish
for thread in threads:
thread.join()
Thread PDS thread 1 started. Thread PDS thread 2 started. Thread PDS thread 3 started. Thread PDS thread 1 generated: 3719 Thread PDS thread 2 generated: 2455 Thread PDS thread 3 generated: 1043 Thread PDS thread 1 generated: 1117 Thread PDS thread 3 generated: 2646 Thread PDS thread 2 generated: 7701 Thread PDS thread 1 generated: 2028 Thread PDS thread 2 generated: 6958 Thread PDS thread 2 generated: 2801 Thread PDS thread 1 generated: 5005 Thread PDS thread 3 generated: 3526 Thread PDS thread 2 generated: 6600 Thread PDS thread 2 stopped. Thread PDS thread 4 started. Thread PDS thread 4 generated: 5984 Thread PDS thread 1 generated: 7160 Thread PDS thread 1 stopped. Thread PDS thread 5 started. Thread PDS thread 3 generated: 8260 Thread PDS thread 4 generated: 3667 Thread PDS thread 5 generated: 8990 Thread PDS thread 3 generated: 6648 Thread PDS thread 3 stopped. Thread PDS thread 6 started. Thread PDS thread 4 generated: 2797 Thread PDS thread 6 generated: 4234 Thread PDS thread 5 generated: 4844 Thread PDS thread 4 generated: 9067 Thread PDS thread 6 generated: 7050 Thread PDS thread 5 generated: 4266 Thread PDS thread 4 generated: 3289 Thread PDS thread 4 stopped. Thread PDS thread 7 started. Thread PDS thread 6 generated: 1301 Thread PDS thread 7 generated: 2785 Thread PDS thread 5 generated: 5079 Thread PDS thread 6 generated: 5576 Thread PDS thread 5 generated: 1447 Thread PDS thread 5 stopped. Thread PDS thread 7 generated: 2268 Thread PDS thread 6 generated: 6287 Thread PDS thread 6 stopped. Thread PDS thread 7 generated: 6368 Thread PDS thread 7 generated: 2012 Thread PDS thread 7 generated: 3647 Thread PDS thread 7 stopped.
module queue — A synchronized queue class¶
The queue module implements multi-producer, multi-consumer queues. It is especially useful in threaded programming when information must be exchanged safely between multiple threads. The Queue class in this module implements all the required locking semantics.
In [3]:
import queue
In [1]:
from queue import Queue, PriorityQueue, LifoQueue #rad, prioritny rad, zasobnik/stack
In [10]:
def test(data_structure):
data_structure.put(5)
data_structure.put(7)
data_structure.put(2)
while not data_structure.empty():
print(data_structure.get())
In [11]:
test(Queue()) # QUEUE / same order
5 7 2
In [12]:
test(LifoQueue()) #STACK / reverse order
2 7 5
In [13]:
test(PriorityQueue()) #by priority
2 5 7
Task1: Create producer and consumer and run them using Queue with maxsize (4)¶
In [5]:
q = Queue(maxsize=4)
In [6]:
q.put(1)
q.put(2)
q.put(3)
In [17]:
print(q.queue) #double ended queue/linked list
deque([1, 2, 3])
In [7]:
q.qsize()
Out[7]:
3
In [18]:
print(*q.queue)
1 2 3
In [ ]:
q.put(5) #WILL freeeze - wait to have space
#you have to check if is not empty before
In [ ]:
def producer(que, count):
#create list of count random numbers
#at random time pop from the list
#put in commom queue
#checking if the queue is not full
def consumer(que, count):
#at random time get value from common queue
#checking if the queue is not empty
print(f'Thread {name} started.')
print(f'Thread {name} is waiting.')
print(f'Thread {name} got {value}.')
print(f'Thread {name} stopped.')
n = 10
q = queue.Queue(maxsize=4)
threads = [threading(target=func, args=(q, n)) for func in [producer, consumer]]
#start
for thread in threads:
thread.start()
#wait for finish
for thread in threads:
thread.join()
TASK 2: Several consumers & producers (e.g. 3 consumer-each 5 numbers and 5 producer-each 3 numbers), consuming with PriorityQueue¶
describe priority queue usage on your example ([5, 7, 2] -> 2) [first "few" put/get commands showing chcecking empty/full]
In [8]:
q
Out[8]:
<queue.Queue at 0x191473b8e10>
In [12]:
print(q.queue)
deque([1, 2, 3])
In [13]:
q.get()
Out[13]:
1
In [14]:
print(q.queue)
deque([2, 3])
In [3]:
import queue
q = queue.Queue(maxsize=4)
q.put(7)
q.put(2)
q.put(5)
q.put(6)
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
from deque([7, 2, 5, 6]) got 7 from deque([2, 5, 6]) got 2 from deque([5, 6]) got 5 from deque([6]) got 6
In [4]:
import queue
q = queue.PriorityQueue(maxsize=4)
q.put(7)
q.put(2)
q.put(5)
q.put(6)
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
print(f'from {q.queue} got {q.get()}')
from [2, 6, 5, 7] got 2 from [5, 6, 7] got 5 from [6, 7] got 6 from [7] got 7