Fix EventletConnection close() race causing EBADF errors (#614)

EventletConnection.close() uses blocking kill() which mostly prevents
races, but if close() is called from a non-eventlet thread the
greenthread may not be interrupted as expected.

- Add is_closed/is_defunct guards in handle_read() and handle_write()
  error paths to silently exit during shutdown
- Set last_error in close() when connected_event is not yet set to
  prevent factory() from returning a dead connection
- Set last_error on server-initiated close (EOF) in handle_read()

Author: dkropachev

cassandra/io/eventletreactor.py

@@ -148,7 +148,10 @@ def close(self):
             msg = "Connection to %s was closed" % self.endpoint
             if self.last_error:
                 msg += ": %s" % (self.last_error,)
-            self.error_all_requests(ConnectionShutdown(msg))
+            shutdown_exc = ConnectionShutdown(msg)
+            self.error_all_requests(shutdown_exc)
+            if not self.connected_event.is_set():
+                self.last_error = shutdown_exc
             # don't leave in-progress operations hanging
             self.connected_event.set()
 
@@ -162,6 +165,8 @@ def handle_write(self):
                 next_msg = self._write_queue.get()
                 self._socket.sendall(next_msg)
             except socket.error as err:
+                if self.is_closed or self.is_defunct:
+                    return
                 log.debug("Exception during socket send for %s: %s", self, err)
                 self.defunct(err)
                 return  # Leave the write loop
@@ -174,6 +179,8 @@ def handle_read(self):
                 buf = self._socket.recv(self.in_buffer_size)
                 self._iobuf.write(buf)
             except socket.error as err:
+                if self.is_closed or self.is_defunct:
+                    return
                 log.debug("Exception during socket recv for %s: %s",
                           self, err)
                 self.defunct(err)
@@ -185,6 +192,8 @@ def handle_read(self):
                 self.process_io_buffer()
             else:
                 log.debug("Connection %s closed by server", self)
+                self.last_error = ConnectionShutdown(
+                    "Connection to %s was closed by server" % self.endpoint)
                 self.close()
                 return