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

GeventConnection.close() uses kill(block=False) which schedules
GreenletExit for the next yield point, then closes the socket
immediately. If close() is called from a non-gevent thread, the
greenlet may still be mid-I/O when the socket is closed, causing
EBADF.

- Add is_closed/is_defunct guards in handle_read() and handle_write()
  error paths to silently exit during shutdown
- Add GreenletExit exception handling in both I/O handlers for
  graceful greenlet termination (matching eventletreactor pattern)
- 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/geventreactor.py

@@ -16,6 +16,7 @@
 from gevent.queue import Queue
 from gevent import socket
 import gevent.ssl
+from greenlet import GreenletExit
 
 import logging
 import time
@@ -98,7 +99,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()
 
@@ -115,6 +119,8 @@ def handle_write(self):
                 log.debug("Exception in send for %s: %s", self, err)
                 self.defunct(err)
                 return
+            except GreenletExit:
+                return
 
     def handle_read(self):
         while True:
@@ -125,11 +131,15 @@ def handle_read(self):
                 log.debug("Exception in read for %s: %s", self, err)
                 self.defunct(err)
                 return  # leave the read loop
+            except GreenletExit:
+                return
 
             if buf and self._iobuf.tell():
                 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