使用 0mq (ZeroMQ) 同步两个简单的 python3 脚本时出现死锁答案

【问题标题】：Deadlock when synchronizing two simple python3 scripts using 0mq (ZeroMQ)使用 0mq (ZeroMQ) 同步两个简单的 python3 脚本时出现死锁
【发布时间】：2016-04-03 21:34:50
【问题描述】：

当我尝试使用 0mq (ZeroMQ) 同步两个 python3 脚本时，我遇到了这个奇怪的死锁。这些脚本可以正常运行数千次迭代，但迟早它们都会停止并相互等待。我在 Windows 7 上从不同的 CMD-Windows 运行这两个脚本。

我不明白
为什么会出现这样的死锁。
这里会出什么问题？

脚本 A：

while (1):
   context = zmq.Context()
   socket = context.socket(zmq.REP)
   socket.bind('tcp://127.0.0.1:10001')
   msg = socket.recv()                        # Waiting for script B to send done
   # ............................................................................
   # ... do something useful (takes only a few millisecs)
   # ............................................................................     
   context = zmq.Context()
   socket = context.socket(zmq.REQ)
   socket.connect('tcp://127.0.0.1:10002')
   socket.send_string("done")                 # Tell script B we are done

脚本 B

while (1):
   # ............................................................................
   # ... do something useful (takes only a few millisecs)
   # ............................................................................
   context = zmq.Context()
   socket = context.socket(zmq.REQ)
   socket.connect('tcp://127.0.0.1:10001')
   socket.send_string("done")               # Tell script A we are done

   context = zmq.Context()
   socket = context.socket(zmq.REP)
   socket.bind('tcp://127.0.0.1:10002')
   msg = socket.recv()                      # Waiting for script A to send done

【问题讨论】：

你一直在打开新的连接，从不关闭一个。您可能已达到操作系统中的最大连接数限制
我尝试了一个快速破解来验证这一点。我稍后会改进代码，但现在我只是在每个“send_string()”和“recv()”之后放置一个“socket.close()”。这至少应该确保一次只打开 2 个连接。现在它崩溃得更快。这是否告诉我达到最大连接数不是问题，还是我看错了？
好吧，您可能已经注意到，仅仅一个 .close() 并不能拯救您。您的两个 while (1): 循环都从未终止的 .Context( 1 ) 中生成至少一对（一个四边形一起 (!!!) ）新的附加 I/O 线程b>-s 进入您的 O/S 计划在每个循环中（！！！）。这迟早会杀死任何一种当代操作系统。

标签： python synchronization deadlock zeromq low-latency

【解决方案1】：

这不是死锁案例

当然，代码仍然需要注意。

消歧：您的场景不会进入资源互锁状态，也就是 DeadLock。是的，当然，您的代码崩溃了，但很可能不是由于 REQ/REP DeadLock （它可能并且确实出现在有损网络 tcp:transport-class 上）。发布的代码崩溃是由于非托管资源处理，而不是由于达到了死锁/活锁的相互阻塞状态。

如何解决？

首先，让我们假设您的超低延迟驱动系统不允许重复实例化任何内容。这也有例外，但让我们成为专业人士。

将您的.Context() 资源设置（或从外部调用继承）移出循环
查看，您是否需要以及您的延迟限制允许您在每次循环运行中设置/拆除 .socket() 资源两次。
决定，一旦第一条消息在传输路径中丢失，您是否可以忍受真正的REQ/REP 死锁
实施优雅的资源使用终止（.socket()-s，O/S port#s，.Context()-s）。不要让他们永远悬而未决，同时创建无限数量的其他人，这会破坏任何“容错”系统。资源永远是无限的。
以非阻塞方式设计信令和传输行为。这使您可以检测和处理远程进程超时，并为本地补救/响应操作提供机会。
重新设计代码到您需要的安全代码级别（以下示例在分布式处理框架中 24/7/365 的软实时控制无限循环中工作了几年使用远程键盘和其他一些本地和远程诊断工具）。

生产级代码缺少什么？

您的代码必须“设想”在分布式系统的任何部分中可能出现的问题。是的，这很难，但很有必要。您的远程节点 - 一个通信对方 - 停止响应，丢失消息，重新启动，由于操作系统崩溃而停止，无论你能想象到什么（加上一些相当令人讨厌的惊喜，你只会在飞行中发现......） .这是这篇小文章中要介绍的另一个潘多拉魔盒，但这并不意味着没有必要。这是你的救生背心。

尽可能以非阻塞方式进行设计，这样您就可以控制事件...

无论如何，总是以优雅的方式释放系统资源和 .term() 所有 ZeroMQ .Context() 实例——“整理”是一种公平的做法——无论是在现实生活中还是在现实生活中代码帝国。

# /\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\/\
#NONSTOP RESPONDER RAW EXAMPLE:
def aMiniRESPONDER( aTarget2Bind2_URL             = "tcp://A.B.C.D:8889",
                    anExternalPREDICTOR           = None,
                    anExternallyManagedZmqCONTEXT = None,
                    aSpreadMinSafetyMUL           = 3.0,
                    aSilentMODE                   = True
                    ):
   try: # RESOURCES LAYER
        # ... SETUP
        # ------------------------------------------------- .Context()
        # can setup a locally-managed context or re-use
        # anExternallyManagedZmqCONTEXT obtained upon a func Call
        aZmqCONTEXT   = anExternallyManagedZmqCONTEXT or zmq.Context( 1 )   

        # localhost:8887 [REP] ... remote [REQ] peer  .connect() + .send()
        aCtrlPORT_URL = "tcp://*:8887"                                      

        # localhost:8890 [PUB] ... remote [SUB] peers .connect() +
        # .subscribe + .recv( zmq.NOBLOCK ) ( MQL4 cannot .poll() so far ...)
        aSIGsPORT_URL = "tcp://*:8890"                                      
        aXmitPORT_URL = aTarget2Bind2_URL

        aListOfSOCKETs = []

        pass # -------------------------------------------------------------# ZMQ
        try: # -------------------------------------------------------------#
            # try: XmitPORT
            aXmitSOCKET = aZmqCONTEXT.socket( zmq.PAIR )

            # XmitPORT
            aXmitSOCKET.bind(      aXmitPORT_URL )                          
            aListOfSOCKETs.append( aXmitSOCKET )
        except:                                                             
            #    EXC: XmitPORT on Failure: GRACEFUL CLEARING XmitPORT

            msg =  "\nEXC. ZmqError({0:s}) on aXmitSOCKET setup / .bind( {1:s} )"
            print msg.format( repr( zmq.ZMQError() ), aTarget2Bind2_URL )
            raise ValueError( "ZMQ_EXC_EXIT @ XmitPORT SETUP" )
        pass # -------------------------------------------------------------# ZMQ
        try: # -------------------------------------------------------------#
            # try: CtrlPORT    
            # CtrlSOCKET [REP] .recv()s<--[REQ] + .send()s--> [REQ]
            aCtrlSOCKET = aZmqCONTEXT.socket( zmq.REP )                     

            # CtrlPORT <-REQ/REP means a remote peer [REQ] has to
            # .send()+.recv() before sending another CtrlCMD
            aCtrlSOCKET.bind(      aCtrlPORT_URL )                          
            aListOfSOCKETs.append( aCtrlSOCKET )
        except:                                                             
            # EXC: CtrlPORT on Failure: GRACEFUL CLEARING both CtrlPORT
            # and XmitPORT
            msg =  "\nEXC. ZmqError({0:s}) on aCtrlSOCKET setup / .bind( {1:s} )"
            print msg.format( repr( zmq.ZMQError() ), aCtrlPORT_URL )
            raise ValueError( "ZMQ_EXC_EXIT @ CtrlPORT SETUP" )
        pass # -------------------------------------------------------------# ZMQ
        try: # -------------------------------------------------------------#
            # try: SIGsPORT

            # SIGsPORT [PUB] .send()s--> [SUB]s
            aSIGsSOCKET= aZmqCONTEXT.socket( zmq.PUB  )                     

            # SIGsPORT -->  PUB/SUB means a remote peer(s) [SUB] .subscribe() + .recv()
            aSIGsSOCKET.bind(      aSIGsPORT_URL )                          
            aListOfSOCKETs.append( aSIGsSOCKET )
        except:                                                             
            # EXC: SIGsPORT on Failure: GRACEFUL CLEARING both CtrlPORT
            # and XmitPORT and SIGsPORT
            msg =  "\nEXC. ZmqError({0:s}) on aSIGsSOCKET setup / .bind( {1:s} )"
            print msg.format( repr( zmq.ZMQError() ), aSIGsPORT_URL )
            raise ValueError( "ZMQ_EXC_EXIT @ SIGsPORT SETUP" )
        pass # -------------------------------------------------------------# ZMQ

        # vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv
        # ... SETUP YOUR APPLICATION CODE

        try:     # APP LAYER ___________________________________________
           #           what you want to do
           #           here you go ...

        except:  # APP LAYER ___________________________________________
           #           handle EXCs

        finally: # APP LAYER ___________________________________________
           #           your own application post-mortem / pre-exit code

        # ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

   except:  # RESOURCES LAYER .............................................
        # ... code shall handle it's own exceptions + externally caused events

   finally: # RESOURCES LAYER .............................................
        # ... always, ALWAYS gracefully exit ( avoid leakages and dirty things )

        [ allSOCKETs.setsockopt( zmq.LINGER, 0 ) for allSOCKETs in aListOfSOCKETs ]
        [ allSOCKETs.close( )                    for allSOCKETs in aListOfSOCKETs ]

        # --------------------------------------------------------------#
        # RESOURCES dismantled, may .term()

        # .TERM(), NOP otherwise
        if not ( aZmqCONTEXT is anExternallyManagedZmqCONTEXT ):        #
                 aZmqCONTEXT.term()                                     #
        return

【讨论】：

哦！澄清这不是我认为的僵局已经有很大帮助。今天我将根据您的建议改进我的代码。应该仔细研究一下 0MQ 的实现方式。
完全同意所有 6 条建议。每一个都很重要（并且需要一些时间来理解和正确实施）。代码示例很难掌握 - 宽（cmets）和长（许多代码行）。顶级通信模式会有所帮助。不确定zmq.PAIR - 规范声称，它被认为是实验性的，并且不支持重新连接，我一直在避免它，也许我会在下一个项目中重新考虑它。

【解决方案2】：

您应该只处理一次 context 和 socket 创建，而不是每次迭代。

另外，你应该重用上下文（除非你打算从另一个代码中的线程）。

try:
    context = zmq.Context()
    rep_sck = context.socket(zmq.REP)
    rep_sck.bind('tcp://127.0.0.1:10001')
    rq_sck = context.socket(zmq.REQ)
    rq_sck.connect('tcp://127.0.0.1:10002')
    while (1):
        msg = rep_sck.recv() # Waiting for script B to send done

        do something useful (takes only a few millisecs)

        rq_sck.send_string("done") # Tell script B we are done
finally:
    rep_sck.close()
    rq_sck.close()

第二个脚本也是如此。

try:
    context = zmq.Context()
    rq_sck = context.socket(zmq.REQ)
    rq_sck.connect('tcp://127.0.0.1:10001')

    rep_sck = context.socket(zmq.REP)
    rep_sck.bind('tcp://127.0.0.1:10002')

    while (1):
        do something useful (takes only a few millisecs)

        rq_sck.send_string("done") # Tell script A we are done

        msg = rep_sck.recv() # Waiting for script A to send done
finally:
    rq_sck.close()
    rep_sck.close()

编辑：更新代码以调用Socket.close()

由于 pyzmq 版本 14.3.0，Socket.close() 和 Context.term() 不会自动调用在垃圾收集期间，添加了正确关闭套接字。

【讨论】：