[pubsubhubbub] r387 committed - use mapper framework for subscription reconfirmation

codesite-noreply Wed, 22 Sep 2010 15:17:15 -0700

Revision: 387
Author: bslatkin
Date: Wed Sep 22 15:16:49 2010
Log: use mapper framework for subscription reconfirmation
http://code.google.com/p/pubsubhubbub/source/detail?r=387


Modified:
 /trunk/hub
 /trunk/hub/async_apiproxy.py
 /trunk/hub/index.yaml
 /trunk/hub/main.py
 /trunk/hub/main_test.py
 /trunk/hub/mapreduce.yaml
 /trunk/hub/offline_jobs.py

=======================================
--- /trunk/hub/async_apiproxy.py        Sun Jul 11 22:58:45 2010
+++ /trunk/hub/async_apiproxy.py        Wed Sep 22 15:16:49 2010
@@ -126,10 +126,8 @@

   def wait(self):
     """Wait for RPCs to finish. Returns True if any were processed."""
-    while self.enqueued:
+    while self.enqueued or self.complete:

# Run the callbacks before even waiting, because a response couldhave

       # come back during any outbound API call.
       self._run_callbacks()
       self._wait_one()
-    # Run them one last time after waiting to pick up the final callback!
-    self._run_callbacks()
=======================================
--- /trunk/hub/index.yaml       Wed Feb  3 09:37:28 2010
+++ /trunk/hub/index.yaml       Wed Sep 22 15:16:49 2010
@@ -24,3 +24,8 @@
 # manually, move them above the marker line.  The index.yaml file is
 # automatically uploaded to the admin console when you next deploy
 # your application using appcfg.py.
+
+- kind: Subscription
+  properties:
+  - name: __key__
+    direction: desc
=======================================
--- /trunk/hub/main.py  Sun Jul 11 22:57:08 2010
+++ /trunk/hub/main.py  Wed Sep 22 15:16:49 2010
@@ -119,6 +119,9 @@
 import fork_join_queue
 import urlfetch_async

+import mapreduce.control
+import mapreduce.model
+
 async_proxy = async_apiproxy.AsyncAPIProxy()

################################################################################

@@ -168,17 +171,17 @@
 # How far before expiration to refresh subscriptions.
 SUBSCRIPTION_CHECK_BUFFER_SECONDS = (24 * 60 * 60)  # 24 hours

-# How many subscriber checking tasks to scheudle at a time.
-SUBSCRIPTION_CHECK_CHUNK_SIZE = 200
+# How many mapper shards to use for reconfirming subscriptions.
+SUBSCRIPTION_RECONFIRM_SHARD_COUNT = 4

 # How often to poll feeds.
 POLLING_BOOTSTRAP_PERIOD = 10800  # in seconds; 3 hours

 # Default expiration time of a lease.
-DEFAULT_LEASE_SECONDS = (30 * 24 * 60 * 60)  # 30 days
+DEFAULT_LEASE_SECONDS = (5 * 24 * 60 * 60)  # 5 days

 # Maximum expiration time of a lease.
-MAX_LEASE_SECONDS = DEFAULT_LEASE_SECONDS * 3  # 90 days
+MAX_LEASE_SECONDS = (10 * 24 * 60 * 60)  # 10 days

 # Maximum number of redirects to follow when feed fetching.
 MAX_REDIRECTS = 7
@@ -2111,62 +2114,41 @@
 class SubscriptionReconfirmHandler(webapp.RequestHandler):

"""Periodic handler causes reconfirmation for almost expiredsubscriptions."""


-  def __init__(self, now=time.time):
+  def __init__(self, now=time.time, start_map=mapreduce.control.start_map):
     """Initializer."""
     webapp.RequestHandler.__init__(self)
     self.now = now
+    self.start_map = start_map

   @work_queue_only
   def get(self):
-    threshold_timestamp = str(
-        int(self.now() - SUBSCRIPTION_CHECK_BUFFER_SECONDS))

- # NOTE: See PollBootstrapHandler as to why we need a named task herefor

-    # the first insertion and the rest of the sequence.
-    name = 'reconfirm-' + threshold_timestamp

+ # Use the name, such that only one of these tasks runs per calendarday.+ name = 'reconfirm-%s' % time.strftime('%Y-%m-%d' ,time.gmtime(self.now()))

     try:
       taskqueue.Task(
           url='/work/reconfirm_subscriptions',
-          name=name,
-          params=dict(time_offset=threshold_timestamp)
+          name=name
       ).add(POLLING_QUEUE)

except (taskqueue.TaskAlreadyExistsError,taskqueue.TombstonedTaskError):

-      logging.exception('Could not enqueue FIRST reconfirmation task')
+      logging.exception('Could not enqueue FIRST reconfirmation task; '
+                        'must have already run today.')

   @work_queue_only
   def post(self):
-    time_offset = self.request.get('time_offset')
-    datetime_offset = datetime.datetime.utcfromtimestamp(int(time_offset))
-    key_offset = self.request.get('key_offset')
-    logging.info('Handling reconfirmations for time_offset = %s, '
-                 'current_key = %s', datetime_offset, key_offset)
-
-    query = (Subscription.all()
-             .filter('subscription_state =', Subscription.STATE_VERIFIED)
-             .order('__key__'))
-    if key_offset:
-      query.filter('__key__ >', db.Key(key_offset))
-
-    subscriptions = query.fetch(SUBSCRIPTION_CHECK_CHUNK_SIZE)
-    if not subscriptions:
-      logging.info('All done with periodic subscription reconfirmations')
-      return
-
-    next_key = str(subscriptions[-1].key())
-    try:
-      taskqueue.Task(
-          url='/work/reconfirm_subscriptions',
-          name='reconfirm-%s-%s' % (time_offset, sha1_hash(next_key)),
-          params=dict(time_offset=time_offset,
-                      key_offset=next_key)).add(POLLING_QUEUE)

- except (taskqueue.TaskAlreadyExistsError,taskqueue.TombstonedTaskError):

-      logging.exception('Continued polling task already present; '
-                        'this work has already been done')
-      return
-
-    for sub in subscriptions:
-      if sub.expiration_time < datetime_offset:
-        sub.request_insert(sub.callback, sub.topic, sub.verify_token,
-                           sub.secret, auto_reconfirm=True)
+    self.start_map(
+        name='Reconfirm expiring subscriptions',
+        handler_spec='offline_jobs.SubscriptionReconfirmMapper.run',
+        reader_spec='mapreduce.input_readers.DatastoreInputReader',
+        reader_parameters=dict(
+            processing_rate=100000,
+            entity_kind='main.Subscription'),
+        shard_count=SUBSCRIPTION_RECONFIRM_SHARD_COUNT,
+        queue_name=POLLING_QUEUE,
+        mapreduce_parameters=dict(
+          threshold_timestamp=int(
+              self.now() + SUBSCRIPTION_CHECK_BUFFER_SECONDS),
+          done_callback='/work/cleanup_mapper',
+          done_callback_queue=POLLING_QUEUE))


 class SubscriptionCleanupHandler(webapp.RequestHandler):
@@ -2184,6 +2166,19 @@

except (db.Error, apiproxy_errors.Error,runtime.DeadlineExceededError):

         logging.exception('Could not clean-up Subscription instances')

+
+class CleanupMapperHandler(webapp.RequestHandler):
+  """Cleans up all data from a Mapper job run."""
+
+  @work_queue_only
+  def post(self):
+    mapreduce_id = self.request.headers.get('mapreduce-id')
+    # TODO: Use Mapper Cleanup API once available.

+db.delete(mapreduce.model.MapreduceControl.get_key_by_job_id(mapreduce_id))

+    shards = mapreduce.model.ShardState.find_by_mapreduce_id(mapreduce_id)
+    db.delete(shards)

+db.delete(mapreduce.model.MapreduceState.get_key_by_job_id(mapreduce_id))

################################################################################

 # Publishing handlers

@@ -2411,7 +2406,6 @@
              getattr(response, 'headers', None),
              getattr(response, 'content', None),
              exception)
-
   urlfetch_async.fetch(fetch_url,
                        headers=headers,
                        follow_redirects=False,
@@ -3445,7 +3439,8 @@
       (r'/work/poll_bootstrap', PollBootstrapHandler),
       (r'/work/event_cleanup', EventCleanupHandler),
       (r'/work/subscription_cleanup', SubscriptionCleanupHandler),
-      (r'/work/reconfirm_subscriptions', SubscriptionReconfirmHandler)
+      (r'/work/reconfirm_subscriptions', SubscriptionReconfirmHandler),
+      (r'/work/cleanup_mapper', CleanupMapperHandler),
     ])
   application = webapp.WSGIApplication(HANDLERS, debug=DEBUG)
   wsgiref.handlers.CGIHandler().run(application)
=======================================
--- /trunk/hub/main_test.py     Sun Jul 11 22:24:18 2010
+++ /trunk/hub/main_test.py     Wed Sep 22 15:16:49 2010
@@ -45,6 +45,9 @@
 import main
 import urlfetch_test_stub

+import mapreduce.control
+import mapreduce.model
+

################################################################################

 # For convenience

@@ -2719,7 +2722,7 @@
         '&hub.challenge=this_is_my_fake_challenge_string'
         '&hub.topic=http%%3A%%2F%%2Fexample.com%%2Fthe-topic'
         '&hub.mode=%s'
-        '&hub.lease_seconds=2592000')
+        '&hub.lease_seconds=432000')

   def tearDown(self):
     """Tears down the test harness."""
@@ -2986,7 +2989,7 @@
         '&hub.challenge=this_is_my_fake_challenge_string'
         '&hub.topic=http%%3A%%2F%%2Fexample.com%%2Fthe-topic'
         '&hub.mode=%s'
-        '&hub.lease_seconds=7776000')
+        '&hub.lease_seconds=864000')
     urlfetch_test_stub.instance.expect(