Unpin member when read preference changes PYTHON-431

Author: A. Jesse Jiryu Davis

pymongo/mongo_replica_set_client.py

@@ -770,18 +770,24 @@ def __update_pools(self):
     def __schedule_refresh(self):
         self.__monitor.schedule_refresh()
 
-    def __pin_host(self, host):
+    def __pin_host(self, host, mode, tag_sets, latency):
         # After first successful read in a request, continue reading from same
         # member until read preferences change, host goes down, or
         # end_request(). This offers a small assurance that reads won't jump
         # around in time.
         self.__threadlocal.host = host
+        self.__threadlocal.read_preference = (mode, tag_sets, latency)
+
+    def __keep_pinned_host(self, mode, tag_sets, latency):
+        # If read preferences have changed, return False
+        return getattr(self.__threadlocal, 'read_preference', None) == (
+            mode, tag_sets, latency)
 
     def __pinned_host(self):
         return getattr(self.__threadlocal, 'host', None)
 
     def __unpin_host(self):
-        self.__threadlocal.host = None
+        self.__threadlocal.host = self.__threadlocal.read_preference = None
 
     def __reset_pinned_hosts(self):
         if self.__use_greenlets:
@@ -1158,7 +1164,7 @@ def _send_message_with_response(self, msg, _connection_to_use=None,
             mode = ReadPreference.PRIMARY
             tag_sets = [{}]
 
-        secondary_acceptable_latency_ms = kwargs.get(
+        latency = kwargs.get(
             'secondary_acceptable_latency_ms',
             self.secondary_acceptable_latency_ms)
 
@@ -1176,12 +1182,18 @@ def _send_message_with_response(self, msg, _connection_to_use=None,
                 self.disconnect()
             raise
 
+        # To provide some monotonic consistency, we use the same member as
+        # long as this thread is in a request and all reads use the same
+        # mode, tags, and latency. The member gets unpinned if pref changes,
+        # if member changes state, if we detect a failover and call
+        # __reset_pinned_hosts(), or if this thread calls end_request().
         errors = []
         pinned_member = self.__members.get(self.__pinned_host())
         if (pinned_member
             and pinned_member.matches_mode(mode)
             and pinned_member.matches_tag_sets(tag_sets)
             and pinned_member.up
+            and self.__keep_pinned_host(mode, tag_sets, latency)
         ):
             try:
                 return (
@@ -1204,7 +1216,7 @@ def _send_message_with_response(self, msg, _connection_to_use=None,
                 members=members,
                 mode=mode,
                 tag_sets=tag_sets,
-                latency=secondary_acceptable_latency_ms)
+                latency=latency)
 
             if not member:
                 # Ran out of members to try
@@ -1218,7 +1230,8 @@ def _send_message_with_response(self, msg, _connection_to_use=None,
                 # Success
                 if self.in_request():
                     # Keep reading from this member in this thread / greenlet
-                    self.__pin_host(member.host)
+                    # unless read preference changes
+                    self.__pin_host(member.host, mode, tag_sets, latency)
                 return member.host, response
             except AutoReconnect, why:
                 errors.append(str(why))

test/high_availability/test_ha.py

@@ -19,6 +19,7 @@
 # a replica set. Thus each method asserts everything we want to assert for a
 # given replica-set configuration.
 
+import itertools
 import time
 import unittest
 
@@ -28,10 +29,12 @@
 
 from pymongo import (ReplicaSetConnection,
                      ReadPreference)
-from pymongo.mongo_replica_set_client import Member, Monitor
+from pymongo.mongo_replica_set_client import (
+    Member, Monitor, MongoReplicaSetClient)
 from pymongo.mongo_client import _partition_node
 from pymongo.connection import Connection
 from pymongo.errors import AutoReconnect, OperationFailure, ConnectionFailure
+from pymongo.read_preferences import modes
 
 from test import utils
 from test.utils import one
@@ -726,6 +729,63 @@ def unpartition_node(node):
 
         self.clear_ping_times()
 
+    def test_pinning(self):
+        # To make the code terser, copy modes into local scope
+        PRIMARY = ReadPreference.PRIMARY
+        PRIMARY_PREFERRED = ReadPreference.PRIMARY_PREFERRED
+        SECONDARY = ReadPreference.SECONDARY
+        SECONDARY_PREFERRED = ReadPreference.SECONDARY_PREFERRED
+        NEAREST = ReadPreference.NEAREST
+
+        c = MongoReplicaSetClient(
+            self.seed, replicaSet=self.name, use_greenlets=use_greenlets,
+            auto_start_request=True)
+
+        # Verify that changing the mode unpins the member. We'll try it for
+        # every relevant change of mode.
+        for mode0, mode1 in itertools.permutations(
+            (PRIMARY, SECONDARY, SECONDARY_PREFERRED, NEAREST), 2
+        ):
+            # Try reading and then changing modes and reading again, see if we
+            # read from a different host
+            for _ in range(1000):
+                # pin to this host
+                host = utils.read_from_which_host(c, mode0)
+                # unpin?
+                new_host = utils.read_from_which_host(c, mode1)
+                if host != new_host:
+                    # Reading with a different mode unpinned, hooray!
+                    break
+            else:
+                self.fail(
+                    "Changing from mode %s to mode %s never unpinned" % (
+                        modes[mode0], modes[mode1]))
+
+        # Now verify changing the tag_sets unpins the member.
+        tags0 = [{'a': 'a'}, {}]
+        tags1 = [{'a': 'x'}, {}]
+        for _ in range(10000):
+            host = utils.read_from_which_host(c, NEAREST, tags0)
+            new_host = utils.read_from_which_host(c, NEAREST, tags1)
+            if host != new_host:
+                break
+        else:
+            self.fail(
+                "Changing from tags %s to tags %s never unpinned" % (
+                    tags0, tags1))
+
+        # Finally, verify changing the secondary_acceptable_latency_ms unpins
+        # the member.
+        for _ in range(10000):
+            host = utils.read_from_which_host(c, SECONDARY, None, 15)
+            new_host = utils.read_from_which_host(c, SECONDARY, None, 20)
+            if host != new_host:
+                break
+        else:
+            self.fail(
+                "Changing secondary_acceptable_latency_ms from 15 to 20"
+                " never unpinned")
+
     def tearDown(self):
         self.c.close()
         ha_tools.kill_all_members()

test/test_replica_set_connection.py

@@ -905,23 +905,19 @@ def test_pinned_member(self):
         self.assertTrue(host in conn.secondaries)
         assertReadFrom(self, conn, host, ReadPreference.SECONDARY)
 
-        # Oddly, we expect PRIMARY_PREFERRED to keep reading from secondary,
-        # since the secondary is pinned and "matches" the preference.
-        assertReadFrom(self, conn, host, ReadPreference.PRIMARY_PREFERRED)
-
-        # Repin
-        primary = read_from_which_host(conn, ReadPreference.PRIMARY)
-        self.assertEqual(conn.primary, primary)
-        assertReadFrom(self, conn, primary, ReadPreference.NEAREST)
+        # Changing any part of read preference (mode, tag_sets, latency)
+        # unpins the current host and pins to a new one
+        primary = conn.primary
         assertReadFrom(self, conn, primary, ReadPreference.PRIMARY_PREFERRED)
 
-        # Since the we're pinned to primary we still use it
-        assertReadFrom(self, conn, primary, ReadPreference.SECONDARY_PREFERRED)
+        host = read_from_which_host(conn, ReadPreference.NEAREST)
+        assertReadFrom(self, conn, host, ReadPreference.NEAREST)
 
-        # Repin again
-        host = read_from_which_host(conn, ReadPreference.SECONDARY)
+        assertReadFrom(self, conn, primary, ReadPreference.PRIMARY_PREFERRED)
+
+        host = read_from_which_host(conn, ReadPreference.SECONDARY_PREFERRED)
         self.assertTrue(host in conn.secondaries)
-        assertReadFrom(self, conn, host, ReadPreference.SECONDARY)
+        assertReadFrom(self, conn, host, ReadPreference.SECONDARY_PREFERRED)
 
         # Unpin
         conn.end_request()