从多个线程同时从单个串行端口读取和写入
Reading and writing from a single serial port simultaneously from multiple threads
我有一个串口，可以从不同的硬件中获取大量不同的数据。我需要向串行端口发送不同的命令以从中接收不同类型的数据。所以，我需要在不同的函数中同时从端口写入和读取数据。有时，我可能需要在 10 个不同的线程中同时从串口读取和写入。在这种情况下，为来自单个端口的同时 reading/writing 数据编写代码的最佳方法是什么？线程、子进程等
从适用于您的案例的最直接的代码开始。如果您的脚本已经启动了多个线程，那么您可以添加 reader/writer 与控制器通信的线程——所有其他线程 send/receive 仅来自这些线程的数据：send_queue 由编写器线程使用向控制器发送命令，每个工作线程都有自己的 receive_queue 由 reader 线程填充：
#!/usr/bin/env python2
"""Communicate with a dummy controller from 10 worker threads.
- start controller subprocess, worker threads
- send all requests and check responses in worker threads concurrently
- wait until all responses are received
"""
import json
import logging
import sys
from Queue import Queue
from subprocess import Popen, PIPE
from threading import Thread
log = logging.getLogger(__name__).debug
def worker(tid, send_queue, receive_queue):
"""Send some dummy requests, get responses."""
log('starting worker tid=%s', tid)
for i in range(5):
log('sending request key=%s, data=%d', tid, i)
send_queue.put(dict(key=tid, data=i)) # send request
response = receive_queue.get() # get response
# check response
log('got response %s, key=%s, data=%d', response, tid, i)
assert response['key'] == tid and response['data'] == i**2
def reader(pipe, thread_queues):
"""Read responses from the controller (*pipe*).
Dispatch them to the corresponding threads.
"""
log('starting reader')
with pipe:
for line in iter(pipe.readline, b''): # read response from the controller
log('got line %r', line)
response = json.loads(line)
# dispatch to the corresponding thread
thread_queues[response['key']].put(response)
def writer(pipe, send_queue):
"""Write requests to the controller (*pipe*) from *send_queue*."""
log('starting writer')
with pipe:
for request in iter(send_queue.get, None):
log('sending request %s', request)
print >>pipe, json.dumps(request)
def start_daemon_thread(*args, **kwargs):
"""Start and return daemonic thread."""
t = Thread(*args, **kwargs)
t.daemon = True # die with the program
t.start()
return t
def main():
nthreads = 10
timeout = 1 # seconds, for Ctrl+C
logging.basicConfig(format="%(asctime)-15s %(threadName)s %(message)s",
datefmt='%F %T', level=logging.DEBUG)
# start controller
controller = Popen([sys.executable, "-u", "dummy_controller.py"],
stdin=PIPE, stdout=PIPE, bufsize=1)
# start threads
send_queue = Queue(maxsize=50) # limit number of queued requests (backlog)
thread_queues = {tid: Queue() for tid in range(1, nthreads + 1)}
threads = [start_daemon_thread(name='worker-' + str(tid),
target=worker, args=[tid, send_queue, q])
for tid, q in thread_queues.items()]
# read/write until there are requests
start_daemon_thread(name='writer', target=writer,
args=[controller.stdin, send_queue])
log('about to start a reader')
reader_thread = start_daemon_thread(name='reader', target=reader,
args=[controller.stdout, thread_queues])
# wait until workers are done
while threads:
for t in threads:
t.join(timeout) # workaround, enable Ctrl+C on Python 2.7
if not t.is_alive():
threads.remove(t)
break
log('no worker threads')
send_queue.put(None) # no more requests
# wait until all responses are received
while reader_thread.is_alive():
reader_thread.join(timeout) # workaround, enable Ctrl+C on Python 2.7
controller.wait() # wait for the subprocess to exit (to avoid zombies)
log('done')
if __name__ == "__main__":
main()
其中 dummy_controller.py 是一个简单的脚本，它为每个输入计算 x**2 x:
#!/usr/bin/env python2
"""Dummy controller."""
import json
import sys
for line in iter(sys.stdin.readline, b''):
request = json.loads(line)
# square data field in the request and send it back
print json.dumps(dict(request, data=int(request['data'])**2))
您的代码不需要子进程，您应该为串行端口调用适当的 read/write 方法，而不是启动子进程并使用管道。
代码冗长且未针对速度进行优化。如果你的工作线程是 CPU-bound 那么你应该找到一种方法来释放 GIL（例如，在 CPython 实现上）：C 扩展或将工作卸载到进程池；否则只有一个 CPU 核心被 python 使用 GIL 实现（CPython、Pypy（默认））。
示例输出
2015-09-27 20:02:38 worker-1 starting worker tid=1
2015-09-27 20:02:38 worker-1 sending request key=1, data=0
2015-09-27 20:02:38 worker-2 starting worker tid=2
2015-09-27 20:02:38 worker-3 starting worker tid=3
2015-09-27 20:02:38 worker-3 sending request key=3, data=0
2015-09-27 20:02:38 worker-2 sending request key=2, data=0
2015-09-27 20:02:38 worker-5 starting worker tid=5
2015-09-27 20:02:38 worker-4 starting worker tid=4
2015-09-27 20:02:38 worker-6 starting worker tid=6
2015-09-27 20:02:38 worker-7 starting worker tid=7
2015-09-27 20:02:38 worker-5 sending request key=5, data=0
2015-09-27 20:02:38 worker-6 sending request key=6, data=0
2015-09-27 20:02:38 worker-4 sending request key=4, data=0
2015-09-27 20:02:38 worker-8 starting worker tid=8
2015-09-27 20:02:38 worker-7 sending request key=7, data=0
2015-09-27 20:02:38 worker-9 starting worker tid=9
2015-09-27 20:02:38 worker-8 sending request key=8, data=0
2015-09-27 20:02:38 worker-9 sending request key=9, data=0
2015-09-27 20:02:38 worker-10 starting worker tid=10
2015-09-27 20:02:38 worker-10 sending request key=10, data=0
2015-09-27 20:02:38 MainThread about to start a reader
2015-09-27 20:02:38 writer starting writer
2015-09-27 20:02:38 reader starting reader
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 1}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 3}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 2}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 5}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 6}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 4}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 7}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 8}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 9}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 10}
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 1}\n'
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 3}\n'
2015-09-27 20:02:38 worker-1 got response {u'data': 0, u'key': 1}, key=1, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 0, u'key': 3}, key=3, data=0
2015-09-27 20:02:38 worker-1 sending request key=1, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 5}\n'
2015-09-27 20:02:38 worker-2 got response {u'data': 0, u'key': 2}, key=2, data=0
2015-09-27 20:02:38 worker-3 sending request key=3, data=1
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 1}
2015-09-27 20:02:38 worker-5 got response {u'data': 0, u'key': 5}, key=5, data=0
2015-09-27 20:02:38 worker-2 sending request key=2, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 3}
2015-09-27 20:02:38 worker-5 sending request key=5, data=1
2015-09-27 20:02:38 worker-6 got response {u'data': 0, u'key': 6}, key=6, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 4}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 2}
2015-09-27 20:02:38 worker-6 sending request key=6, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 0, u'key': 4}, key=4, data=0
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 5}
2015-09-27 20:02:38 worker-4 sending request key=4, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 8}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 6}
2015-09-27 20:02:38 worker-7 got response {u'data': 0, u'key': 7}, key=7, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 0, u'key': 8}, key=8, data=0
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 4}
2015-09-27 20:02:38 worker-7 sending request key=7, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 0, u'key': 9}, key=9, data=0
2015-09-27 20:02:38 worker-8 sending request key=8, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 1}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 7}
2015-09-27 20:02:38 worker-10 got response {u'data': 0, u'key': 10}, key=10, data=0
2015-09-27 20:02:38 worker-9 sending request key=9, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 3}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 8}
2015-09-27 20:02:38 worker-1 got response {u'data': 1, u'key': 1}, key=1, data=1
2015-09-27 20:02:38 worker-10 sending request key=10, data=1
2015-09-27 20:02:38 worker-3 got response {u'data': 1, u'key': 3}, key=3, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 2}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 9}
2015-09-27 20:02:38 worker-1 sending request key=1, data=2
2015-09-27 20:02:38 worker-3 sending request key=3, data=2
2015-09-27 20:02:38 worker-2 got response {u'data': 1, u'key': 2}, key=2, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 10}
2015-09-27 20:02:38 worker-2 sending request key=2, data=2
2015-09-27 20:02:38 worker-5 got response {u'data': 1, u'key': 5}, key=5, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 1}
2015-09-27 20:02:38 worker-5 sending request key=5, data=2
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 4}\n'
2015-09-27 20:02:38 worker-6 got response {u'data': 1, u'key': 6}, key=6, data=1
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 3}
2015-09-27 20:02:38 worker-4 got response {u'data': 1, u'key': 4}, key=4, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 7}\n'
2015-09-27 20:02:38 worker-6 sending request key=6, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 2}
2015-09-27 20:02:38 worker-4 sending request key=4, data=2
2015-09-27 20:02:38 worker-7 got response {u'data': 1, u'key': 7}, key=7, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 8}\n'
2015-09-27 20:02:38 worker-7 sending request key=7, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 5}
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 1, u'key': 8}, key=8, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 10}\n'
2015-09-27 20:02:38 worker-8 sending request key=8, data=2
2015-09-27 20:02:38 worker-9 got response {u'data': 1, u'key': 9}, key=9, data=1
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 6}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 1}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 1, u'key': 10}, key=10, data=1
2015-09-27 20:02:38 worker-9 sending request key=9, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 4}
2015-09-27 20:02:38 worker-1 got response {u'data': 4, u'key': 1}, key=1, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 3}\n'
2015-09-27 20:02:38 worker-10 sending request key=10, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 7}
2015-09-27 20:02:38 worker-1 sending request key=1, data=3
2015-09-27 20:02:38 worker-3 got response {u'data': 4, u'key': 3}, key=3, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 2}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 8}
2015-09-27 20:02:38 worker-3 sending request key=3, data=3
2015-09-27 20:02:38 worker-2 got response {u'data': 4, u'key': 2}, key=2, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 9}
2015-09-27 20:02:38 worker-2 sending request key=2, data=3
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 10}
2015-09-27 20:02:38 worker-5 got response {u'data': 4, u'key': 5}, key=5, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 4}\n'
2015-09-27 20:02:38 worker-6 got response {u'data': 4, u'key': 6}, key=6, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 1}
2015-09-27 20:02:38 worker-5 sending request key=5, data=3
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 4, u'key': 4}, key=4, data=2
2015-09-27 20:02:38 worker-6 sending request key=6, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 3}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 8}\n'
2015-09-27 20:02:38 worker-7 got response {u'data': 4, u'key': 7}, key=7, data=2
2015-09-27 20:02:38 worker-4 sending request key=4, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 2}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 9}\n'
2015-09-27 20:02:38 worker-7 sending request key=7, data=3
2015-09-27 20:02:38 worker-8 got response {u'data': 4, u'key': 8}, key=8, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 5}
2015-09-27 20:02:38 worker-9 got response {u'data': 4, u'key': 9}, key=9, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 10}\n'
2015-09-27 20:02:38 worker-8 sending request key=8, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 6}
2015-09-27 20:02:38 worker-9 sending request key=9, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 1}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 4, u'key': 10}, key=10, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 4}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 3}\n'
2015-09-27 20:02:38 worker-10 sending request key=10, data=3
2015-09-27 20:02:38 worker-1 got response {u'data': 9, u'key': 1}, key=1, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 7}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 9, u'key': 3}, key=3, data=3
2015-09-27 20:02:38 worker-3 sending request key=3, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 8}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 5}\n'
2015-09-27 20:02:38 worker-2 got response {u'data': 9, u'key': 2}, key=2, data=3
2015-09-27 20:02:38 worker-2 sending request key=2, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 9}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 6}\n'
2015-09-27 20:02:38 worker-5 got response {u'data': 9, u'key': 5}, key=5, data=3
2015-09-27 20:02:38 worker-1 sending request key=1, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 10}
2015-09-27 20:02:38 worker-6 got response {u'data': 9, u'key': 6}, key=6, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 4}\n'
2015-09-27 20:02:38 worker-5 sending request key=5, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 3}
2015-09-27 20:02:38 worker-6 sending request key=6, data=4
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 9, u'key': 4}, key=4, data=3
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 2}
2015-09-27 20:02:38 worker-7 got response {u'data': 9, u'key': 7}, key=7, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 8}\n'
2015-09-27 20:02:38 worker-4 sending request key=4, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 1}
2015-09-27 20:02:38 worker-7 sending request key=7, data=4
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 9, u'key': 8}, key=8, data=3
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 5}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 9, u'key': 9}, key=9, data=3
2015-09-27 20:02:38 worker-8 sending request key=8, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 6}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 3}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 9, u'key': 10}, key=10, data=3
2015-09-27 20:02:38 worker-9 sending request key=9, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 4}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 16, u'key': 3}, key=3, data=4
2015-09-27 20:02:38 worker-10 sending request key=10, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 7}
2015-09-27 20:02:38 worker-2 got response {u'data': 16, u'key': 2}, key=2, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 1}\n'
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 8}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 9}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 6}\n'
2015-09-27 20:02:38 worker-5 got response {u'data': 16, u'key': 5}, key=5, data=4
2015-09-27 20:02:38 worker-1 got response {u'data': 16, u'key': 1}, key=1, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 10}
2015-09-27 20:02:38 worker-6 got response {u'data': 16, u'key': 6}, key=6, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 4}\n'
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 16, u'key': 4}, key=4, data=4
2015-09-27 20:02:38 worker-7 got response {u'data': 16, u'key': 7}, key=7, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 8}\n'
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 16, u'key': 8}, key=8, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 16, u'key': 9}, key=9, data=4
2015-09-27 20:02:38 worker-10 got response {u'data': 16, u'key': 10}, key=10, data=4
2015-09-27 20:02:38 MainThread no worker threads
2015-09-27 20:02:38 MainThread done
根据体系结构和驱动程序的类型，可以有不同的方法。其中之一可以如下：
一旦您通过接收中断接收到数据，post 数据就会进入主接收缓冲区队列。可以有一个名为 Rx dispatcher/Rx 管理器的线程，它始终读取主接收缓冲区队列和 post/dispatch 到基于消息 type/id 的 reader 线程的相应接收队列以进行进一步处理.
在传输端，各个线程可以post将数据连同消息type/id发送到主传输缓冲区队列，该队列应由Tx线程通过传输中断发送出去。
我有一个串口，可以从不同的硬件中获取大量不同的数据。我需要向串行端口发送不同的命令以从中接收不同类型的数据。所以，我需要在不同的函数中同时从端口写入和读取数据。有时，我可能需要在 10 个不同的线程中同时从串口读取和写入。在这种情况下，为来自单个端口的同时 reading/writing 数据编写代码的最佳方法是什么？线程、子进程等
从适用于您的案例的最直接的代码开始。如果您的脚本已经启动了多个线程，那么您可以添加 reader/writer 与控制器通信的线程——所有其他线程 send/receive 仅来自这些线程的数据：send_queue 由编写器线程使用向控制器发送命令，每个工作线程都有自己的 receive_queue 由 reader 线程填充：
#!/usr/bin/env python2
"""Communicate with a dummy controller from 10 worker threads.
- start controller subprocess, worker threads
- send all requests and check responses in worker threads concurrently
- wait until all responses are received
"""
import json
import logging
import sys
from Queue import Queue
from subprocess import Popen, PIPE
from threading import Thread
log = logging.getLogger(__name__).debug
def worker(tid, send_queue, receive_queue):
"""Send some dummy requests, get responses."""
log('starting worker tid=%s', tid)
for i in range(5):
log('sending request key=%s, data=%d', tid, i)
send_queue.put(dict(key=tid, data=i)) # send request
response = receive_queue.get() # get response
# check response
log('got response %s, key=%s, data=%d', response, tid, i)
assert response['key'] == tid and response['data'] == i**2
def reader(pipe, thread_queues):
"""Read responses from the controller (*pipe*).
Dispatch them to the corresponding threads.
"""
log('starting reader')
with pipe:
for line in iter(pipe.readline, b''): # read response from the controller
log('got line %r', line)
response = json.loads(line)
# dispatch to the corresponding thread
thread_queues[response['key']].put(response)
def writer(pipe, send_queue):
"""Write requests to the controller (*pipe*) from *send_queue*."""
log('starting writer')
with pipe:
for request in iter(send_queue.get, None):
log('sending request %s', request)
print >>pipe, json.dumps(request)
def start_daemon_thread(*args, **kwargs):
"""Start and return daemonic thread."""
t = Thread(*args, **kwargs)
t.daemon = True # die with the program
t.start()
return t
def main():
nthreads = 10
timeout = 1 # seconds, for Ctrl+C
logging.basicConfig(format="%(asctime)-15s %(threadName)s %(message)s",
datefmt='%F %T', level=logging.DEBUG)
# start controller
controller = Popen([sys.executable, "-u", "dummy_controller.py"],
stdin=PIPE, stdout=PIPE, bufsize=1)
# start threads
send_queue = Queue(maxsize=50) # limit number of queued requests (backlog)
thread_queues = {tid: Queue() for tid in range(1, nthreads + 1)}
threads = [start_daemon_thread(name='worker-' + str(tid),
target=worker, args=[tid, send_queue, q])
for tid, q in thread_queues.items()]
# read/write until there are requests
start_daemon_thread(name='writer', target=writer,
args=[controller.stdin, send_queue])
log('about to start a reader')
reader_thread = start_daemon_thread(name='reader', target=reader,
args=[controller.stdout, thread_queues])
# wait until workers are done
while threads:
for t in threads:
t.join(timeout) # workaround, enable Ctrl+C on Python 2.7
if not t.is_alive():
threads.remove(t)
break
log('no worker threads')
send_queue.put(None) # no more requests
# wait until all responses are received
while reader_thread.is_alive():
reader_thread.join(timeout) # workaround, enable Ctrl+C on Python 2.7
controller.wait() # wait for the subprocess to exit (to avoid zombies)
log('done')
if __name__ == "__main__":
main()
其中 dummy_controller.py 是一个简单的脚本，它为每个输入计算 x**2 x:
#!/usr/bin/env python2
"""Dummy controller."""
import json
import sys
for line in iter(sys.stdin.readline, b''):
request = json.loads(line)
# square data field in the request and send it back
print json.dumps(dict(request, data=int(request['data'])**2))
您的代码不需要子进程，您应该为串行端口调用适当的 read/write 方法，而不是启动子进程并使用管道。
代码冗长且未针对速度进行优化。如果你的工作线程是 CPU-bound 那么你应该找到一种方法来释放 GIL（例如，在 CPython 实现上）：C 扩展或将工作卸载到进程池；否则只有一个 CPU 核心被 python 使用 GIL 实现（CPython、Pypy（默认））。
示例输出
2015-09-27 20:02:38 worker-1 starting worker tid=1
2015-09-27 20:02:38 worker-1 sending request key=1, data=0
2015-09-27 20:02:38 worker-2 starting worker tid=2
2015-09-27 20:02:38 worker-3 starting worker tid=3
2015-09-27 20:02:38 worker-3 sending request key=3, data=0
2015-09-27 20:02:38 worker-2 sending request key=2, data=0
2015-09-27 20:02:38 worker-5 starting worker tid=5
2015-09-27 20:02:38 worker-4 starting worker tid=4
2015-09-27 20:02:38 worker-6 starting worker tid=6
2015-09-27 20:02:38 worker-7 starting worker tid=7
2015-09-27 20:02:38 worker-5 sending request key=5, data=0
2015-09-27 20:02:38 worker-6 sending request key=6, data=0
2015-09-27 20:02:38 worker-4 sending request key=4, data=0
2015-09-27 20:02:38 worker-8 starting worker tid=8
2015-09-27 20:02:38 worker-7 sending request key=7, data=0
2015-09-27 20:02:38 worker-9 starting worker tid=9
2015-09-27 20:02:38 worker-8 sending request key=8, data=0
2015-09-27 20:02:38 worker-9 sending request key=9, data=0
2015-09-27 20:02:38 worker-10 starting worker tid=10
2015-09-27 20:02:38 worker-10 sending request key=10, data=0
2015-09-27 20:02:38 MainThread about to start a reader
2015-09-27 20:02:38 writer starting writer
2015-09-27 20:02:38 reader starting reader
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 1}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 3}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 2}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 5}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 6}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 4}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 7}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 8}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 9}
2015-09-27 20:02:38 writer sending request {'data': 0, 'key': 10}
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 1}\n'
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 3}\n'
2015-09-27 20:02:38 worker-1 got response {u'data': 0, u'key': 1}, key=1, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 0, u'key': 3}, key=3, data=0
2015-09-27 20:02:38 worker-1 sending request key=1, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 5}\n'
2015-09-27 20:02:38 worker-2 got response {u'data': 0, u'key': 2}, key=2, data=0
2015-09-27 20:02:38 worker-3 sending request key=3, data=1
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 1}
2015-09-27 20:02:38 worker-5 got response {u'data': 0, u'key': 5}, key=5, data=0
2015-09-27 20:02:38 worker-2 sending request key=2, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 3}
2015-09-27 20:02:38 worker-5 sending request key=5, data=1
2015-09-27 20:02:38 worker-6 got response {u'data': 0, u'key': 6}, key=6, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 4}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 2}
2015-09-27 20:02:38 worker-6 sending request key=6, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 0, u'key': 4}, key=4, data=0
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 5}
2015-09-27 20:02:38 worker-4 sending request key=4, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 8}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 6}
2015-09-27 20:02:38 worker-7 got response {u'data': 0, u'key': 7}, key=7, data=0
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 0, u'key': 8}, key=8, data=0
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 4}
2015-09-27 20:02:38 worker-7 sending request key=7, data=1
2015-09-27 20:02:38 reader got line '{"data": 0, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 0, u'key': 9}, key=9, data=0
2015-09-27 20:02:38 worker-8 sending request key=8, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 1}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 7}
2015-09-27 20:02:38 worker-10 got response {u'data': 0, u'key': 10}, key=10, data=0
2015-09-27 20:02:38 worker-9 sending request key=9, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 3}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 8}
2015-09-27 20:02:38 worker-1 got response {u'data': 1, u'key': 1}, key=1, data=1
2015-09-27 20:02:38 worker-10 sending request key=10, data=1
2015-09-27 20:02:38 worker-3 got response {u'data': 1, u'key': 3}, key=3, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 2}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 9}
2015-09-27 20:02:38 worker-1 sending request key=1, data=2
2015-09-27 20:02:38 worker-3 sending request key=3, data=2
2015-09-27 20:02:38 worker-2 got response {u'data': 1, u'key': 2}, key=2, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 1, 'key': 10}
2015-09-27 20:02:38 worker-2 sending request key=2, data=2
2015-09-27 20:02:38 worker-5 got response {u'data': 1, u'key': 5}, key=5, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 1}
2015-09-27 20:02:38 worker-5 sending request key=5, data=2
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 4}\n'
2015-09-27 20:02:38 worker-6 got response {u'data': 1, u'key': 6}, key=6, data=1
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 3}
2015-09-27 20:02:38 worker-4 got response {u'data': 1, u'key': 4}, key=4, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 7}\n'
2015-09-27 20:02:38 worker-6 sending request key=6, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 2}
2015-09-27 20:02:38 worker-4 sending request key=4, data=2
2015-09-27 20:02:38 worker-7 got response {u'data': 1, u'key': 7}, key=7, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 8}\n'
2015-09-27 20:02:38 worker-7 sending request key=7, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 5}
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 1, u'key': 8}, key=8, data=1
2015-09-27 20:02:38 reader got line '{"data": 1, "key": 10}\n'
2015-09-27 20:02:38 worker-8 sending request key=8, data=2
2015-09-27 20:02:38 worker-9 got response {u'data': 1, u'key': 9}, key=9, data=1
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 6}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 1}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 1, u'key': 10}, key=10, data=1
2015-09-27 20:02:38 worker-9 sending request key=9, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 4}
2015-09-27 20:02:38 worker-1 got response {u'data': 4, u'key': 1}, key=1, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 3}\n'
2015-09-27 20:02:38 worker-10 sending request key=10, data=2
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 7}
2015-09-27 20:02:38 worker-1 sending request key=1, data=3
2015-09-27 20:02:38 worker-3 got response {u'data': 4, u'key': 3}, key=3, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 2}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 8}
2015-09-27 20:02:38 worker-3 sending request key=3, data=3
2015-09-27 20:02:38 worker-2 got response {u'data': 4, u'key': 2}, key=2, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 9}
2015-09-27 20:02:38 worker-2 sending request key=2, data=3
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 6}\n'
2015-09-27 20:02:38 writer sending request {'data': 2, 'key': 10}
2015-09-27 20:02:38 worker-5 got response {u'data': 4, u'key': 5}, key=5, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 4}\n'
2015-09-27 20:02:38 worker-6 got response {u'data': 4, u'key': 6}, key=6, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 1}
2015-09-27 20:02:38 worker-5 sending request key=5, data=3
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 4, u'key': 4}, key=4, data=2
2015-09-27 20:02:38 worker-6 sending request key=6, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 3}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 8}\n'
2015-09-27 20:02:38 worker-7 got response {u'data': 4, u'key': 7}, key=7, data=2
2015-09-27 20:02:38 worker-4 sending request key=4, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 2}
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 9}\n'
2015-09-27 20:02:38 worker-7 sending request key=7, data=3
2015-09-27 20:02:38 worker-8 got response {u'data': 4, u'key': 8}, key=8, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 5}
2015-09-27 20:02:38 worker-9 got response {u'data': 4, u'key': 9}, key=9, data=2
2015-09-27 20:02:38 reader got line '{"data": 4, "key": 10}\n'
2015-09-27 20:02:38 worker-8 sending request key=8, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 6}
2015-09-27 20:02:38 worker-9 sending request key=9, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 1}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 4, u'key': 10}, key=10, data=2
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 4}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 3}\n'
2015-09-27 20:02:38 worker-10 sending request key=10, data=3
2015-09-27 20:02:38 worker-1 got response {u'data': 9, u'key': 1}, key=1, data=3
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 7}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 9, u'key': 3}, key=3, data=3
2015-09-27 20:02:38 worker-3 sending request key=3, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 8}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 5}\n'
2015-09-27 20:02:38 worker-2 got response {u'data': 9, u'key': 2}, key=2, data=3
2015-09-27 20:02:38 worker-2 sending request key=2, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 9}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 6}\n'
2015-09-27 20:02:38 worker-5 got response {u'data': 9, u'key': 5}, key=5, data=3
2015-09-27 20:02:38 worker-1 sending request key=1, data=4
2015-09-27 20:02:38 writer sending request {'data': 3, 'key': 10}
2015-09-27 20:02:38 worker-6 got response {u'data': 9, u'key': 6}, key=6, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 4}\n'
2015-09-27 20:02:38 worker-5 sending request key=5, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 3}
2015-09-27 20:02:38 worker-6 sending request key=6, data=4
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 9, u'key': 4}, key=4, data=3
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 2}
2015-09-27 20:02:38 worker-7 got response {u'data': 9, u'key': 7}, key=7, data=3
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 8}\n'
2015-09-27 20:02:38 worker-4 sending request key=4, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 1}
2015-09-27 20:02:38 worker-7 sending request key=7, data=4
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 9, u'key': 8}, key=8, data=3
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 5}
2015-09-27 20:02:38 reader got line '{"data": 9, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 9, u'key': 9}, key=9, data=3
2015-09-27 20:02:38 worker-8 sending request key=8, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 6}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 3}\n'
2015-09-27 20:02:38 worker-10 got response {u'data': 9, u'key': 10}, key=10, data=3
2015-09-27 20:02:38 worker-9 sending request key=9, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 4}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 2}\n'
2015-09-27 20:02:38 worker-3 got response {u'data': 16, u'key': 3}, key=3, data=4
2015-09-27 20:02:38 worker-10 sending request key=10, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 7}
2015-09-27 20:02:38 worker-2 got response {u'data': 16, u'key': 2}, key=2, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 1}\n'
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 8}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 5}\n'
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 9}
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 6}\n'
2015-09-27 20:02:38 worker-5 got response {u'data': 16, u'key': 5}, key=5, data=4
2015-09-27 20:02:38 worker-1 got response {u'data': 16, u'key': 1}, key=1, data=4
2015-09-27 20:02:38 writer sending request {'data': 4, 'key': 10}
2015-09-27 20:02:38 worker-6 got response {u'data': 16, u'key': 6}, key=6, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 4}\n'
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 7}\n'
2015-09-27 20:02:38 worker-4 got response {u'data': 16, u'key': 4}, key=4, data=4
2015-09-27 20:02:38 worker-7 got response {u'data': 16, u'key': 7}, key=7, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 8}\n'
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 9}\n'
2015-09-27 20:02:38 worker-8 got response {u'data': 16, u'key': 8}, key=8, data=4
2015-09-27 20:02:38 reader got line '{"data": 16, "key": 10}\n'
2015-09-27 20:02:38 worker-9 got response {u'data': 16, u'key': 9}, key=9, data=4
2015-09-27 20:02:38 worker-10 got response {u'data': 16, u'key': 10}, key=10, data=4
2015-09-27 20:02:38 MainThread no worker threads
2015-09-27 20:02:38 MainThread done
根据体系结构和驱动程序的类型，可以有不同的方法。其中之一可以如下：
一旦您通过接收中断接收到数据，post 数据就会进入主接收缓冲区队列。可以有一个名为 Rx dispatcher/Rx 管理器的线程，它始终读取主接收缓冲区队列和 post/dispatch 到基于消息 type/id 的 reader 线程的相应接收队列以进行进一步处理.
在传输端，各个线程可以post将数据连同消息type/id发送到主传输缓冲区队列，该队列应由Tx线程通过传输中断发送出去。