java/src/main/com/google/appengine/api/datastore/DatastoreBackedEntityCachingStrategy.java

   1 package com.google.appengine.api.datastore;
   2
   3 import static com.google.appengine.api.datastore.FutureHelper.quietGet;
   4
   5 import com.google.appengine.api.datastore.EntityCache.CasCacheValues;
   6 import com.google.appengine.api.datastore.EntityCache.IdentifiableCacheValue;
   7 import com.google.appengine.api.datastore.ReadPolicy.Consistency;
   8 import com.google.appengine.api.memcache.MemcacheService.SetPolicy;
   9 import com.google.apphosting.datastore.EntityStorage.CacheValue;
  10 import com.google.apphosting.datastore.EntityStorage.CacheValue.State;
  11 import com.google.common.collect.ImmutableList;
  12 import com.google.common.collect.ImmutableMap;
  13 import com.google.common.collect.ImmutableSet;
  14 import com.google.common.collect.Maps;
  15 import com.google.common.collect.Sets;
  16
  17 import java.util.HashSet;
  18 import java.util.Iterator;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Set;
  22 import java.util.concurrent.Future;
  23
  24 /**
  25  * A datastore-backed entity caching strategy implementation. Implements the policy specified by
  26  * {@link EntityCachePolicy#CACHE}.
  27  *
  28  */
  29 final class DatastoreBackedEntityCachingStrategy extends EntityCachingStrategy {
  30
  31   private final EntityCache cache;
  32   private final DatastoreServiceConfig datastoreServiceConfig;
  33
  34   DatastoreBackedEntityCachingStrategy(DatastoreServiceConfig datastoreServiceConfig) {
  35     MemcacheServiceHelper memcacheServiceHelper =
  36         MemcacheServiceHelper.Builder.withLogAndAbsorbMemcacheServiceErrors();
  37     double datastoreRpcDeadline = (datastoreServiceConfig.getDeadline() == null) ?
  38         DEFAULT_DATASTORE_RPC_DEADLINE_SECS : datastoreServiceConfig.getDeadline();
  39     cache = new EntityCache(memcacheServiceHelper, datastoreRpcDeadline);
  40     this.datastoreServiceConfig = datastoreServiceConfig;
  41   }
  42
  43   DatastoreBackedEntityCachingStrategy(EntityCache cache,
  44       DatastoreServiceConfig datastoreServiceConfig) {
  45     this.cache = cache;
  46     this.datastoreServiceConfig = datastoreServiceConfig;
  47   }
  48
  49   @Override
  50   public PreGetCachingResult preGet(CurrentTransactionProvider currentTxnProvider,
  51       List<Key> keysToGet, Map<Key, Entity> resultMap) {
  52     Set<Key> keysToSkipLoading = Sets.newHashSet();
  53     Future<Set<Key>> readInProgressKeys =
  54         new FutureHelper.FakeFuture<Set<Key>>(ImmutableSet.<Key>of());
  55     if ((currentTxnProvider.getCurrentTransaction(null) != null) ||
  56         (datastoreServiceConfig.getReadPolicy().getConsistency() == Consistency.EVENTUAL)) {
  57       return new DatastoreBackedPreGetCachingResult(keysToSkipLoading, readInProgressKeys);
  58     }
  59     List<Key> cacheableKeys = getCacheableKeys(datastoreServiceConfig, keysToGet);
  60     Map<Key, IdentifiableCacheValue> getResults = quietGet(
  61         cache.getIdentifiableAsync(cacheableKeys));
  62     HashSet<Key> uncachedKeys = Sets.newHashSet(cacheableKeys);
  63     for (Map.Entry<Key, IdentifiableCacheValue> result : getResults.entrySet()) {
  64       Key key = result.getKey();
  65       uncachedKeys.remove(key);
  66       CacheValue value = result.getValue().getValue();
  67       if (value.getStateEnum() == State.ENTITY) {
  68         keysToSkipLoading.add(key);
  69         if (value.hasEntity()) {
  70           resultMap.put(key, EntityTranslator.createFromPb(value.getEntity().getV3Entity()));
  71         }
  72       }
  73     }
  74     if (!uncachedKeys.isEmpty()) {
  75       readInProgressKeys = cache.putStateAsync(
  76           uncachedKeys, State.READ_IN_PROGRESS, SetPolicy.ADD_ONLY_IF_NOT_PRESENT);
  77     }
  78     return new DatastoreBackedPreGetCachingResult(keysToSkipLoading, readInProgressKeys);
  79   }
  80
  81   @Override
  82   protected void postGet(PreGetCachingResult preGetResult, Map<Key, Entity> resultMap) {
  83     DatastoreBackedPreGetCachingResult preGetStrategyResult =
  84         (DatastoreBackedPreGetCachingResult) preGetResult;
  85     Set<Key> readInProgressKeys = quietGet(preGetStrategyResult.getReadInProgressKeys());
  86     if (!readInProgressKeys.isEmpty()) {
  87       Map<Key, IdentifiableCacheValue> readInProgressCacheValues =
  88           quietGet(cache.getIdentifiableAsync(readInProgressKeys));
  89       Map<Key, CasCacheValues> cacheValuesToUpdate =
  90           Maps.newHashMapWithExpectedSize(readInProgressCacheValues.size());
  91       for (Map.Entry<Key, IdentifiableCacheValue> readInProgressMapEntry :
  92           readInProgressCacheValues.entrySet()) {
  93         Key key = readInProgressMapEntry.getKey();
  94         IdentifiableCacheValue oldIdCacheValue = readInProgressMapEntry.getValue();
  95         CacheValue oldCacheValue = oldIdCacheValue.getValue();
  96         if (oldCacheValue.getStateEnum() != State.READ_IN_PROGRESS) {
  97           continue;
  98         }
  99         Entity entity = resultMap.get(key);
 100         CacheValue newCacheValue = CacheValueUtil.createCacheValue(State.ENTITY,
 101             (entity == null) ? null : EntityTranslator.convertToPb(entity));
 102         cacheValuesToUpdate.put(key, new CasCacheValues(oldIdCacheValue, newCacheValue));
 103       }
 104       if (!cacheValuesToUpdate.isEmpty()) {
 105         cache.putIfUntouchedAsync(cacheValuesToUpdate);
 106       }
 107     }
 108   }
 109
 110   @Override
 111   public PreMutationCachingResult prePut(CurrentTransactionProvider currentTxnProvider,
 112       List<Entity> entitiesToPut) {
 113     if (currentTxnProvider.getCurrentTransaction(null) == null) {
 114       return preCommit(entitiesToPut, ImmutableList.<Key>of());
 115     } else {
 116       return new PreMutationCachingResult(ImmutableSet.<Key>of());
 117     }
 118   }
 119
 120   @Override
 121   public PreMutationCachingResult preDelete(CurrentTransactionProvider currentTxnProvider,
 122       List<Key> keysToDelete) {
 123     if (currentTxnProvider.getCurrentTransaction(null) == null) {
 124       return preCommit(ImmutableList.<Entity>of(), keysToDelete);
 125     } else {
 126       return new PreMutationCachingResult(ImmutableSet.<Key>of());
 127     }
 128   }
 129
 130   @Override
 131   public PreMutationCachingResult preCommit(List<Entity> entitiesToPut, List<Key> keysToDelete) {
 132     List<Entity> cacheableEntitiesToPut =
 133         getCacheableEntities(datastoreServiceConfig, entitiesToPut);
 134     List<Key> cacheableKeysToDelete = getCacheableKeys(datastoreServiceConfig, keysToDelete);
 135     Set<Key> completeKeys = Sets.newHashSetWithExpectedSize(
 136         cacheableEntitiesToPut.size() + cacheableKeysToDelete.size());
 137     completeKeys.addAll(cacheableKeysToDelete);
 138     for (Entity entity : cacheableEntitiesToPut) {
 139       Key key = entity.getKey();
 140       if (key.isComplete()) {
 141         completeKeys.add(key);
 142       }
 143     }
 144     Map<Key, IdentifiableCacheValue> mutationInProgressCacheValues;
 145     if (completeKeys.isEmpty()) {
 146       mutationInProgressCacheValues = ImmutableMap.of();
 147     } else {
 148       mutationInProgressCacheValues = cache.putStateIdentifiable(
 149           completeKeys, State.MUTATION_IN_PROGRESS, SetPolicy.SET_ALWAYS);
 150     }
 151     return new DatastoreBackedPreMutationCachingResult(completeKeys, mutationInProgressCacheValues);
 152   }
 153
 154   @Override
 155   protected void postMutation(PreMutationCachingResult preMutationResult) {
 156     DatastoreBackedPreMutationCachingResult preMutationStrategyResult =
 157         (DatastoreBackedPreMutationCachingResult) preMutationResult;
 158     if (preMutationStrategyResult.getMutationKeys().isEmpty()) {
 159       return;
 160     }
 161     Set<Key> updatedKeys = ImmutableSet.of();
 162     if (!preMutationStrategyResult.getMutationInProgressCacheValues().isEmpty()) {
 163       updatedKeys = quietGet(cache.evictIfUntouchedAsync(
 164           preMutationStrategyResult.getMutationInProgressCacheValues()));
 165     }
 166     Set<Key> nonUpdatedKeys = preMutationStrategyResult.getMutationKeys();
 167     nonUpdatedKeys.removeAll(updatedKeys);
 168     if (!nonUpdatedKeys.isEmpty()) {
 169       Map<Key, IdentifiableCacheValue> cacheEntriesNotUpdated =
 170           quietGet(cache.getIdentifiableAsync(nonUpdatedKeys));
 171       Iterator<Map.Entry<Key, IdentifiableCacheValue>> cacheEntriesNotUpdatedIter =
 172           cacheEntriesNotUpdated.entrySet().iterator();
 173       while (cacheEntriesNotUpdatedIter.hasNext()) {
 174         Map.Entry<Key, IdentifiableCacheValue> cacheEntry =
 175             cacheEntriesNotUpdatedIter.next();
 176         CacheValue value = cacheEntry.getValue().getValue();
 177         if ((value.getStateEnum() != State.ENTITY) &&
 178             (value.getStateEnum() != State.READ_IN_PROGRESS)) {
 179           cacheEntriesNotUpdatedIter.remove();
 180         }
 181       }
 182       quietGet(cache.evictIfUntouchedAsync(cacheEntriesNotUpdated));
 183     }
 184   }
 185
 186   static final class DatastoreBackedPreGetCachingResult extends PreGetCachingResult {
 187     private final Future<Set<Key>> readInProgressKeys;
 188
 189     public DatastoreBackedPreGetCachingResult(Set<Key> keysToSkipLoading,
 190         Future<Set<Key>> readInProgressKeys) {
 191       super(keysToSkipLoading);
 192       this.readInProgressKeys = readInProgressKeys;
 193     }
 194
 195     public Future<Set<Key>> getReadInProgressKeys() {
 196       return readInProgressKeys;
 197     }
 198   }
 199
 200   static final class DatastoreBackedPreMutationCachingResult
 201       extends PreMutationCachingResult {
 202     private final Set<Key> mutationKeys;
 203     private final Map<Key, IdentifiableCacheValue> mutationInProgressCacheValues;
 204
 205     public DatastoreBackedPreMutationCachingResult(Set<Key> mutationKeys,
 206         Map<Key, IdentifiableCacheValue> mutationInProgressCacheValues) {
 207       super(ImmutableSet.<Key>of());
 208       this.mutationKeys = mutationKeys;
 209       this.mutationInProgressCacheValues = mutationInProgressCacheValues;
 210     }
 211
 212     public Set<Key> getMutationKeys() {
 213       return mutationKeys;
 214     }
 215
 216     public Map<Key, IdentifiableCacheValue> getMutationInProgressCacheValues() {
 217       return mutationInProgressCacheValues;
 218     }
 219   }
 220 }