add minimumSharedCache param + more tests.

src/java/voldemort/server/VoldemortConfig.java

@@ -86,6 +86,7 @@ public class VoldemortConfig implements Serializable {
     private boolean bdbFairLatches;
     private long bdbStatsCacheTtlMs;
     private boolean bdbExposeSpaceUtilization;
+    private long bdbMinimumSharedCache;
 
     private String mysqlUsername;
     private String mysqlPassword;
@@ -228,6 +229,7 @@ public VoldemortConfig(Props props) {
         this.bdbReadUncommitted = props.getBoolean("bdb.lock.read_uncommitted", true);
         this.bdbStatsCacheTtlMs = props.getLong("bdb.stats.cache.ttl.ms", 5 * Time.MS_PER_SECOND);
         this.bdbExposeSpaceUtilization = props.getBoolean("bdb.expose.space.utilization", true);
+        this.bdbMinimumSharedCache = props.getLong("bdb.minimum.shared.cache", 0);
 
         this.readOnlyBackups = props.getInt("readonly.backups", 1);
         this.readOnlySearchStrategy = props.getString("readonly.search.strategy",
@@ -1167,6 +1169,14 @@ public void setBdbStatsCacheTtlMs(long statsCacheTtlMs) {
         this.bdbStatsCacheTtlMs = statsCacheTtlMs;
     }
 
+    public long getBdbMinimumSharedCache() {
+        return this.bdbMinimumSharedCache;
+    }
+
+    public void setBdbMinimumSharedCache(long minimumSharedCache) {
+        this.bdbMinimumSharedCache = minimumSharedCache;
+    }
+
     public int getSchedulerThreads() {
         return schedulerThreads;
     }

src/java/voldemort/server/protocol/admin/AdminServiceRequestHandler.java

@@ -1510,6 +1510,7 @@ public VAdminProto.ReserveMemoryResponse handleReserveMemory(VAdminProto.Reserve
 
                             storeDefList.set(i, newStoreDef);
                             storageService.updateStore(newStoreDef);
+                            break;
                         }
                     }
 

src/java/voldemort/store/StoreDefinition.java

@@ -388,7 +388,8 @@ && getRoutingPolicy() == def.getRoutingPolicy()
                                 def.getSerializerFactory() != null ? def.getSerializerFactory()
                                                                   : null)
                && Objects.equal(getHintedHandoffStrategyType(), def.getHintedHandoffStrategyType())
-               && Objects.equal(getHintPrefListSize(), def.getHintPrefListSize());
+               && Objects.equal(getHintPrefListSize(), def.getHintPrefListSize())
+               && Objects.equal(getMemoryFootprintMB(), def.getMemoryFootprintMB());
     }
 
     @Override
@@ -419,7 +420,8 @@ public int hashCode() {
                                 hasHintedHandoffStrategyType() ? getHintedHandoffStrategyType()
                                                               : null,
                                 hasHintPreflistSize() ? getHintPrefListSize() : null,
-                                getOwners());
+                                getOwners(),
+                                getMemoryFootprintMB());
     }
 
     @Override
@@ -439,6 +441,6 @@ public String toString() {
                + getZoneCountWrites() + ", serializer factory = " + getSerializerFactory() + ")"
                + ", hinted-handoff-strategy = " + getHintedHandoffStrategyType()
                + ", hint-preflist-size = " + getHintPrefListSize() + ", owners = " + getOwners()
-               + ")";
+               + ", memory-footprint(MB)" + getMemoryFootprintMB() + ")";
     }
 }

src/java/voldemort/store/bdb/BdbStorageConfiguration.java

@@ -32,6 +32,7 @@
 import voldemort.store.StorageInitializationException;
 import voldemort.store.StoreDefinition;
 import voldemort.utils.ByteArray;
+import voldemort.utils.ByteUtils;
 import voldemort.utils.JmxUtils;
 import voldemort.utils.Time;
 
@@ -58,24 +59,20 @@ public class BdbStorageConfiguration implements StorageConfiguration {
     private static final String SHARED_ENV_KEY = "shared";
 
     private static Logger logger = Logger.getLogger(BdbStorageConfiguration.class);
-    private static final long BYTES_PER_MB = 1048576;
-
     private final Object lock = new Object();
     private final Map<String, Environment> environments = Maps.newHashMap();
     private final EnvironmentConfig environmentConfig;
     private final DatabaseConfig databaseConfig;
     private final String bdbMasterDir;
     private final boolean useOneEnvPerStore;
     private final VoldemortConfig voldemortConfig;
-    private final long totalCacheSize;
     private long reservedCacheSize = 0;
     private Set<Environment> unreservedStores;
 
     public BdbStorageConfiguration(VoldemortConfig config) {
         this.voldemortConfig = config;
         environmentConfig = new EnvironmentConfig();
         environmentConfig.setTransactional(true);
-        totalCacheSize = config.getBdbCacheSize();
         if(config.isBdbWriteTransactionsEnabled() && config.isBdbFlushTransactionsEnabled()) {
             environmentConfig.setDurability(Durability.COMMIT_SYNC);
         } else if(config.isBdbWriteTransactionsEnabled() && !config.isBdbFlushTransactionsEnabled()) {
@@ -155,7 +152,7 @@ public StorageEngine<ByteArray, byte[], byte[]> getStore(StoreDefinition storeDe
      * 
      */
     private void adjustCacheSizes() {
-        long newSharedCacheSize = this.totalCacheSize - this.reservedCacheSize;
+        long newSharedCacheSize = voldemortConfig.getBdbCacheSize() - this.reservedCacheSize;
         logger.info("Setting the shared cache size to " + newSharedCacheSize);
         for(Environment environment: unreservedStores) {
             EnvironmentMutableConfig mConfig = environment.getMutableConfig();
@@ -180,14 +177,27 @@ public Environment getEnvironment(StoreDefinition storeDef) throws DatabaseExcep
                 // configure the BDB cache
                 if(storeDef.hasMemoryFootprint()) {
                     // make room for the reservation, by adjusting other stores
-                    long reservedBytes = storeDef.getMemoryFootprintMB() * BYTES_PER_MB;
-                    this.reservedCacheSize += reservedBytes;
+                    long reservedBytes = storeDef.getMemoryFootprintMB() * ByteUtils.BYTES_PER_MB;
+                    long newReservedCacheSize = this.reservedCacheSize + reservedBytes;
+
+                    // check that we leave a 'minimum' shared cache
+                    if((voldemortConfig.getBdbCacheSize() - newReservedCacheSize) < voldemortConfig.getBdbMinimumSharedCache()) {
+                        throw new StorageInitializationException("Reservation of "
+                                                                 + storeDef.getMemoryFootprintMB()
+                                                                 + " MB for store "
+                                                                 + storeName
+                                                                 + " violates minimum shared cache size of "
+                                                                 + voldemortConfig.getBdbMinimumSharedCache());
+                    }
+
+                    this.reservedCacheSize = newReservedCacheSize;
                     adjustCacheSizes();
                     environmentConfig.setSharedCache(false);
                     environmentConfig.setCacheSize(reservedBytes);
                 } else {
                     environmentConfig.setSharedCache(true);
-                    environmentConfig.setCacheSize(this.totalCacheSize - this.reservedCacheSize);
+                    environmentConfig.setCacheSize(voldemortConfig.getBdbCacheSize()
+                                                   - this.reservedCacheSize);
                 }
 
                 Environment environment = new Environment(bdbDir, environmentConfig);
@@ -307,14 +317,28 @@ public void update(StoreDefinition storeDef) {
         if(!useOneEnvPerStore)
             throw new VoldemortException("Memory foot print can be set only when using different environments per store");
 
-        Environment environment = environments.get(storeDef.getName());
+        String storeName = storeDef.getName();
+        Environment environment = environments.get(storeName);
         // change reservation amount of reserved store
         if(!unreservedStores.contains(environment) && storeDef.hasMemoryFootprint()) {
             EnvironmentMutableConfig mConfig = environment.getMutableConfig();
             long currentCacheSize = mConfig.getCacheSize();
-            long newCacheSize = storeDef.getMemoryFootprintMB() * BYTES_PER_MB;
+            long newCacheSize = storeDef.getMemoryFootprintMB() * ByteUtils.BYTES_PER_MB;
             if(currentCacheSize != newCacheSize) {
-                this.reservedCacheSize = this.reservedCacheSize - currentCacheSize + newCacheSize;
+                long newReservedCacheSize = this.reservedCacheSize - currentCacheSize
+                                            + newCacheSize;
+
+                // check that we leave a 'minimum' shared cache
+                if((voldemortConfig.getBdbCacheSize() - newReservedCacheSize) < voldemortConfig.getBdbMinimumSharedCache()) {
+                    throw new StorageInitializationException("Reservation of "
+                                                             + storeDef.getMemoryFootprintMB()
+                                                             + " MB for store "
+                                                             + storeName
+                                                             + " violates minimum shared cache size of "
+                                                             + voldemortConfig.getBdbMinimumSharedCache());
+                }
+
+                this.reservedCacheSize = newReservedCacheSize;
                 adjustCacheSizes();
                 mConfig.setCacheSize(newCacheSize);
                 environment.setMutableConfig(mConfig);
@@ -327,4 +351,8 @@ public void update(StoreDefinition storeDef) {
             throw new VoldemortException("Cannot switch between shared and private cache dynamically");
         }
     }
+
+    public long getReservedCacheSize() {
+        return this.reservedCacheSize;
+    }
 }

src/java/voldemort/utils/ByteUtils.java

@@ -53,6 +53,9 @@ public class ByteUtils {
     public static final int MASK_00111111 = Integer.parseInt("00111111", 2);
     public static final int MASK_00011111 = Integer.parseInt("00011111", 2);
 
+    public static final int BYTES_PER_MB = 1048576;
+    public static final long BYTES_PER_GB = 1073741824;
+
     public static MessageDigest getDigest(String algorithm) {
         try {
             return MessageDigest.getInstance(algorithm);