backends/counter/counter_v3_with_write_behind.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2011 Google Inc. All Rights Reserved.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 # vim: set ts=4 sw=4 et tw=79:
  17
  18 """Implementation of a counter with persistent storage."""
  19
  20 __author__ = 'Greg Darke <darke@google.com>'
  21
  22 import logging
  23 import math
  24 import time
  25
  26 from google.appengine.api import backends
  27 from google.appengine.api import runtime
  28 from google.appengine.api import taskqueue
  29 from google.appengine.ext import db
  30 from google.appengine.ext import webapp
  31 from google.appengine.ext.webapp.util import run_wsgi_app
  32
  33
  34 _MEMORY_LIMIT = 128  # The memory limit for a B1 server
  35 _CULL_AMOUNT = 0.15
  36
  37
  38 def _get_taskqueue_target():
  39     return '%d.%s' % (backends.get_instance(), backends.get_backend())
  40
  41
  42 class CounterModel(db.Model):
  43     value = db.IntegerProperty(default=0)
  44     _dirty = False
  45     _last_accessed = None
  46
  47     @classmethod
  48     def get_or_new(cls, name):
  49         model = cls.get_by_key_name(name)
  50         if model is None:
  51             model = cls(key_name=name)
  52         return model
  53
  54
  55 class CounterStore(object):
  56     def __init__(self):
  57         self._store = {}
  58         self._has_shutdown = False
  59         self._dirty = False
  60         self._batch_size = 100
  61
  62     def get_value(self, name):
  63         self._ensure_within_memory_limit()
  64
  65         if name not in self._store:
  66             model = CounterModel.get_or_new(name)
  67             self._store[name] = model
  68         else:
  69             model = self._store[name]
  70
  71         model._last_accessed = time.time()
  72         return model
  73
  74     def inc_value(self, name, delta):
  75         if not self._dirty:
  76             # Enqueue a task with a 'target' specifying traffic be sent to this
  77             # instance of the backend.
  78             taskqueue.add(url='/backend/counter/flush',
  79                           target=_get_taskqueue_target(),
  80                           countdown=2)
  81             self._dirty = True
  82
  83         model = self.get_value(name)
  84         model.value += delta
  85         model._dirty = True
  86         if self._has_shutdown:
  87             # Since the shutdown hook may be called at any time, we need to
  88             # protect ourselves against this.
  89             model.put()
  90         return model
  91
  92     def _ensure_within_memory_limit(self):
  93         memory_limit = _MEMORY_LIMIT * 0.8
  94
  95         memory_usage = runtime.memory_usage().current()
  96         if memory_usage >= memory_limit:
  97             # Create a list of candidate counters to remove. We remove counters
  98             # that have not been modified before those that have been modified,
  99             # then order them by the last time they were accessed.
 100             counters = self._store.values()
 101             counters.sort(key=lambda counter: (counter._dirty,
 102                                                counter._last_accessed))
 103             counters_to_cull = int(math.ceil(len(counters) * _CULL_AMOUNT))
 104             counters = counters[:counters_to_cull]
 105
 106             logging.info('Removing %d entries as we are over the memory limit '
 107                          'by %dMB.',
 108                          counters_to_cull, memory_limit - memory_usage)
 109
 110             self._write_in_batches(counters)
 111             for counter in counters:
 112                 del self._store[counter.key().name()]
 113
 114     def _put_counters(self, counters):
 115         db.put(counters)
 116         for counter in counters:
 117             counter._dirty = False
 118
 119     def _write_in_batches(self, counters):
 120         """Write out the dirty entries from 'counters' in batches.
 121
 122         The batch size is determined by self._batch_size.
 123
 124         Args:
 125           counters: An iterable containing instances of CounterModel.
 126         """
 127         to_put = []
 128         for counter in counters:
 129             if counter._dirty:
 130                 to_put.append(counter)
 131
 132                 if len(to_put) >= self._batch_size:
 133                     self._put_counters(to_put)
 134                     to_put = []
 135
 136         if to_put:
 137             self._put_counters(to_put)
 138
 139     def flush_to_datastore(self):
 140         """Write the dirty entries from _store to datastore."""
 141         self._write_in_batches(self._store.itervalues())
 142         self._dirty = False
 143
 144     def shutdown_hook(self):
 145         """Ensures all counters are written to datastore."""
 146         if self._dirty:
 147             self.flush_to_datastore()
 148         self._has_shutdown = True
 149
 150
 151 class StartHandler(webapp.RequestHandler):
 152     """Handler for '/_ah/start'.
 153
 154     This url is called once when the backend is started.
 155     """
 156     def get(self):
 157         runtime.set_shutdown_hook(_counter_store.shutdown_hook)
 158
 159
 160 class FlushHandler(webapp.RequestHandler):
 161     """Handler for counter/flush.
 162
 163     This handler is protected by login: admin in app.yaml.
 164     """
 165     def post(self):
 166         _counter_store.flush_to_datastore()
 167
 168 class CounterHandler(webapp.RequestHandler):
 169     """Handler for counter/{get,inc,dec}.
 170
 171     This handler is protected by login: admin in app.yaml.
 172     """
 173
 174     def _write_error(self, error_message):
 175         self.response.error(400)
 176         self.response.out.write(error_message)
 177
 178     def post(self, method):
 179         """Handler a post to counter/{get,inc,dec}.
 180
 181         The 'method' parameter is parsed from the url by a regex capture group.
 182
 183         Args:
 184           method: The type of operation to perform against the counter store.
 185               It must be one of 'get', 'inc' or 'dec'.
 186         """
 187         key_name = self.request.get('name')
 188         delta = self.request.get('delta')
 189
 190         if not key_name:
 191             self._write_error('Request did not have a "key_name" parameter.')
 192             return
 193
 194         if method == 'get':
 195             model = _counter_store.get_value(key_name)
 196         else:
 197             if not delta:
 198                 self._write_error('Request did not have a "delta" parameter.')
 199                 return
 200             try:
 201                 delta = int(delta)
 202             except ValueError:
 203                 self._write_error('"delta" is not an integer.')
 204                 return
 205
 206             if method == 'inc':
 207                 model = _counter_store.inc_value(key_name, delta)
 208             elif method == 'dec':
 209                 model = _counter_store.inc_value(key_name, -delta)
 210
 211         self.response.headers['Content-Type'] = 'text/plain'
 212         self.response.out.write('%d' % model.value)
 213
 214
 215 _handlers = [(r'/_ah/start', StartHandler),
 216              (r'/backend/counter/flush$', FlushHandler),
 217              (r'/backend/counter/(get|inc|dec)$', CounterHandler)]
 218
 219 application = webapp.WSGIApplication(_handlers)
 220 _counter_store = CounterStore()
 221
 222
 223 def main():
 224     run_wsgi_app(application)
 225
 226 if __name__ == '__main__':
 227     main()