From 14bb181e48799331bdfe68565df34c8d75249901 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Wed, 16 Jan 2013 01:25:06 -0600 Subject: [PATCH 001/441] adding basic grep example --- example/grep.py | 44 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 example/grep.py diff --git a/example/grep.py b/example/grep.py new file mode 100644 index 0000000..1733837 --- /dev/null +++ b/example/grep.py @@ -0,0 +1,44 @@ +import logging +import os +import os.path + +import webapp2 + +from furious.async import Async + +class GrepHandler(webapp2.RequestHandler): + def get(self): + query = self.request.get('query') + curdir = os.getcwd() + build_and_start(query, curdir) + self.response.out.write('starting grep for query: %s' % query) + +def build_and_start(query, directory): + async_task = Async( + target=grep, args=[query, directory], callbacks={'success': all_done} + ) + async_task.start() + +def grep_file(query, item): + return ['%s: %s' % (item, line) for line in open(item) if query in line] + +def grep(query, directory): + dir_contents = os.listdir(directory) + results = [] + for item in dir_contents: + path = os.path.join(directory, item) + if os.path.isdir(path): + build_and_start(query, path) + else: + if item.endswith('.py'): + results.extend(grep_file(query, path)) + return results + +def all_done(): + from furious.context import get_current_async + + async = get_current_async() + + for result in async.result: + logging.info(result) + From 2e288633661890640bb28a7ef094d53963a08fe7 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Wed, 16 Jan 2013 01:28:13 -0600 Subject: [PATCH 002/441] adding handler to app --- example/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/example/__init__.py b/example/__init__.py index ac5296d..b5b809a 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -266,6 +266,7 @@ def state_machine_success(): logging.info('Done working, stop now.') +from . import grep app = webapp2.WSGIApplication([ ('/', AsyncIntroHandler), @@ -275,5 +276,6 @@ def state_machine_success(): ('/callback/async', AsyncAsyncCallbackHandler), ('/workflow', SimpleWorkflowHandler), ('/workflow/complex', ComplexWorkflowHandler), + ('/grep', grep.GrepHandler), ]) From ae4a7f56ba7935fb62014d433f156e04ca0d7f8f Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Wed, 16 Jan 2013 01:46:23 -0600 Subject: [PATCH 003/441] changing searching to regex --- example/grep.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/example/grep.py b/example/grep.py index 1733837..bdf6994 100644 --- a/example/grep.py +++ b/example/grep.py @@ -1,6 +1,7 @@ import logging import os import os.path +import re import webapp2 @@ -20,7 +21,8 @@ def build_and_start(query, directory): async_task.start() def grep_file(query, item): - return ['%s: %s' % (item, line) for line in open(item) if query in line] + return ['%s: %s' % (item, line) for line in open(item) + if re.search(query, line)] def grep(query, directory): dir_contents = os.listdir(directory) From 57e9bc629f562f0b4499db41fe9ae014863b384c Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Wed, 16 Jan 2013 01:51:26 -0600 Subject: [PATCH 004/441] renaming callback from all_done to log_result --- example/grep.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/example/grep.py b/example/grep.py index bdf6994..5ca32f9 100644 --- a/example/grep.py +++ b/example/grep.py @@ -16,7 +16,8 @@ def get(self): def build_and_start(query, directory): async_task = Async( - target=grep, args=[query, directory], callbacks={'success': all_done} + target=grep, args=[query, directory], + callbacks={'success': log_results} ) async_task.start() @@ -36,7 +37,7 @@ def grep(query, directory): results.extend(grep_file(query, path)) return results -def all_done(): +def log_results(): from furious.context import get_current_async async = get_current_async() From 8374468bce45af29412e9e5c27b6ae1b7cd48467 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Wed, 16 Jan 2013 15:00:23 -0600 Subject: [PATCH 005/441] fixing typo in README and moving grep import to top of __init__.py --- README.md | 2 +- example/__init__.py | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 7ee93b6..7792ac0 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ In the simplest form, usage looks like: # Create an Async object. async = Async( target="your.module.func", - args=("positstional", "args"), + args=("positional", "args"), kwargs={"kwargs": "too"}) # Tell the async to insert itself to be run. diff --git a/example/__init__.py b/example/__init__.py index b5b809a..9dae1db 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -25,6 +25,7 @@ import webapp2 +from . import grep from furious.async import defaults @@ -266,8 +267,6 @@ def state_machine_success(): logging.info('Done working, stop now.') -from . import grep - app = webapp2.WSGIApplication([ ('/', AsyncIntroHandler), ('/context', ContextIntroHandler), From 4f560935dfcb33be2897131bad9ff67bcd08395f Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Wed, 16 Jan 2013 23:55:41 -0700 Subject: [PATCH 006/441] Correct Async.from_dict to not mutate source options dict. Async.from_dict mutated the options passed in as the argument. Such unexpected mutations are not friendly and may be dangerous. --- furious/async.py | 3 ++- furious/tests/test_async.py | 3 +++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/furious/async.py b/furious/async.py index 48931c5..f2a04c7 100644 --- a/furious/async.py +++ b/furious/async.py @@ -251,8 +251,9 @@ def to_dict(self): @classmethod def from_dict(cls, async): """Return an async job from a dict output by Async.to_dict.""" + import copy - async_options = async.copy() + async_options = copy.deepcopy(async) # JSON don't like datetimes. eta = async_options.get('task_args', {}).get('eta') diff --git a/furious/tests/test_async.py b/furious/tests/test_async.py index ac71217..11007c0 100644 --- a/furious/tests/test_async.py +++ b/furious/tests/test_async.py @@ -460,6 +460,9 @@ def test_to_task(self): self.assertEqual(expected_url, task.url) self.assertEqual(full_headers, task.headers) + options['task_args']['eta'] = datetime.datetime.fromtimestamp( + eta_posix) + self.assertEqual( options, Async.from_dict(json.loads(task.payload)).get_options()) From c87d46c9f070437c2308af67b28c6c67ca729a12 Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Wed, 16 Jan 2013 23:58:25 -0700 Subject: [PATCH 007/441] Transplant encode / decode callback logic to job_utils. The callback encoding logic is required by other objects as well, such as Context. Because of this, they should be split into a more "neutral" location than async.py. --- furious/async.py | 42 +++++------------------------------------- furious/job_utils.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 37 deletions(-) diff --git a/furious/async.py b/furious/async.py index f2a04c7..cfa7b62 100644 --- a/furious/async.py +++ b/furious/async.py @@ -72,6 +72,8 @@ def run_me(*args, **kwargs): import json +from .job_utils import decode_callbacks +from .job_utils import encode_callbacks from .job_utils import get_function_path_and_options @@ -244,7 +246,7 @@ def to_dict(self): callbacks = self._options.get('callbacks') if callbacks: - options['callbacks'] = _encode_callbacks(callbacks) + options['callbacks'] = encode_callbacks(callbacks) return options @@ -267,9 +269,9 @@ def from_dict(cls, async): # If there are callbacks, reconstitute them. callbacks = async_options.get('callbacks', {}) if callbacks: - async_options['callbacks'] = _decode_callbacks(callbacks) + async_options['callbacks'] = decode_callbacks(callbacks) - return Async(target, args, kwargs, **async_options) + return cls(target, args, kwargs, **async_options) def defaults(**options): @@ -299,37 +301,3 @@ def _check_options(options): assert 'job' not in options #assert 'callbacks' not in options - -def _encode_callbacks(callbacks): - """Encode callbacks to as a dict suitable for JSON encoding.""" - if not callbacks: - return - - encoded_callbacks = {} - for event, callback in callbacks.iteritems(): - if callable(callback): - callback, _ = get_function_path_and_options(callback) - - elif isinstance(callback, Async): - callback = callback.to_dict() - - encoded_callbacks[event] = callback - - return encoded_callbacks - - -def _decode_callbacks(encoded_callbacks): - """Decode the callbacks to an executable form.""" - from furious.job_utils import function_path_to_reference - - callbacks = {} - for event, callback in encoded_callbacks.iteritems(): - if isinstance(callback, dict): - callback = Async.from_dict(callback) - else: - callback = function_path_to_reference(callback) - - callbacks[event] = callback - - return callbacks - diff --git a/furious/job_utils.py b/furious/job_utils.py index ffd7dec..9005431 100644 --- a/furious/job_utils.py +++ b/furious/job_utils.py @@ -105,3 +105,39 @@ def function_path_to_reference(function_path): raise BadFunctionPathError( 'Unable to find function "%s".' % (function_path,)) + +def encode_callbacks(callbacks): + """Encode callbacks to as a dict suitable for JSON encoding.""" + from .async import Async + + if not callbacks: + return + + encoded_callbacks = {} + for event, callback in callbacks.iteritems(): + if callable(callback): + callback, _ = get_function_path_and_options(callback) + + elif isinstance(callback, Async): + callback = callback.to_dict() + + encoded_callbacks[event] = callback + + return encoded_callbacks + + +def decode_callbacks(encoded_callbacks): + """Decode the callbacks to an executable form.""" + from .async import Async + + callbacks = {} + for event, callback in encoded_callbacks.iteritems(): + if isinstance(callback, dict): + callback = Async.from_dict(callback) + else: + callback = function_path_to_reference(callback) + + callbacks[event] = callback + + return callbacks + From d9b08095a38235a82da0a47a4513de635e78aa84 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 00:43:28 -0700 Subject: [PATCH 008/441] Add bump_batch method for incremeting the batch count. Increments memcahce buy a key. --- furious/batcher.py | 15 +++++++++------ furious/tests/test_batcher.py | 16 ++++++++++++++++ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index 173eb7c..61db28f 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -7,6 +7,7 @@ MESSAGE_DEFAULT_QUEUE = 'default_pull' MESSAGE_PROCESSOR_NAME = 'processor' +MESSAGE_BATCH_NAME = 'agg-batch' METHOD_TYPE = 'PULL' @@ -121,7 +122,7 @@ def to_task(self): @property def group_key(self): """Return the :class: `str` group key based off of the tag.""" - return 'agg-batch-%s' % (self.tag) + return "%s-%s" % (MESSAGE_BATCH_NAME, self.tag) @property def current_batch(self): @@ -148,9 +149,11 @@ def time_throttle(self): return int(time.time() / max(1, self.frequency)) -def fetch_messages(): - pass +def bump_batch(work_group): + """Return the incremented batch id for the work group + :param work_group: :class: `str` - -def bump_batch(): - pass + :return: :class: `int` current batch id. + """ + key = "%s-%s" % (MESSAGE_BATCH_NAME, work_group) + return memcache.incr(key) diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 1427933..cf52123 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -353,3 +353,19 @@ def test_curent_batch_key_doesnt_exist_in_cache(self, cache): cache.get.assert_called_once_with('agg-batch-processor') cache.add.assert_called_once_with('agg-batch-processor', 1) + + +class BumpBatchTestCase(unittest.TestCase): + + @patch('furious.batcher.memcache') + def test_cache_incremented_by_key(self, cache): + """Ensure that the cache object is incremented by the key passed in.""" + from furious.batcher import bump_batch + + cache.incr.return_value = 2 + + val = bump_batch('group') + + self.assertEqual(val, 2) + + cache.incr.assert_called_once_with('agg-batch-group') From 53e536892f4c80595acae7d74e036a7dc077a656 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 17 Jan 2013 02:17:18 -0600 Subject: [PATCH 009/441] Adding get_current_context() method --- furious/context/__init__.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/furious/context/__init__.py b/furious/context/__init__.py index e8a1e47..9d5b233 100644 --- a/furious/context/__init__.py +++ b/furious/context/__init__.py @@ -80,3 +80,14 @@ def get_current_async(): raise NotInContextError('Not in an _ExecutionContext.') + +def get_current_context(): + """Return a reference to the currently Context object. + """ + local_context = _local.get_local_context() + + if local_context._executing_async: + return local_context.registry[-1] + + raise NotInContextError('Not in an _ExecutionContext.') + From a9727d999597e42bba4ba5b2e03ccc5b89e29233 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Wed, 16 Jan 2013 22:28:11 -0700 Subject: [PATCH 010/441] Fix broken unit test around default pull queue for messages. --- furious/tests/test_batcher.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 9e98c95..21c19cc 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -69,7 +69,7 @@ def test_get_default_queue(self): message = Message() - self.assertEqual('default', message.get_queue()) + self.assertEqual('default_pull', message.get_queue()) def test_get_task_args(self): """Ensure get_task_args returns the message task_args.""" From 0338c6601f12ccf4a1ca5be7df7522d87750fbc8 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 00:04:04 -0700 Subject: [PATCH 011/441] Add MessageProcessor and insert_messsage_processor to batcher Add MessageProcessor Async object to the batcher module. This object will insert a task to pull tasks off a pull queue. It will try to insert them serially using a memcache marker based off a time frequency formula. Add insert_messsage_processor function that wraps creating a MessageProcessor and inserting it. --- furious/batcher.py | 100 +++++++++++++- furious/tests/test_batcher.py | 240 +++++++++++++++++++++++++++++++++- 2 files changed, 335 insertions(+), 5 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index 58d7eec..3846dca 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -1,6 +1,12 @@ import json +import time + +from google.appengine.api import memcache + +from .async import Async MESSAGE_DEFAULT_QUEUE = 'default_pull' +MESSAGE_PROCESSOR_NAME = 'processor' METHOD_TYPE = 'PULL' @@ -64,8 +70,6 @@ def to_dict(self): # JSON don't like datetimes. eta = options.get('task_args', {}).get('eta') if eta: - import time - options['task_args']['eta'] = time.mktime(eta.timetuple()) return options @@ -86,8 +90,96 @@ def from_dict(cls, message): return Message(**message_options) -def insert_messsage_processor(): - pass +class MessageProcessor(Async): + """Async message processor for processing messages in the pull queue.""" + + def __init__(self, target, args=None, kwargs=None, tag=None, freq=30, + **options): + super(MessageProcessor, self).__init__(target, args, kwargs, **options) + + self.frequency = freq + self.tag = tag if tag else MESSAGE_PROCESSOR_NAME + + def to_task(self): + """Return a task object representing this MessageProcessor job.""" + task_args = self.get_task_args() + + # check for name in task args + name = task_args.get('name', MESSAGE_PROCESSOR_NAME) + + # if the countdown isn't in the task_args set it to the frequency + if not 'countdown' in task_args: + task_args['countdown'] = self.frequency + + task_args['name'] = "%s-%s-%s-%s" % ( + name, self.tag, self.current_batch, self.time_throttle) + + self.update_options(task_args=task_args) + + return super(MessageProcessor, self).to_task() + + @property + def group_key(self): + """Return the :class: `str` group key based off of the tag.""" + return 'agg-batch-%s' % (self.tag) + + @property + def current_batch(self): + """Return the batch id for the tag. + + :return: :class: `int` current batch id + """ + current_batch = memcache.get(self.group_key) + + if not current_batch: + memcache.add(self.group_key, 1) + current_batch = 1 + + return current_batch + + @property + def time_throttle(self): + """Return an :class: `int` of the currrent time divided by the + processor frequency. Frequency cannot be less than one and will default + to one if it is. + + :return: :class: `int` + """ + return int(time.time() / max(1, self.frequency)) + + +def insert_messsage_processor(job, args, kwargs, queue_name, freq=30, tag=None, + countdown=None, name=None): + """Return the :class: `MessageProcessor` created and started. Will set the + countdown and name as task_args if they aren't None. + + :param job: Python function to run within the Async task + :param args: :class: `tuple` of arguments to pass to the job when it runs. + :param kwargs: :class: `dict` of optional arguments to pass to the job + when it runs. + :param queue_name: :class: `str` Queue that the async task will run within. + :param freq: :class: `int` The frequency throttle for how many proccessing + jobs can be inserted at one time. + :param tag: :class: `str` Pull queue tag used for fetching the work items. + :param countdown: :class: `int` Delay before running the job. + :param name: :class: `str` Part of the unique name of the task. + + :return: :class: `MessageProcessor` created and started. + """ + task_args = {} + + if countdown: + task_args['countdown'] = countdown + + if name: + task_args['name'] = name + + processor = MessageProcessor(job, args, kwargs, queue=queue_name, + freq=freq, tag=tag, task_args=task_args) + + processor.start() + + return processor def fetch_messages(): diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 21c19cc..04be1b9 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -1,7 +1,7 @@ import json import unittest -from mock import patch +from mock import Mock, patch class MessageTestCase(unittest.TestCase): @@ -190,3 +190,241 @@ def test_insert(self, queue_mock): queue_mock.assert_called_once_with(name='my_queue') self.assertTrue(queue_mock.return_value.add.called) + + +class MessageProcessorTestCase(unittest.TestCase): + + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_with_no_name_passed_in(self, memcache, time): + """Ensure that if no name is passed into the MessageProcessor that it + creates a default unique name when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue') + + task = processor.to_task() + + self.assertEqual(task.name, 'processor-processor-current-batch-3') + + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_with_frequency_passed_in(self, memcache, time): + """Ensure that if a frequency is passed into the MessageProcessor that + it uses that frequency when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue', freq=100) + + task = processor.to_task() + + self.assertEqual(task.name, 'processor-processor-current-batch-1') + + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_with_name_passed_in(self, memcache, time): + """Ensure that if a name is passed into the MessageProcessor that it + uses that name when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue', + task_args={'name': 'test-name'}) + + task = processor.to_task() + + self.assertEqual(task.name, 'test-name-processor-current-batch-3') + + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_with_tag_passed_in(self, memcache, time): + """Ensure that if a tag is passed into the MessageProcessor that it + uses that tag when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue', + tag='test-tag') + + task = processor.to_task() + + self.assertEqual(task.name, 'processor-test-tag-current-batch-3') + + memcache.get.assert_called_once_with('agg-batch-test-tag') + + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_with_tag_not_passed_in(self, memcache, time): + """Ensure that if a tag is not passed into the MessageProcessor that it + uses a default value when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue') + + task = processor.to_task() + + self.assertEqual(task.name, 'processor-processor-current-batch-3') + + memcache.get.assert_called_once_with('agg-batch-processor') + + @patch('google.appengine.api.taskqueue.Task', autospec=True) + @patch('furious.batcher.time') + @patch('furious.batcher.memcache') + def test_to_task_has_correct_arguments(self, memcache, time, task): + """Ensure that if no name is passed into the MessageProcessor that it + creates a default unique name when creating the task. + """ + from furious.batcher import MessageProcessor + + memcache.get.return_value = 'current-batch' + time.time.return_value = 100 + + processor = MessageProcessor('something', queue='test_queue') + + processor.to_task() + + task_args = { + 'url': '/_ah/queue/async/something', + 'headers': {}, + 'payload': json.dumps({ + 'queue': 'test_queue', + 'job': ["something", None, None], + 'task_args': { + 'countdown': 30, + 'name': 'processor-processor-current-batch-3' + }, + }), + 'countdown': 30, + 'name': 'processor-processor-current-batch-3' + } + + task.assert_called_once_with(**task_args) + + @patch('furious.batcher.memcache') + def test_curent_batch_key_exists_in_cache(self, cache): + """Ensure that if the current batch key exists in cache that it uses it + and doesn't update it. + """ + from furious.batcher import MessageProcessor + + cache.get.return_value = 1 + + processor = MessageProcessor('something') + + current_batch = processor.current_batch + + cache.get.assert_called_once_with('agg-batch-processor') + + self.assertEqual(current_batch, 1) + self.assertFalse(cache.add.called) + + @patch('furious.batcher.memcache') + def test_curent_batch_key_doesnt_exist_in_cache(self, cache): + """Ensure that if the current batch key doesn't exist in cache that it + inserts the default value of 1 into cache and returns it. + """ + from furious.batcher import MessageProcessor + + cache.get.return_value = None + + processor = MessageProcessor('something') + + current_batch = processor.current_batch + + self.assertEqual(current_batch, 1) + + cache.get.assert_called_once_with('agg-batch-processor') + cache.add.assert_called_once_with('agg-batch-processor', 1) + + +class InsertMessageProcessorTestCase(unittest.TestCase): + + def test_no_countdown_or_name(self): + """Ensure that if no countdown or name are set that they aren't passed + in as task_args to the MessageProcessor. + """ + from furious.batcher import insert_messsage_processor + from furious.batcher import MessageProcessor + + with patch.object(MessageProcessor, 'start') as start: + processor = insert_messsage_processor( + 'job', None, None, 'queue-name') + + options = { + 'queue': 'queue-name', + 'job': ('job', None, None), + 'task_args': {} + } + self.assertEqual(processor.get_options(), options) + self.assertEqual(processor.tag, 'processor') + self.assertEqual(processor.frequency, 30) + + start.assert_called_once_with() + + def test_countdown_passed_in(self): + """Ensure that if countdown is set that it's passed as a task_arg to + the MessageProcessor. + """ + from furious.batcher import insert_messsage_processor + from furious.batcher import MessageProcessor + + with patch.object(MessageProcessor, 'start') as start: + processor = insert_messsage_processor( + 'job', None, None, 'queue-name', countdown=100) + + task_args = { + 'countdown': 100 + } + self.assertEqual(processor.get_task_args(), task_args) + + start.assert_called_once_with() + + def test_name_passed_in(self): + """Ensure that if name is set that it's passed as a task_arg to the + MessageProcessor. + """ + from furious.batcher import insert_messsage_processor + from furious.batcher import MessageProcessor + + with patch.object(MessageProcessor, 'start') as start: + processor = insert_messsage_processor( + 'job', None, None, 'queue-name', name='name') + + task_args = { + 'name': 'name' + } + self.assertEqual(processor.get_task_args(), task_args) + + start.assert_called_once_with() + + def test_tag_passed_in(self): + """Ensure that if tag is set that it's set on the MessageProcessor.""" + + from furious.batcher import insert_messsage_processor + from furious.batcher import MessageProcessor + + with patch.object(MessageProcessor, 'start') as start: + processor = insert_messsage_processor( + 'job', None, None, 'queue-name', tag='tag') + + self.assertEqual(processor.tag, 'tag') + + start.assert_called_once_with() From 0cddbbb1ab4f0c21a90d140253f1a58ff45b35e5 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 00:27:12 -0700 Subject: [PATCH 012/441] Removing helper function for batch processor. --- furious/batcher.py | 34 ---------------- furious/tests/test_batcher.py | 77 +---------------------------------- 2 files changed, 1 insertion(+), 110 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index 3846dca..173eb7c 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -148,40 +148,6 @@ def time_throttle(self): return int(time.time() / max(1, self.frequency)) -def insert_messsage_processor(job, args, kwargs, queue_name, freq=30, tag=None, - countdown=None, name=None): - """Return the :class: `MessageProcessor` created and started. Will set the - countdown and name as task_args if they aren't None. - - :param job: Python function to run within the Async task - :param args: :class: `tuple` of arguments to pass to the job when it runs. - :param kwargs: :class: `dict` of optional arguments to pass to the job - when it runs. - :param queue_name: :class: `str` Queue that the async task will run within. - :param freq: :class: `int` The frequency throttle for how many proccessing - jobs can be inserted at one time. - :param tag: :class: `str` Pull queue tag used for fetching the work items. - :param countdown: :class: `int` Delay before running the job. - :param name: :class: `str` Part of the unique name of the task. - - :return: :class: `MessageProcessor` created and started. - """ - task_args = {} - - if countdown: - task_args['countdown'] = countdown - - if name: - task_args['name'] = name - - processor = MessageProcessor(job, args, kwargs, queue=queue_name, - freq=freq, tag=tag, task_args=task_args) - - processor.start() - - return processor - - def fetch_messages(): pass diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 04be1b9..1427933 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -1,7 +1,7 @@ import json import unittest -from mock import Mock, patch +from mock import patch class MessageTestCase(unittest.TestCase): @@ -353,78 +353,3 @@ def test_curent_batch_key_doesnt_exist_in_cache(self, cache): cache.get.assert_called_once_with('agg-batch-processor') cache.add.assert_called_once_with('agg-batch-processor', 1) - - -class InsertMessageProcessorTestCase(unittest.TestCase): - - def test_no_countdown_or_name(self): - """Ensure that if no countdown or name are set that they aren't passed - in as task_args to the MessageProcessor. - """ - from furious.batcher import insert_messsage_processor - from furious.batcher import MessageProcessor - - with patch.object(MessageProcessor, 'start') as start: - processor = insert_messsage_processor( - 'job', None, None, 'queue-name') - - options = { - 'queue': 'queue-name', - 'job': ('job', None, None), - 'task_args': {} - } - self.assertEqual(processor.get_options(), options) - self.assertEqual(processor.tag, 'processor') - self.assertEqual(processor.frequency, 30) - - start.assert_called_once_with() - - def test_countdown_passed_in(self): - """Ensure that if countdown is set that it's passed as a task_arg to - the MessageProcessor. - """ - from furious.batcher import insert_messsage_processor - from furious.batcher import MessageProcessor - - with patch.object(MessageProcessor, 'start') as start: - processor = insert_messsage_processor( - 'job', None, None, 'queue-name', countdown=100) - - task_args = { - 'countdown': 100 - } - self.assertEqual(processor.get_task_args(), task_args) - - start.assert_called_once_with() - - def test_name_passed_in(self): - """Ensure that if name is set that it's passed as a task_arg to the - MessageProcessor. - """ - from furious.batcher import insert_messsage_processor - from furious.batcher import MessageProcessor - - with patch.object(MessageProcessor, 'start') as start: - processor = insert_messsage_processor( - 'job', None, None, 'queue-name', name='name') - - task_args = { - 'name': 'name' - } - self.assertEqual(processor.get_task_args(), task_args) - - start.assert_called_once_with() - - def test_tag_passed_in(self): - """Ensure that if tag is set that it's set on the MessageProcessor.""" - - from furious.batcher import insert_messsage_processor - from furious.batcher import MessageProcessor - - with patch.object(MessageProcessor, 'start') as start: - processor = insert_messsage_processor( - 'job', None, None, 'queue-name', tag='tag') - - self.assertEqual(processor.tag, 'tag') - - start.assert_called_once_with() From 28b286da36cba986d1c84a0a25c44d056e6f78f2 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 00:43:28 -0700 Subject: [PATCH 013/441] Add bump_batch method for incremeting the batch count. Increments memcahce buy a key. --- furious/batcher.py | 15 +++++++++------ furious/tests/test_batcher.py | 16 ++++++++++++++++ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index 173eb7c..61db28f 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -7,6 +7,7 @@ MESSAGE_DEFAULT_QUEUE = 'default_pull' MESSAGE_PROCESSOR_NAME = 'processor' +MESSAGE_BATCH_NAME = 'agg-batch' METHOD_TYPE = 'PULL' @@ -121,7 +122,7 @@ def to_task(self): @property def group_key(self): """Return the :class: `str` group key based off of the tag.""" - return 'agg-batch-%s' % (self.tag) + return "%s-%s" % (MESSAGE_BATCH_NAME, self.tag) @property def current_batch(self): @@ -148,9 +149,11 @@ def time_throttle(self): return int(time.time() / max(1, self.frequency)) -def fetch_messages(): - pass +def bump_batch(work_group): + """Return the incremented batch id for the work group + :param work_group: :class: `str` - -def bump_batch(): - pass + :return: :class: `int` current batch id. + """ + key = "%s-%s" % (MESSAGE_BATCH_NAME, work_group) + return memcache.incr(key) diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 1427933..cf52123 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -353,3 +353,19 @@ def test_curent_batch_key_doesnt_exist_in_cache(self, cache): cache.get.assert_called_once_with('agg-batch-processor') cache.add.assert_called_once_with('agg-batch-processor', 1) + + +class BumpBatchTestCase(unittest.TestCase): + + @patch('furious.batcher.memcache') + def test_cache_incremented_by_key(self, cache): + """Ensure that the cache object is incremented by the key passed in.""" + from furious.batcher import bump_batch + + cache.incr.return_value = 2 + + val = bump_batch('group') + + self.assertEqual(val, 2) + + cache.incr.assert_called_once_with('agg-batch-group') From c254d919eb2bb30a87829d04320e56f65084130d Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Thu, 17 Jan 2013 00:52:53 -0700 Subject: [PATCH 014/441] Convert Context to take options kwargs. To facilitate future extensions, switch from named insert_tasks injection parameter to fetching it from options. This will keep the API cleaner as additional injection points are added. --- furious/context/context.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/furious/context/context.py b/furious/context/context.py index d5de125..7058bfb 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -55,11 +55,12 @@ class Context(object): NOTE: Use the module's new function to get a context, do not manually instantiate. """ - def __init__(self, insert_tasks=None): + def __init__(self, **options): self._tasks = [] - self.insert_tasks = insert_tasks or _insert_tasks + insert_tasks = options.get('insert_tasks') + self._insert_tasks = insert_tasks or _insert_tasks self._tasks_inserted = False def __enter__(self): @@ -79,7 +80,7 @@ def _handle_tasks(self): task_map = self._get_tasks_by_queue() for queue, tasks in task_map.iteritems(): - self.insert_tasks(tasks, queue=queue) + self._insert_tasks(tasks, queue=queue) self._tasks_inserted = True From e1c497dc18ef5ed7389ea20074d40bd3bbdb5cc0 Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Thu, 17 Jan 2013 01:01:42 -0700 Subject: [PATCH 015/441] Add `id` to Context. `id` will be used to group asyncs withing the Context and also to facilitate access to the Context's state once persistence is implemented. --- furious/context/context.py | 12 +++++++++++- furious/tests/context/test_context.py | 12 ++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/furious/context/context.py b/furious/context/context.py index 7058bfb..8ef63da 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -42,6 +42,7 @@ """ +import uuid class ContextAlreadyStartedError(Exception): """Attempt to set context on an Async that is already executing in a @@ -58,10 +59,19 @@ class Context(object): def __init__(self, **options): self._tasks = [] + id = options.get('id') + if not id: + id = uuid.uuid4().hex + self._tasks_inserted = False + self._id = id insert_tasks = options.get('insert_tasks') self._insert_tasks = insert_tasks or _insert_tasks - self._tasks_inserted = False + + + @property + def id(self): + return self._id def __enter__(self): return self diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index 3388a69..5abc382 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -63,6 +63,18 @@ def test_context_works(self): with Context(): pass + def test_context_gets_id(self): + """Ensure a new Context gets an id generated.""" + from furious.context import Context + + self.assertTrue(Context().id) + + def test_context_gets_assigned_id(self): + """Ensure a new Context keeps its assigned id.""" + from furious.context import Context + + self.assertEqual('test_id_weee', Context(id='test_id_weee').id) + @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 636968d6399fc781ff1df7f293df0ef4d8edaf47 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 01:32:44 -0700 Subject: [PATCH 016/441] Add MessageIterator for pulling work off a pull queue. Add a MessageIterator which pulls messages off a specified pull queue in specified batch amounts. It exposes an iterator to go over the messages and send out the payload from the messages --- furious/batcher.py | 128 +++++++++++++++++++++++++++ furious/tests/test_batcher.py | 157 +++++++++++++++++++++++++++++++++- 2 files changed, 284 insertions(+), 1 deletion(-) diff --git a/furious/batcher.py b/furious/batcher.py index 61db28f..ad5e51d 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -1,4 +1,5 @@ import json +import logging import time from google.appengine.api import memcache @@ -149,6 +150,117 @@ def time_throttle(self): return int(time.time() / max(1, self.frequency)) +class MessageIterator(object): + """This iterator will return a batch of messages for a given group. + + This iterator should be directly used when trying to avoid the lease + operation inside a transaction, or when other flows are needed. + """ + + def __init__(self, tag, queue_name, size, duration=60, auto_delete=True): + """The generator will yield json deserialized payloads from tasks with + the corresponding tag. + + :param tag: :class: `str` Pull queue tag to query against + :param queue_name: :class: `str` Name of PULL queue holding tasks to + lease. + :param size: :class: `int` The number of items to pull at once + :param duration: :class: `int` After this time, the tasks may be leased + again. Tracked in seconds + :param auto_delete: :class: `bool` Delete tasks when iteration is + complete. + + :return: :class: `iterator` of json deserialized payloads + """ + from google.appengine.api.taskqueue import Queue + + self.queue_name = queue_name + self.queue = Queue(name=self.queue_name) + + self.tag = tag + self.size = size + self.duration = duration + self.auto_delete = auto_delete + + self._messages = [] + self._processed_messages = [] + self._fetched = False + + def fetch_messages(self): + """Fetch messages from the specified pull-queue. + + This should only be called a single time by a given MessageIterator + object. If the MessageIterator is iterated over again, it should + return the originally leased messages. + """ + if self._fetched: + return + + loaded_messages = self.queue.lease_tasks_by_tag( + self.duration, self.size, tag=self.tag) + + self._messages.extend(loaded_messages) + + self._fetched = True + + logging.debug("Calling fetch messages with %s:%s:%s:%s:%s:%s" % ( + len(self._messages), len(loaded_messages), + len(self._processed_messages), self.duration, self.size, self.tag)) + + def __iter__(self): + """Initialize this MessageIterator for iteration. + + If messages have not been fetched, fetch them. If messages have been + fetched, reset self._messages and self._processed_messages for + re-iteration. The reset is done to prevent deleting messages that were + never applied. + """ + if self._processed_messages: + # If the iterator is used within a transaction, and there is a + # retry we need to re-process the original messages, not new + # messages. + self._messages = list( + set(self._messages) | set(self._processed_messages)) + self._processed_messages = [] + + if not self._messages: + self.fetch_messages() + + return self + + def next(self): + """Get the next batch of messages from the previously fetched messages. + + If there's no more messages, check if we should auto-delete the + messages and raise StopIteration. + """ + if not self._messages: + if self.auto_delete: + self.delete_messages() + raise StopIteration + + message = self._messages.pop(0) + self._processed_messages.append(message) + return json.loads(message.payload) + + def delete_messages(self, only_processed=True): + """Delete the messages previously leased. + + Unless otherwise directed, only the messages iterated over will be + deleted. + """ + messages = self._processed_messages + if not only_processed: + messages += self._messages + + if messages: + try: + self.queue.delete_tasks(messages) + except Exception: + logging.exception("Error deleting messages") + raise + + def bump_batch(work_group): """Return the incremented batch id for the work group :param work_group: :class: `str` @@ -157,3 +269,19 @@ def bump_batch(work_group): """ key = "%s-%s" % (MESSAGE_BATCH_NAME, work_group) return memcache.incr(key) + + +def get_messages(tag, queue_name, size, duration=60): + """Yield out the paylod from the task pulled from the pull queue by its + tag. + + :param tag: :class: `str` Pull queue tag to query against + :param queue_name: :class: `str` + :param size: :class: `int` The number of items to pull at once + :param duration: :class: `int` + + :return: iterator of json deserialized payloads + """ + iterator = MessageIterator(tag, queue_name, size, duration) + for message in iterator: + yield message diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index cf52123..2faf30f 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -1,7 +1,7 @@ import json import unittest -from mock import patch +from mock import Mock, patch class MessageTestCase(unittest.TestCase): @@ -369,3 +369,158 @@ def test_cache_incremented_by_key(self, cache): self.assertEqual(val, 2) cache.incr.assert_called_once_with('agg-batch-group') + + +class MessageIteratorTestCase(unittest.TestCase): + + def test_raise_stopiteration_if_no_messages(self): + """Ensure MessageIterator raises StopIteration if no messages.""" + from furious.batcher import MessageIterator + + iterator = MessageIterator('tag', 'qn', 1) + with patch.object(iterator, 'queue') as queue: + queue.lease_tasks_by_tag.return_value = [] + + self.assertRaises(StopIteration, iterator.next) + + def test_iterates(self): + """Ensure MessageIterator instances iterate in loop.""" + from furious.batcher import MessageIterator + + payload = '["test"]' + task = Mock(payload=payload, tag='tag') + task1 = Mock(payload=payload, tag='tag') + task2 = Mock(payload=payload, tag='tag') + + iterator = MessageIterator('tag', 'qn', 1) + + with patch.object(iterator, 'queue') as queue: + queue.lease_tasks_by_tag.return_value = [task, task1, task2] + + results = [payload for payload in iterator] + + self.assertEqual(results, [payload, payload, payload]) + + def test_calls_lease_exactly_once(self): + """Ensure MessageIterator calls lease only once.""" + from furious.batcher import MessageIterator + + payload = '["test"]' + task = Mock(payload=payload, tag='tag') + + message_iterator = MessageIterator('tag', 'qn', 1) + + with patch.object(message_iterator, 'queue') as queue: + queue.lease_tasks_by_tag.return_value = [task] + + iterator = iter(message_iterator) + iterator.next() + + self.assertRaises(StopIteration, iterator.next) + self.assertRaises(StopIteration, iterator.next) + + queue.lease_tasks_by_tag.assert_called_once_with( + 60, 1, tag='tag') + + def test_rerun_after_depletion_calls_once(self): + """Ensure MessageIterator works when used manually.""" + from furious.batcher import MessageIterator + + payload = '["test"]' + task = Mock(payload=payload, tag='tag') + + iterator = MessageIterator('tag', 'qn', 1) + + with patch.object(iterator, 'queue') as queue: + queue.lease_tasks_by_tag.return_value = [task] + + results = [payload for payload in iterator] + self.assertEqual(results, [payload]) + + results = [payload for payload in iterator] + + queue.lease_tasks_by_tag.assert_called_once_with( + 60, 1, tag='tag') + + def test_rerun_after_depletion_doesnt_delete_too_much(self): + """Ensure MessageIterator works when used manually.""" + from furious.batcher import MessageIterator + + payload = '["test"]' + task = Mock(payload=payload, tag='tag') + + iterator = MessageIterator('tag', 'qn', 1, auto_delete=False) + + with patch.object(iterator, 'queue') as queue: + queue.lease_tasks_by_tag.return_value = [task] + + results = [payload for payload in iterator] + self.assertEqual(results, [payload]) + + # This new work should never be leased, but simulates new pending + # work. + task_1 = Mock(payload='["task_1"]', tag='tag') + queue.lease_tasks_by_tag.return_value = [task_1] + + # Iterating again should return the "originally leased" work, not + # new work. + results = [payload for payload in iterator] + self.assertEqual(results, [payload]) + + # Lease should only have been called a single time. + queue.lease_tasks_by_tag.assert_called_once_with( + 60, 1, tag='tag') + + # The delete call should delete only the original work. + iterator.delete_messages() + queue.delete_tasks.assert_called_once_with([task]) + + +class GetMessagesTestCase(unittest.TestCase): + + @patch('furious.batcher.MessageIterator', autospec=True) + def test_no_messages(self, iterator): + """Ensure that no payloads are returned if no messages are found.""" + from furious.batcher import get_messages + + iterator.return_value = iter([]) + + result = get_messages('t', 'qn', 1) + + from types import GeneratorType + self.assertIsInstance(result, GeneratorType) + + self.assertFalse(list(result)) + + @patch('furious.batcher.MessageIterator', autospec=True) + def test_has_a_single_message(self, iterator): + """Ensure that a single payload is returned if one message is found.""" + from furious.batcher import get_messages + + payload = ["payload"] + + iterator.return_value = iter([payload]) + + result = get_messages('t', 'qn', 1) + + from types import GeneratorType + self.assertIsInstance(result, GeneratorType) + + self.assertEqual([payload], list(result)) + + @patch('furious.batcher.MessageIterator', autospec=True) + def test_has_a_multiple_messages(self, iterator): + """Ensure that no messages are returned if none are found.""" + from furious.batcher import get_messages + + payload = ["payload"] + payload1 = ["payload"] + + iterator.return_value = iter([payload, payload1]) + + result = get_messages('t', 'qn', 1) + + from types import GeneratorType + self.assertIsInstance(result, GeneratorType) + + self.assertEqual([payload, payload1], list(result)) From f0e15a48242e23ef78cec5427d0c61c028553cf0 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 17 Jan 2013 02:36:08 -0600 Subject: [PATCH 017/441] removing further reference to _executing_async in get_current_context --- furious/context/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/furious/context/__init__.py b/furious/context/__init__.py index 9d5b233..6b1e8ab 100644 --- a/furious/context/__init__.py +++ b/furious/context/__init__.py @@ -86,7 +86,7 @@ def get_current_context(): """ local_context = _local.get_local_context() - if local_context._executing_async: + if local_context: return local_context.registry[-1] raise NotInContextError('Not in an _ExecutionContext.') From 2419cb4e10ce659f5992055c240cb613ec80e894 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 01:40:59 -0700 Subject: [PATCH 018/441] Remove get_messages call as the MessageIterator does what's needed. get_messages is no longer needed. --- furious/batcher.py | 16 ----------- furious/tests/test_batcher.py | 50 ----------------------------------- 2 files changed, 66 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index ad5e51d..fcf4862 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -269,19 +269,3 @@ def bump_batch(work_group): """ key = "%s-%s" % (MESSAGE_BATCH_NAME, work_group) return memcache.incr(key) - - -def get_messages(tag, queue_name, size, duration=60): - """Yield out the paylod from the task pulled from the pull queue by its - tag. - - :param tag: :class: `str` Pull queue tag to query against - :param queue_name: :class: `str` - :param size: :class: `int` The number of items to pull at once - :param duration: :class: `int` - - :return: iterator of json deserialized payloads - """ - iterator = MessageIterator(tag, queue_name, size, duration) - for message in iterator: - yield message diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 2faf30f..7b00c90 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -474,53 +474,3 @@ def test_rerun_after_depletion_doesnt_delete_too_much(self): # The delete call should delete only the original work. iterator.delete_messages() queue.delete_tasks.assert_called_once_with([task]) - - -class GetMessagesTestCase(unittest.TestCase): - - @patch('furious.batcher.MessageIterator', autospec=True) - def test_no_messages(self, iterator): - """Ensure that no payloads are returned if no messages are found.""" - from furious.batcher import get_messages - - iterator.return_value = iter([]) - - result = get_messages('t', 'qn', 1) - - from types import GeneratorType - self.assertIsInstance(result, GeneratorType) - - self.assertFalse(list(result)) - - @patch('furious.batcher.MessageIterator', autospec=True) - def test_has_a_single_message(self, iterator): - """Ensure that a single payload is returned if one message is found.""" - from furious.batcher import get_messages - - payload = ["payload"] - - iterator.return_value = iter([payload]) - - result = get_messages('t', 'qn', 1) - - from types import GeneratorType - self.assertIsInstance(result, GeneratorType) - - self.assertEqual([payload], list(result)) - - @patch('furious.batcher.MessageIterator', autospec=True) - def test_has_a_multiple_messages(self, iterator): - """Ensure that no messages are returned if none are found.""" - from furious.batcher import get_messages - - payload = ["payload"] - payload1 = ["payload"] - - iterator.return_value = iter([payload, payload1]) - - result = get_messages('t', 'qn', 1) - - from types import GeneratorType - self.assertIsInstance(result, GeneratorType) - - self.assertEqual([payload, payload1], list(result)) From 1eaac54ec4e80f1867fa0d54a789151c454efb2e Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 17 Jan 2013 09:51:49 -0600 Subject: [PATCH 019/441] Fixed typos in the comments and the Exception. Also fixed if condition to check registry --- furious/context/__init__.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/furious/context/__init__.py b/furious/context/__init__.py index 6b1e8ab..6c2e201 100644 --- a/furious/context/__init__.py +++ b/furious/context/__init__.py @@ -82,12 +82,12 @@ def get_current_async(): def get_current_context(): - """Return a reference to the currently Context object. + """Return a reference to the current Context object. """ local_context = _local.get_local_context() - if local_context: + if local_context.registry: return local_context.registry[-1] - raise NotInContextError('Not in an _ExecutionContext.') + raise NotInContextError('Not in a Context.') From 8b7a0a5801057020d14f6f38ef9a5ee5fd4f28de Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 10:32:04 -0700 Subject: [PATCH 020/441] Add an async_intro module with the basic intro async example. --- example/async_intro.py | 50 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 example/async_intro.py diff --git a/example/async_intro.py b/example/async_intro.py new file mode 100644 index 0000000..a48a737 --- /dev/null +++ b/example/async_intro.py @@ -0,0 +1,50 @@ +# +# 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. +# + +"""A very basic async example. + +This will insert an async task to run the example_function. +""" + +import logging + +import webapp2 + + +class AsyncIntroHandler(webapp2.RequestHandler): + """Demonstrate the creation and insertion of a single furious task.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object. + async_task = Async( + target=example_function, args=[1], kwargs={'some': 'value'}) + + # Insert the task to run the Async object, note that it may begin + # executing immediately or with some delay. + async_task.start() + + logging.info('Async job kicked off.') + + self.response.out.write('Successfully inserted Async job.') + + +def example_function(*args, **kwargs): + """This function is called by furious tasks to demonstrate usage.""" + logging.info('example_function executed with args: %r, kwargs: %r', + args, kwargs) + + return args From 558995e3199a5ea94d2f0142869c0998c5b74c6d Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 10:53:45 -0700 Subject: [PATCH 021/441] Add a basic context_intro module to examples. --- example/context_intro.py | 63 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 example/context_intro.py diff --git a/example/context_intro.py b/example/context_intro.py new file mode 100644 index 0000000..db4c80e --- /dev/null +++ b/example/context_intro.py @@ -0,0 +1,63 @@ +# +# 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. +# + +"""A very basic async context example. + +This example will create a context that adds an Async manually to the context. +It then adds 5 Async jobs through the shorthand style. +""" + + +import logging + +import webapp2 + + +class ContextIntroHandler(webapp2.RequestHandler): + """Demonstrate using a Context to batch insert a group of furious tasks.""" + def get(self): + from furious.async import Async + from furious import context + + # Create a new furious Context. + with context.new() as ctx: + # "Manually" instantiate and add an Async object to the Context. + async_task = Async( + target=example_function, kwargs={'first': 'async'}) + ctx.add(async_task) + logging.info('Added manual job to context.') + + # Use the shorthand style, note that add returns the Async object. + for i in xrange(5): + ctx.add(target=example_function, args=[i]) + logging.info('Added job %d to context.', i) + + # When the Context is exited, the tasks are inserted (if there are no + # errors). + + logging.info('Async jobs for context batch inserted.') + + self.response.out.write('Successfully inserted a group of Async jobs.') + + +def example_function(*args, **kwargs): + """This function is called by furious tasks to demonstrate usage.""" + logging.info('example_function executed with args: %r, kwargs: %r', + args, kwargs) + + return args + + From cd8e0c1ef9aabd9b5203050a3e0142a99d0b4065 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 10:54:27 -0700 Subject: [PATCH 022/441] Add a callback module to exmamples for showing the different callback styles This has 3 examples of the different ways to wire up callbacks. --- example/callback.py | 139 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 139 insertions(+) create mode 100644 example/callback.py diff --git a/example/callback.py b/example/callback.py new file mode 100644 index 0000000..7ac56cb --- /dev/null +++ b/example/callback.py @@ -0,0 +1,139 @@ +# +# 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. +# + +"""Async Callback Examples. + +There are 3 examples below. + +1. AsyncCallbackHandler +Registering a function as a callback to be triggered when the job has +completed. + +2. AsyncErrorCallbackHandler +Registering a function as an error callback to be triggered when an error has +been hit in the Async process. + +3. AsyncAsyncCallbackHandler +Registering another Async object as a callback to be inserted when the job has +complated. +""" + + +import logging + +import webapp2 + + +class AsyncCallbackHandler(webapp2.RequestHandler): + """Demonstrate setting an Async callback.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object, specifying a 'success' callback. + async_task = Async( + target=example_function, args=[1], kwargs={'some': 'value'}, + callbacks={'success': all_done} + ) + + # Insert the task to run the Async object. The success callback will + # be executed in the furious task after the job is executed. + async_task.start() + + logging.info('Async job kicked off.') + + self.response.out.write('Successfully inserted Async job.') + + +class AsyncErrorCallbackHandler(webapp2.RequestHandler): + """Demonstrate handling an error using an Async callback.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object, specifying a 'error' callback. + async_task = Async( + target=dir, args=[1, 2, 3], + callbacks={'error': handle_an_error} + ) + + # Insert the task to run the Async object. The error callback will be + # executed in the furious task after the job has raised an exception. + async_task.start() + + logging.info('Erroneous Async job kicked off.') + + self.response.out.write('Successfully inserted Async job.') + + +class AsyncAsyncCallbackHandler(webapp2.RequestHandler): + """Demonstrate using an Async as a callback for another Async.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object to act as our success callback. + # NOTE: Your async.result is not directly available from the + # success_callback Async, you will need to persist the result + # and fetch it from the other Async if needed. + success_callback = Async( + target=example_function, kwargs={'it': 'worked'} + ) + + # Instantiate an Async object, setting the success_callback to the + # above Async object. + async_task = Async( + target=example_function, kwargs={'trigger': 'job'}, + callbacks={'success': success_callback} + ) + + # Insert the task to run the Async object. + async_task.start() + + logging.info('Async job kicked off.') + + self.response.out.write('Successfully inserted Async job.') + + +def example_function(*args, **kwargs): + """This function is called by furious tasks to demonstrate usage.""" + logging.info('example_function executed with args: %r, kwargs: %r', + args, kwargs) + + return args + + +def all_done(): + """Will be run if the async task runs successfully.""" + from furious.context import get_current_async + + async = get_current_async() + + logging.info('async task complete, value returned: %r', async.result) + + +def handle_an_error(): + """Will be run if the async task raises an unhandled exception.""" + import os + + from furious.context import get_current_async + + exception_info = get_current_async().result + + logging.info('async job blew up, exception info: %r', exception_info) + + retries = int(os.environ['HTTP_X_APPENGINE_TASKRETRYCOUNT']) + if retries < 2: + raise exception_info.exception + else: + logging.info('Caught too many errors, giving up now.') From f0f39a94671ac34ce958c018651008272bb37300 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 11:02:39 -0700 Subject: [PATCH 023/441] Add a simple_workflow module to examples. This example shows a simple flow of chaining Asyncs together. --- example/simple_workflow.py | 57 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 example/simple_workflow.py diff --git a/example/simple_workflow.py b/example/simple_workflow.py new file mode 100644 index 0000000..d6f92ac --- /dev/null +++ b/example/simple_workflow.py @@ -0,0 +1,57 @@ +# +# 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. +# + +"""A Simple Workflow Example. + +Inserts a Async targetting the simple_state_machine function. This function +does a random number generation. If that number falls above a certain range +it will return another Aysnc pointed at the same simple_state_machine function. +If the simple_state_machine random falls below the range it will return and +complete the simple workflow. +""" + +import logging + +import webapp2 + + +class SimpleWorkflowHandler(webapp2.RequestHandler): + """Demonstrate constructing a simple state machine.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object to start the chain. + Async(target=simple_state_machine).start() + + logging.info('Async chain kicked off.') + + self.response.out.write('Successfully inserted Async chain starter.') + + +def simple_state_machine(): + """Pick a number, if it is more than some cuttoff continue the chain.""" + from random import random + + from furious.async import Async + + number = random() + logging.info('Generating a number... %s', number) + + if number > 0.25: + logging.info('Continuing to do stuff.') + return Async(target=simple_state_machine) + + return number From 22b271a78bd1497a448d8232e24f8af7bffe7d93 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 11:18:21 -0700 Subject: [PATCH 024/441] Add complex_workflow module to the examples. --- example/complex_workflow.py | 93 +++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) create mode 100644 example/complex_workflow.py diff --git a/example/complex_workflow.py b/example/complex_workflow.py new file mode 100644 index 0000000..95cc38f --- /dev/null +++ b/example/complex_workflow.py @@ -0,0 +1,93 @@ +# +# 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. +# + +"""Contained within this module are several working examples showing basic +usage and complex context-based chaining. + +The examples demonstrate basic task execution, and also the basics of creating +more complicated processing pipelines. +""" + +import logging + +import webapp2 + +from furious.async import defaults + + +class ComplexWorkflowHandler(webapp2.RequestHandler): + """Demonstrate constructing a more complex state machine.""" + def get(self): + from furious.async import Async + + # Instantiate an Async object to start the machine in state alpha. + Async(target=complex_state_generator_alpha).start() + + logging.info('Async chain kicked off.') + + self.response.out.write('Successfully inserted Async chain starter.') + + +@defaults( + callbacks={'success': "example.complex_workflow.state_machine_success"}) +def complex_state_generator_alpha(last_state=''): + """Pick a state.""" + from random import choice + + states = ['ALPHA', 'ALPHA', 'ALPHA', 'BRAVO', 'BRAVO', 'DONE'] + if last_state: + states.remove(last_state) # Slightly lower chances of previous state. + + state = choice(states) + + logging.info('Generating a state... %s', state) + + return state + + +def state_machine_success(): + """A positive result! Iterate!""" + from furious.async import Async + from furious.context import get_current_async + + result = get_current_async().result + + if result == 'ALPHA': + logging.info('Inserting continuation for state %s.', result) + return Async(target=complex_state_generator_alpha, args=[result]) + + elif result == 'BRAVO': + logging.info('Inserting continuation for state %s.', result) + return Async(target=complex_state_generator_bravo, args=[result]) + + logging.info('Done working, stop now.') + + +@defaults( + callbacks={'success': "example.complex_workflow.state_machine_success"}) +def complex_state_generator_bravo(last_state=''): + """Pick a state.""" + from random import choice + + states = ['ALPHA', 'BRAVO', 'BRAVO', 'DONE'] + if last_state: + states.remove(last_state) # Slightly lower chances of previous state. + + state = choice(states) + + logging.info('Generating a state... %s', state) + + return state From 7675a7eff5e7be6eaa4e94c5e1468bb69387b328 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 11:19:13 -0700 Subject: [PATCH 025/441] Update the example app to point to the broken out example modules --- example/__init__.py | 250 ++------------------------------------- example/context_intro.py | 2 - 2 files changed, 7 insertions(+), 245 deletions(-) diff --git a/example/__init__.py b/example/__init__.py index ac5296d..101e96b 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -21,250 +21,15 @@ more complicated processing pipelines. """ -import logging - import webapp2 -from furious.async import defaults - - -class AsyncIntroHandler(webapp2.RequestHandler): - """Demonstrate the creation and insertion of a single furious task.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object. - async_task = Async( - target=example_function, args=[1], kwargs={'some': 'value'}) - - # Insert the task to run the Async object, note that it may begin - # executing immediately or with some delay. - async_task.start() - - logging.info('Async job kicked off.') - - self.response.out.write('Successfully inserted Async job.') - - -class ContextIntroHandler(webapp2.RequestHandler): - """Demonstrate using a Context to batch insert a group of furious tasks.""" - def get(self): - from furious.async import Async - from furious import context - - # Create a new furious Context. - with context.new() as ctx: - # "Manually" instantiate and add an Async object to the Context. - async_task = Async( - target=example_function, kwargs={'first': 'async'}) - ctx.add(async_task) - logging.info('Added manual job to context.') - - # Use the shorthand style, note that add returns the Async object. - for i in xrange(5): - ctx.add(target=example_function, args=[i]) - logging.info('Added job %d to context.', i) - - # When the Context is exited, the tasks are inserted (if there are no - # errors). - - logging.info('Async jobs for context batch inserted.') - - self.response.out.write('Successfully inserted a group of Async jobs.') - - -class AsyncCallbackHandler(webapp2.RequestHandler): - """Demonstrate setting an Async callback.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object, specifying a 'success' callback. - async_task = Async( - target=example_function, args=[1], kwargs={'some': 'value'}, - callbacks={'success': all_done} - ) - - # Insert the task to run the Async object. The success callback will - # be executed in the furious task after the job is executed. - async_task.start() - - logging.info('Async job kicked off.') - - self.response.out.write('Successfully inserted Async job.') - - -class AsyncErrorCallbackHandler(webapp2.RequestHandler): - """Demonstrate handling an error using an Async callback.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object, specifying a 'error' callback. - async_task = Async( - target=dir, args=[1, 2, 3], - callbacks={'error': handle_an_error} - ) - - # Insert the task to run the Async object. The error callback will be - # executed in the furious task after the job has raised an exception. - async_task.start() - - logging.info('Erroneous Async job kicked off.') - - self.response.out.write('Successfully inserted Async job.') - - -class AsyncAsyncCallbackHandler(webapp2.RequestHandler): - """Demonstrate using an Async as a callback for another Async.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object to act as our success callback. - # NOTE: Your async.result is not directly available from the - # success_callback Async, you will need to persist the result - # and fetch it from the other Async if needed. - success_callback = Async( - target=example_function, kwargs={'it': 'worked'} - ) - - # Instantiate an Async object, setting the success_callback to the - # above Async object. - async_task = Async( - target=example_function, kwargs={'trigger': 'job'}, - callbacks={'success': success_callback} - ) - - # Insert the task to run the Async object. - async_task.start() - - logging.info('Async job kicked off.') - - self.response.out.write('Successfully inserted Async job.') - - -class SimpleWorkflowHandler(webapp2.RequestHandler): - """Demonstrate constructing a simple state machine.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object to start the chain. - Async(target=simple_state_machine).start() - - logging.info('Async chain kicked off.') - - self.response.out.write('Successfully inserted Async chain starter.') - - -class ComplexWorkflowHandler(webapp2.RequestHandler): - """Demonstrate constructing a more complex state machine.""" - def get(self): - from furious.async import Async - - # Instantiate an Async object to start the machine in state alpha. - Async(target=complex_state_generator_alpha).start() - - logging.info('Async chain kicked off.') - - self.response.out.write('Successfully inserted Async chain starter.') - - -def example_function(*args, **kwargs): - """This function is called by furious tasks to demonstrate usage.""" - logging.info('example_function executed with args: %r, kwargs: %r', - args, kwargs) - - return args - - -def all_done(): - """Will be run if the async task runs successfully.""" - from furious.context import get_current_async - - async = get_current_async() - - logging.info('async task complete, value returned: %r', async.result) - - -def handle_an_error(): - """Will be run if the async task raises an unhandled exception.""" - import os - - from furious.context import get_current_async - - exception_info = get_current_async().result - - logging.info('async job blew up, exception info: %r', exception_info) - - retries = int(os.environ['HTTP_X_APPENGINE_TASKRETRYCOUNT']) - if retries < 2: - raise exception_info.exception - else: - logging.info('Caught too many errors, giving up now.') - - -def simple_state_machine(): - """Pick a number, if it is more than some cuttoff continue the chain.""" - from random import random - - from furious.async import Async - - number = random() - logging.info('Generating a number... %s', number) - - if number > 0.25: - logging.info('Continuing to do stuff.') - return Async(target=simple_state_machine) - - return number - - -@defaults(callbacks={'success': "example.state_machine_success"}) -def complex_state_generator_alpha(last_state=''): - """Pick a state.""" - from random import choice - - states = ['ALPHA', 'ALPHA', 'ALPHA', 'BRAVO', 'BRAVO', 'DONE'] - if last_state: - states.remove(last_state) # Slightly lower chances of previous state. - - state = choice(states) - - logging.info('Generating a state... %s', state) - - return state - - -@defaults(callbacks={'success': "example.state_machine_success"}) -def complex_state_generator_bravo(last_state=''): - """Pick a state.""" - from random import choice - - states = ['ALPHA', 'BRAVO', 'BRAVO', 'DONE'] - if last_state: - states.remove(last_state) # Slightly lower chances of previous state. - - state = choice(states) - - logging.info('Generating a state... %s', state) - - return state - - -def state_machine_success(): - """A positive result! Iterate!""" - from furious.async import Async - from furious.context import get_current_async - - result = get_current_async().result - - if result == 'ALPHA': - logging.info('Inserting continuation for state %s.', result) - return Async(target=complex_state_generator_alpha, args=[result]) - - elif result == 'BRAVO': - logging.info('Inserting continuation for state %s.', result) - return Async(target=complex_state_generator_bravo, args=[result]) - - logging.info('Done working, stop now.') +from .async_intro import AsyncIntroHandler +from .context_intro import ContextIntroHandler +from .callback import AsyncCallbackHandler +from .callback import AsyncErrorCallbackHandler +from .callback import AsyncAsyncCallbackHandler +from .simple_workflow import SimpleWorkflowHandler +from .complex_workflow import ComplexWorkflowHandler app = webapp2.WSGIApplication([ @@ -276,4 +41,3 @@ def state_machine_success(): ('/workflow', SimpleWorkflowHandler), ('/workflow/complex', ComplexWorkflowHandler), ]) - diff --git a/example/context_intro.py b/example/context_intro.py index db4c80e..3ab0d08 100644 --- a/example/context_intro.py +++ b/example/context_intro.py @@ -59,5 +59,3 @@ def example_function(*args, **kwargs): args, kwargs) return args - - From e257ecc7ef8dfc88c9d500c8c586e67c8fb425e4 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 11:23:34 -0700 Subject: [PATCH 026/441] Cleanup the complex workflow example. --- example/complex_workflow.py | 40 ++++++++++++++++++------------------- 1 file changed, 19 insertions(+), 21 deletions(-) diff --git a/example/complex_workflow.py b/example/complex_workflow.py index 95cc38f..281e356 100644 --- a/example/complex_workflow.py +++ b/example/complex_workflow.py @@ -14,11 +14,9 @@ # limitations under the License. # -"""Contained within this module are several working examples showing basic -usage and complex context-based chaining. +"""A complex workflow example. -The examples demonstrate basic task execution, and also the basics of creating -more complicated processing pipelines. +This example uses a couple different functions to chain together in a worklow. """ import logging @@ -58,6 +56,23 @@ def complex_state_generator_alpha(last_state=''): return state +@defaults( + callbacks={'success': "example.complex_workflow.state_machine_success"}) +def complex_state_generator_bravo(last_state=''): + """Pick a state.""" + from random import choice + + states = ['ALPHA', 'BRAVO', 'BRAVO', 'DONE'] + if last_state: + states.remove(last_state) # Slightly lower chances of previous state. + + state = choice(states) + + logging.info('Generating a state... %s', state) + + return state + + def state_machine_success(): """A positive result! Iterate!""" from furious.async import Async @@ -74,20 +89,3 @@ def state_machine_success(): return Async(target=complex_state_generator_bravo, args=[result]) logging.info('Done working, stop now.') - - -@defaults( - callbacks={'success': "example.complex_workflow.state_machine_success"}) -def complex_state_generator_bravo(last_state=''): - """Pick a state.""" - from random import choice - - states = ['ALPHA', 'BRAVO', 'BRAVO', 'DONE'] - if last_state: - states.remove(last_state) # Slightly lower chances of previous state. - - state = choice(states) - - logging.info('Generating a state... %s', state) - - return state From 998fb8a7407837250006a74a453e878f3716a957 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 17 Jan 2013 17:08:44 -0600 Subject: [PATCH 027/441] changing import style in __init__.py --- example/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/example/__init__.py b/example/__init__.py index 9dae1db..3ccc844 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -25,7 +25,7 @@ import webapp2 -from . import grep +from .grep import GrepHandler from furious.async import defaults From 580c0e65c841fe88fe8c316da7044a8804293e3c Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 16:16:09 -0700 Subject: [PATCH 028/441] Add an example runner for easily running the examples. The script takes in the url to run and optionally the sdk path. Right now this only works for localhost. More robustness for hitting GAE appspots coming later. --- example/runner.py | 51 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 example/runner.py diff --git a/example/runner.py b/example/runner.py new file mode 100644 index 0000000..e2c50eb --- /dev/null +++ b/example/runner.py @@ -0,0 +1,51 @@ +#!/usr/bin/python + +import argparse +import sys + + +def args(): + parser = argparse.ArgumentParser(description='Run the Furious Examples.') + + parser.add_argument('--gae-sdk-path', metavar='S', dest="gae_lib_path", + default="/usr/local/google_appengine", + help='path to the GAE SDK') + + parser.add_argument('--url', metavar='U', dest="url", default="", + help="the endpoint to run") + + return parser.parse_args() + + +def setup(options): + sys.path.insert(0, options.gae_lib_path) + + from dev_appserver import fix_sys_path + fix_sys_path() + + +def run(options): + from google.appengine.tools import appengine_rpc + from google.appengine.tools import appcfg + + source = 'furious' + user_agent = appcfg.GetUserAgent() + server = appengine_rpc.HttpRpcServer( + 'localhost:8080', lambda: ('test@example.com', 'password'), user_agent, + source, secure=False) + + server._DevAppServerAuthenticate() + server.Send(options.url, content_type="text/html; charset=utf-8", + payload=None) + + +def main(): + options = args() + + setup(options) + + run(options) + + +if __name__ == "__main__": + main() From 4e31e5c776c40997cccd76d4ce592d7f3d5de752 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Thu, 17 Jan 2013 16:21:50 -0700 Subject: [PATCH 029/441] Update the way the url is handled. --- example/runner.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/example/runner.py b/example/runner.py index e2c50eb..68ceb85 100644 --- a/example/runner.py +++ b/example/runner.py @@ -11,7 +11,7 @@ def args(): default="/usr/local/google_appengine", help='path to the GAE SDK') - parser.add_argument('--url', metavar='U', dest="url", default="", + parser.add_argument('url', metavar='U', default="", nargs=1, help="the endpoint to run") return parser.parse_args() @@ -34,8 +34,12 @@ def run(options): 'localhost:8080', lambda: ('test@example.com', 'password'), user_agent, source, secure=False) + url = "/" + if options.url: + url += options.url[0] + server._DevAppServerAuthenticate() - server.Send(options.url, content_type="text/html; charset=utf-8", + server.Send(url, content_type="text/html; charset=utf-8", payload=None) From bcfa509f35b01506b7cb8dd9d8799449d0b28447 Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Thu, 17 Jan 2013 17:59:15 -0700 Subject: [PATCH 030/441] Alter Context _insert_task setting logic to validate function. Require a callable function to be provided for insert_tasks, or nothing at all. Specifying any non-callable value is considered an error. --- furious/context/context.py | 6 ++++-- furious/tests/context/test_context.py | 6 ++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/furious/context/context.py b/furious/context/context.py index 8ef63da..cd2bd99 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -65,8 +65,10 @@ def __init__(self, **options): self._tasks_inserted = False self._id = id - insert_tasks = options.get('insert_tasks') - self._insert_tasks = insert_tasks or _insert_tasks + + self._insert_tasks = options.pop('insert_tasks', _insert_tasks) + if not callable(self._insert_tasks): + raise TypeError('You must provide a valid insert_tasks function.') @property diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index 5abc382..cd09749 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -63,6 +63,12 @@ def test_context_works(self): with Context(): pass + def test_context_requires_insert_tasks(self): + """Ensure Contexts require a callable insert_tasks function.""" + from furious.context import Context + + self.assertRaises(TypeError, Context, insert_tasks='nope') + def test_context_gets_id(self): """Ensure a new Context gets an id generated.""" from furious.context import Context From 8558ee27bc18b6be36e9eee6d56ac1d62abf73de Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Thu, 17 Jan 2013 18:34:17 -0700 Subject: [PATCH 031/441] Adding to_dict method to Context to facilitate persistence. Context objects need to be encoded so that they can be passed between requests and / or persisted. A first step of this process is encoding Contexts to a JSON serializable format. --- furious/context/context.py | 31 +++++++++++++++ furious/tests/context/test_context.py | 56 +++++++++++++++++++++++++++ furious/tests/test_async.py | 3 -- 3 files changed, 87 insertions(+), 3 deletions(-) diff --git a/furious/context/context.py b/furious/context/context.py index cd2bd99..27ec21e 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -44,6 +44,10 @@ import uuid +from ..job_utils import encode_callbacks +from ..job_utils import get_function_path_and_options + + class ContextAlreadyStartedError(Exception): """Attempt to set context on an Async that is already executing in a context. @@ -65,11 +69,13 @@ def __init__(self, **options): self._tasks_inserted = False self._id = id + self._options = options self._insert_tasks = options.pop('insert_tasks', _insert_tasks) if not callable(self._insert_tasks): raise TypeError('You must provide a valid insert_tasks function.') + self._persistence_engine = options.pop('persistence_engine', None) @property def id(self): @@ -132,6 +138,31 @@ def start(self): self._handle_tasks() + def to_dict(self): + """Return this Context as a dict suitable for json encoding.""" + import copy + + options = copy.deepcopy(self._options) + + if self._insert_tasks: + options['insert_tasks'], _ = get_function_path_and_options( + self._insert_tasks) + + if self._persistence_engine: + options['persistence_engine'], _ = get_function_path_and_options( + self._persistence_engine) + + options.update({ + '_tasks_inserted': self._tasks_inserted, + '_task_ids': [async.id for async in self._tasks] + }) + + callbacks = self._options.get('callbacks') + if callbacks: + options['callbacks'] = encode_callbacks(callbacks) + + return options + def _insert_tasks(tasks, queue, transactional=False): """Insert a batch of tasks into the specified queue. If an error occurs during insertion, split the batch and retry until they are successfully diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index cd09749..710ebe6 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -175,6 +175,62 @@ def add(self, *args, **kwargs): self.assertEqual(1, len(queue_registry['B']._calls[0][0][0])) self.assertEqual(1, len(queue_registry['C']._calls[0][0][0])) + def test_to_dict(self): + """Ensure to_dict returns a dictionary representation of the Context. + """ + import copy + + from furious.context import Context + + options = { + 'persistence_engine': 'persistence_engine', + 'unkown': True, + } + + context = Context(**copy.deepcopy(options)) + + # This stuff gets dumped out by to_dict(). + options.update({ + 'insert_tasks': 'furious.context.context._insert_tasks', + '_tasks_inserted': False, + '_task_ids': [], + }) + + self.assertEqual(options, context.to_dict()) + + def test_to_dict_with_callbacks(self): + """Ensure to_dict correctly encodes callbacks.""" + import copy + + from furious.async import Async + from furious.context import Context + + options = { + 'persistence_engine': 'persistence_engine', + 'callbacks': { + 'success': self.__class__.test_to_dict_with_callbacks, + 'failure': "failure_function", + 'exec': Async(target=dir) + } + } + + context = Context(**copy.deepcopy(options)) + + # This stuff gets dumped out by to_dict(). + options.update({ + 'insert_tasks': 'furious.context.context._insert_tasks', + 'persistence_engine': 'persistence_engine', + '_tasks_inserted': False, + '_task_ids': [], + 'callbacks': { + 'success': ("furious.tests.context.test_context." + "TestContext.test_to_dict_with_callbacks"), + 'failure': "failure_function", + 'exec': {'job': ('dir', None, None)} + } + }) + + self.assertEqual(options, context.to_dict()) class TestInsertTasks(unittest.TestCase): """Test that _insert_tasks behaves as expected.""" diff --git a/furious/tests/test_async.py b/furious/tests/test_async.py index 11007c0..f9c7ef0 100644 --- a/furious/tests/test_async.py +++ b/furious/tests/test_async.py @@ -343,9 +343,6 @@ def test_to_dict_with_callbacks(self): """Ensure to_dict correctly encodes callbacks.""" from furious.async import Async - def success_function(): - pass - options = {'callbacks': { 'success': self.__class__.test_to_dict_with_callbacks, 'failure': "failure_function", From 3fd1dbfcc5bc050304e5ffb82d9047885d9a1bf0 Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Thu, 17 Jan 2013 19:39:35 -0700 Subject: [PATCH 032/441] Add from_dict method for Context to produce a Context from a dict. from_dict is the inverse function of to_dict on Context. This is required to facilitate the reconstitution of a Context that has been encoded and persisted to disk. --- furious/context/context.py | 35 ++++++++++++++ furious/tests/context/test_context.py | 66 +++++++++++++++++++++++++++ 2 files changed, 101 insertions(+) diff --git a/furious/context/context.py b/furious/context/context.py index 27ec21e..1fb113a 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -44,7 +44,9 @@ import uuid +from ..job_utils import decode_callbacks from ..job_utils import encode_callbacks +from ..job_utils import function_path_to_reference from ..job_utils import get_function_path_and_options @@ -163,6 +165,39 @@ def to_dict(self): return options + @classmethod + def from_dict(cls, context): + """Return a context job from a dict output by Context.to_dict.""" + import copy + + context_options = copy.deepcopy(context) + + tasks_inserted = context_options.pop('_tasks_inserted', False) + task_ids = context_options.pop('_task_ids', []) + + insert_tasks = context_options.pop('insert_tasks', None) + if insert_tasks: + context_options['insert_tasks'] = function_path_to_reference( + insert_tasks) + + persistence_engine = context_options.pop('persistence_engine', None) + if persistence_engine: + context_options['persistence_engine'] = function_path_to_reference( + persistence_engine) + + # If there are callbacks, reconstitute them. + callbacks = context_options.pop('callbacks', None) + if callbacks: + context_options['callbacks'] = decode_callbacks(callbacks) + + context = cls(**context_options) + + context._tasks_inserted = tasks_inserted + context._task_ids = task_ids + + return context + + def _insert_tasks(tasks, queue, transactional=False): """Insert a batch of tasks into the specified queue. If an error occurs during insertion, split the batch and retry until they are successfully diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index 710ebe6..15add70 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -232,6 +232,72 @@ def test_to_dict_with_callbacks(self): self.assertEqual(options, context.to_dict()) + def test_from_dict(self): + """Ensure from_dict returns the correct Context object.""" + from furious.context import Context + + from furious.context.context import _insert_tasks + + # TODO: persistence_engine needs set to a real persistence module. + + options = { + 'id': 123456, + 'insert_tasks': 'furious.context.context._insert_tasks', + 'random_option': 'avalue', + '_tasks_inserted': True, + '_task_ids': [1, 2, 3, 4], + 'persistence_engine': 'furious.context.Context' + } + + context = Context.from_dict(options) + + self.assertEqual(123456, context.id) + self.assertEqual([1, 2, 3, 4], context._task_ids) + self.assertEqual(True, context._tasks_inserted) + self.assertEqual('avalue', context._options.get('random_option')) + self.assertEqual(_insert_tasks, context._insert_tasks) + self.assertEqual(Context, context._persistence_engine) + + def test_from_dict_with_callbacks(self): + """Ensure from_dict reconstructs the Context callbacks correctly.""" + from furious.context import Context + + callbacks = { + 'success': ("furious.tests.context.test_context." + "TestContext.test_to_dict_with_callbacks"), + 'failure': "dir", + 'exec': {'job': ('id', None, None)} + } + + context = Context.from_dict({'callbacks': callbacks}) + + check_callbacks = { + 'success': TestContext.test_to_dict_with_callbacks, + 'failure': dir + } + + callbacks = context._options.get('callbacks') + exec_callback = callbacks.pop('exec') + + self.assertEqual(check_callbacks, callbacks) + self.assertEqual({'job': ('id', None, None)}, exec_callback.to_dict()) + + def test_reconstitution(self): + """Ensure to_dict(job.from_dict()) returns the same thing.""" + from furious.context import Context + + options = { + 'id': 123098, + 'insert_tasks': 'furious.context.context._insert_tasks', + 'persistence_engine': 'furious.job_utils.get_function_path_and_options', + '_tasks_inserted': True, + '_task_ids': [] + } + + context = Context.from_dict(options) + + self.assertEqual(options, context.to_dict()) + class TestInsertTasks(unittest.TestCase): """Test that _insert_tasks behaves as expected.""" def setUp(self): From 245cabedd448b21382082587fb77670295ca5519 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 00:58:20 -0700 Subject: [PATCH 033/441] Add basic error handling for start on Async. Switch the insert task logic in Async.start to match the context insert_tasks function by handling some standard queue exceptions. --- furious/async.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/furious/async.py b/furious/async.py index 48931c5..087fd55 100644 --- a/furious/async.py +++ b/furious/async.py @@ -223,10 +223,17 @@ def to_task(self): def start(self): """Insert the task into the requested queue, 'default' if non given.""" - from google.appengine.api.taskqueue import Queue + from google.appengine.api import taskqueue task = self.to_task() - Queue(name=self.get_queue()).add(task) + + try: + taskqueue.Queue(name=self.get_queue()).add(task) + except (taskqueue.TransientError, + taskqueue.TaskAlreadyExistsError, + taskqueue.TombstonedTaskError): + return + # TODO: Return a "result" object. def to_dict(self): From 2622cc84f77073e77c6e4d74cd1e6cc2784b3637 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 00:58:20 -0700 Subject: [PATCH 034/441] Add basic error handling for start on Async. Switch the insert task logic in Async.start to match the context insert_tasks function by handling some standard queue exceptions. --- furious/async.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/furious/async.py b/furious/async.py index 48931c5..087fd55 100644 --- a/furious/async.py +++ b/furious/async.py @@ -223,10 +223,17 @@ def to_task(self): def start(self): """Insert the task into the requested queue, 'default' if non given.""" - from google.appengine.api.taskqueue import Queue + from google.appengine.api import taskqueue task = self.to_task() - Queue(name=self.get_queue()).add(task) + + try: + taskqueue.Queue(name=self.get_queue()).add(task) + except (taskqueue.TransientError, + taskqueue.TaskAlreadyExistsError, + taskqueue.TombstonedTaskError): + return + # TODO: Return a "result" object. def to_dict(self): From 2f5447772b1d54e6b6ca4de173355b492b8e591c Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:01:48 -0700 Subject: [PATCH 035/441] Update batcher default queue to proper name. --- furious/batcher.py | 19 ++++++++++++++++++- furious/tests/test_batcher.py | 2 +- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/furious/batcher.py b/furious/batcher.py index fcf4862..1477dee 100644 --- a/furious/batcher.py +++ b/furious/batcher.py @@ -1,3 +1,19 @@ +# +# 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 json import logging import time @@ -6,7 +22,7 @@ from .async import Async -MESSAGE_DEFAULT_QUEUE = 'default_pull' +MESSAGE_DEFAULT_QUEUE = 'default-pull' MESSAGE_PROCESSOR_NAME = 'processor' MESSAGE_BATCH_NAME = 'agg-batch' METHOD_TYPE = 'PULL' @@ -61,6 +77,7 @@ def insert(self): from google.appengine.api.taskqueue import Queue task = self.to_task() + Queue(name=self.get_queue()).add(task) def to_dict(self): diff --git a/furious/tests/test_batcher.py b/furious/tests/test_batcher.py index 7b00c90..76210cd 100644 --- a/furious/tests/test_batcher.py +++ b/furious/tests/test_batcher.py @@ -69,7 +69,7 @@ def test_get_default_queue(self): message = Message() - self.assertEqual('default_pull', message.get_queue()) + self.assertEqual('default-pull', message.get_queue()) def test_get_task_args(self): """Ensure get_task_args returns the message task_args.""" From f7fc6fd7c62c0fd619a3622357e96fcfb90aa79c Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:02:46 -0700 Subject: [PATCH 036/441] Add batcher.Message to context.add check Add batcher.Message to the context.add isinstance check to allow batch inserts of Messages similar to Asyncs. --- furious/context/context.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/furious/context/context.py b/furious/context/context.py index d5de125..c21e680 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -101,12 +101,13 @@ def add(self, target, args=None, kwargs=None, **options): object as argumnets. Returns the newly added Async object. """ from ..async import Async + from ..batcher import Message if self._tasks_inserted: raise ContextAlreadyStartedError( "This Context has already had its tasks inserted.") - if not isinstance(target, Async): + if not isinstance(target, (Async, Message)): target = Async(target, args, kwargs, **options) self._tasks.append(target) From 99a43a6f4291cd1f73cda4ab915ee93074a77f9d Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:03:46 -0700 Subject: [PATCH 037/441] Add batcher example to examples Created a batcher example that allows the insertion of many messages and batch processing of those messages. --- example/__init__.py | 13 +- example/batcher/__init__.py | 232 +++++++++++++++++++++++++++++++++ example/templates/batcher.html | 193 +++++++++++++++++++++++++++ 3 files changed, 437 insertions(+), 1 deletion(-) create mode 100644 example/batcher/__init__.py create mode 100644 example/templates/batcher.html diff --git a/example/__init__.py b/example/__init__.py index 101e96b..d653585 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -30,7 +30,15 @@ from .callback import AsyncAsyncCallbackHandler from .simple_workflow import SimpleWorkflowHandler from .complex_workflow import ComplexWorkflowHandler +from .batcher import BatcherViewHandler +from .batcher import BatcherHandler +from .batcher import BatcherStatsHandler +config = { + 'webapp2_extras.jinja2': { + 'template_path': 'example/templates' + } +} app = webapp2.WSGIApplication([ ('/', AsyncIntroHandler), @@ -40,4 +48,7 @@ ('/callback/async', AsyncAsyncCallbackHandler), ('/workflow', SimpleWorkflowHandler), ('/workflow/complex', ComplexWorkflowHandler), -]) + ('/batcher', BatcherViewHandler), + ('/batcher/run', BatcherHandler), + ('/batcher/stats', BatcherStatsHandler), +], config=config) diff --git a/example/batcher/__init__.py b/example/batcher/__init__.py new file mode 100644 index 0000000..5c1325b --- /dev/null +++ b/example/batcher/__init__.py @@ -0,0 +1,232 @@ +# +# 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 datetime +import time +import json +import logging + +import webapp2 + +from webapp2_extras import jinja2 + + +class BatcherViewHandler(webapp2.RequestHandler): + """Batcher view to insert tasks and view the stats. + + url: host/batcher + """ + + @webapp2.cached_property + def jinja2(self): + # Returns a Jinja2 renderer cached in the app registry. + return jinja2.get_jinja2(app=self.app) + + def render_response(self, _template, **context): + # Renders a template and writes the result to the response. + rv = self.jinja2.render_template(_template, **context) + self.response.write(rv) + + def get(self): + context = {} + + self.render_response('batcher.html', **context) + + +class BatcherHandler(webapp2.RequestHandler): + """Handler for inserting batch messages. Takes urlparams of color, value + and count. The color is just a tag and value is an incrementing number for + that tag. The count is the number of messages with that payload to insert. + """ + + def get_params(self): + color = self.request.GET['color'] + value = self.request.GET['value'] + count = int(self.request.GET['count']) + + assert color + assert value + assert count + + return color, value, count + + def get(self): + from furious import context + from furious.batcher import Message + from furious.batcher import MessageProcessor + + try: + color, value, count = self.get_params() + except (KeyError, AssertionError): + response = { + "success": False, + "message": "Invalid paramters." + } + self.response.write(json.dumps(response)) + return + + payload = { + "color": color, + "value": value, + "timestamp": time.mktime(datetime.datetime.utcnow().timetuple()) + } + + tag = "color" + + # create a context to insert multiple Messages + with context.new() as ctx: + # loop through the count adding a task to the context per increment + for _ in xrange(count): + # insert the message with the payload + ctx.add(Message(task_args={"payload": payload, "tag": tag})) + + # insert a processor to fetch the messages in batches + # this should always be inserted. the logic will keep it from inserting + # too many processors + processor = MessageProcessor( + target=process_messages, args=(tag,), tag=tag, + task_args={"countdown": 0}) + processor.start() + + response = { + "success": True, + "message": "Task inserted successfully with %s" % (payload,) + } + + self.response.write(json.dumps(response)) + + +class BatcherStatsHandler(webapp2.RequestHandler): + """Handler for returing the stats to the client. Returns as a json payload. + Pulls the stats from memcache. + """ + + def get(self): + from google.appengine.api import memcache + + stats = memcache.get('color') + stats = stats if stats else json.dumps(get_default_stats()) + self.response.write(stats) + + +def process_messages(tag, retries=0): + """Processes the messages pulled fromm a queue based off the tag passed in. + Will insert another processor if any work was processed or the retry count + is under the max retry count. Will update a aggregated stats object with + the data in the payload of the messages processed. + + :param tag: :class: `str` Tag to query the queue on + :param retry: :class: `int` Number of retries the job has processed + """ + from furious.batcher import bump_batch + from furious.batcher import MESSAGE_DEFAULT_QUEUE + from furious.batcher import MessageIterator + from furious.batcher import MessageProcessor + + from google.appengine.api import memcache + + # since we don't have a flag for checking complete we'll re-insert a + # processor task with a retry count to catch any work that may still be + # filtering in. If we've hit our max retry count we just bail out and + # consider the job complete. + if retries > 5: + logging.info("Process messages hit max retry and is exiting") + return + + # create a message iteragor for the tag in batches of 500 + message_iterator = MessageIterator(tag, MESSAGE_DEFAULT_QUEUE, 500) + + # get the stats object from cache + stats = memcache.get(tag) + + # json decode it if it exists otherwise get the default state. + stats = json.loads(stats) if stats else get_default_stats() + + work_processed = False + + # loop through the messages pulled from the queue. + for message in message_iterator: + work_processed = True + + value = int(message.get("value", 0)) + color = message.get("color").lower() + + # update the total stats with the value pulled + set_stats(stats["totals"], value) + + # update the specific color status via the value pulled + set_stats(stats["colors"][color], value) + + # insert the stats back into cache + memcache.set(tag, json.dumps(stats)) + + # bump the process batch id + bump_batch(tag) + + if work_processed: + # reset the retries as we've processed work + retries = 0 + else: + # no work was processed so increment the retries + retries += 1 + + # insert another processor + processor = MessageProcessor( + target=process_messages, args=("colors",), + kwargs={'retries': retries}, tag="colors") + + processor.start() + + +def set_stats(stats, value): + """Updates the stats with the value passed in. + + :param stats: :class: `dict` + :param value: :class: `int` + """ + stats["total_count"] += 1 + stats["value"] += value + stats["average"] = stats["value"] / stats["total_count"] + + if value > stats["max"]: + stats["max"] = value + + if value < stats["min"] or stats["min"] == 0: + stats["min"] = value + + +def get_default_stats(): + """Returns a :class: `dict` of the default stats structure.""" + + default_stats = { + "total_count": 0, + "max": 0, + "min": 0, + "value": 0, + "average": 0, + "last_update": None, + } + + return { + "totals": default_stats, + "colors": { + "red": default_stats.copy(), + "blue": default_stats.copy(), + "yellow": default_stats.copy(), + "green": default_stats.copy(), + "black": default_stats.copy(), + } + } diff --git a/example/templates/batcher.html b/example/templates/batcher.html new file mode 100644 index 0000000..527b92d --- /dev/null +++ b/example/templates/batcher.html @@ -0,0 +1,193 @@ + + + + + + Async Batcher Example + + +

Async Batcher Example

+ +
+ + +
+
+ + +
+
+ + +
+ + +

Color Stats

+ +
+ + +
+
+ + + From 177148b7349f2a68de405f655decc782749b6861 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:16:09 -0700 Subject: [PATCH 038/441] Add docstrings to the example/runner.py module. --- example/runner.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/example/runner.py b/example/runner.py index 68ceb85..8e8f1f4 100644 --- a/example/runner.py +++ b/example/runner.py @@ -5,6 +5,12 @@ def args(): + """Add and parse the arguments for the script. + + url: the url of the example to run + gae-sdk-path: this allows a user to point the script to their GAE SDK + if it's not in /usr/local/google_appengine. + """ parser = argparse.ArgumentParser(description='Run the Furious Examples.') parser.add_argument('--gae-sdk-path', metavar='S', dest="gae_lib_path", @@ -18,6 +24,12 @@ def args(): def setup(options): + """Grabs the gae_lib_path from the options and inserts it into the first + index of the sys.path. Then calls GAE's fix_sys_path to get all the proper + GAE paths included. + + :param options: + """ sys.path.insert(0, options.gae_lib_path) from dev_appserver import fix_sys_path @@ -25,25 +37,45 @@ def setup(options): def run(options): + """Run the passed in url of the example using GAE's rpc runner. + + Uses appengine_rpc.HttpRpcServer to send a request to the url passed in + via the options. + + :param options: + """ from google.appengine.tools import appengine_rpc from google.appengine.tools import appcfg source = 'furious' + + # use the same user agent that GAE uses in appcfg user_agent = appcfg.GetUserAgent() + + # Since we're only using the dev server for now we can hard code these + # values. This will need to change and accept these values as variables + # when this is wired up to hit appspots. server = appengine_rpc.HttpRpcServer( 'localhost:8080', lambda: ('test@example.com', 'password'), user_agent, source, secure=False) + # if no url is passed in just use the top level. url = "/" if options.url: url += options.url[0] + # use the dev server authentication for now. server._DevAppServerAuthenticate() + + # send a simple GET request to the url server.Send(url, content_type="text/html; charset=utf-8", payload=None) def main(): + """Send a request to the url passed in via the options using GAE's rpc + server. + """ options = args() setup(options) From 7ded28455daf1895ff838ae5fcf76324c5b19bf0 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:35:40 -0700 Subject: [PATCH 039/441] Have Async.start retry on TransientError When hitting a TransientError Async.start will now try to re-insert the task. Also added tests for the error types. --- furious/async.py | 15 ++++++++-- furious/tests/test_async.py | 56 +++++++++++++++++++++++++++++++++++-- 2 files changed, 65 insertions(+), 6 deletions(-) diff --git a/furious/async.py b/furious/async.py index 087fd55..42f4ccb 100644 --- a/furious/async.py +++ b/furious/async.py @@ -222,15 +222,24 @@ def to_task(self): return Task(**kwargs) def start(self): - """Insert the task into the requested queue, 'default' if non given.""" + """Insert the task into the requested queue, 'default' if non given. + + If a TransientError is hit the task will re-insert the task. + + If a TaskAlreadyExistsError or TombstonedTaskError is hit the task will + silently fail. + """ from google.appengine.api import taskqueue task = self.to_task() try: taskqueue.Queue(name=self.get_queue()).add(task) - except (taskqueue.TransientError, - taskqueue.TaskAlreadyExistsError, + + except taskqueue.TransientError: + taskqueue.Queue(name=self.get_queue()).add(task) + + except (taskqueue.TaskAlreadyExistsError, taskqueue.TombstonedTaskError): return diff --git a/furious/tests/test_async.py b/furious/tests/test_async.py index ac71217..b03c6e0 100644 --- a/furious/tests/test_async.py +++ b/furious/tests/test_async.py @@ -511,14 +511,64 @@ def test_setting_result(self): self.assertTrue(job.executed) @patch('google.appengine.api.taskqueue.Queue', autospec=True) - def test_start(self, queue_mock): + def test_start_hits_transient_error(self, queue_mock): + """Ensure the task retries if a transient error is hit.""" + from google.appengine.api.taskqueue import TransientError + from furious.async import Async + + def add(task, *args, **kwargs): + def add_second(task, *args, **kwargs): + assert task + + queue_mock.return_value.add.side_effect = add_second + raise TransientError() + + queue_mock.return_value.add.side_effect = add + + async_job = Async("something", queue='my_queue') + async_job.start() + + queue_mock.assert_called_with(name='my_queue') + self.assertEqual(2, queue_mock.return_value.add.call_count) + + @patch('google.appengine.api.taskqueue.Queue', autospec=True) + def test_start_hits_task_already_exists_error_error(self, queue_mock): + """Ensure the task retries if a task already exists error is hit.""" + from google.appengine.api.taskqueue import TaskAlreadyExistsError + from furious.async import Async + + queue_mock.return_value.add.side_effect = TaskAlreadyExistsError() + + async_job = Async("something", queue='my_queue') + async_job.start() + + queue_mock.assert_called_with(name='my_queue') + self.assertEqual(1, queue_mock.return_value.add.call_count) + + @patch('google.appengine.api.taskqueue.Queue', autospec=True) + def test_start_hits_tombstoned_task_error_error(self, queue_mock): + """Ensure the task retries if a tombstoned task error is hit.""" + from google.appengine.api.taskqueue import TombstonedTaskError + from furious.async import Async + + queue_mock.return_value.add.side_effect = TombstonedTaskError() + + async_job = Async("something", queue='my_queue') + async_job.start() + + queue_mock.assert_called_with(name='my_queue') + self.assertEqual(1, queue_mock.return_value.add.call_count) + + @patch('google.appengine.api.taskqueue.Queue', autospec=True) + def test_start_runs_successfully(self, queue_mock): """Ensure the Task is inserted into the specified queue.""" from furious.async import Async async_job = Async("something", queue='my_queue') - # task = async_job.to_task() async_job.start() + queue_mock.assert_called_once_with(name='my_queue') + self.assertTrue(queue_mock.return_value.add.called) + # TODO: Check that the task is the same. # self.assertEqual(task, queue_mock.add.call_args) - From 71a893c989f52b0626192d27ebf3a6c69acaa2be Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:37:21 -0700 Subject: [PATCH 040/441] Add module docstring to example/runner.py --- example/runner.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/example/runner.py b/example/runner.py index 8e8f1f4..85304ad 100644 --- a/example/runner.py +++ b/example/runner.py @@ -1,4 +1,14 @@ #!/usr/bin/python +""" +This script will aid in running the examples and "integration" tests included +with furious. + +To run: + + python example/runner.py workflow + +This will hit the /workflow url, causing the "workflow" example to run. +""" import argparse import sys From 961c6864e4ce02a6a5cf742f2e4127717b731b20 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Fri, 18 Jan 2013 01:42:39 -0700 Subject: [PATCH 041/441] Fix test comment for test_async --- furious/tests/test_async.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/furious/tests/test_async.py b/furious/tests/test_async.py index b03c6e0..44a24bf 100644 --- a/furious/tests/test_async.py +++ b/furious/tests/test_async.py @@ -533,7 +533,7 @@ def add_second(task, *args, **kwargs): @patch('google.appengine.api.taskqueue.Queue', autospec=True) def test_start_hits_task_already_exists_error_error(self, queue_mock): - """Ensure the task retries if a task already exists error is hit.""" + """Ensure the task returns if a task already exists error is hit.""" from google.appengine.api.taskqueue import TaskAlreadyExistsError from furious.async import Async @@ -547,7 +547,7 @@ def test_start_hits_task_already_exists_error_error(self, queue_mock): @patch('google.appengine.api.taskqueue.Queue', autospec=True) def test_start_hits_tombstoned_task_error_error(self, queue_mock): - """Ensure the task retries if a tombstoned task error is hit.""" + """Ensure the task returns if a tombstoned task error is hit.""" from google.appengine.api.taskqueue import TombstonedTaskError from furious.async import Async From bc6e350f7b531f716cc7f9681362c35536f327af Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Fri, 18 Jan 2013 02:02:27 -0700 Subject: [PATCH 042/441] Add persistence API to Context. Add high-level persistence API interaction to Context. The API is very minimal, requiring the persistence engine only to implement a "store_context" function. --- furious/context/context.py | 7 +++++++ furious/tests/context/test_context.py | 24 ++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/furious/context/context.py b/furious/context/context.py index 1fb113a..5560d59 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -139,6 +139,13 @@ def start(self): if self._tasks: self._handle_tasks() + def persist(self): + """Store the context.""" + if not self._persistence_engine: + raise RuntimeError( + 'Specify a valid persistence_engine to persist this context.') + + return self._persistence_engine.store_context(self.id, self.to_dict()) def to_dict(self): """Return this Context as a dict suitable for json encoding.""" diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index 15add70..d7d762f 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -18,6 +18,7 @@ from google.appengine.ext import testbed +from mock import Mock from mock import patch @@ -298,6 +299,29 @@ def test_reconstitution(self): self.assertEqual(options, context.to_dict()) + def test_persist_with_no_engine(self): + """Calling persist with no engine should blow up.""" + from furious.context import Context + + context = Context() + self.assertRaises(RuntimeError, context.persist) + + def test_persist_persists(self): + """Calling persist with an engine persists the Context.""" + from furious.context import Context + + persistence_engine = Mock() + persistence_engine.func_name = 'persistence_engine' + persistence_engine.im_class.__name__ = 'engine' + + context = Context(persistence_engine=persistence_engine) + + context.persist() + + persistence_engine.store_context.assert_called_once_with( + context.id, context.to_dict()) + + class TestInsertTasks(unittest.TestCase): """Test that _insert_tasks behaves as expected.""" def setUp(self): From 70d3448bb547112581c44f8f3387eb423a872804 Mon Sep 17 00:00:00 2001 From: Robert Kluin Date: Fri, 18 Jan 2013 02:11:28 -0700 Subject: [PATCH 043/441] Add Context.load method to load a context from a persistence_engine. Context.load will load a Context object from the given persistence_engine. The API requires only that the persistence_engine implement a load_context function that returns a dict loadable by Context.from_dict. --- furious/context/context.py | 9 +++++++++ furious/tests/context/test_context.py | 14 ++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/furious/context/context.py b/furious/context/context.py index 5560d59..25c40b5 100644 --- a/furious/context/context.py +++ b/furious/context/context.py @@ -147,6 +147,15 @@ def persist(self): return self._persistence_engine.store_context(self.id, self.to_dict()) + @classmethod + def load(cls, context_id, persistence_engine): + """Load and instantiate a Context from the persistence_engine.""" + if not persistence_engine: + raise RuntimeError( + 'Specify a valid persistence_engine to load the context.') + + return cls.from_dict(persistence_engine.load_context(context_id)) + def to_dict(self): """Return this Context as a dict suitable for json encoding.""" import copy diff --git a/furious/tests/context/test_context.py b/furious/tests/context/test_context.py index d7d762f..76f2505 100644 --- a/furious/tests/context/test_context.py +++ b/furious/tests/context/test_context.py @@ -321,6 +321,20 @@ def test_persist_persists(self): persistence_engine.store_context.assert_called_once_with( context.id, context.to_dict()) + def test_load_context(self): + """Calling load with an engine attempts to load the Context.""" + from furious.context import Context + + persistence_engine = Mock() + persistence_engine.func_name = 'persistence_engine' + persistence_engine.im_class.__name__ = 'engine' + persistence_engine.load_context.return_value = {'id': 'ABC123'} + + context = Context.load('ABC123', persistence_engine) + + persistence_engine.load_context.assert_called_once_with('ABC123') + self.assertEqual('ABC123', context.id) + class TestInsertTasks(unittest.TestCase): """Test that _insert_tasks behaves as expected.""" From c1e60f6c8cd17e57560922c936b8677c35a5b401 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 24 Jan 2013 12:28:08 -0600 Subject: [PATCH 044/441] remove os.path import and add defaults decorator to build_and_start --- example/grep.py | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/example/grep.py b/example/grep.py index 5ca32f9..5feae95 100644 --- a/example/grep.py +++ b/example/grep.py @@ -1,30 +1,37 @@ +#TODO: add copyright notice + import logging import os -import os.path import re import webapp2 +from furious import context +from furious import defaults from furious.async import Async + class GrepHandler(webapp2.RequestHandler): def get(self): query = self.request.get('query') curdir = os.getcwd() - build_and_start(query, curdir) + + with context.new(): + build_and_start(query, curdir) self.response.out.write('starting grep for query: %s' % query) + +@defaults(callbacks={'success': log_results}) def build_and_start(query, directory): - async_task = Async( - target=grep, args=[query, directory], - callbacks={'success': log_results} - ) + async_task = Async(target=grep, args=[query, directory]) async_task.start() + def grep_file(query, item): return ['%s: %s' % (item, line) for line in open(item) if re.search(query, line)] + def grep(query, directory): dir_contents = os.listdir(directory) results = [] @@ -37,6 +44,7 @@ def grep(query, directory): results.extend(grep_file(query, path)) return results + def log_results(): from furious.context import get_current_async @@ -44,4 +52,3 @@ def log_results(): for result in async.result: logging.info(result) - From df7c2d33e05467ca057cc26450dd0a4e78f46a56 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 24 Jan 2013 12:33:23 -0600 Subject: [PATCH 045/441] added copyright notice to grep example --- example/grep.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/example/grep.py b/example/grep.py index 5feae95..d6810db 100644 --- a/example/grep.py +++ b/example/grep.py @@ -1,4 +1,18 @@ -#TODO: add copyright notice +# +# 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 logging import os From 616920a0c39632523d62af32159abeccbd05b8e6 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Thu, 24 Jan 2013 12:53:17 -0600 Subject: [PATCH 046/441] added docstrs and comments to the code. --- example/grep.py | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/example/grep.py b/example/grep.py index d6810db..42d26f0 100644 --- a/example/grep.py +++ b/example/grep.py @@ -14,6 +14,9 @@ # limitations under the License. # +"""This is a Furious example that greps the source code of the application and +logs a message with each line that satisifes the input regular expression.""" + import logging import os import re @@ -26,6 +29,10 @@ class GrepHandler(webapp2.RequestHandler): + """This is the handler that starts off a grep run. It takes the query out + of the query string and starts a new Furious Context to run all of the + Asyncs in.""" + def get(self): query = self.request.get('query') curdir = os.getcwd() @@ -37,16 +44,24 @@ def get(self): @defaults(callbacks={'success': log_results}) def build_and_start(query, directory): - async_task = Async(target=grep, args=[query, directory]) - async_task.start() + """This function will create and then start a new Async task with the + default callbacks argument defined in the decorator.""" + + Async(target=grep, args=[query, directory]).start() def grep_file(query, item): + """This function performs the actual grep on a given file.""" return ['%s: %s' % (item, line) for line in open(item) if re.search(query, line)] def grep(query, directory): + """This function will search through the directory structure of the + application and for each directory it finds it launches an Async task to + run itself. For each .py file it finds, it actually greps the file and then + returns the found output.""" + dir_contents = os.listdir(directory) results = [] for item in dir_contents: @@ -60,6 +75,8 @@ def grep(query, directory): def log_results(): + """This is the callback that is run once the Async task is finished. It + takes the output from grep and logs it.""" from furious.context import get_current_async async = get_current_async() From 393b60dc36608b2387c208b00975f047275d5cb1 Mon Sep 17 00:00:00 2001 From: Tanner Miller Date: Tue, 29 Jan 2013 10:23:18 -0600 Subject: [PATCH 047/441] added a few more comments --- example/grep.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/example/grep.py b/example/grep.py index 42d26f0..0522227 100644 --- a/example/grep.py +++ b/example/grep.py @@ -37,8 +37,10 @@ def get(self): query = self.request.get('query') curdir = os.getcwd() + # create the context and start the first Async with context.new(): build_and_start(query, curdir) + self.response.out.write('starting grep for query: %s' % query) @@ -79,7 +81,10 @@ def log_results(): takes the output from grep and logs it.""" from furious.context import get_current_async + # Get the recently finished Async object. async = get_current_async() + # Pull out the result data and log it. for result in async.result: logging.info(result) + From 4e319a248c55fdc2b67f46249c9a5d2df841aae0 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Tue, 29 Jan 2013 14:23:02 -0600 Subject: [PATCH 048/441] Re-order the imports to make Robert happy :) --- example/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/example/__init__.py b/example/__init__.py index d653585..0f61360 100644 --- a/example/__init__.py +++ b/example/__init__.py @@ -30,9 +30,9 @@ from .callback import AsyncAsyncCallbackHandler from .simple_workflow import SimpleWorkflowHandler from .complex_workflow import ComplexWorkflowHandler -from .batcher import BatcherViewHandler from .batcher import BatcherHandler from .batcher import BatcherStatsHandler +from .batcher import BatcherViewHandler config = { 'webapp2_extras.jinja2': { From 0fc87b2830f8d80c55c0e892df5b54567bedfc38 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Tue, 29 Jan 2013 14:23:57 -0600 Subject: [PATCH 049/441] Clean up some of the batcher example and switch the caching to CAS --- example/batcher/__init__.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/example/batcher/__init__.py b/example/batcher/__init__.py index 5c1325b..e7e9adf 100644 --- a/example/batcher/__init__.py +++ b/example/batcher/__init__.py @@ -61,7 +61,7 @@ def get_params(self): assert value assert count - return color, value, count + return color.strip(), value.strip(), count def get(self): from furious import context @@ -73,7 +73,7 @@ def get(self): except (KeyError, AssertionError): response = { "success": False, - "message": "Invalid paramters." + "message": "Invalid parameters." } self.response.write(json.dumps(response)) return @@ -150,7 +150,8 @@ def process_messages(tag, retries=0): message_iterator = MessageIterator(tag, MESSAGE_DEFAULT_QUEUE, 500) # get the stats object from cache - stats = memcache.get(tag) + client = memcache.Client() + stats = client.get(tag) # json decode it if it exists otherwise get the default state. stats = json.loads(stats) if stats else get_default_stats() @@ -171,7 +172,8 @@ def process_messages(tag, retries=0): set_stats(stats["colors"][color], value) # insert the stats back into cache - memcache.set(tag, json.dumps(stats)) + if not client.cas(tag, json.dumps(stats)): + raise Exception("Transaction Collision.") # bump the process batch id bump_batch(tag) @@ -201,6 +203,8 @@ def set_stats(stats, value): stats["value"] += value stats["average"] = stats["value"] / stats["total_count"] + # this is just a basic example and not the best way to track aggregation. + # for max and min old there are cases where this will not work correctly. if value > stats["max"]: stats["max"] = value From 45c360c7e276b6ce40dcac4e555ee0693e29cd81 Mon Sep 17 00:00:00 2001 From: Beau Lyddon Date: Tue, 29 Jan 2013 14:24:19 -0600 Subject: [PATCH 050/441] Clean up the batcher example html and javascrpt --- example/templates/batcher.html | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/example/templates/batcher.html b/example/templates/batcher.html index 527b92d..dc1c8c8 100644 --- a/example/templates/batcher.html +++ b/example/templates/batcher.html @@ -20,12 +20,12 @@

Async Batcher Example

-
-
@@ -39,9 +39,7 @@

Color Stats