| # -*- coding: utf-8 -*- |
| # Copyright 2013 Google Inc. All Rights Reserved. |
| # |
| # Permission is hereby granted, free of charge, to any person obtaining a |
| # copy of this software and associated documentation files (the |
| # "Software"), to deal in the Software without restriction, including |
| # without limitation the rights to use, copy, modify, merge, publish, dis- |
| # tribute, sublicense, and/or sell copies of the Software, and to permit |
| # persons to whom the Software is furnished to do so, subject to the fol- |
| # lowing conditions: |
| # |
| # The above copyright notice and this permission notice shall be included |
| # in all copies or substantial portions of the Software. |
| # |
| # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
| # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
| # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
| # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
| # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
| # IN THE SOFTWARE. |
| """Unit tests for gsutil parallelism framework.""" |
| |
| from __future__ import absolute_import |
| |
| import functools |
| import os |
| import signal |
| import threading |
| import time |
| |
| from boto.storage_uri import BucketStorageUri |
| from gslib import cs_api_map |
| from gslib.command import Command |
| from gslib.command import CreateGsutilLogger |
| from gslib.command import DummyArgChecker |
| import gslib.tests.testcase as testcase |
| from gslib.tests.testcase.base import RequiresIsolation |
| from gslib.tests.util import unittest |
| from gslib.util import CheckMultiprocessingAvailableAndInit |
| from gslib.util import IS_WINDOWS |
| |
| |
| # Amount of time for an individual test to run before timing out. We need a |
| # reasonably high value since if many tests are running in parallel, an |
| # individual test may take a while to complete. |
| _TEST_TIMEOUT_SECONDS = 120 |
| |
| |
| def Timeout(func): |
| """Decorator used to provide a timeout for functions.""" |
| @functools.wraps(func) |
| def Wrapper(*args, **kwargs): |
| if not IS_WINDOWS: |
| signal.signal(signal.SIGALRM, _HandleAlarm) |
| signal.alarm(_TEST_TIMEOUT_SECONDS) |
| try: |
| func(*args, **kwargs) |
| finally: |
| if not IS_WINDOWS: |
| signal.alarm(0) # Cancel the alarm. |
| return Wrapper |
| |
| |
| # pylint: disable=unused-argument |
| def _HandleAlarm(signal_num, cur_stack_frame): |
| raise Exception('Test timed out.') |
| |
| |
| class CustomException(Exception): |
| |
| def __init__(self, exception_str): |
| super(CustomException, self).__init__(exception_str) |
| |
| |
| def _ReturnOneValue(cls, args, thread_state=None): |
| return 1 |
| |
| |
| def _ReturnProcAndThreadId(cls, args, thread_state=None): |
| return (os.getpid(), threading.currentThread().ident) |
| |
| |
| def _SleepThenReturnProcAndThreadId(cls, args, thread_state=None): |
| # This can fail if the total time to spawn new processes and threads takes |
| # longer than 5 seconds, but if that occurs, then we have a performance |
| # problem that needs to be addressed. |
| time.sleep(5) |
| return _ReturnProcAndThreadId(cls, args, thread_state=thread_state) |
| |
| |
| def _FailureFunc(cls, args, thread_state=None): |
| raise CustomException('Failing on purpose.') |
| |
| |
| def _FailingExceptionHandler(cls, e): |
| cls.failure_count += 1 |
| raise CustomException('Exception handler failing on purpose.') |
| |
| |
| def _ExceptionHandler(cls, e): |
| cls.logger.exception(e) |
| cls.failure_count += 1 |
| |
| |
| def _IncrementByLength(cls, args, thread_state=None): |
| cls.arg_length_sum += len(args) |
| |
| |
| def _AdjustProcessCountIfWindows(process_count): |
| if IS_WINDOWS: |
| return 1 |
| else: |
| return process_count |
| |
| |
| def _ReApplyWithReplicatedArguments(cls, args, thread_state=None): |
| """Calls Apply with arguments repeated seven times. |
| |
| The first two elements of args should be the process and thread counts, |
| respectively, to be used for the recursive calls. |
| |
| Args: |
| cls: The Command class to call Apply on. |
| args: Arguments to pass to Apply. |
| thread_state: Unused, required by function signature. |
| |
| Returns: |
| Number of values returned by the two calls to Apply. |
| """ |
| new_args = [args] * 7 |
| process_count = _AdjustProcessCountIfWindows(args[0]) |
| thread_count = args[1] |
| return_values = cls.Apply(_PerformNRecursiveCalls, new_args, |
| _ExceptionHandler, arg_checker=DummyArgChecker, |
| process_count=process_count, |
| thread_count=thread_count, |
| should_return_results=True) |
| ret = sum(return_values) |
| |
| return_values = cls.Apply(_ReturnOneValue, new_args, |
| _ExceptionHandler, arg_checker=DummyArgChecker, |
| process_count=process_count, |
| thread_count=thread_count, |
| should_return_results=True) |
| |
| return len(return_values) + ret |
| |
| |
| def _PerformNRecursiveCalls(cls, args, thread_state=None): |
| """Calls Apply to perform N recursive calls. |
| |
| The first two elements of args should be the process and thread counts, |
| respectively, to be used for the recursive calls, while N is the third element |
| (the number of recursive calls to make). |
| |
| Args: |
| cls: The Command class to call Apply on. |
| args: Arguments to pass to Apply. |
| thread_state: Unused, required by function signature. |
| |
| Returns: |
| Number of values returned by the call to Apply. |
| """ |
| process_count = _AdjustProcessCountIfWindows(args[0]) |
| thread_count = args[1] |
| return_values = cls.Apply(_ReturnOneValue, [()] * args[2], _ExceptionHandler, |
| arg_checker=DummyArgChecker, |
| process_count=process_count, |
| thread_count=thread_count, |
| should_return_results=True) |
| return len(return_values) |
| |
| |
| def _SkipEvenNumbersArgChecker(cls, arg): |
| return arg % 2 != 0 |
| |
| |
| class FailingIterator(object): |
| |
| def __init__(self, size, failure_indices): |
| self.size = size |
| self.failure_indices = failure_indices |
| self.current_index = 0 |
| |
| def __iter__(self): |
| return self |
| |
| def next(self): |
| if self.current_index == self.size: |
| raise StopIteration('') |
| elif self.current_index in self.failure_indices: |
| self.current_index += 1 |
| raise CustomException( |
| 'Iterator failing on purpose at index %d.' % self.current_index) |
| else: |
| self.current_index += 1 |
| return self.current_index - 1 |
| |
| |
| class FakeCommand(Command): |
| """Fake command class for overriding command instance state.""" |
| command_spec = Command.CreateCommandSpec( |
| 'fake', |
| command_name_aliases=[], |
| ) |
| # Help specification. See help_provider.py for documentation. |
| help_spec = Command.HelpSpec( |
| help_name='fake', |
| help_name_aliases=[], |
| help_type='command_help', |
| help_one_line_summary='Something to take up space.', |
| help_text='Something else to take up space.', |
| subcommand_help_text={}, |
| ) |
| |
| def __init__(self, do_parallel): |
| self.bucket_storage_uri_class = BucketStorageUri |
| support_map = { |
| 'gs': ['JSON'], |
| 's3': ['XML'] |
| } |
| default_map = { |
| 'gs': 'JSON', |
| 's3': 'XML' |
| } |
| self.gsutil_api_map = cs_api_map.GsutilApiMapFactory.GetApiMap( |
| cs_api_map.GsutilApiClassMapFactory, support_map, default_map) |
| self.logger = CreateGsutilLogger('FakeCommand') |
| self.parallel_operations = do_parallel |
| self.failure_count = 0 |
| self.multiprocessing_is_available = ( |
| CheckMultiprocessingAvailableAndInit().is_available) |
| self.debug = 0 |
| |
| |
| class FakeCommandWithoutMultiprocessingModule(FakeCommand): |
| |
| def __init__(self, do_parallel): |
| super(FakeCommandWithoutMultiprocessingModule, self).__init__(do_parallel) |
| self.multiprocessing_is_available = False |
| |
| |
| # TODO: Figure out a good way to test that ctrl+C really stops execution, |
| # and also that ctrl+C works when there are still tasks enqueued. |
| class TestParallelismFramework(testcase.GsUtilUnitTestCase): |
| """gsutil parallelism framework test suite.""" |
| |
| command_class = FakeCommand |
| |
| def _RunApply(self, func, args_iterator, process_count, thread_count, |
| command_inst=None, shared_attrs=None, fail_on_error=False, |
| thr_exc_handler=None, arg_checker=DummyArgChecker): |
| command_inst = command_inst or self.command_class(True) |
| exception_handler = thr_exc_handler or _ExceptionHandler |
| |
| return command_inst.Apply(func, args_iterator, exception_handler, |
| thread_count=thread_count, |
| process_count=process_count, |
| arg_checker=arg_checker, |
| should_return_results=True, |
| shared_attrs=shared_attrs, |
| fail_on_error=fail_on_error) |
| |
| @RequiresIsolation |
| def testBasicApplySingleProcessSingleThread(self): |
| self._TestBasicApply(1, 1) |
| |
| @RequiresIsolation |
| def testBasicApplySingleProcessMultiThread(self): |
| self._TestBasicApply(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testBasicApplyMultiProcessSingleThread(self): |
| self._TestBasicApply(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testBasicApplyMultiProcessMultiThread(self): |
| self._TestBasicApply(3, 3) |
| |
| @Timeout |
| def _TestBasicApply(self, process_count, thread_count): |
| args = [()] * (17 * process_count * thread_count + 1) |
| |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(len(args), len(results)) |
| |
| @RequiresIsolation |
| def testNoTasksSingleProcessSingleThread(self): |
| self._TestApplyWithNoTasks(1, 1) |
| |
| @RequiresIsolation |
| def testNoTasksSingleProcessMultiThread(self): |
| self._TestApplyWithNoTasks(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testNoTasksMultiProcessSingleThread(self): |
| self._TestApplyWithNoTasks(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testNoTasksMultiProcessMultiThread(self): |
| self._TestApplyWithNoTasks(3, 3) |
| |
| @Timeout |
| def _TestApplyWithNoTasks(self, process_count, thread_count): |
| """Tests that calling Apply with no tasks releases locks/semaphores.""" |
| empty_args = [()] |
| |
| for _ in range(process_count * thread_count + 1): |
| self._RunApply(_ReturnOneValue, empty_args, process_count, thread_count) |
| |
| # Ensure that work can still be performed. |
| self._TestBasicApply(process_count, thread_count) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testApplySaturatesMultiProcessSingleThread(self): |
| self._TestApplySaturatesAvailableProcessesAndThreads(3, 1) |
| |
| @RequiresIsolation |
| def testApplySaturatesSingleProcessMultiThread(self): |
| self._TestApplySaturatesAvailableProcessesAndThreads(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testApplySaturatesMultiProcessMultiThread(self): |
| self._TestApplySaturatesAvailableProcessesAndThreads(3, 3) |
| |
| @RequiresIsolation |
| def _TestApplySaturatesAvailableProcessesAndThreads(self, process_count, |
| thread_count): |
| """Tests that created processes and threads evenly share tasks.""" |
| calls_per_thread = 2 |
| args = [()] * (process_count * thread_count * calls_per_thread) |
| expected_calls_per_thread = calls_per_thread |
| |
| if not self.command_class(True).multiprocessing_is_available: |
| # When multiprocessing is unavailable, only a single process is used. |
| # Calls should be evenly distributed across threads. |
| expected_calls_per_thread = calls_per_thread * process_count |
| |
| results = self._RunApply(_SleepThenReturnProcAndThreadId, args, |
| process_count, thread_count) |
| usage_dict = {} # (process_id, thread_id): number of tasks performed |
| for (process_id, thread_id) in results: |
| usage_dict[(process_id, thread_id)] = ( |
| usage_dict.get((process_id, thread_id), 0) + 1) |
| |
| for (id_tuple, num_tasks_completed) in usage_dict.iteritems(): |
| self.assertEqual(num_tasks_completed, expected_calls_per_thread, |
| 'Process %s thread %s completed %s tasks. Expected: %s' % |
| (id_tuple[0], id_tuple[1], num_tasks_completed, |
| expected_calls_per_thread)) |
| |
| @RequiresIsolation |
| def testIteratorFailureSingleProcessSingleThread(self): |
| self._TestIteratorFailure(1, 1) |
| |
| @RequiresIsolation |
| def testIteratorFailureSingleProcessMultiThread(self): |
| self._TestIteratorFailure(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testIteratorFailureMultiProcessSingleThread(self): |
| self._TestIteratorFailure(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testIteratorFailureMultiProcessMultiThread(self): |
| self._TestIteratorFailure(3, 3) |
| |
| @Timeout |
| def _TestIteratorFailure(self, process_count, thread_count): |
| """Tests apply with a failing iterator.""" |
| # Tests for fail_on_error == False. |
| |
| args = FailingIterator(10, [0]) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(9, len(results)) |
| |
| args = FailingIterator(10, [5]) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(9, len(results)) |
| |
| args = FailingIterator(10, [9]) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(9, len(results)) |
| |
| if process_count * thread_count > 1: |
| # In this case, we should ignore the fail_on_error flag. |
| args = FailingIterator(10, [9]) |
| results = self._RunApply(_ReturnOneValue, args, process_count, |
| thread_count, fail_on_error=True) |
| self.assertEqual(9, len(results)) |
| |
| args = FailingIterator(10, range(10)) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(0, len(results)) |
| |
| args = FailingIterator(0, []) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.assertEqual(0, len(results)) |
| |
| @RequiresIsolation |
| def testTestSharedAttrsWorkSingleProcessSingleThread(self): |
| self._TestSharedAttrsWork(1, 1) |
| |
| @RequiresIsolation |
| def testTestSharedAttrsWorkSingleProcessMultiThread(self): |
| self._TestSharedAttrsWork(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testTestSharedAttrsWorkMultiProcessSingleThread(self): |
| self._TestSharedAttrsWork(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testTestSharedAttrsWorkMultiProcessMultiThread(self): |
| self._TestSharedAttrsWork(3, 3) |
| |
| @Timeout |
| def _TestSharedAttrsWork(self, process_count, thread_count): |
| """Tests that Apply successfully uses shared_attrs.""" |
| command_inst = self.command_class(True) |
| command_inst.arg_length_sum = 19 |
| args = ['foo', ['bar', 'baz'], [], ['x', 'y'], [], 'abcd'] |
| self._RunApply(_IncrementByLength, args, process_count, |
| thread_count, command_inst=command_inst, |
| shared_attrs=['arg_length_sum']) |
| expected_sum = 19 |
| for arg in args: |
| expected_sum += len(arg) |
| self.assertEqual(expected_sum, command_inst.arg_length_sum) |
| |
| # Test that shared variables work when the iterator fails at the beginning, |
| # middle, and end. |
| for (failing_iterator, expected_failure_count) in ( |
| (FailingIterator(5, [0]), 1), |
| (FailingIterator(10, [1, 3, 5]), 3), |
| (FailingIterator(5, [4]), 1)): |
| command_inst = self.command_class(True) |
| args = failing_iterator |
| self._RunApply(_ReturnOneValue, args, process_count, thread_count, |
| command_inst=command_inst, shared_attrs=['failure_count']) |
| self.assertEqual( |
| expected_failure_count, command_inst.failure_count, |
| msg='Failure count did not match. Expected: %s, actual: %s ' |
| 'for failing iterator of size %s, failing indices %s' % |
| (expected_failure_count, command_inst.failure_count, |
| failing_iterator.size, failing_iterator.failure_indices)) |
| |
| @RequiresIsolation |
| def testThreadsSurviveExceptionsInFuncSingleProcessSingleThread(self): |
| self._TestThreadsSurviveExceptionsInFunc(1, 1) |
| |
| @RequiresIsolation |
| def testThreadsSurviveExceptionsInFuncSingleProcessMultiThread(self): |
| self._TestThreadsSurviveExceptionsInFunc(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testThreadsSurviveExceptionsInFuncMultiProcessSingleThread(self): |
| self._TestThreadsSurviveExceptionsInFunc(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testThreadsSurviveExceptionsInFuncMultiProcessMultiThread(self): |
| self._TestThreadsSurviveExceptionsInFunc(3, 3) |
| |
| @Timeout |
| def _TestThreadsSurviveExceptionsInFunc(self, process_count, thread_count): |
| command_inst = self.command_class(True) |
| args = ([()] * 5) |
| self._RunApply(_FailureFunc, args, process_count, thread_count, |
| command_inst=command_inst, shared_attrs=['failure_count'], |
| thr_exc_handler=_FailingExceptionHandler) |
| self.assertEqual(len(args), command_inst.failure_count) |
| |
| @RequiresIsolation |
| def testThreadsSurviveExceptionsInHandlerSingleProcessSingleThread(self): |
| self._TestThreadsSurviveExceptionsInHandler(1, 1) |
| |
| @RequiresIsolation |
| def testThreadsSurviveExceptionsInHandlerSingleProcessMultiThread(self): |
| self._TestThreadsSurviveExceptionsInHandler(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testThreadsSurviveExceptionsInHandlerMultiProcessSingleThread(self): |
| self._TestThreadsSurviveExceptionsInHandler(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testThreadsSurviveExceptionsInHandlerMultiProcessMultiThread(self): |
| self._TestThreadsSurviveExceptionsInHandler(3, 3) |
| |
| @Timeout |
| def _TestThreadsSurviveExceptionsInHandler(self, process_count, thread_count): |
| command_inst = self.command_class(True) |
| args = ([()] * 5) |
| self._RunApply(_FailureFunc, args, process_count, thread_count, |
| command_inst=command_inst, shared_attrs=['failure_count'], |
| thr_exc_handler=_FailingExceptionHandler) |
| self.assertEqual(len(args), command_inst.failure_count) |
| |
| @RequiresIsolation |
| @Timeout |
| def testFailOnErrorFlag(self): |
| """Tests that fail_on_error produces the correct exception on failure.""" |
| def _ExpectCustomException(test_func): |
| try: |
| test_func() |
| self.fail( |
| 'Setting fail_on_error should raise any exception encountered.') |
| except CustomException, e: |
| pass |
| except Exception, e: |
| self.fail('Got unexpected error: ' + str(e)) |
| |
| def _RunFailureFunc(): |
| command_inst = self.command_class(True) |
| args = ([()] * 5) |
| self._RunApply(_FailureFunc, args, 1, 1, command_inst=command_inst, |
| shared_attrs=['failure_count'], fail_on_error=True) |
| _ExpectCustomException(_RunFailureFunc) |
| |
| def _RunFailingIteratorFirstPosition(): |
| args = FailingIterator(10, [0]) |
| results = self._RunApply(_ReturnOneValue, args, 1, 1, fail_on_error=True) |
| self.assertEqual(0, len(results)) |
| _ExpectCustomException(_RunFailingIteratorFirstPosition) |
| |
| def _RunFailingIteratorPositionMiddlePosition(): |
| args = FailingIterator(10, [5]) |
| results = self._RunApply(_ReturnOneValue, args, 1, 1, fail_on_error=True) |
| self.assertEqual(5, len(results)) |
| _ExpectCustomException(_RunFailingIteratorPositionMiddlePosition) |
| |
| def _RunFailingIteratorLastPosition(): |
| args = FailingIterator(10, [9]) |
| results = self._RunApply(_ReturnOneValue, args, 1, 1, fail_on_error=True) |
| self.assertEqual(9, len(results)) |
| _ExpectCustomException(_RunFailingIteratorLastPosition) |
| |
| def _RunFailingIteratorMultiplePositions(): |
| args = FailingIterator(10, [1, 3, 5]) |
| results = self._RunApply(_ReturnOneValue, args, 1, 1, fail_on_error=True) |
| self.assertEqual(1, len(results)) |
| _ExpectCustomException(_RunFailingIteratorMultiplePositions) |
| |
| @RequiresIsolation |
| def testRecursiveDepthThreeDifferentFunctionsSingleProcessSingleThread(self): |
| self._TestRecursiveDepthThreeDifferentFunctions(1, 1) |
| |
| @RequiresIsolation |
| def testRecursiveDepthThreeDifferentFunctionsSingleProcessMultiThread(self): |
| self._TestRecursiveDepthThreeDifferentFunctions(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testRecursiveDepthThreeDifferentFunctionsMultiProcessSingleThread(self): |
| self._TestRecursiveDepthThreeDifferentFunctions(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testRecursiveDepthThreeDifferentFunctionsMultiProcessMultiThread(self): |
| self._TestRecursiveDepthThreeDifferentFunctions(3, 3) |
| |
| @Timeout |
| def _TestRecursiveDepthThreeDifferentFunctions(self, process_count, |
| thread_count): |
| """Tests recursive application of Apply. |
| |
| Calls Apply(A), where A calls Apply(B) followed by Apply(C) and B calls |
| Apply(C). |
| |
| Args: |
| process_count: Number of processes to use. |
| thread_count: Number of threads to use. |
| """ |
| base_args = [3, 1, 4, 1, 5] |
| args = [[process_count, thread_count, count] for count in base_args] |
| |
| results = self._RunApply(_ReApplyWithReplicatedArguments, args, |
| process_count, thread_count) |
| self.assertEqual(7 * (sum(base_args) + len(base_args)), sum(results)) |
| |
| @RequiresIsolation |
| def testExceptionInProducerRaisesAndTerminatesSingleProcessSingleThread(self): |
| self._TestExceptionInProducerRaisesAndTerminates(1, 1) |
| |
| @RequiresIsolation |
| def testExceptionInProducerRaisesAndTerminatesSingleProcessMultiThread(self): |
| self._TestExceptionInProducerRaisesAndTerminates(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testExceptionInProducerRaisesAndTerminatesMultiProcessSingleThread(self): |
| self._TestExceptionInProducerRaisesAndTerminates(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testExceptionInProducerRaisesAndTerminatesMultiProcessMultiThread(self): |
| self._TestExceptionInProducerRaisesAndTerminates(3, 3) |
| |
| @Timeout |
| def _TestExceptionInProducerRaisesAndTerminates(self, process_count, |
| thread_count): |
| args = self # The ProducerThread will try and fail to iterate over this. |
| try: |
| self._RunApply(_ReturnOneValue, args, process_count, thread_count) |
| self.fail('Did not raise expected exception.') |
| except TypeError: |
| pass |
| |
| @RequiresIsolation |
| def testSkippedArgumentsSingleThreadSingleProcess(self): |
| self._TestSkippedArguments(1, 1) |
| |
| @RequiresIsolation |
| def testSkippedArgumentsMultiThreadSingleProcess(self): |
| self._TestSkippedArguments(1, 3) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testSkippedArgumentsSingleThreadMultiProcess(self): |
| self._TestSkippedArguments(3, 1) |
| |
| @RequiresIsolation |
| @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') |
| def testSkippedArgumentsMultiThreadMultiProcess(self): |
| self._TestSkippedArguments(3, 3) |
| |
| @Timeout |
| def _TestSkippedArguments(self, process_count, thread_count): |
| |
| # Skip a proper subset of the arguments. |
| n = 2 * process_count * thread_count |
| args = range(1, n + 1) |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count, |
| arg_checker=_SkipEvenNumbersArgChecker) |
| self.assertEqual(n / 2, len(results)) # We know n is even. |
| self.assertEqual(n / 2, sum(results)) |
| |
| # Skip all arguments. |
| args = [2 * x for x in args] |
| results = self._RunApply(_ReturnOneValue, args, process_count, thread_count, |
| arg_checker=_SkipEvenNumbersArgChecker) |
| self.assertEqual(0, len(results)) |
| |
| |
| class TestParallelismFrameworkWithoutMultiprocessing(TestParallelismFramework): |
| """Tests parallelism framework works with multiprocessing module unavailable. |
| |
| Notably, this test has no way to override previous calls |
| to gslib.util.CheckMultiprocessingAvailableAndInit to prevent the |
| initialization of all of the global variables in command.py, so this still |
| behaves slightly differently than the behavior one would see on a machine |
| where the multiprocessing functionality is actually not available (in |
| particular, it will not catch the case where a global variable that is not |
| available for the sequential path is referenced before initialization). |
| """ |
| command_class = FakeCommandWithoutMultiprocessingModule |
