@@ -1,25 +1,26 @@ -Sharded, memcached counter for Google App Engine +Sharded, memcached counter for Google App Engine with error tolerance Copyright 2008, William T Katz. Released under Apache License, version 2.0 -This is one way of implementing counters based on a Google best practices talk: -http://sites.google.com/site/io/building-scalable-web-applications-with-google-app-engine +Should be used for counters that handle a lot of concurrent use. +Follows pattern described in Google I/O talk: + http://sites.google.com/site/io/building-scalable-web-applications-with-google-app-engine -It shards a counter to decrease contentions on writes. Rather than use a -Model for the counter configuration as in the Google example, this version -uses a memcached counter that can be fed with an expandable shard count. - -Memcache is used for caching counts, although you can force non-cached -counts. +Memcache is used for caching counts and if a cached count is available, it is +the most correct. If there are datastore put issues, we store the un-put +values into a delayed_incr memcache that will be applied as soon as the next shard put is successful. Changes will only be lost if we lose memcache before +a successful datastore shard put or there's a failure/error in memcache. Usage: hits = Counter('hits') hits.increment() my_hits = hits.count - hits.get_count(nocache=True) # Forces non-cached count. - hits.decrement() - hits.delete() # Deletes all associated shards. + hits.get_count(nocache=True) # Forces non-cached count of all shards + hits.count = 6 # Set the counter to arbitrary value + hits.increment(incr=-1) # Decrement + hits.increment(10) -For an example of its use, you can see the Tag model in the Bloog open-source -blog software for App Engine (http://github.com/DocSavage/bloog). \ No newline at end of file +For an example of its use (early version), you can see the Tag model in the +Bloog open-source blog software for App Engine +(http://github.com/DocSavage/bloog). \ No newline at end of file README @@ -14,100 +14,136 @@ # # THIS LICENSE INFORMATION/ATTRIBUTION must be left in place. +import string import random import logging from google.appengine.api import memcache from google.appengine.ext import db +from google.appengine.runtime import apiproxy_errors +class MemcachedCount(object): + DELTA_ZERO = 500000 # Allows negative numbers in unsigned memcache + + def __init__(self, name): + self.key = 'MemcachedCount' + name + + def get_count(self): + value = memcache.get(self.key) + if value is None: + return 0 + else: + return string.atoi(value) - MemcachedCount.DELTA_ZERO + + def set_count(self, value): + memcache.set(self.key, str(MemcachedCount.DELTA_ZERO + value)) + + def delete_count(self): + memcache.delete(self.key) + + count = property(get_count, set_count, delete_count) + + def increment(self, incr=1): + value = memcache.get(self.key) + if value is None: + self.count = incr + elif incr > 0: + memcache.incr(self.key, incr) + elif incr < 0: + memcache.decr(self.key, -incr) + +# Naming function for shard key names in datastore and memcache +def get_shardname(name, shard): + return 'Shard' + name + str(shard) + class Counter(object): """A counter using sharded writes to prevent contentions. - This is one way of implementing counters based on a Google best practices - talk: - http://sites.google.com/site/io/building-scalable-web-applications-with-google-app-engine - - It shards a counter to decrease contentions on writes. Rather than use a - Model for the counter configuration as in the Google example, this version - uses a memcached counter that can be fed with an expandable shard count. + Should be used for counters that handle a lot of concurrent use. + Follows pattern described in Google I/O talk: + http://sites.google.com/site/io/building-scalable-web-applications-with-google-app-engine - Memcache is used for caching counts, although you can force non-cached - counts. + Memcache is used for caching counts and if a cached count is available, it is + the most correct. If there are datastore put issues, we store the un-put values + into a delayed_incr memcache that will be applied as soon as the next shard put + is successful. Changes will only be lost if we lose memcache before a successful + datastore shard put or there's a failure/error in memcache. Usage: hits = Counter('hits') hits.increment() my_hits = hits.count - hits.get_count(nocache=True) # Forces non-cached count. - hits.decrement() - hits.delete() # Deletes all associated shards. + hits.get_count(nocache=True) # Forces non-cached count of all shards + hits.count = 6 # Set the counter to arbitrary value + hits.increment(incr=-1) # Decrement + hits.increment(10) """ - MAX_SHARDS = 50 + NUM_SHARDS = 20 - def __init__(self, name, num_shards=5, cache_time=30): + def __init__(self, name): self.name = name - self.num_shards = min(num_shards, Counter.MAX_SHARDS) - self.cache_time = cache_time + self.memcached = MemcachedCount('Counter' + name) + self.delayed_incr = MemcachedCount('DelayedIncr' + name) def delete(self): q = db.Query(CounterShard).filter('name =', self.name) - # Need to use MAX_SHARDS since current number of shards - # may be smaller than previous value. - shards = q.fetch(limit=Counter.MAX_SHARDS) - for shard in shards: - shard.delete() + shards = q.fetch(limit=Counter.NUM_SHARDS) + db.delete(shards) - def memcache_key(self): - return 'Counter' + self.name + def get_count_and_cache(self): + q = db.Query(CounterShard).filter('name =', self.name) + shards = q.fetch(limit=Counter.NUM_SHARDS) + datastore_count = 0 + for shard in shards: + datastore_count += shard.count + count = datastore_count + self.delayed_incr.count + self.memcached.count = count + return count def get_count(self, nocache=False): - total = memcache.get(self.memcache_key()) + total = self.memcached.count if nocache or total is None: - total = 0 - q = db.Query(CounterShard).filter('name =', self.name) - shards = q.fetch(limit=Counter.MAX_SHARDS) - for shard in shards: - total += shard.count - memcache.add(self.memcache_key(), str(total), - self.cache_time) - return total + return self.get_count_and_cache() else: return int(total) - count = property(get_count) - def increment(self): - CounterShard.increment(self.name, self.num_shards) - return memcache.incr(self.memcache_key()) + def set_count(self, value): + cur_value = self.get_count() + self.memcached.count = value + delta = value - cur_value + if delta != 0: + CounterShard.increment(self, incr=delta) + count = property(get_count, set_count) + + def increment(self, incr=1, refresh=False): + CounterShard.increment(self, incr) + self.memcached.increment(incr) - def decrement(self): - CounterShard.increment(self.name, self.num_shards, - downward=True) - return memcache.decr(self.memcache_key()) class CounterShard(db.Model): name = db.StringProperty(required=True) count = db.IntegerProperty(default=0) @classmethod - def increment(cls, name, num_shards, downward=False): - index = random.randint(1, num_shards) - shard_key_name = 'Shard' + name + str(index) + def increment(cls, counter, incr=1): + index = random.randint(1, Counter.NUM_SHARDS) + shard_key_name = get_shardname(counter.name, index) + counter_name = counter.name + delayed_incr = counter.delayed_incr.count def get_or_create_shard(): shard = CounterShard.get_by_key_name(shard_key_name) if shard is None: - shard = CounterShard(key_name=shard_key_name, - name=name) - if downward: - shard.count -= 1 - else: - shard.count += 1 + shard = CounterShard(key_name=shard_key_name, name=counter_name) + shard.count += incr + delayed_incr key = shard.put() try: db.run_in_transaction(get_or_create_shard) - return True - except db.TransactionFailedError(): - logging.error("CounterShard (%s, %d) - can't increment", - name, num_shards) + except (db.Error, apiproxy_errors.Error), e: + counter.delayed_incr.increment(incr) + logging.error("CounterShard (%s) delayed increment %d: %s", + counter_name, incr, e) return False - + if delayed_incr: + counter.delayed_incr.count = 0 + return True counter.py 14d2e5b3516d7ec9009d189aff07226d05f1052d Bill Katz billkatz@gmail.com http://github.com/DocSavage/sharded_counter/commit/c5ec5d99c4c745ed2f0ced2bc0d938246a4065d0 c5ec5d99c4c745ed2f0ced2bc0d938246a4065d0 2008-10-07T01:28:07-07:00 2008-10-07T01:28:07-07:00 Cleaned up code. Added ability to tolerate errors on shard put by using memcache. b45dffd5b5c920d50941f3622a62f1dd6a5f6311 Bill Katz billkatz@gmail.com