| #!/usr/bin/env python |
| # |
| # Copyright 2007 Google Inc. |
| # |
| # 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. |
| # |
| |
| |
| |
| |
| """Stub version of the memcache API, keeping all data in process memory.""" |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| import logging |
| import time |
| |
| from google.appengine.api import apiproxy_stub |
| from google.appengine.api import memcache |
| from google.appengine.api.memcache import memcache_service_pb |
| from google.appengine.runtime import apiproxy_errors |
| |
| |
| MemcacheSetResponse = memcache_service_pb.MemcacheSetResponse |
| MemcacheSetRequest = memcache_service_pb.MemcacheSetRequest |
| MemcacheIncrementRequest = memcache_service_pb.MemcacheIncrementRequest |
| MemcacheIncrementResponse = memcache_service_pb.MemcacheIncrementResponse |
| MemcacheDeleteResponse = memcache_service_pb.MemcacheDeleteResponse |
| |
| |
| MAX_REQUEST_SIZE = 32 << 20 |
| |
| |
| class CacheEntry(object): |
| """An entry in the cache.""" |
| |
| def __init__(self, value, expiration, flags, cas_id, gettime): |
| """Initializer. |
| |
| Args: |
| value: String containing the data for this entry. |
| expiration: Number containing the expiration time or offset in seconds |
| for this entry. |
| flags: Opaque flags used by the memcache implementation. |
| cas_id: Unique Compare-And-Set ID. |
| gettime: Used for testing. Function that works like time.time(). |
| """ |
| assert isinstance(value, basestring) |
| assert len(value) <= memcache.MAX_VALUE_SIZE |
| assert isinstance(expiration, (int, long)) |
| |
| self._gettime = gettime |
| self.value = value |
| self.flags = flags |
| self.cas_id = cas_id |
| self.created_time = self._gettime() |
| self.will_expire = expiration != 0 |
| self.locked = False |
| self._SetExpiration(expiration) |
| |
| def _SetExpiration(self, expiration): |
| """Sets the expiration for this entry. |
| |
| Args: |
| expiration: Number containing the expiration time or offset in seconds |
| for this entry. If expiration is above one month, then it's considered |
| an absolute time since the UNIX epoch. |
| """ |
| if expiration > (86400 * 30): |
| self.expiration_time = expiration |
| else: |
| self.expiration_time = self._gettime() + expiration |
| |
| def CheckExpired(self): |
| """Returns True if this entry has expired; False otherwise.""" |
| return self.will_expire and self._gettime() >= self.expiration_time |
| |
| def ExpireAndLock(self, timeout): |
| """Marks this entry as deleted and locks it for the expiration time. |
| |
| Used to implement memcache's delete timeout behavior. |
| |
| Args: |
| timeout: Parameter originally passed to memcache.delete or |
| memcache.delete_multi to control deletion timeout. |
| """ |
| self.will_expire = True |
| self.locked = True |
| self._SetExpiration(timeout) |
| |
| def CheckLocked(self): |
| """Returns True if this entry was deleted but has not yet timed out.""" |
| return self.locked and not self.CheckExpired() |
| |
| |
| class MemcacheServiceStub(apiproxy_stub.APIProxyStub): |
| """Python only memcache service stub. |
| |
| This stub keeps all data in the local process' memory, not in any |
| external servers. |
| """ |
| |
| THREADSAFE = True |
| |
| def __init__(self, gettime=time.time, service_name='memcache'): |
| """Initializer. |
| |
| Args: |
| gettime: time.time()-like function used for testing. |
| service_name: Service name expected for all calls. |
| """ |
| super(MemcacheServiceStub, self).__init__(service_name, |
| max_request_size=MAX_REQUEST_SIZE) |
| self._next_cas_id = 1 |
| self._gettime = lambda: int(gettime()) |
| self._ResetStats() |
| |
| |
| self._the_cache = {} |
| |
| |
| |
| |
| |
| |
| def _ResetStats(self): |
| """Resets statistics information. |
| |
| Must be called while the current thread holds self._mutex (with an exception |
| for __init__). |
| """ |
| |
| self._hits = 0 |
| self._misses = 0 |
| self._byte_hits = 0 |
| |
| self._cache_creation_time = self._gettime() |
| |
| @apiproxy_stub.Synchronized |
| def _GetKey(self, namespace, key): |
| """Retrieves a CacheEntry from the cache if it hasn't expired. |
| |
| Does not take deletion timeout into account. |
| |
| Args: |
| namespace: The namespace that keys are stored under. |
| key: The key to retrieve from the cache. |
| |
| Returns: |
| The corresponding CacheEntry instance, or None if it was not found or |
| has already expired. |
| """ |
| namespace_dict = self._the_cache.get(namespace, None) |
| if namespace_dict is None: |
| return None |
| entry = namespace_dict.get(key, None) |
| if entry is None: |
| return None |
| elif entry.CheckExpired(): |
| del namespace_dict[key] |
| return None |
| else: |
| return entry |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_Get(self, request, response): |
| """Implementation of MemcacheService::Get(). |
| |
| Args: |
| request: A MemcacheGetRequest. |
| response: A MemcacheGetResponse. |
| """ |
| namespace = request.name_space() |
| keys = set(request.key_list()) |
| for key in keys: |
| entry = self._GetKey(namespace, key) |
| if entry is None or entry.CheckLocked(): |
| self._misses += 1 |
| continue |
| self._hits += 1 |
| self._byte_hits += len(entry.value) |
| item = response.add_item() |
| item.set_key(key) |
| item.set_value(entry.value) |
| item.set_flags(entry.flags) |
| if request.for_cas(): |
| item.set_cas_id(entry.cas_id) |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_Set(self, request, response): |
| """Implementation of MemcacheService::Set(). |
| |
| Args: |
| request: A MemcacheSetRequest. |
| response: A MemcacheSetResponse. |
| """ |
| namespace = request.name_space() |
| for item in request.item_list(): |
| key = item.key() |
| set_policy = item.set_policy() |
| old_entry = self._GetKey(namespace, key) |
| |
| set_status = MemcacheSetResponse.NOT_STORED |
| if ((set_policy == MemcacheSetRequest.SET) or |
| (set_policy == MemcacheSetRequest.ADD and old_entry is None) or |
| (set_policy == MemcacheSetRequest.REPLACE and old_entry is not None)): |
| |
| |
| if (old_entry is None or |
| set_policy == MemcacheSetRequest.SET |
| or not old_entry.CheckLocked()): |
| set_status = MemcacheSetResponse.STORED |
| |
| elif (set_policy == MemcacheSetRequest.CAS and item.has_cas_id()): |
| if old_entry is None or old_entry.CheckLocked(): |
| set_status = MemcacheSetResponse.NOT_STORED |
| elif old_entry.cas_id != item.cas_id(): |
| set_status = MemcacheSetResponse.EXISTS |
| else: |
| set_status = MemcacheSetResponse.STORED |
| |
| if set_status == MemcacheSetResponse.STORED: |
| if namespace not in self._the_cache: |
| self._the_cache[namespace] = {} |
| self._the_cache[namespace][key] = CacheEntry(item.value(), |
| item.expiration_time(), |
| item.flags(), |
| self._next_cas_id, |
| gettime=self._gettime) |
| self._next_cas_id += 1 |
| |
| response.add_set_status(set_status) |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_Delete(self, request, response): |
| """Implementation of MemcacheService::Delete(). |
| |
| Args: |
| request: A MemcacheDeleteRequest. |
| response: A MemcacheDeleteResponse. |
| """ |
| namespace = request.name_space() |
| for item in request.item_list(): |
| key = item.key() |
| entry = self._GetKey(namespace, key) |
| |
| delete_status = MemcacheDeleteResponse.DELETED |
| if entry is None: |
| delete_status = MemcacheDeleteResponse.NOT_FOUND |
| elif item.delete_time() == 0: |
| del self._the_cache[namespace][key] |
| else: |
| |
| entry.ExpireAndLock(item.delete_time()) |
| |
| response.add_delete_status(delete_status) |
| |
| @apiproxy_stub.Synchronized |
| def _internal_increment(self, namespace, request): |
| """Internal function for incrementing from a MemcacheIncrementRequest. |
| |
| Args: |
| namespace: A string containing the namespace for the request, if any. |
| Pass an empty string if there is no namespace. |
| request: A MemcacheIncrementRequest instance. |
| |
| Returns: |
| An integer or long if the offset was successful, None on error. |
| """ |
| key = request.key() |
| entry = self._GetKey(namespace, key) |
| if entry is None or entry.CheckLocked(): |
| if not request.has_initial_value(): |
| return None |
| if namespace not in self._the_cache: |
| self._the_cache[namespace] = {} |
| flags = 0 |
| if request.has_initial_flags(): |
| flags = request.initial_flags() |
| self._the_cache[namespace][key] = CacheEntry(str(request.initial_value()), |
| expiration=0, |
| flags=flags, |
| cas_id=self._next_cas_id, |
| gettime=self._gettime) |
| self._next_cas_id += 1 |
| entry = self._GetKey(namespace, key) |
| assert entry is not None |
| |
| try: |
| old_value = long(entry.value) |
| if old_value < 0: |
| |
| |
| |
| |
| raise ValueError |
| except ValueError: |
| logging.error('Increment/decrement failed: Could not interpret ' |
| 'value for key = "%s" as an unsigned integer.', key) |
| return None |
| |
| delta = request.delta() |
| if request.direction() == MemcacheIncrementRequest.DECREMENT: |
| delta = -delta |
| |
| |
| new_value = max(old_value + delta, 0) % (2**64) |
| |
| entry.value = str(new_value) |
| return new_value |
| |
| def _Dynamic_Increment(self, request, response): |
| """Implementation of MemcacheService::Increment(). |
| |
| Args: |
| request: A MemcacheIncrementRequest. |
| response: A MemcacheIncrementResponse. |
| """ |
| namespace = request.name_space() |
| new_value = self._internal_increment(namespace, request) |
| if new_value is None: |
| raise apiproxy_errors.ApplicationError( |
| memcache_service_pb.MemcacheServiceError.UNSPECIFIED_ERROR) |
| response.set_new_value(new_value) |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_BatchIncrement(self, request, response): |
| """Implementation of MemcacheService::BatchIncrement(). |
| |
| Args: |
| request: A MemcacheBatchIncrementRequest. |
| response: A MemcacheBatchIncrementResponse. |
| """ |
| namespace = request.name_space() |
| for request_item in request.item_list(): |
| new_value = self._internal_increment(namespace, request_item) |
| item = response.add_item() |
| if new_value is None: |
| item.set_increment_status(MemcacheIncrementResponse.NOT_CHANGED) |
| else: |
| item.set_increment_status(MemcacheIncrementResponse.OK) |
| item.set_new_value(new_value) |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_FlushAll(self, request, response): |
| """Implementation of MemcacheService::FlushAll(). |
| |
| Args: |
| request: A MemcacheFlushRequest. |
| response: A MemcacheFlushResponse. |
| """ |
| self._the_cache.clear() |
| self._ResetStats() |
| |
| @apiproxy_stub.Synchronized |
| def _Dynamic_Stats(self, request, response): |
| """Implementation of MemcacheService::Stats(). |
| |
| Args: |
| request: A MemcacheStatsRequest. |
| response: A MemcacheStatsResponse. |
| """ |
| stats = response.mutable_stats() |
| stats.set_hits(self._hits) |
| stats.set_misses(self._misses) |
| stats.set_byte_hits(self._byte_hits) |
| items = 0 |
| total_bytes = 0 |
| for namespace in self._the_cache.itervalues(): |
| items += len(namespace) |
| for entry in namespace.itervalues(): |
| total_bytes += len(entry.value) |
| stats.set_items(items) |
| stats.set_bytes(total_bytes) |
| |
| |
| stats.set_oldest_item_age(self._gettime() - self._cache_creation_time) |
