a1cf366a0b
xrange is not supported in python 3, so replace it with range since no large size lists involved. Closes-Bug: #1237717 Change-Id: I4e5e0782153d32c8beee03f7d871722ed4352859
266 lines
11 KiB
Python
266 lines
11 KiB
Python
# Copyright (c) 2010-2012 OpenStack, LLC.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
from itertools import chain
|
|
import sys
|
|
from time import sleep
|
|
from Queue import Queue
|
|
from threading import Thread
|
|
from traceback import format_exception
|
|
|
|
from swiftclient.exceptions import ClientException
|
|
|
|
|
|
class StopWorkerThreadSignal(object):
|
|
pass
|
|
|
|
|
|
class QueueFunctionThread(Thread):
|
|
"""
|
|
Calls `func`` for each item in ``queue``; ``func`` is called with a
|
|
de-queued item as the first arg followed by ``*args`` and ``**kwargs``.
|
|
|
|
Any exceptions raised by ``func`` are stored in :attr:`self.exc_infos`.
|
|
|
|
If the optional kwarg ``store_results`` is specified, it must be a list and
|
|
each result of invoking ``func`` will be appended to that list.
|
|
|
|
Putting a :class:`StopWorkerThreadSignal` instance into queue will cause
|
|
this thread to exit.
|
|
"""
|
|
|
|
def __init__(self, queue, func, *args, **kwargs):
|
|
"""
|
|
:param queue: A :class:`Queue` object from which work jobs will be
|
|
pulled.
|
|
:param func: A callable which will be invoked with a dequeued item
|
|
followed by ``*args`` and ``**kwargs``.
|
|
:param \*args: Optional positional arguments for ``func``.
|
|
:param \*\*kwargs: Optional kwargs for func. If the kwarg
|
|
``store_results`` is specified, its value must be a
|
|
list, and every result from invoking ``func`` will
|
|
be appended to the supplied list. The kwarg
|
|
``store_results`` will not be passed into ``func``.
|
|
"""
|
|
Thread.__init__(self)
|
|
self.queue = queue
|
|
self.func = func
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
self.exc_infos = []
|
|
self.store_results = kwargs.pop('store_results', None)
|
|
|
|
def run(self):
|
|
while True:
|
|
item = self.queue.get()
|
|
if isinstance(item, StopWorkerThreadSignal):
|
|
break
|
|
try:
|
|
result = self.func(item, *self.args, **self.kwargs)
|
|
if self.store_results is not None:
|
|
self.store_results.append(result)
|
|
except Exception:
|
|
self.exc_infos.append(sys.exc_info())
|
|
|
|
|
|
class QueueFunctionManager(object):
|
|
"""
|
|
A context manager to handle the life-cycle of a single :class:`Queue`
|
|
and a list of associated :class:`QueueFunctionThread` instances.
|
|
|
|
This class is not usually instantiated directly. Instead, call the
|
|
:meth:`MultiThreadingManager.queue_manager` object method,
|
|
which will return an instance of this class.
|
|
|
|
When entering the context, ``thread_count`` :class:`QueueFunctionThread`
|
|
instances are created and started. The input queue is returned. Inside
|
|
the context, any work item put into the queue will get worked on by one of
|
|
the :class:`QueueFunctionThread` instances.
|
|
|
|
When the context is exited, all threads are sent a
|
|
:class:`StopWorkerThreadSignal` instance and then all threads are waited
|
|
upon. Finally, any exceptions from any of the threads are reported on via
|
|
the supplied ``thread_manager``'s :meth:`error` method. If an
|
|
``error_counter`` list was supplied on instantiation, its first element is
|
|
incremented once for every exception which occurred.
|
|
"""
|
|
|
|
def __init__(self, func, thread_count, thread_manager, thread_args=None,
|
|
thread_kwargs=None, error_counter=None,
|
|
connection_maker=None):
|
|
"""
|
|
:param func: The worker function which will be passed into each
|
|
:class:`QueueFunctionThread`'s constructor.
|
|
:param thread_count: The number of worker threads to run.
|
|
:param thread_manager: An instance of :class:`MultiThreadingManager`.
|
|
:param thread_args: Optional positional arguments to be passed into
|
|
each invocation of ``func`` after the de-queued
|
|
work item.
|
|
:param thread_kwargs: Optional keyword arguments to be passed into each
|
|
invocation of ``func``. If a list is supplied as
|
|
the ``store_results`` keyword argument, it will
|
|
be filled with every result of invoking ``func``
|
|
in all threads.
|
|
:param error_counter: Optional list containing one integer. If
|
|
supplied, the list's first element will be
|
|
incremented once for each exception in any
|
|
thread. This happens only when exiting the
|
|
context.
|
|
:param connection_maker: Optional callable. If supplied, this callable
|
|
will be invoked once per created thread, and
|
|
the result will be passed into func after the
|
|
de-queued work item but before ``thread_args``
|
|
and ``thread_kwargs``. This is used to ensure
|
|
each thread has its own connection to Swift.
|
|
"""
|
|
self.func = func
|
|
self.thread_count = thread_count
|
|
self.thread_manager = thread_manager
|
|
self.error_counter = error_counter
|
|
self.connection_maker = connection_maker
|
|
self.queue = Queue(10000)
|
|
self.thread_list = []
|
|
self.thread_args = thread_args if thread_args else ()
|
|
self.thread_kwargs = thread_kwargs if thread_kwargs else {}
|
|
|
|
def __enter__(self):
|
|
for _junk in range(self.thread_count):
|
|
if self.connection_maker:
|
|
thread_args = (self.connection_maker(),) + self.thread_args
|
|
else:
|
|
thread_args = self.thread_args
|
|
qf_thread = QueueFunctionThread(self.queue, self.func,
|
|
*thread_args, **self.thread_kwargs)
|
|
qf_thread.start()
|
|
self.thread_list.append(qf_thread)
|
|
return self.queue
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
for thread in [t for t in self.thread_list if t.isAlive()]:
|
|
self.queue.put(StopWorkerThreadSignal())
|
|
|
|
while any(map(QueueFunctionThread.is_alive, self.thread_list)):
|
|
sleep(0.05)
|
|
|
|
for thread in self.thread_list:
|
|
for info in thread.exc_infos:
|
|
if self.error_counter:
|
|
self.error_counter[0] += 1
|
|
if isinstance(info[1], ClientException):
|
|
self.thread_manager.error(str(info[1]))
|
|
else:
|
|
self.thread_manager.error(''.join(format_exception(*info)))
|
|
|
|
|
|
class MultiThreadingManager(object):
|
|
"""
|
|
One object to manage context for multi-threading. This should make
|
|
bin/swift less error-prone and allow us to test this code.
|
|
|
|
This object is a context manager and returns itself into the context. When
|
|
entering the context, two printing threads are created (see below) and they
|
|
are waited on and cleaned up when exiting the context.
|
|
|
|
A convenience method, :meth:`queue_manager`, is provided to create a
|
|
:class:`QueueFunctionManager` context manager (a thread-pool with an
|
|
associated input queue for work items).
|
|
|
|
Also, thread-safe printing to two streams is provided. The
|
|
:meth:`print_msg` method will print to the supplied ``print_stream``
|
|
(defaults to ``sys.stdout``) and the :meth:`error` method will print to the
|
|
supplied ``error_stream`` (defaults to ``sys.stderr``). Both of these
|
|
printing methods will format the given string with any supplied ``*args``
|
|
(a la printf) and encode the result to utf8 if necessary.
|
|
|
|
The attribute :attr:`self.error_count` is incremented once per error
|
|
message printed, so an application can tell if any worker threads
|
|
encountered exceptions or otherwise called :meth:`error` on this instance.
|
|
The swift command-line tool uses this to exit non-zero if any error strings
|
|
were printed.
|
|
"""
|
|
|
|
def __init__(self, print_stream=sys.stdout, error_stream=sys.stderr):
|
|
"""
|
|
:param print_stream: The stream to which :meth:`print_msg` sends
|
|
formatted messages, encoded to utf8 if necessary.
|
|
:param error_stream: The stream to which :meth:`error` sends formatted
|
|
messages, encoded to utf8 if necessary.
|
|
"""
|
|
self.print_stream = print_stream
|
|
self.printer = QueueFunctionManager(self._print, 1, self)
|
|
self.error_stream = error_stream
|
|
self.error_printer = QueueFunctionManager(self._print_error, 1, self)
|
|
self.error_count = 0
|
|
|
|
def __enter__(self):
|
|
self.printer.__enter__()
|
|
self.error_printer.__enter__()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
self.error_printer.__exit__(exc_type, exc_value, traceback)
|
|
self.printer.__exit__(exc_type, exc_value, traceback)
|
|
|
|
def queue_manager(self, func, thread_count, *args, **kwargs):
|
|
connection_maker = kwargs.pop('connection_maker', None)
|
|
error_counter = kwargs.pop('error_counter', None)
|
|
return QueueFunctionManager(func, thread_count, self, thread_args=args,
|
|
thread_kwargs=kwargs,
|
|
connection_maker=connection_maker,
|
|
error_counter=error_counter)
|
|
|
|
def print_msg(self, msg, *fmt_args):
|
|
if fmt_args:
|
|
msg = msg % fmt_args
|
|
self.printer.queue.put(msg)
|
|
|
|
def print_items(self, items, offset=14, skip_missing=False):
|
|
lines = []
|
|
template = '%%%ds: %%s' % offset
|
|
for k, v in items:
|
|
if skip_missing and not v:
|
|
continue
|
|
lines.append((template % (k, v)).rstrip())
|
|
self.print_msg('\n'.join(lines))
|
|
|
|
def print_headers(self, headers, meta_prefix='', exclude_headers=None,
|
|
offset=14):
|
|
exclude_headers = exclude_headers or []
|
|
meta_headers = []
|
|
other_headers = []
|
|
template = '%%%ds: %%s' % offset
|
|
for key, value in headers.items():
|
|
if key.startswith(meta_prefix):
|
|
meta_key = 'Meta %s' % key[len(meta_prefix):].title()
|
|
meta_headers.append(template % (meta_key, value))
|
|
elif key not in exclude_headers:
|
|
other_headers.append(template % (key.title(), value))
|
|
self.print_msg('\n'.join(chain(meta_headers, other_headers)))
|
|
|
|
def error(self, msg, *fmt_args):
|
|
if fmt_args:
|
|
msg = msg % fmt_args
|
|
self.error_printer.queue.put(msg)
|
|
|
|
def _print(self, item, stream=None):
|
|
if stream is None:
|
|
stream = self.print_stream
|
|
if isinstance(item, unicode):
|
|
item = item.encode('utf8')
|
|
print >>stream, item
|
|
|
|
def _print_error(self, item):
|
|
self.error_count += 1
|
|
return self._print(item, stream=self.error_stream)
|