finer timeouts and partial getalls.

src/java/voldemort/client/AbstractStoreClientFactory.java

@@ -114,7 +114,7 @@ public AbstractStoreClientFactory(ClientConfig config) {
         this.clientZoneId = config.getClientZoneId();
         this.routedStoreFactory = new RoutedStoreFactory(config.isPipelineRoutedStoreEnabled(),
                                                          threadPool,
-                                                         config.getRoutingTimeout(TimeUnit.MILLISECONDS));
+                                                         config.getTimeoutConfig());
 
         if(this.isJmxEnabled) {
             JmxUtils.registerMbean(threadPool,

src/java/voldemort/client/ClientConfig.java

@@ -36,6 +36,7 @@
 import voldemort.utils.ConfigurationException;
 import voldemort.utils.Props;
 import voldemort.utils.ReflectUtils;
+import voldemort.utils.TimeoutConfig;
 import voldemort.utils.Utils;
 
 /**
@@ -55,6 +56,7 @@ public class ClientConfig {
     private volatile boolean socketKeepAlive = false;
     private volatile int selectors = 8;
     private volatile long routingTimeoutMs = 15000;
+    private volatile TimeoutConfig timeoutConfig = new TimeoutConfig(routingTimeoutMs, false);
     private volatile int socketBufferSize = 64 * 1024;
     private volatile SerializerFactory serializerFactory = new DefaultSerializerFactory();
     private volatile List<String> bootstrapUrls = null;
@@ -91,6 +93,12 @@ public ClientConfig() {}
     public static final String SOCKET_KEEPALIVE_PROPERTY = "socket_keepalive";
     public static final String SELECTORS_PROPERTY = "selectors";
     public static final String ROUTING_TIMEOUT_MS_PROPERTY = "routing_timeout_ms";
+    public static final String GETALL_ROUTING_TIMEOUT_MS_PROPERTY = "getall_routing_timeout_ms";
+    public static final String PUT_ROUTING_TIMEOUT_MS_PROPERTY = "put_routing_timeout_ms";
+    public static final String GET_ROUTING_TIMEOUT_MS_PROPERTY = "get_routing_timeout_ms";
+    public static final String GET_VERSIONS_ROUTING_TIMEOUT_MS_PROPERTY = "getversions_routing_timeout_ms";
+    public static final String DELETE_ROUTING_TIMEOUT_MS_PROPERTY = "delete_routing_timeout_ms";
+    public static final String ALLOW_PARTIAL_GETALLS_PROPERTY = "allow_partial_getalls";
     public static final String NODE_BANNAGE_MS_PROPERTY = "node_bannage_ms";
     public static final String SOCKET_BUFFER_SIZE_PROPERTY = "socket_buffer_size";
     public static final String SERIALIZER_FACTORY_CLASS_PROPERTY = "serializer_factory_class";
@@ -174,6 +182,36 @@ private void setProperties(Properties properties) {
         if(props.containsKey(ROUTING_TIMEOUT_MS_PROPERTY))
             this.setRoutingTimeout(props.getInt(ROUTING_TIMEOUT_MS_PROPERTY), TimeUnit.MILLISECONDS);
 
+        // By default, make all the timeouts equal to routing timeout
+        timeoutConfig = new TimeoutConfig(routingTimeoutMs, false);
+
+        if(props.containsKey(GETALL_ROUTING_TIMEOUT_MS_PROPERTY))
+            timeoutConfig.getAllTimeoutMs(props.getInt(GETALL_ROUTING_TIMEOUT_MS_PROPERTY),
+                                          TimeUnit.MILLISECONDS);
+
+        if(props.containsKey(GET_ROUTING_TIMEOUT_MS_PROPERTY))
+            timeoutConfig.getTimeoutMs(props.getInt(GET_ROUTING_TIMEOUT_MS_PROPERTY),
+                                       TimeUnit.MILLISECONDS);
+
+        if(props.containsKey(PUT_ROUTING_TIMEOUT_MS_PROPERTY)) {
+            long putTimeoutMs = props.getInt(PUT_ROUTING_TIMEOUT_MS_PROPERTY);
+            timeoutConfig.putTimeoutMs(putTimeoutMs, TimeUnit.MILLISECONDS);
+            // By default, use the same thing for getVersions() also
+            timeoutConfig.getVersionsTimeoutMs(putTimeoutMs, TimeUnit.MILLISECONDS);
+        }
+
+        // of course, if someone overrides it, we will respect that
+        if(props.containsKey(GET_VERSIONS_ROUTING_TIMEOUT_MS_PROPERTY))
+            timeoutConfig.getVersionsTimeoutMs(props.getInt(GET_VERSIONS_ROUTING_TIMEOUT_MS_PROPERTY),
+                                               TimeUnit.MILLISECONDS);
+
+        if(props.containsKey(DELETE_ROUTING_TIMEOUT_MS_PROPERTY))
+            timeoutConfig.deleteTimeoutMs(props.getInt(DELETE_ROUTING_TIMEOUT_MS_PROPERTY),
+                                          TimeUnit.MILLISECONDS);
+
+        if(props.containsKey(ALLOW_PARTIAL_GETALLS_PROPERTY))
+            timeoutConfig.setPartialGetAllAllowed(props.getBoolean(ALLOW_PARTIAL_GETALLS_PROPERTY));
+
         if(props.containsKey(SOCKET_BUFFER_SIZE_PROPERTY))
             this.setSocketBufferSize(props.getInt(SOCKET_BUFFER_SIZE_PROPERTY));
 
@@ -324,6 +362,26 @@ public ClientConfig setRoutingTimeout(int routingTimeout, TimeUnit unit) {
         return this;
     }
 
+    /**
+     * Set the timeout configuration for the voldemort operations
+     * 
+     * @param tConfig
+     * @return
+     */
+    public ClientConfig setTimeoutConfig(TimeoutConfig tConfig) {
+        this.timeoutConfig = tConfig;
+        return this;
+    }
+
+    /**
+     * Get the timeouts for voldemort operations
+     * 
+     * @return
+     */
+    public TimeoutConfig getTimeoutConfig() {
+        return timeoutConfig;
+    }
+
     /**
      * @deprecated Use {@link #getFailureDetectorBannagePeriod()} instead
      */
@@ -517,7 +575,7 @@ public boolean isLazyEnabled() {
 
     /**
      * Enable lazy initialization of clients?
-     *
+     * 
      * @param enableLazy If true clients will be lazily initialized
      */
     public ClientConfig setEnableLazy(boolean enableLazy) {

src/java/voldemort/server/VoldemortConfig.java

@@ -21,6 +21,7 @@
 import java.io.Serializable;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
 import voldemort.client.protocol.RequestFormatType;
 import voldemort.cluster.failuredetector.FailureDetectorConfig;
@@ -34,6 +35,7 @@
 import voldemort.utils.ConfigurationException;
 import voldemort.utils.Props;
 import voldemort.utils.Time;
+import voldemort.utils.TimeoutConfig;
 import voldemort.utils.UndefinedPropertyException;
 import voldemort.utils.Utils;
 
@@ -112,6 +114,7 @@ public class VoldemortConfig implements Serializable {
 
     private int clientSelectors;
     private int clientRoutingTimeoutMs;
+    private TimeoutConfig clientTimeoutConfig;
     private int clientMaxConnectionsPerNode;
     private int clientConnectionTimeoutMs;
     private int clientMaxThreads;
@@ -275,6 +278,24 @@ public VoldemortConfig(Props props) {
         this.clientMaxConnectionsPerNode = props.getInt("client.max.connections.per.node", 50);
         this.clientConnectionTimeoutMs = props.getInt("client.connection.timeout.ms", 500);
         this.clientRoutingTimeoutMs = props.getInt("client.routing.timeout.ms", 15000);
+        this.clientTimeoutConfig = new TimeoutConfig(this.clientRoutingTimeoutMs, false);
+        this.clientTimeoutConfig.getTimeoutMs(props.getInt("client.routing.get.timeout.ms",
+                                                           this.clientRoutingTimeoutMs),
+                                              TimeUnit.MILLISECONDS);
+        this.clientTimeoutConfig.getAllTimeoutMs(props.getInt("client.routing.getall.timeout.ms",
+                                                              this.clientRoutingTimeoutMs),
+                                                 TimeUnit.MILLISECONDS);
+        this.clientTimeoutConfig.putTimeoutMs(props.getInt("client.routing.put.timeout.ms",
+                                                           this.clientRoutingTimeoutMs),
+                                              TimeUnit.MILLISECONDS);
+        this.clientTimeoutConfig.getVersionsTimeoutMs(props.getLong("client.routing.getversions.timeout.ms",
+                                                                    this.clientTimeoutConfig.putTimeoutMs()),
+                                                      TimeUnit.MILLISECONDS);
+        this.clientTimeoutConfig.deleteTimeoutMs(props.getInt("client.routing.delete.timeout.ms",
+                                                              this.clientRoutingTimeoutMs),
+                                                 TimeUnit.MILLISECONDS);
+        this.clientTimeoutConfig.setPartialGetAllAllowed(props.getBoolean("client.routing.allow.partial.getall",
+                                                                          false));
         this.clientMaxThreads = props.getInt("client.max.threads", 500);
         this.clientThreadIdleMs = props.getInt("client.thread.idle.ms", 100000);
         this.clientMaxQueuedRequests = props.getInt("client.max.queued.requests", 1000);
@@ -997,6 +1018,10 @@ public void setClientRoutingTimeoutMs(int routingTimeoutMs) {
         this.clientRoutingTimeoutMs = routingTimeoutMs;
     }
 
+    public TimeoutConfig getTimeoutConfig() {
+        return this.clientTimeoutConfig;
+    }
+
     public int getClientMaxConnectionsPerNode() {
         return clientMaxConnectionsPerNode;
     }

src/java/voldemort/server/storage/StorageService.java

@@ -155,7 +155,7 @@ public StorageService(StoreRepository storeRepository,
         this.storeStats = new StoreStats();
         this.routedStoreFactory = new RoutedStoreFactory(voldemortConfig.isPipelineRoutedStoreEnabled(),
                                                          this.clientThreadPool,
-                                                         voldemortConfig.getClientRoutingTimeoutMs());
+                                                         voldemortConfig.getTimeoutConfig());
 
         /*
          * Initialize the dynamic throttle limit based on the per node limit