1# Copyright 2015 gRPC authors. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""A thread pool that logs exceptions raised by tasks executed within it.""" 15 16from concurrent import futures 17import logging 18 19_LOGGER = logging.getLogger(__name__) 20 21 22def _wrap(behavior): 23 """Wraps an arbitrary callable behavior in exception-logging.""" 24 25 def _wrapping(*args, **kwargs): 26 try: 27 return behavior(*args, **kwargs) 28 except Exception: 29 _LOGGER.exception( 30 'Unexpected exception from %s executed in logging pool!', 31 behavior) 32 raise 33 34 return _wrapping 35 36 37class _LoggingPool(object): 38 """An exception-logging futures.ThreadPoolExecutor-compatible thread pool.""" 39 40 def __init__(self, backing_pool): 41 self._backing_pool = backing_pool 42 43 def __enter__(self): 44 return self 45 46 def __exit__(self, exc_type, exc_val, exc_tb): 47 self._backing_pool.shutdown(wait=True) 48 49 def submit(self, fn, *args, **kwargs): 50 return self._backing_pool.submit(_wrap(fn), *args, **kwargs) 51 52 def map(self, func, *iterables, **kwargs): 53 return self._backing_pool.map(_wrap(func), 54 *iterables, 55 timeout=kwargs.get('timeout', None)) 56 57 def shutdown(self, wait=True): 58 self._backing_pool.shutdown(wait=wait) 59 60 61def pool(max_workers): 62 """Creates a thread pool that logs exceptions raised by the tasks within it. 63 64 Args: 65 max_workers: The maximum number of worker threads to allow the pool. 66 67 Returns: 68 A futures.ThreadPoolExecutor-compatible thread pool that logs exceptions 69 raised by the tasks executed within it. 70 """ 71 return _LoggingPool(futures.ThreadPoolExecutor(max_workers)) 72