Размер очереди и блокировка для прерывистых соединений

Question

Этот вопрос о том, как работают очереди Python (из модуля очереди), когда они достигли своего максимального размера, но дополнительные элементы продолжают появляться при их вводе.

Я работаю над программой Python, которая читает данныеиз нескольких последовательных портов разделяет данные на несколько очередей и предоставляет сервер сокетов TCP, который принимает соединения на несколько портов, каждый из которых обслуживает одну очередь. Программа работает следующим образом:

• Каждый из четырех потоков последовательного порта непрерывно сбрасывает данные для двух каналов в общую очередь ввода (всего восемь записей по 25-30 символов в секунду).

• Другой поток get () определяет входную очередь и разбивает данные на девять выходных очередей (по одной для каждого канала плюс одна, содержащая все каналы).

• Затем каждый из девяти потоков сокетов при установлении соединения получает () данные из своей очереди и отправляет их клиенту.

Удивительно, но мой код кажетсяработать, но я хочу убедиться, что это надежно. Проблема заключается в том, что последовательные порты непрерывно выводят данные, но клиентские соединения для извлечения этих данных из очередей являются прерывистыми.

В таком случае, как будет выглядеть вывод при подключении клиента? Содержимое очереди будет отправлено немедленно, но что произойдет, когда старые данные исчезнут? например, если размер очереди равен 10, но поток производителя закачал в него 1000 элементов без вызовов get (), как только будут использованы десять элементов, что будет дальше - элемент 11 или элемент 1001? Другими словами, отброшены ли элементы, которые поступили с момента блокировки очереди, при следующей вставке - первые данные, которые поступают, когда есть место, или они вставлены по порядку?

Я пробовалНесколько тестов, и все, что я смог определить, это то, что если размеры очереди заданы слишком маленькими, я получаю начальный вывод, а затем (после периода ожидания) программа завершается с сообщением «Пустая очередь». Если размер очереди установлен на 1000 или около того, программа, кажется, работает непрерывно, но трудно сказать, теряю ли я данные.

Это нормально, если я потеряю самые старые сэмплы, но я не могуесть пробелы - за пунктом 10 должен следовать пункт 11, а не пункт 1001. Работает ли механизм очереди таким образом, или мне нужно что-то сделать для обеспечения такого поведения? (Я предполагаю, что размер очереди может быть бесконечным, но я подозреваю, что это может вызвать другие проблемы после запуска программы некоторое время.)

Спасибо!

Answer 1

Отвечая самому себе ...

Резюме: когда очередь заполнена, вы должны либо (а) блокировать, останавливать все вверх по течению, либо (б) бросать новые входы на этаж, пока не останется места.

Сначала моя программа остановилась через некоторое (переменное) время. Это произошло, когда одна из выходных очередей, которые не читались, была заполнена. Это исправлено тестированием перед помещением элемента в очередь:

if not q.full():
    q.put(item)

Если очередь заполнена, бросайте новые предметы на пол. Когда вы начинаете извлекать данные из очереди, вы получаете содержимое очереди до тех пор, пока оно не будет заполнено, а затем новые данные, когда есть место. Это проблема, потому что мне нужна непрерывная последовательность элементов в моих выходных данных.

Чтобы это исправить, я очищаю очередь перед считыванием данных:

while not q.empty():
    line = q.get()    # toss this
while True:
    line = q.get()
    conn.send(line)
    q.task_done()

Я теряю старую очередьданные (но это нормально в моем приложении), и теперь я получаю последовательный вывод без пропусков.