PYTHON-1660 Clear MongoClient session pool after a fork

Note that a MongoClient instance is still not fork-safe. This change
avoids "Cannot start transaction X on session <SID> because a newer
transaction Y has already started" errors and other incorrect command
results caused by duplicate sessions in the child process.

Author: ShaneHarvey

pymongo/client_session.py

@@ -95,6 +95,7 @@
 """
 
 import collections
+import os
 import sys
 import uuid
 
@@ -834,12 +835,13 @@ def _start_retryable_write(self):
 
 
 class _ServerSession(object):
-    def __init__(self):
+    def __init__(self, pool_id):
         # Ensure id is type 4, regardless of CodecOptions.uuid_representation.
         self.session_id = {'id': Binary(uuid.uuid4().bytes, 4)}
         self.last_use = monotonic.time()
         self._transaction_id = 0
         self.dirty = False
+        self.pool_id = pool_id
 
     def mark_dirty(self):
         """Mark this session as dirty.
@@ -869,6 +871,14 @@ class _ServerSessionPool(collections.deque):
 
     This class is not thread-safe, access it while holding the Topology lock.
     """
+    def __init__(self, *args, **kwargs):
+        super(_ServerSessionPool, self).__init__(*args, **kwargs)
+        self.pool_id = 0
+
+    def reset(self):
+        self.pool_id += 1
+        self.clear()
+
     def pop_all(self):
         ids = []
         while self:
@@ -889,15 +899,17 @@ def get_server_session(self, session_timeout_minutes):
             if not s.timed_out(session_timeout_minutes):
                 return s
 
-        return _ServerSession()
+        return _ServerSession(self.pool_id)
 
     def return_server_session(self, server_session, session_timeout_minutes):
         self._clear_stale(session_timeout_minutes)
         if not server_session.timed_out(session_timeout_minutes):
             self.return_server_session_no_lock(server_session)
 
     def return_server_session_no_lock(self, server_session):
-        if not server_session.dirty:
+        # Discard sessions from an old pool to avoid duplicate sessions in the
+        # child process after a fork.
+        if server_session.pool_id == self.pool_id and not server_session.dirty:
             self.appendleft(server_session)
 
     def _clear_stale(self, session_timeout_minutes):

pymongo/topology.py

@@ -156,6 +156,10 @@ def open(self):
                     "after forking. See PyMongo's documentation for details: "
                     "http://api.mongodb.org/python/current/faq.html#"
                     "is-pymongo-fork-safe")
+                with self._lock:
+                    # Reset the session pool to avoid duplicate sessions in
+                    # the child process.
+                    self._session_pool.reset()
 
         with self._lock:
             self._ensure_opened()