| 72 | |
| 73 | |
| 74 | class UnsafeThreadPoolExecutor( _base.Executor ): |
| 75 | def __init__( self, max_workers ): |
| 76 | """Initializes a new ThreadPoolExecutor instance. |
| 77 | |
| 78 | Args: |
| 79 | max_workers: The maximum number of threads that can be used to |
| 80 | execute the given calls. |
| 81 | """ |
| 82 | self._max_workers = max_workers |
| 83 | self._work_queue = queue.Queue() |
| 84 | self._threads = set() |
| 85 | self._shutdown = False |
| 86 | self._shutdown_lock = threading.Lock() |
| 87 | |
| 88 | def submit( self, fn, *args, **kwargs ): |
| 89 | with self._shutdown_lock: |
| 90 | if self._shutdown: |
| 91 | raise RuntimeError( 'cannot schedule new futures after shutdown' ) |
| 92 | |
| 93 | f = _base.Future() |
| 94 | w = _WorkItem( f, fn, args, kwargs ) |
| 95 | |
| 96 | self._work_queue.put( w ) |
| 97 | self._adjust_thread_count() |
| 98 | return f |
| 99 | submit.__doc__ = _base.Executor.submit.__doc__ |
| 100 | |
| 101 | def _adjust_thread_count( self ): |
| 102 | # When the executor gets lost, the weakref callback will wake up |
| 103 | # the worker threads. |
| 104 | def weakref_cb( _, q=self._work_queue ): |
| 105 | q.put( None ) |
| 106 | # TODO(bquinlan): Should avoid creating new threads if there are more |
| 107 | # idle threads than items in the work queue. |
| 108 | if len( self._threads ) < self._max_workers: |
| 109 | t = threading.Thread( target=_worker, |
| 110 | args=( weakref.ref( self, weakref_cb ), |
| 111 | self._work_queue ) ) |
| 112 | t.daemon = True |
| 113 | t.start() |
| 114 | self._threads.add( t ) |
| 115 | |
| 116 | def shutdown( self, wait=True ): |
| 117 | with self._shutdown_lock: |
| 118 | self._shutdown = True |
| 119 | self._work_queue.put( None ) |
| 120 | if wait: |
| 121 | for t in self._threads: |
| 122 | t.join() |
| 123 | shutdown.__doc__ = _base.Executor.shutdown.__doc__ |