PYTHON-1799 Don't iterate _ENCODERS dict when encoding bson

(cherry picked from commit eb4a047278)

bson/__init__.py

@@ -752,6 +752,9 @@ if not PY3:
     _ENCODERS[long] = _encode_long
 
 
+_BUILT_IN_TYPES = tuple(t for t in _ENCODERS)
+
+
 def _name_value_to_bson(name, value, check_keys, opts,
                         in_fallback_call=False):
     """Encode a single name, value pair."""
@@ -783,7 +786,7 @@ def _name_value_to_bson(name, value, check_keys, opts,
 
     # If all else fails test each base type. This will only happen once for
     # a subtype of a supported base type.
-    for base in _ENCODERS:
+    for base in _BUILT_IN_TYPES:
         if isinstance(value, base):
             func = _ENCODERS[base]
             # Cache this type for faster subsequent lookup.

test/test_bson.py

@@ -50,6 +50,7 @@ from bson.tz_util import (FixedOffset,
                           utc)
 
 from test import qcheck, SkipTest, unittest
+from test.utils import ExceptionCatchingThread
 
 if PY3:
     long = int
@@ -904,6 +905,24 @@ class TestBSON(unittest.TestCase):
                           {"_id": {'$oid': "52d0b971b3ba219fdeb4170e"}}, True)
         BSON.encode({"_id": {'$oid': "52d0b971b3ba219fdeb4170e"}})
 
+    def test_bson_encode_thread_safe(self):
+
+        def target(i):
+            for j in range(1000):
+                my_int = type('MyInt_%s_%s' % (i, j), (int,), {})
+                bson.BSON.encode({'my_int': my_int()})
+
+        threads = [ExceptionCatchingThread(target=target, args=(i,))
+                   for i in range(3)]
+        for t in threads:
+            t.start()
+
+        for t in threads:
+            t.join()
+
+        for t in threads:
+            self.assertIsNone(t.exc)
+
 
 class TestCodecOptions(unittest.TestCase):
     def test_document_class(self):

test/utils.py

@@ -503,3 +503,17 @@ def enable_replication(client):
         secondary = single_client(host, port)
         secondary.admin.command('configureFailPoint', 'stopReplProducer',
                                 mode='off')
+
+
+class ExceptionCatchingThread(threading.Thread):
+    """A thread that stores any exception encountered from run()."""
+    def __init__(self, *args, **kwargs):
+        self.exc = None
+        super(ExceptionCatchingThread, self).__init__(*args, **kwargs)
+
+    def run(self):
+        try:
+            super(ExceptionCatchingThread, self).run()
+        except BaseException as exc:
+            self.exc = exc
+            raise