pool: call cluster.on_down() after pool shutdown()

In this commit, a logic in return_connection() is modified. This
particular piece of code is executed when a connection is being closed 
or it is defunct. 

The on_down() call is a call to method decorated with @run_in_executor,
meaning it can be executed in a separate thread. The shutdown() call
is a synchronous method.

This can cause a race:

- If shutdown() is executed before on_down(), then on_down() will see
there are no valid connections and will reconnect. This is good.
- But if on_down() is faster and is executed before shutdown(), the
on_down() method will see that there is still a valid connection in the
pool (because the pool wasn't shut down) and will NOT reconnect. 
Afterwards, the shutdown() completes and there are no valid connections
in the pool. After that, you will not be able to send queries to that
host, as it will result in "ConnectionException('Pool is shutdown')" 
exception.

The problem is fixed in this commit by moving on_down() calls after
a call to shutdown(). Because shutdown() is a synchronous method, this
means that first shutdown() is executed and then on_down() is executed.
This is the first case described in the previous paragraph and it is
a correct order.

You can also think about this change as a reduction from two possible
orderings (shutdown(), on_down() or on_down(), shutdown()) to a single
possible ordering (shutdown(), on_down()).

Fixes #170

Author: avelanarius

cassandra/pool.py

@@ -543,16 +543,14 @@ def return_connection(self, connection, stream_was_orphaned=False):
                 log.debug("Defunct or closed connection (%s) returned to pool, potentially "
                           "marking host %s as down", id(connection), self.host)
                 is_down = self.host.signal_connection_failure(connection.last_error)
-                if is_down:
-                     self._session.cluster.on_down(self.host, False, False)
                 connection.signaled_error = True
 
             if self.shutdown_on_error and not is_down:
                 is_down = True
-                self._session.cluster.on_down(self.host, is_host_addition=False)
 
             if is_down:
                 self.shutdown()
+                self._session.cluster.on_down(self.host, is_host_addition=False)
             else:
                 connection.close()
                 with self._lock: