140 lines
3.6 KiB
Text
140 lines
3.6 KiB
Text
@startuml
|
||
title Работа multiprocessing.Queue с большими объектами\n(2 worker процесса)
|
||
|
||
participant "Главный процесс" as Main
|
||
participant "Queue\n(pipe buffer ~65KB)" as Queue
|
||
participant "Worker 1" as W1
|
||
participant "Worker 2" as W2
|
||
|
||
== Запуск процессов ==
|
||
|
||
Main -> W1 ** : start()
|
||
Main -> W2 ** : start()
|
||
|
||
note over Main
|
||
Оба worker'а работают
|
||
параллельно в фоне
|
||
end note
|
||
|
||
== Worker 1: Отправка больших данных ==
|
||
|
||
W1 -> W1 : Обработка данных
|
||
activate W1 #lightblue
|
||
W1 -> W1 : events = [...] (большой массив)
|
||
W1 -> Queue : put({"events": events})\n[Начало сериализации]
|
||
activate Queue #orange
|
||
|
||
note right of W1
|
||
pickle сериализует объект:
|
||
ОБЪЕКТ_1 → [500 KB байт]
|
||
end note
|
||
|
||
W1 -> Queue : Запись chunk 1 (64 KB)
|
||
W1 -> Queue : Запись chunk 2 (64 KB)
|
||
|
||
note over Queue
|
||
Pipe buffer заполнен!
|
||
Больше нельзя записать
|
||
end note
|
||
|
||
W1 -[#red]> W1 : ⏸️ БЛОКИРОВКА на put()
|
||
note right of W1 #pink
|
||
Worker 1 ЖДЕТ,
|
||
пока освободится место
|
||
end note
|
||
|
||
== Worker 2: Тоже пытается отправить ==
|
||
|
||
W2 -> W2 : Обработка данных
|
||
activate W2 #lightgreen
|
||
W2 -> W2 : events = [...] (большой массив)
|
||
W2 -> Queue : put({"events": events})
|
||
W2 -[#red]> W2 : ⏸️ БЛОКИРОВКА\n(очередь занята Worker 1)
|
||
|
||
== Главный процесс: Чтение (итерация 1) ==
|
||
|
||
Main -> Main : for i in range(2):\n result = queue.get()
|
||
activate Main #yellow
|
||
|
||
Main -> Queue : get() [вызов]
|
||
note right of Main #lightgreen
|
||
Первая итерация цикла
|
||
i = 0
|
||
end note
|
||
|
||
Queue -> Queue : Читает ВСЕ байты ОБЪЕКТА_1
|
||
note over Queue
|
||
✓ Читает chunk 1 (64 KB)
|
||
✓ Читает chunk 2 (64 KB)
|
||
✓ Читает chunk 3...
|
||
...
|
||
✓ Читает последний chunk
|
||
|
||
ВАЖНО: get() не вернется,
|
||
пока не прочитает ВЕСЬ объект!
|
||
end note
|
||
|
||
Queue -> Main : return ПОЛНЫЙ ОБЪЕКТ_1
|
||
note right of Main
|
||
Получен ЦЕЛЫЙ объект
|
||
от Worker 1
|
||
end note
|
||
|
||
note over Queue #lightgreen
|
||
Pipe buffer освобожден!
|
||
Теперь Worker 1 может
|
||
продолжить запись
|
||
end note
|
||
|
||
W1 -[#green]> W1 : ✓ Разблокировка!
|
||
W1 -> W1 : put() завершен
|
||
deactivate W1
|
||
|
||
note over Queue
|
||
Теперь очередь пустая,
|
||
Worker 2 может записать
|
||
end note
|
||
|
||
W2 -[#green]> W2 : ✓ Разблокировка!
|
||
W2 -> Queue : Запись ОБЪЕКТА_2
|
||
activate Queue #orange
|
||
W2 -> W2 : put() завершен
|
||
deactivate W2
|
||
|
||
== Главный процесс: Чтение (итерация 2) ==
|
||
|
||
Main -> Queue : get() [второй вызов]
|
||
note right of Main #lightgreen
|
||
Вторая итерация цикла
|
||
i = 1
|
||
end note
|
||
|
||
Queue -> Queue : Читает ВСЕ байты ОБЪЕКТА_2
|
||
Queue -> Main : return ПОЛНЫЙ ОБЪЕКТ_2
|
||
deactivate Queue
|
||
|
||
note right of Main
|
||
Получен ЦЕЛЫЙ объект
|
||
от Worker 2
|
||
|
||
ДАННЫЕ НЕ СМЕШАЛИСЬ!
|
||
Каждый get() вернул
|
||
полный объект
|
||
end note
|
||
|
||
== Завершение ==
|
||
|
||
Main -> W1 : join()
|
||
W1 --> Main : ✓ завершен
|
||
Main -> W2 : join()
|
||
W2 --> Main : ✓ завершен
|
||
|
||
deactivate Main
|
||
|
||
note over Main, W2 #lightgreen
|
||
✓ Все данные получены полностью
|
||
✓ Ничего не потеряно
|
||
✓ Объекты не смешались
|
||
end note
|
||
|
||
@enduml
|