distbuild: Handle errors from socket

We found a distbuild controller stuck in a busy loop, with the logs
full of the same error message repeated:

    ... _flush(): Exception 'IOError: [Errno 32] Broken pipe' from sock.write()

We suspect this came about because the initiator disconnected without
sending an EOF. The initiator was in a VM on a laptop so it seems
possible that the host OS turned off the wireless adaptor without giving
the VM a chance to close its connections gracefully.

The busy loop is because nothing in the SocketBuffer class handles the
SocketError events queued by the _flush() method. Unhandled events are
ignored. So the SocketBuffer stays in 'w' state without ever shifting
any data and never returns. Adding transitions to handle the SocketError
event will fix the problem.

If a socket error happens now in the same scenario, it will be handled
as if the initiator disconnected.

Change-Id: I0f6834f7186a01ca2bc74aef899a4cccbc891e51

4 files changed, 25 insertions, 3 deletions

diff --git a/distbuild/__init__.py b/distbuild/__init__.py
index 145dc40a..bc5bc9b5 100644
--- a/distbuild/__init__.py
+++ b/distbuild/__init__.py
@@ -25,7 +25,7 @@ from sockbuf import (SocketBufferNewData, SocketBufferEof,
                      SocketBufferClosed, SocketBuffer)
 from mainloop import MainLoop
 from sockserv import ListenServer
-from jm import JsonMachine, JsonNewMessage, JsonEof
+from jm import JsonMachine, JsonNewMessage, JsonEof, JsonError
 
 from artifact_reference import (encode_artifact,
                                 encode_artifact_reference,
diff --git a/distbuild/initiator_connection.py b/distbuild/initiator_connection.py
index 6e6ea833..4efcfdb1 100644
--- a/distbuild/initiator_connection.py
+++ b/distbuild/initiator_connection.py
@@ -75,6 +75,8 @@ class InitiatorConnection(distbuild.StateMachine):
             # state, source, event_class, new_state, callback
             ('idle', self.jm, distbuild.JsonNewMessage, 'idle', 
                 self._handle_msg),
+            ('idle', self.jm, distbuild.JsonError, 'closing',
+                self._disconnect),
             ('idle', self.jm, distbuild.JsonEof, 'closing', self._disconnect),
             ('idle', distbuild.BuildController, distbuild.BuildFinished,
                 'idle', self._send_build_finished_message),
diff --git a/distbuild/jm.py b/distbuild/jm.py
index c08a1958..d1e5fab7 100644
--- a/distbuild/jm.py
+++ b/distbuild/jm.py
@@ -38,8 +38,17 @@ class JsonNewMessage(object):
 class JsonEof(object):
 
     pass
-    
-    
+
+
+class JsonError(object):
+
+    '''An error has occured with a socket used for communication.'''
+
+    def __init__(self, sock, exception):
+        self.sock = sock
+        self.exception = exception
+
+
 class _Close2(object):
 
     pass
@@ -71,9 +80,12 @@ class JsonMachine(StateMachine):
             # state, source, event_class, new_state, callback
             ('rw', sockbuf, SocketBufferNewData, 'rw', self._parse),
             ('rw', sockbuf, SocketBufferEof, 'w', self._send_eof),
+            ('rw', sockbuf, SocketError, 'error', self._send_error),
             ('rw', self, _Close2, None, self._really_close),
             
             ('w', self, _Close2, None, self._really_close),
+
+            ('error', self, _Close2, None, self._really_close)
         ]
         self.add_transitions(spec)
         
@@ -119,6 +131,9 @@ class JsonMachine(StateMachine):
     def _send_eof(self, event_source, event):
         self.mainloop.queue_event(self, JsonEof())
 
+    def _send_error(self, event_source, event):
+        self.mainloop.queue_event(self, JsonError(event.sock, event.exception))
+
     def _really_close(self, event_source, event):
         self.sockbuf.close()
         self._send_eof(event_source, event)
diff --git a/distbuild/sockbuf.py b/distbuild/sockbuf.py
index 6feb8669..be17a8f4 100644
--- a/distbuild/sockbuf.py
+++ b/distbuild/sockbuf.py
@@ -105,22 +105,27 @@ class SocketBuffer(StateMachine):
             ('reading', self, _WriteBufferNotEmpty, 'rw', 
                 self._start_writing),
             ('reading', self, SocketBufferEof, 'idle', None),
+            ('reading', self, SocketError, None, self._really_close),
             ('reading', self, _Close, None, self._really_close),
             
             ('rw', src, SocketReadable, 'rw', self._fill),
             ('rw', src, SocketWriteable, 'rw', self._flush),
             ('rw', self, _WriteBufferIsEmpty, 'reading', self._stop_writing),
             ('rw', self, SocketBufferEof, 'w', None),
+            ('rw', self, SocketError, None, self._really_close),
             ('rw', self, _Close, 'wc', None),
             
             ('idle', self, _WriteBufferNotEmpty, 'w', self._start_writing),
+            ('idle', self, SocketError, None, self._really_close),
             ('idle', self, _Close, None, self._really_close),
             
             ('w', src, SocketWriteable, 'w', self._flush),
             ('w', self, _WriteBufferIsEmpty, 'idle', self._stop_writing),
+            ('w', self, SocketError, None, self._really_close),
 
             ('wc', src, SocketWriteable, 'wc', self._flush),
             ('wc', self, _WriteBufferIsEmpty, None, self._really_close),
+            ('wc', self, SocketError, None, self._really_close),
         ]
         self.add_transitions(spec)