gh-274: Handle blocking I/O errors in GreenSocket

vstinner · jstasiak · commit bc4d1b5d362e · 2016-01-10T02:35:24.000+01:00
Fix recv_into(), recvfrom(), recvfrom_into() and sendto() methods of GreenSocket to handle blocking I/O errors (ex: BlockingIOError on Python 3). Even if the trampoline was called, the socket method can still fails with an I/O errors for various reasons (see manual pages of the C functions for examples). Ref: #274
diff --git a/eventlet/greenio/base.py b/eventlet/greenio/base.py
@@ -304,59 +304,61 @@ def makeGreenFile(self, *args, **kw):
                       "makefile instead", DeprecationWarning, stacklevel=2)
         return self.makefile(*args, **kw)
 
-    def recv(self, buflen, flags=0):
+    def _read_trampoline(self):
+        self._trampoline(
+            self.fd,
+            read=True,
+            timeout=self.gettimeout(),
+            timeout_exc=socket.timeout("timed out"))
+
+    def _recv_loop(self, recv_meth, *args):
         fd = self.fd
         if self.act_non_blocking:
-            return fd.recv(buflen, flags)
+            return recv_meth(*args)
+
         while True:
             try:
-                return fd.recv(buflen, flags)
+                # recv: bufsize=0?
+                # recv_into: buffer is empty?
+                if not args[0]:
+                    self._read_trampoline()
+                return recv_meth(*args)
             except socket.error as e:
                 if get_errno(e) in SOCKET_BLOCKING:
                     pass
                 elif get_errno(e) in SOCKET_CLOSED:
                     return b''
                 else:
                     raise
+
             try:
-                self._trampoline(
-                    fd,
-                    read=True,
-                    timeout=self.gettimeout(),
-                    timeout_exc=socket.timeout("timed out"))
+                self._read_trampoline()
             except IOClosed as e:
                 # Perhaps we should return '' instead?
                 raise EOFError()
 
-    def recvfrom(self, *args):
-        if not self.act_non_blocking:
-            self._trampoline(self.fd, read=True, timeout=self.gettimeout(),
-                             timeout_exc=socket.timeout("timed out"))
-        return self.fd.recvfrom(*args)
+    def recv(self, bufsize, flags=0):
+        return self._recv_loop(self.fd.recv, bufsize, flags)
 
-    def recvfrom_into(self, *args):
-        if not self.act_non_blocking:
-            self._trampoline(self.fd, read=True, timeout=self.gettimeout(),
-                             timeout_exc=socket.timeout("timed out"))
-        return self.fd.recvfrom_into(*args)
+    def recvfrom(self, bufsize, flags=0):
+        return self._recv_loop(self.fd.recvfrom, bufsize, flags)
 
-    def recv_into(self, *args):
-        if not self.act_non_blocking:
-            self._trampoline(self.fd, read=True, timeout=self.gettimeout(),
-                             timeout_exc=socket.timeout("timed out"))
-        return self.fd.recv_into(*args)
+    def recv_into(self, buffer, nbytes=0, flags=0):
+        return self._recv_loop(self.fd.recv_into, buffer, nbytes, flags)
 
-    def send(self, data, flags=0):
-        fd = self.fd
+    def recvfrom_into(self, buffer, nbytes=0, flags=0):
+        return self._recv_loop(self.fd.recvfrom_into, buffer, nbytes, flags)
+
+    def _send_loop(self, send_method, data, *args):
         if self.act_non_blocking:
-            return fd.send(data, flags)
+            return send_method(data, *args)
 
         # blocking socket behavior - sends all, blocks if the buffer is full
         total_sent = 0
         len_data = len(data)
         while 1:
             try:
-                total_sent += fd.send(data[total_sent:], flags)
+                total_sent += send_method(data[total_sent:], *args)
             except socket.error as e:
                 eno = get_errno(e)
                 if eno == errno.ENOTCONN or eno not in SOCKET_BLOCKING:
@@ -373,16 +375,18 @@ def send(self, data, flags=0):
 
         return total_sent
 
+    def send(self, data, flags=0):
+        return self._send_loop(self.fd.send, data, flags)
+
+    def sendto(self, data, address, flags=0):
+        return self._send_loop(self.fd.sendto, data, address, flags)
+
     def sendall(self, data, flags=0):
         tail = self.send(data, flags)
         len_data = len(data)
         while tail < len_data:
             tail += self.send(data[tail:], flags)
 
-    def sendto(self, *args):
-        self._trampoline(self.fd, write=True)
-        return self.fd.sendto(*args)
-
     def setblocking(self, flag):
         if flag:
             self.act_non_blocking = False
