Merge pull request #416 from rollbar/added-thread-pool

Author: danielmorell

.github/workflows/ci.yml

@@ -205,6 +205,28 @@ jobs:
       - name: Install dependencies
         run: pip install setuptools==39.2.0 --force-reinstall
 
+      - name: Install Python 2 dependencies
+        if: ${{ contains(matrix.python-version, '2.7') }}
+        # certifi dropped support for Python 2 in 2020.4.5.2 but only started
+        # using Python 3 syntax in 2022.5.18. 2021.10.8 is the last release with
+        # Python 2 support.
+        run: pip install certifi==2021.10.8 requests==2.27.1 incremental==21.3.0
+
+      - name: Install Python 3.4 dependencies
+        if: ${{ contains(matrix.python-version, '3.4') }}
+        # certifi uses the 'typing' from Python 3.5 module starting in 2022.5.18
+        run: pip install certifi==2021.10.8 "typing-extensions<4"  incremental==21.3.0
+
+      - name: Install Python 3.5 dependencies
+        if: ${{ contains(matrix.python-version, '3.5') }}
+        # typing-extensions dropped support for Python 3.5 in version 4
+        run: pip install "typing-extensions<4"
+
+      - name: Install Python 3.6 dependencies
+        if: ${{ contains(matrix.python-version, '3.6') }}
+        # typing-extensions dropped support for Python 3.6 in version 4.2
+        run: pip install "typing-extensions<4.2"
+
       - name: Set the framework
         run: echo ${{ matrix.framework }} >> $GITHUB_ENV
 
@@ -232,27 +254,5 @@ jobs:
         if: ${{ contains(matrix.framework, 'FASTAPI_VERSION') }}
         run: pip install fastapi==$FASTAPI_VERSION
 
-      - name: Install Python 2 dependencies
-        if: ${{ contains(matrix.python-version, '2.7') }}
-        # certifi dropped support for Python 2 in 2020.4.5.2 but only started
-        # using Python 3 syntax in 2022.5.18. 2021.10.8 is the last release with
-        # Python 2 support.
-        run: pip install certifi==2021.10.8 requests==2.27.1
-
-      - name: Install Python 3.4 dependencies
-        if: ${{ contains(matrix.python-version, '3.4') }}
-        # certifi uses the 'typing' from Python 3.5 module starting in 2022.5.18
-        run: pip install certifi==2021.10.8 "typing-extensions<4"
-
-      - name: Install Python 3.5 dependencies
-        if: ${{ contains(matrix.python-version, '3.5') }}
-        # typing-extensions dropped support for Python 3.5 in version 4
-        run: pip install "typing-extensions<4"
-
-      - name: Install Python 3.6 dependencies
-        if: ${{ contains(matrix.python-version, '3.6') }}
-        # typing-extensions dropped support for Python 3.6 in version 4.2
-        run: pip install "typing-extensions<4.2"
-
       - name: Run tests
         run: python setup.py test

rollbar/__init__.py

@@ -23,7 +23,7 @@
 from rollbar.lib import events, filters, dict_merge, parse_qs, text, transport, urljoin, iteritems, defaultJSONEncode
 
 
-__version__ = '0.16.3'
+__version__ = '0.16.4beta'
 __log_name__ = 'rollbar'
 log = logging.getLogger(__log_name__)
 
@@ -124,7 +124,7 @@ def wrap(*args, **kwargs):
     from twisted.internet.ssl import CertificateOptions
     from twisted.internet import task, defer, ssl, reactor
     from zope.interface import implementer
-    
+
     @implementer(IPolicyForHTTPS)
     class VerifyHTTPS(object):
         def __init__(self):
@@ -275,7 +275,12 @@ def _get_fastapi_request():
     'root': None,  # root path to your code
     'branch': None,  # git branch name
     'code_version': None,
-    'handler': 'default',  # 'blocking', 'thread' (default), 'async', 'agent', 'tornado', 'gae', 'twisted' or 'httpx'
+    # 'blocking', 'thread' (default), 'async', 'agent', 'tornado', 'gae', 'twisted', 'httpx' or 'thread_pool'
+    # 'async' requires Python 3.4 or higher.
+    # 'httpx' requires Python 3.7 or higher.
+    # 'thread_pool' requires Python 3.2 or higher.
+    'handler': 'default',
+    'thread_pool_workers': None,
     'endpoint': DEFAULT_ENDPOINT,
     'timeout': DEFAULT_TIMEOUT,
     'agent.log_file': 'log.rollbar',
@@ -383,6 +388,9 @@ def init(access_token, environment='production', scrub_fields=None, url_fields=N
 
     if SETTINGS.get('handler') == 'agent':
         agent_log = _create_agent_log()
+    elif SETTINGS.get('handler') == 'thread_pool':
+        from rollbar.lib.thread_pool import init_pool
+        init_pool(SETTINGS.get('thread_pool_workers', None))
 
     if not SETTINGS['locals']['safelisted_types'] and SETTINGS['locals']['whitelisted_types']:
         warnings.warn('whitelisted_types deprecated use safelisted_types instead', DeprecationWarning)
@@ -523,6 +531,7 @@ def send_payload(payload, access_token):
     - 'gae': calls _send_payload_appengine() (which makes a blocking call to Google App Engine)
     - 'twisted': calls _send_payload_twisted() (which makes an async HTTP request using Twisted and Treq)
     - 'httpx': calls _send_payload_httpx() (which makes an async HTTP request using HTTPX)
+    - 'thread_pool': uses a pool of worker threads to make HTTP requests off the main thread. Returns immediately.
     """
     payload = events.on_payload(payload)
     if payload is False:
@@ -569,6 +578,8 @@ def send_payload(payload, access_token):
         _send_payload_async(payload_str, access_token)
     elif handler == 'thread':
         _send_payload_thread(payload_str, access_token)
+    elif handler == 'thread_pool':
+        _send_payload_thread_pool(payload_str, access_token)
     else:
         # default to 'thread'
         _send_payload_thread(payload_str, access_token)
@@ -1510,6 +1521,18 @@ def _send_payload_thread(payload_str, access_token):
     thread.start()
 
 
+def _send_payload_pool(payload_str, access_token):
+    try:
+        _post_api('item/', payload_str, access_token=access_token)
+    except Exception as e:
+        log.exception('Exception while posting item %r', e)
+
+
+def _send_payload_thread_pool(payload_str, access_token):
+    from rollbar.lib.thread_pool import submit
+    submit(_send_payload_pool, payload_str, access_token)
+
+
 def _send_payload_appengine(payload_str, access_token):
     try:
         _post_api_appengine('item/', payload_str, access_token=access_token)

rollbar/lib/thread_pool.py

@@ -0,0 +1,38 @@
+import logging
+import os
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+_pool = None  # type: ThreadPoolExecutor|None
+
+log = logging.getLogger(__name__)
+
+
+def init_pool(max_workers):
+    """
+    Creates the thread pool with the max workers.
+
+    :type max_workers: int|None
+    :param max_workers: If max_workers is None it will use the logic from the standard library to calculate the number
+                        of threads. However, we ported the logic from Python 3.5 to earlier versions.
+    """
+    if max_workers is None and sys.version_info < (3, 5):
+        max_workers = (os.cpu_count() or 1) * 5
+
+    global _pool
+    _pool = ThreadPoolExecutor(max_workers)
+
+
+def submit(worker, payload_str, access_token):
+    """
+    Submit a new task to the thread pool.
+
+    :type worker: function
+    :type payload_str: str
+    :type access_token: str
+    """
+    global _pool
+    if _pool is None:
+        log.warning('pyrollbar: Thead pool not initialized. Please ensure init_pool() is called prior to submit().')
+        return
+    _pool.submit(worker, payload_str, access_token)

rollbar/test/test_rollbar.py

@@ -1035,6 +1035,32 @@ def _raise():
 
         send_payload_httpx.assert_called_once()
 
+    @unittest.skipUnless(sys.version_info >= (3, 6), 'assert_called_once support requires Python3.6+')
+    @mock.patch('rollbar._send_payload_thread_pool')
+    def test_thread_pool_handler(self, send_payload_thread_pool):
+        def _raise():
+            try:
+                raise Exception('foo')
+            except:
+                rollbar.report_exc_info()
+        rollbar.SETTINGS['handler'] = 'thread_pool'
+        _raise()
+
+        send_payload_thread_pool.assert_called_once()
+
+    @unittest.skipUnless(sys.version_info >= (3, 2), 'concurrent.futures support requires Python3.2+')
+    def test_thread_pool_submit(self):
+        from rollbar.lib.thread_pool import init_pool, submit
+        init_pool(1)
+        ran = {'nope': True}  # dict used so it is not shadowed in run
+
+        def run(payload_str, access_token):
+            ran['nope'] = False
+
+        submit(run, 'foo', 'bar')
+        self.assertFalse(ran['nope'])
+
+
     @mock.patch('rollbar.send_payload')
     def test_args_constructor(self, send_payload):