From 15ce27ecd5f7a6c95bf10ed505bc5dfab730bc9e Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Wed, 7 Aug 2013 18:54:49 -0500 Subject: [PATCH 1/6] First pass at AutoContext Intended to automatically insert asyncs in batches (asynchronously) as they are added to the context. --- furious/context/auto_context.py | 64 +++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 furious/context/auto_context.py diff --git a/furious/context/auto_context.py b/furious/context/auto_context.py new file mode 100644 index 0000000..85f72f6 --- /dev/null +++ b/furious/context/auto_context.py @@ -0,0 +1,64 @@ + +from furious.context.context import Context + + +class AutoContext(Context): + """Similar to context, but automatically inserts tasks asynchronously as + they are added to the context. Inserted in batches if specified. + """ + + def __init__(self, batch_size=None, **options): + """Setup this context in addition to accepting a batch_size.""" + + Context.__init__(self, **options) + + self.batch_size = batch_size + + self._num_tasks_inserted = 0 + + def add(self, target, args=None, kwargs=None, **options): + """Add an Async job to this context. + + The same as Context.add, but calls _auto_insert_check() to + insert tasks automatically. + """ + + target = super( + AutoContext, self).add(target, args, kwargs, **options) + + self._auto_insert_check() + + return target + + def _auto_insert_check(self): + """Automatically insert tasks asynchronously. + Depending on batch_size, insert or wait until next call. + """ + + if not self.batch_size: + self._handle_tasks() + return + + if len(self.tasks) >= self.batch_size: + self._handle_tasks() + + def _handle_tasks(self): + """Convert Async's into tasks, then insert them into queues. + Similar to the default _handle_tasks, but don't mark all + tasks inserted. + """ + + self._handle_tasks_insert(batch_size=self.batch_size) + + def __exit__(self, exc_type, exc_val, exc_tb): + """In addition to the default __exit__(), also mark all tasks + inserted. + """ + + super(AutoContext, self).__exit__(exc_type, exc_val, exc_tb) + + # Mark all tasks inserted. + self._tasks_inserted = True + + return False + From 4173f1dfcc7703756358613ffe62893709987540 Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Thu, 8 Aug 2013 11:54:44 -0500 Subject: [PATCH 2/6] Separate _tasks_inserted set from insertion Move most of handle_tasks() into _handle_tasks() to allow AutoContext to insert some tasks before fully exiting. Also add a parameter to allow different batch_sizes. --- furious/context/context.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/furious/context/context.py b/furious/context/context.py index 30be95f..8265a2d 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -92,7 +92,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): return False - def _handle_tasks(self): + def _handle_tasks_insert(self, batch_size=None): """Convert all Async's into tasks, then insert them into queues.""" if self._tasks_inserted: raise errors.ContextAlreadyStartedError( @@ -100,9 +100,15 @@ def _handle_tasks(self): task_map = self._get_tasks_by_queue() for queue, tasks in task_map.iteritems(): - for batch in _task_batcher(tasks): + for batch in _task_batcher(tasks, batch_size=batch_size): self._insert_tasks(batch, queue=queue) + def _handle_tasks(self): + """Convert all Async's into tasks, then insert them into queues. + Also mark all tasks inserted to ensure they are not reinserted later. + """ + self._handle_tasks_insert() + self._tasks_inserted = True def _get_tasks_by_queue(self): @@ -119,8 +125,8 @@ def _get_tasks_by_queue(self): def add(self, target, args=None, kwargs=None, **options): """Add an Async job to this context. - Takes an Async object or the argumnets to construct an Async - object as argumnets. Returns the newly added Async object. + Takes an Async object or the arguments to construct an Async + object as arguments. Returns the newly added Async object. """ from ..async import Async from ..batcher import Message @@ -239,12 +245,15 @@ def _insert_tasks(tasks, queue, transactional=False): _insert_tasks(tasks[count / 2:], queue, transactional) -def _task_batcher(tasks): +def _task_batcher(tasks, batch_size=None): """Batches large task lists into groups of 100 so that they can all be inserted. """ from itertools import izip_longest - args = [iter(tasks)] * 100 + if not batch_size: + batch_size = DEFAULT_TASK_BATCH_SIZE + + args = [iter(tasks)] * batch_size return ([task for task in group if task] for group in izip_longest(*args)) From d9892a2f0f4f277f162750a4270cb5767fcbdb73 Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Thu, 8 Aug 2013 12:51:56 -0500 Subject: [PATCH 3/6] Add constant --- furious/context/context.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/furious/context/context.py b/furious/context/context.py index 8265a2d..7ee952a 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -51,6 +51,8 @@ from .. import errors +DEFAULT_TASK_BATCH_SIZE = 100 + class Context(object): """Furious context object. From ba6c3b2096adf21219d7a2f34743aa10dd769f27 Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Thu, 8 Aug 2013 16:24:39 -0500 Subject: [PATCH 4/6] Add test to ensure AutoContext adds tasks to queue Also ensure they are added correctly in batches. --- furious/context/auto_context.py | 31 ++++++++- furious/tests/context/test_auto_context.py | 80 ++++++++++++++++++++++ 2 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 furious/tests/context/test_auto_context.py diff --git a/furious/context/auto_context.py b/furious/context/auto_context.py index 85f72f6..0ab12ec 100644 --- a/furious/context/auto_context.py +++ b/furious/context/auto_context.py @@ -1,3 +1,25 @@ +# +# Copyright 2012 WebFilings, 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. +# + +import unittest + +from google.appengine.ext import testbed + +from mock import Mock +from mock import patch from furious.context.context import Context @@ -19,10 +41,12 @@ def __init__(self, batch_size=None, **options): def add(self, target, args=None, kwargs=None, **options): """Add an Async job to this context. - The same as Context.add, but calls _auto_insert_check() to - insert tasks automatically. + Like Context.add(): creates an Async and adds it to our list of tasks. + but also calls _auto_insert_check() to add tasks to queues + automatically. """ + # In superclass, add new task to our list of tasks target = super( AutoContext, self).add(target, args, kwargs, **options) @@ -39,7 +63,7 @@ def _auto_insert_check(self): self._handle_tasks() return - if len(self.tasks) >= self.batch_size: + if len(self._tasks) >= self.batch_size: self._handle_tasks() def _handle_tasks(self): @@ -49,6 +73,7 @@ def _handle_tasks(self): """ self._handle_tasks_insert(batch_size=self.batch_size) + self._tasks = [] def __exit__(self, exc_type, exc_val, exc_tb): """In addition to the default __exit__(), also mark all tasks diff --git a/furious/tests/context/test_auto_context.py b/furious/tests/context/test_auto_context.py new file mode 100644 index 0000000..d775be6 --- /dev/null +++ b/furious/tests/context/test_auto_context.py @@ -0,0 +1,80 @@ +# +# Copyright 2012 WebFilings, 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. +# + +import unittest + +from google.appengine.ext import testbed + +from mock import Mock +from mock import patch + + +class TestAutoContextTestCase(unittest.TestCase): + """Test the AutoContext class.""" + + def setUp(self): + """Setup the test harness and a request hash.""" + import os + import uuid + + self.harness = testbed.Testbed() + self.harness.activate() + self.harness.init_taskqueue_stub() + + # Backup environment + self._orig_environ = os.environ.copy() + # Ensure each test looks like it is in a new request. + os.environ['REQUEST_ID_HASH'] = uuid.uuid4().hex + + def tearDown(self): + """Deactive the test harness, and restore the environment.""" + import os + + self.harness.deactivate() + os.environ.clear() + os.environ.update(self._orig_environ) + + @patch('google.appengine.api.taskqueue.Queue.add', auto_spec=True) + def test_add_job_to_context_works(self, queue_add_mock): + """Ensure adding a job works.""" + from furious.async import Async + from furious.context.auto_context import AutoContext + + batch_size = 2 + + with AutoContext(batch_size) as ctx: + # First batch + job1 = ctx.add('test', args=[1, 2]) + job2 = ctx.add('test2', args=[1, 2]) + + # Second batch (not added until "with" section ends)) + job3 = ctx.add('test3', args=[1, 2]) + + # Ensure the first two jobs were inserted in the first batch. + self.assertIsInstance(job1, Async) + self.assertIsInstance(job2, Async) + queue_add_mock.assert_called_once() + #Ensure only two tasks were inserted + tasks_added = queue_add_mock.call_args[0][0] + self.assertEqual(2, len(tasks_added)) + + # Ensure the third job was inserted when the context exited. + self.assertIsInstance(job3, Async) + # Ensure add has now been called twice. + self.assertEqual(2, queue_add_mock.call_count) + # Ensure only one task was inserted + tasks_added = queue_add_mock.call_args[0][0] + self.assertEqual(1, len(tasks_added)) From c7ea56cc4344fa2a9ee8e51b7a4ca724ba224462 Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Thu, 8 Aug 2013 17:42:59 -0500 Subject: [PATCH 5/6] Add tests with None batch_size and not tasks When batch_size is None, AutoContext will act like Context, where all tasks are inserted at once when the context is exited. Ensujre that when no tasks are added, there are no errors. --- furious/context/auto_context.py | 1 - furious/tests/context/test_auto_context.py | 52 +++++++++++++++++++++- 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/furious/context/auto_context.py b/furious/context/auto_context.py index 0ab12ec..f7299b1 100644 --- a/furious/context/auto_context.py +++ b/furious/context/auto_context.py @@ -60,7 +60,6 @@ def _auto_insert_check(self): """ if not self.batch_size: - self._handle_tasks() return if len(self._tasks) >= self.batch_size: diff --git a/furious/tests/context/test_auto_context.py b/furious/tests/context/test_auto_context.py index d775be6..09d3716 100644 --- a/furious/tests/context/test_auto_context.py +++ b/furious/tests/context/test_auto_context.py @@ -48,8 +48,16 @@ def tearDown(self): os.environ.update(self._orig_environ) @patch('google.appengine.api.taskqueue.Queue.add', auto_spec=True) - def test_add_job_to_context_works(self, queue_add_mock): - """Ensure adding a job works.""" + def test_add_job_to_context_multiple_batches(self, queue_add_mock): + """Ensure adding more tasks than the batch_size causes multiple batches + to get inserted. + + Adding 3 asyncs with a batch_size of 2 should result in two queue.adds, + containing 2 and 1 task respectively. + + Also ensure that the remaining task is inserted upon exiting the + context. + """ from furious.async import Async from furious.context.auto_context import AutoContext @@ -78,3 +86,43 @@ def test_add_job_to_context_works(self, queue_add_mock): # Ensure only one task was inserted tasks_added = queue_add_mock.call_args[0][0] self.assertEqual(1, len(tasks_added)) + + @patch('google.appengine.api.taskqueue.Queue.add', auto_spec=True) + def test_add_job_to_context_batch_size_unspecified(self, queue_add_mock): + """When batch_size is None or 0, the default behavior of Context is + used. All the tasks are added to the queue when the context is exited. + """ + from furious.async import Async + from furious.context.auto_context import AutoContext + + with AutoContext() as ctx: + + job1 = ctx.add('test', args=[1, 2]) + job2 = ctx.add('test2', args=[1, 2]) + job3 = ctx.add('test3', args=[1, 2]) + + self.assertIsInstance(job1, Async) + self.assertIsInstance(job2, Async) + self.assertIsInstance(job3, Async) + + # Ensure no batches of tasks are inserted yet. + self.assertFalse(queue_add_mock.called) + + # Ensure the list of tasks added when the context exited. + self.assertEqual(1, queue_add_mock.call_count) + # Ensure the three tasks were inserted + tasks_added = queue_add_mock.call_args[0][0] + self.assertEqual(3, len(tasks_added)) + + @patch('google.appengine.api.taskqueue.Queue.add', auto_spec=True) + def test_no_jobs(self, queue_add_mock): + """When no Asyncs are added to the context, ensure that there are no + errors and nothing is added to the task queue. + """ + from furious.context.auto_context import AutoContext + + with AutoContext(3): + pass + + # Ensure queue.add() was never called. + self.assertEqual(0, queue_add_mock.call_count) From 9212ed1e88e9e7fb362cba68a70ef6bba977e75c Mon Sep 17 00:00:00 2001 From: Eric Olson Date: Thu, 8 Aug 2013 18:32:41 -0500 Subject: [PATCH 6/6] Add comment and remove unused variable and imports --- furious/context/auto_context.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/furious/context/auto_context.py b/furious/context/auto_context.py index f7299b1..12dbd25 100644 --- a/furious/context/auto_context.py +++ b/furious/context/auto_context.py @@ -14,12 +14,13 @@ # limitations under the License. # -import unittest +""" +Furious AutoContext is used batch inserts of tasks into groups. -from google.appengine.ext import testbed +It is similar to Context, but inserts automatically before the context +is exited. +""" -from mock import Mock -from mock import patch from furious.context.context import Context @@ -36,8 +37,6 @@ def __init__(self, batch_size=None, **options): self.batch_size = batch_size - self._num_tasks_inserted = 0 - def add(self, target, args=None, kwargs=None, **options): """Add an Async job to this context.