Review API betweeen connections and connectors

- Review error handling. Only 2 exceptions remain in connector.py:

  - Drop useless exception handling for EAGAIN since it should not happen
    if the kernel says the socket is ready.
  - Do not distinguish other socket errors. Just close and log in a generic way.
  - No need to raise a specific exception for EOF.
  - Make 'connect' return a boolean instead of raising an exception.
  - Raise appropriate exception when answer/ask/notify is called on a closed
    non-MT connection.

- Add support for more complex connectors, which may need to write for a read
  operation, or to read when there's pending data to send. This will be
  required for SSL support (more exactly, the handshake will be done in
  a transparent way):

  - Move write buffer to connector.
  - Make 'receive' fill the read buffer, instead of returning the read data.
  - Make 'receive' & 'send' return a boolean to switch polling for writing.
  - Tolerate that sockets return 0 as number of bytes sent.

- In testConnection, simply delete all failing tests, as announced
  in commit 71e30fb9.

neo/lib/connection.py

@@ -18,9 +18,7 @@ from functools import wraps
 from time import time
 
 from . import attributeTracker, logging
-from .connector import ConnectorException, ConnectorTryAgainException, \
-        ConnectorInProgressException, ConnectorConnectionRefusedException, \
-        ConnectorConnectionClosedException, ConnectorDelayedConnection
+from .connector import ConnectorException, ConnectorDelayedConnection
 from .locking import RLock
 from .protocol import uuid_str, Errors, \
         PacketMalformedError, Packets, ParserState
@@ -31,14 +29,6 @@ CRITICAL_TIMEOUT = 30
 class ConnectionClosed(Exception):
     pass
 
-def not_closed(func):
-    def decorator(self, *args, **kw):
-        if self.connector is None:
-            raise ConnectorConnectionClosedException
-        return func(self, *args, **kw)
-    return wraps(func)(decorator)
-
-
 class HandlerSwitcher(object):
     _is_handling = False
     _next_timeout = None
@@ -316,14 +306,11 @@ class ListeningConnection(BaseConnection):
         self.em.register(self)
 
     def readable(self):
-        try:
         connector, addr = self.connector.accept()
         logging.debug('accepted a connection from %s:%d', *addr)
         handler = self.getHandler()
         new_conn = ServerConnection(self.em, handler, connector, addr)
         handler.connectionAccepted(new_conn)
-        except ConnectorTryAgainException:
-            pass
 
     def getAddress(self):
         return self.connector.getAddress()
@@ -347,7 +334,6 @@ class Connection(BaseConnection):
     def __init__(self, event_manager, *args, **kw):
         BaseConnection.__init__(self, event_manager, *args, **kw)
         self.read_buf = ReadBuffer()
-        self.write_buf = []
         self.cur_id = 0
         self.aborted = False
         self.uuid = None
@@ -444,39 +430,47 @@ class Connection(BaseConnection):
         """Abort dealing with this connection."""
         logging.debug('aborting a connector for %r', self)
         self.aborted = True
-        assert self.write_buf
+        assert self.pending()
         if self._on_close is not None:
             self._on_close()
             self._on_close = None
 
     def writable(self):
         """Called when self is writable."""
-        self._send()
-        if not self.write_buf and self.connector is not None:
+        try:
+            if self.connector.send():
                 if self.aborted:
                     self.close()
                 else:
                     self.em.removeWriter(self)
+        except ConnectorException:
+            self._closure()
 
     def readable(self):
         """Called when self is readable."""
-        self._recv()
-        self._analyse()
-        if self.aborted:
-            self.em.removeReader(self)
-        return not not self._queue
-
-    def _analyse(self):
-        """Analyse received data."""
+        # last known remote activity
+        self._next_timeout = time() + self._timeout
+        read_buf = self.read_buf
         try:
-            while True:
-                packet = Packets.parse(self.read_buf, self._parser_state)
+            try:
+                if self.connector.receive(read_buf):
+                    self.em.addWriter(self)
+            finally:
+                # A connector may read some data
+                # before raising ConnectorException
+                while 1:
+                    packet = Packets.parse(read_buf, self._parser_state)
                     if packet is None:
                         break
                     self._queue.append(packet)
+        except ConnectorException:
+            self._closure()
         except PacketMalformedError, e:
             logging.error('malformed packet from %r: %s', self, e)
             self._closure()
+        if self.aborted:
+            self.em.removeReader(self)
+        return not not self._queue
 
     def hasPendingMessages(self):
         """
@@ -493,7 +487,8 @@ class Connection(BaseConnection):
         self.updateTimeout()
 
     def pending(self):
-        return self.connector is not None and self.write_buf
+        connector = self.connector
+        return connector is not None and connector.queued
 
     @property
     def setReconnectionNoDelay(self):
@@ -503,7 +498,6 @@ class Connection(BaseConnection):
         if self.connector is None:
             assert self._on_close is None
             assert not self.read_buf
-            assert not self.write_buf
             assert not self.isPending()
             return
         # process the network events with the last registered handler to
@@ -514,7 +508,6 @@ class Connection(BaseConnection):
         if self._on_close is not None:
             self._on_close()
             self._on_close = None
-        del self.write_buf[:]
         self.read_buf.clear()
         try:
             if self.connecting:
@@ -531,89 +524,28 @@ class Connection(BaseConnection):
             self._handlers.handle(self, self._queue.pop(0))
         self.close()
 
-    def _recv(self):
-        """Receive data from a connector."""
-        try:
-            data = self.connector.receive()
-        except ConnectorTryAgainException:
-            pass
-        except ConnectorConnectionRefusedException:
-            assert self.connecting
-            self._closure()
-        except ConnectorConnectionClosedException:
-            # connection resetted by peer, according to the man, this error
-            # should not occurs but it seems it's false
-            logging.debug('Connection reset by peer: %r', self.connector)
-            self._closure()
-        except:
-            logging.debug('Unknown connection error: %r', self.connector)
-            self._closure()
-            # unhandled connector exception
-            raise
-        else:
-            if not data:
-                logging.debug('Connection %r closed in recv', self.connector)
-                self._closure()
-                return
-            # last known remote activity
-            self._next_timeout = time() + self._timeout
-            self.read_buf.append(data)
-
-    def _send(self):
-        """Send data to a connector."""
-        if not self.write_buf:
-            return
-        msg = ''.join(self.write_buf)
-        try:
-            n = self.connector.send(msg)
-        except ConnectorTryAgainException:
-            pass
-        except ConnectorConnectionClosedException:
-            # connection resetted by peer
-            logging.debug('Connection reset by peer: %r', self.connector)
-            self._closure()
-        except:
-            logging.debug('Unknown connection error: %r', self.connector)
-            # unhandled connector exception
-            self._closure()
-            raise
-        else:
-            if not n:
-                logging.debug('Connection %r closed in send', self.connector)
-                self._closure()
-                return
-            if n == len(msg):
-                del self.write_buf[:]
-            else:
-                self.write_buf = [msg[n:]]
-
     def _addPacket(self, packet):
         """Add a packet into the write buffer."""
-        if self.connector is None:
-            return
-
-        was_empty = not self.write_buf
-
-        self.write_buf.extend(packet.encode())
-
-        if was_empty:
+        if self.connector.queue(packet.encode()):
             # enable polling for writing.
             self.em.addWriter(self)
         logging.packet(self, packet, True)
 
-    @not_closed
     def notify(self, packet):
         """ Then a packet with a new ID """
+        if self.isClosed():
+            raise ConnectionClosed
         msg_id = self._getNextId()
         packet.setId(msg_id)
         self._addPacket(packet)
         return msg_id
 
-    @not_closed
     def ask(self, packet, timeout=CRITICAL_TIMEOUT, on_timeout=None, **kw):
         """
         Send a packet with a new ID and register the expectation of an answer
         """
+        if self.isClosed():
+            raise ConnectionClosed
         msg_id = self._getNextId()
         packet.setId(msg_id)
         self._addPacket(packet)
@@ -627,9 +559,10 @@ class Connection(BaseConnection):
                 self.em.wakeup()
         return msg_id
 
-    @not_closed
     def answer(self, packet, msg_id=None):
         """ Answer to a packet by re-using its ID for the packet answer """
+        if self.isClosed():
+            raise ConnectionClosed
         if msg_id is None:
             msg_id = self.getPeerId()
         packet.setId(msg_id)
@@ -656,32 +589,25 @@ class ClientConnection(Connection):
 
     def _connect(self):
         try:
-            self.connector.makeClientConnection()
-        except ConnectorInProgressException:
-            self.em.register(self)
-            self.em.addWriter(self)
+            connected = self.connector.makeClientConnection()
         except ConnectorDelayedConnection, c:
             connect_limit, = c.args
             self.getTimeout = lambda: connect_limit
             self.onTimeout = self._delayedConnect
             self.em.register(self, timeout_only=True)
-            # Fake _addPacket so that if does not
-            # try to reenable polling for writing.
-            self.write_buf.insert(0, '')
-        except ConnectorConnectionRefusedException:
-            self._closure()
         except ConnectorException:
-            # unhandled connector exception
             self._closure()
-            raise
         else:
             self.em.register(self)
-            if self.write_buf:
-                self.em.addWriter(self)
+            if connected:
                 self._connectionCompleted()
+                # A client connection usually has a pending packet to send
+                # from the beginning. It would be too smart to detect when
+                # it's not required to poll for writing.
+            self.em.addWriter(self)
 
     def _delayedConnect(self):
-        del self.getTimeout, self.onTimeout, self.write_buf[0]
+        del self.getTimeout, self.onTimeout
         self._connect()
 
     def writable(self):

neo/lib/connector.py

@@ -17,6 +17,7 @@
 import socket
 import errno
 from time import time
+from . import logging
 
 # Global connector registry.
 # Fill by calling registerConnectorHandler.
@@ -56,8 +57,19 @@ class SocketConnector(object):
         s.setblocking(0)
         # disable Nagle algorithm to reduce latency
         s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        self.queued = []
         return self
 
+    def queue(self, data):
+        was_empty = not self.queued
+        self.queued += data
+        return was_empty
+
+    def _error(self, op, exc):
+        logging.debug("%s failed for %s: %s (%s)",
+            op, self, errno.errorcode[exc.errno], exc.strerror)
+        raise ConnectorException
+
     # Threaded tests monkey-patch the following 2 operations.
     _connect = lambda self, addr: self.socket.connect(addr)
     _bind = lambda self, addr: self.socket.bind(addr)
@@ -68,20 +80,23 @@ class SocketConnector(object):
         try:
             connect_limit = self.connect_limit[addr]
             if time() < connect_limit:
+                # Next call to queue() must return False
+                # in order not to enable polling for writing.
+                self.queued or self.queued.append('')
                 raise ConnectorDelayedConnection(connect_limit)
+            if self.queued and not self.queued[0]:
+                del self.queued[0]
         except KeyError:
             pass
         self.connect_limit[addr] = time() + self.CONNECT_LIMIT
         self.is_server = self.is_closed = False
         try:
             self._connect(addr)
-        except socket.error, (err, errmsg):
-            if err == errno.EINPROGRESS:
-                raise ConnectorInProgressException
-            if err == errno.ECONNREFUSED:
-                raise ConnectorConnectionRefusedException
-            raise ConnectorException, 'makeClientConnection to %s failed:' \
-                ' %s:%s' % (addr, err, errmsg)
+        except socket.error, e:
+            if e.errno == errno.EINPROGRESS:
+                return False
+            self._error('connect', e)
+        return True
 
     def makeListeningConnection(self):
         assert self.is_closed is None
@@ -90,10 +105,9 @@ class SocketConnector(object):
             self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             self._bind(self.addr)
             self.socket.listen(5)
-        except socket.error, (err, errmsg):
+        except socket.error, e:
             self.socket.close()
-            raise ConnectorException, 'makeListeningConnection on %s failed:' \
-                    ' %s:%s' % (addr, err, errmsg)
+            self._error('listen', e)
 
     def getError(self):
         return self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
@@ -116,33 +130,39 @@ class SocketConnector(object):
             s, addr = self.socket.accept()
             s = self.__class__(addr, s)
             return s, s.addr
-        except socket.error, (err, errmsg):
-            if err == errno.EAGAIN:
-                raise ConnectorTryAgainException
-            raise ConnectorException, 'accept failed: %s:%s' % \
-                (err, errmsg)
+        except socket.error, e:
+            self._error('accept', e)
 
-    def receive(self):
+    def receive(self, read_buf):
         try:
-            return self.socket.recv(4096)
-        except socket.error, (err, errmsg):
-            if err == errno.EAGAIN:
-                raise ConnectorTryAgainException
-            if err in (errno.ECONNREFUSED, errno.EHOSTUNREACH):
-                raise ConnectorConnectionRefusedException
-            if err in (errno.ECONNRESET, errno.ETIMEDOUT):
-                raise ConnectorConnectionClosedException
-            raise ConnectorException, 'receive failed: %s:%s' % (err, errmsg)
-
-    def send(self, msg):
+            data = self.socket.recv(4096)
+        except socket.error, e:
+            self._error('recv', e)
+        if data:
+            read_buf.append(data)
+            return
+        logging.debug('%r closed in recv', self)
+        raise ConnectorException
+
+    def send(self):
+        msg = ''.join(self.queued)
+        if msg:
             try:
-            return self.socket.send(msg)
-        except socket.error, (err, errmsg):
-            if err == errno.EAGAIN:
-                raise ConnectorTryAgainException
-            if err in (errno.ECONNRESET, errno.ETIMEDOUT, errno.EPIPE):
-                raise ConnectorConnectionClosedException
-            raise ConnectorException, 'send failed: %s:%s' % (err, errmsg)
+                n = self.socket.send(msg)
+            except socket.error, e:
+                self._error('send', e)
+            # Do nothing special if n == 0:
+            # - it never happens for simple sockets;
+            # - for SSL sockets, this is always the case unless everything
+            #   could be sent.
+            if n != len(msg):
+                self.queued[:] = msg[n:],
+                return False
+            del self.queued[:]
+        else:
+            assert not self.queued
+        return True
+
 
     def close(self):
         self.is_closed = True
@@ -195,17 +215,5 @@ registerConnectorHandler(SocketConnectorIPv6)
 class ConnectorException(Exception):
     pass
 
-class ConnectorTryAgainException(ConnectorException):
-    pass
-
-class ConnectorInProgressException(ConnectorException):
-    pass
-
-class ConnectorConnectionClosedException(ConnectorException):
-    pass
-
-class ConnectorConnectionRefusedException(ConnectorException):
-    pass
-
 class ConnectorDelayedConnection(ConnectorException):
     pass

neo/storage/checker.py

@@ -16,8 +16,7 @@
 
 from collections import deque
 from neo.lib import logging
-from neo.lib.connection import ClientConnection
-from neo.lib.connector import ConnectorConnectionClosedException
+from neo.lib.connection import ClientConnection, ConnectionClosed
 from neo.lib.protocol import NodeTypes, Packets, ZERO_OID
 from neo.lib.util import add64, dump
 from .handlers.storage import StorageOperationHandler
@@ -85,7 +84,7 @@ class Checker(object):
                         if self.conn_dict:
                             break
                         msg = "no replica"
-                    except ConnectorConnectionClosedException:
+                    except ConnectionClosed:
                         msg = "connection closed"
                     finally:
                         conn_set.update(self.conn_dict)

neo/storage/handlers/storage.py

@@ -16,7 +16,7 @@
 
 import weakref
 from functools import wraps
-from neo.lib.connector import ConnectorConnectionClosedException
+from neo.lib.connection import ConnectionClosed
 from neo.lib.handler import EventHandler
 from neo.lib.protocol import Errors, NodeStates, Packets, ProtocolError, \
     ZERO_HASH
@@ -154,7 +154,7 @@ class StorageOperationHandler(EventHandler):
             r = app.dm.checkTIDRange(*args)
             try:
                 conn.answer(Packets.AnswerCheckTIDRange(*r), msg_id)
-            except (weakref.ReferenceError, ConnectorConnectionClosedException):
+            except (weakref.ReferenceError, ConnectionClosed):
                 pass
             yield
         app.newTask(check())
@@ -170,7 +170,7 @@ class StorageOperationHandler(EventHandler):
             r = app.dm.checkSerialRange(*args)
             try:
                 conn.answer(Packets.AnswerCheckSerialRange(*r), msg_id)
-            except (weakref.ReferenceError, ConnectorConnectionClosedException):
+            except (weakref.ReferenceError, ConnectionClosed):
                 pass
             yield
         app.newTask(check())
@@ -211,7 +211,7 @@ class StorageOperationHandler(EventHandler):
                 conn.answer(Packets.AnswerFetchTransactions(
                     pack_tid, next_tid, peer_tid_set), msg_id)
                 yield
-            except (weakref.ReferenceError, ConnectorConnectionClosedException):
+            except (weakref.ReferenceError, ConnectionClosed):
                 pass
         app.newTask(push())
 
@@ -253,6 +253,6 @@ class StorageOperationHandler(EventHandler):
                 conn.answer(Packets.AnswerFetchObjects(
                     pack_tid, next_tid, next_oid, object_dict), msg_id)
                 yield
-            except (weakref.ReferenceError, ConnectorConnectionClosedException):
+            except (weakref.ReferenceError, ConnectionClosed):
                 pass
         app.newTask(push())

neo/tests/threaded/__init__.py

@@ -31,8 +31,7 @@ import neo.client.app, neo.neoctl.app
 from neo.client import Storage
 from neo.lib import logging
 from neo.lib.connection import BaseConnection, Connection
-from neo.lib.connector import SocketConnector, \
-    ConnectorConnectionRefusedException
+from neo.lib.connector import SocketConnector, ConnectorException
 from neo.lib.locking import SimpleQueue
 from neo.lib.protocol import CellStates, ClusterStates, NodeStates, NodeTypes
 from neo.lib.util import cached_property, parseMasterList, p64
@@ -322,7 +321,7 @@ class ServerNode(Node):
         try:
             return self.listening_conn.getAddress()
         except AttributeError:
-            raise ConnectorConnectionRefusedException
+            raise ConnectorException
 
 class AdminApplication(ServerNode, neo.admin.app.Application):
     pass