thiell/test — Issue #207: EnginePort queue full deadlock
Author: thiell
Created: 2012-03-03 16:36:35+00:00
State: open
Labels: Lib/Worker

At the moment, an EnginePort full of messages makes the {{{msg()}}} or {{{msg_send()}}} calls to block, due to the {{{block=True}}} parameter of the {{{Queue.put()}}} method. This is the easiest way to guarantee proper msg delivery, but leads to deadlock when the queue is full and the engine is not running (in most case, when running, calls to {{{Queue.get()}}} to read the port messages will unblock the caller).

{{{
#!python
def msg(self, send_msg, send_once=False):
"""
Port message send method that will wait for acknowledgement
unless the send_once parameter if set.
"""
pmsg = EnginePort._Msg(send_msg, not send_once)
self._msgq.put(pmsg, block=True, timeout=None)

}}}

The most annoying situation, is, for exemple, when you schedule a lot of works from another thread (as {{{Task.schedule()}}} is tasksyncmethod-decorated, it will send a message through the task's port) *and* task/engine is not running.

We should switch to a non-blocking put() to avoid any library deadlock. Indeed, when switching "as is" to non-blocking mode, an exception is raised due to the Queue being full:

{{{
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
None
msg (, (,), {})
Error in sys.excepthook:
Traceback (most recent call last):
File "/clustershell/branches/1.4/lib/ClusterShell/CLI/Clush.py", line 625, in clush_excepthook
raise extype, value
Queue.Full

Original exception was:
Traceback (most recent call last):
File "./clush.py", line 11, in
main()
File "/clustershell/branches/1.4/lib/ClusterShell/CLI/Clush.py", line 843, in main
options.preserve_flag, display)
File "/clustershell/branches/1.4/lib/ClusterShell/CLI/Clush.py", line 597, in run_rcopy
preserve=preserve_flag)
File "/clustershell/branches/1.4/lib/ClusterShell/Task.py", line 532, in rcopy
return self.copy(source, dest, nodes, **kwargs)
File "/clustershell/branches/1.4/lib/ClusterShell/Task.py", line 524, in copy
self.schedule(worker)
File "/clustershell/branches/1.4/lib/ClusterShell/Task.py", line 166, in taskfunc
task._dispatch_port.msg_send((f, fargs, kwargs))
File "/clustershell/branches/1.4/lib/ClusterShell/Worker/EngineClient.py", line 444, in msg_send
self.msg(send_msg, send_once=True)
File "/clustershell/branches/1.4/lib/ClusterShell/Worker/EngineClient.py", line 432, in msg
print self._msgq.put(pmsg, block=False, timeout=None)
File "/usr/lib64/python2.6/Queue.py", line 123, in put
raise Full
Queue.Full
}}}