PYTHON-2123 Streaming heartbeat protocol

MongoClient now requires 2 connections and 2 threads to each MongoDB 4.4+ server.
With one connection, the server streams (or pushes) updated heartbeat info.
With the other connection, the client periodically pings the server to
establish an accurate round-trip time (RTT). This change optimizes the
discovery of server state changes such as replica set elections.

Additional changes:
- Mark server Unknown before retrying isMaster check.
- Always reset the pool _after_ marking the server unknown.
- Configure fail point before creating the client in test SpecRunner.
- Unfreeze with replSetFreeze:0 to ensure a speedy elections in test suite.

Author: ShaneHarvey

pymongo/errors.py

@@ -282,3 +282,9 @@ def __init__(self, cause):
     def cause(self):
         """The exception that caused this encryption or decryption error."""
         return self.__cause
+
+
+class _OperationCancelled(AutoReconnect):
+    """Internal error raised when a socket operation is cancelled.
+    """
+    pass

pymongo/ismaster.py

@@ -46,9 +46,10 @@ def _get_server_type(doc):
 
 
 class IsMaster(object):
-    __slots__ = ('_doc', '_server_type', '_is_writable', '_is_readable')
+    __slots__ = ('_doc', '_server_type', '_is_writable', '_is_readable',
+                 '_awaitable')
 
-    def __init__(self, doc):
+    def __init__(self, doc, awaitable=False):
         """Parse an ismaster response from the server."""
         self._server_type = _get_server_type(doc)
         self._doc = doc
@@ -60,6 +61,7 @@ def __init__(self, doc):
         self._is_readable = (
             self.server_type == SERVER_TYPE.RSSecondary
             or self._is_writable)
+        self._awaitable = awaitable
 
     @property
     def document(self):
@@ -177,3 +179,7 @@ def speculative_authenticate(self):
     @property
     def topology_version(self):
         return self._doc.get('topologyVersion')
+
+    @property
+    def awaitable(self):
+        return self._awaitable

pymongo/message.py

@@ -1563,6 +1563,11 @@ def raw_command_response(self):
         # This should never be called on _OpReply.
         raise NotImplementedError
 
+    @property
+    def more_to_come(self):
+        """Is the moreToCome bit set on this response?"""
+        return False
+
     @classmethod
     def unpack(cls, msg):
         """Construct an _OpReply from raw bytes."""
@@ -1583,6 +1588,11 @@ class _OpMsg(object):
     UNPACK_FROM = struct.Struct("<IBi").unpack_from
     OP_CODE = 2013
 
+    # Flag bits.
+    CHECKSUM_PRESENT = 1
+    MORE_TO_COME = 1 << 1
+    EXHAUST_ALLOWED = 1 << 16  # Only present on requests.
+
     def __init__(self, flags, payload_document):
         self.flags = flags
         self.payload_document = payload_document
@@ -1613,15 +1623,28 @@ def raw_command_response(self):
         """Return the bytes of the command response."""
         return self.payload_document
 
+    @property
+    def more_to_come(self):
+        """Is the moreToCome bit set on this response?"""
+        return self.flags & self.MORE_TO_COME
+
     @classmethod
     def unpack(cls, msg):
         """Construct an _OpMsg from raw bytes."""
         flags, first_payload_type, first_payload_size = cls.UNPACK_FROM(msg)
         if flags != 0:
-            raise ProtocolError("Unsupported OP_MSG flags (%r)" % (flags,))
+            if flags & cls.CHECKSUM_PRESENT:
+                raise ProtocolError(
+                    "Unsupported OP_MSG flag checksumPresent: "
+                    "0x%x" % (flags,))
+
+            if flags ^ cls.MORE_TO_COME:
+                raise ProtocolError(
+                    "Unsupported OP_MSG flags: 0x%x" % (flags,))
         if first_payload_type != 0:
             raise ProtocolError(
-                "Unsupported OP_MSG payload type (%r)" % (first_payload_type,))
+                "Unsupported OP_MSG payload type: "
+                "0x%x" % (first_payload_type,))
 
         if len(msg) != first_payload_size + 5:
             raise ProtocolError("Unsupported OP_MSG reply: >1 section")