Changeset 14:8b4b6177d6a2

Timestamp:

04/01/08 15:16:49 (5 months ago)

Author:

ryan@…

Message:

Added: cas/gets, and noreply option on sets

Files:

• jmemcached-cli/src/main/java/com/thimbleware/jmemcached/Main.java (modified) (1 diff)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/Cache.java (modified) (7 diffs)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/CommandDecoder.java (modified) (2 diffs)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/CommandMessage.java (modified) (1 diff)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/Commands.java (modified) (2 diffs)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/MCElement.java (modified) (1 diff)
• jmemcached-core/src/main/java/com/thimbleware/jmemcached/ServerSessionHandler.java (modified) (18 diffs)

jmemcached-cli/src/main/java/com/thimbleware/jmemcached/Main.java

@@ -126 +126 @@
         // create daemon and start it
         MemCacheDaemon daemon = new MemCacheDaemon();
-        LRUCacheStorageDelegate cacheStorage = new LRUCacheStorageDelegate(max_size, max_size, 1024000);
+        LRUCacheStorageDelegate cacheStorage = new LRUCacheStorageDelegate(max_size, maxBytes, 1024000);
         daemon.setCache(new Cache(cacheStorage));
         daemon.setAddr(addr);

jmemcached-core/src/main/java/com/thimbleware/jmemcached/Cache.java

@@ -32 +32 @@
     public int getMisses;
 
+    public long casCounter;
+
     protected CacheStorage cacheStorage;
+
+
+    public enum StoreResponse {
+        STORED, NOT_STORED, EXISTS, NOT_FOUND
+    }
 
 
@@ -40 +47 @@
      */
     private final ReadWriteLock cacheReadWriteLock;
-
 
     /**
@@ -90 +96 @@
      * @return the message response string
      */
-    protected boolean add(MCElement e) {
-        try {
-            startCacheRead();
-            return !is_there(e.keystring) && set(e);
+    protected StoreResponse add(MCElement e) {
+        try {
+            startCacheRead();
+            if (is_there(e.keystring)) return set(e);
+            else return StoreResponse.NOT_STORED;
         } finally {
             finishCacheRead();
@@ -105 +112 @@
      * @return the message response string
      */
-    public boolean replace(MCElement e) {
-        try {
-            startCacheRead();
-            return is_there(e.keystring) && set(e);
-        } finally {
-            finishCacheRead();
-        }
-    }
+    public StoreResponse replace(MCElement e) {
+        try {
+            startCacheRead();
+            if (is_there(e.keystring)) return set(e);
+            else return StoreResponse.NOT_STORED;
+        } finally {
+            finishCacheRead();
+        }
+    }
+
 
     /**
@@ -120 +129 @@
      * @return the message response string
      */
-    protected boolean set(MCElement e) {
+    protected StoreResponse set(MCElement e) {
         try {
             startCacheWrite();
             setCmds += 1;//update stats
+
+            // increment the CAS counter; put in the new CAS
+            e.cas_unique = casCounter++;
+
             this.cacheStorage.put(e.keystring, e);
-            return true;
+            return StoreResponse.STORED;
+        } finally {
+            finishCacheWrite();
+        }
+    }
+
+    public StoreResponse cas(Long cas_key, MCElement e) {
+        try {
+            startCacheWrite();
+            // have to get the element
+            MCElement element = get(e.keystring);
+            if (element == null)
+                return StoreResponse.NOT_FOUND;
+
+            if (element.cas_unique == cas_key) {
+                // cas_unique matches, now set the element
+                return set(e);
+            } else {
+                // cas didn't match; someone else beat us to it
+                return StoreResponse.EXISTS;
+            }
+
         } finally {
             finishCacheWrite();
@@ -157 +191 @@
             e.data = valueOf(old_val).getBytes(); // toString
             e.data_length = e.data.length;
+
+            // assign new cas id
+            e.cas_unique = casCounter++;
+
             this.cacheStorage.put(e.keystring, e); // save new value
             return old_val;
@@ -345 +383 @@
         return getMisses;
     }
+
 }

jmemcached-core/src/main/java/com/thimbleware/jmemcached/CommandDecoder.java

@@ -178 +178 @@
         if (cmd.cmd == Commands.ADD ||
                 cmd.cmd == Commands.SET ||
-                cmd.cmd == Commands.REPLACE) {
+                cmd.cmd == Commands.REPLACE ||
+                cmd.cmd == Commands.CAS) {
 
             // if we don't have all the parts, it's malformed
-            if (parts.size() != 5) {
+            if (parts.size() < 5) {
                 return new SessionStatus(ERROR);
             }
+
 
             int size = Integer.parseInt(parts.get(4));
@@ -196 +198 @@
             cmd.element.data_length = size;
 
+            // look for cas and "noreply" elements
+            if (parts.size() > 5) {
+                int noreply = cmd.cmd == Commands.CAS ? 6 : 5;
+                if (cmd.cmd == Commands.CAS) {
+                    cmd.cas_key = Long.valueOf(parts.get(5));
+                }
+
+                if (parts.size() == noreply + 1 && parts.get(noreply).equalsIgnoreCase("noreply"))
+                    cmd.noreply = true;
+
+            }
+
             return new SessionStatus(WAITING_FOR_DATA, size, cmd);
 
         } else if (cmd.cmd == Commands.GET ||
+                cmd.cmd == Commands.GETS ||
                 cmd.cmd == Commands.INCR ||
                 cmd.cmd == Commands.DECR ||

jmemcached-core/src/main/java/com/thimbleware/jmemcached/CommandMessage.java

@@ -26 +26 @@
     public MCElement element;
     public ArrayList<String> keys;
+    public boolean noreply;
+    public Long cas_key;
 
     public CommandMessage(String cmd) {

jmemcached-core/src/main/java/com/thimbleware/jmemcached/Commands.java

@@ -20 +20 @@
 public interface Commands {
     String GET = "GET".intern();
+    String GETS = "GETS".intern();
     String DELETE = "DELETE".intern();
     String DECR = "DECR".intern();
@@ -26 +27 @@
     String ADD = "ADD".intern();
     String SET = "SET".intern();
+    String CAS = "CAS".intern();
     String STATS = "STATS".intern();
     String VERSION = "VERSION".intern();

jmemcached-core/src/main/java/com/thimbleware/jmemcached/MCElement.java

@@ -27 +27 @@
     public byte[] data;
     public String keystring;
+    public long cas_unique;
 }

jmemcached-core/src/main/java/com/thimbleware/jmemcached/ServerSessionHandler.java

@@ -22 +22 @@
 import org.slf4j.LoggerFactory;
 
+import static java.lang.Integer.parseInt;
+import static java.lang.String.valueOf;
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.Charset;
 import java.nio.charset.CharsetEncoder;
 import java.util.Iterator;
-import static java.lang.String.valueOf;
-import static java.lang.Integer.*;
 
 /**
@@ -50 +50 @@
     public boolean verbose;
 
-    public static CharsetEncoder ENCODER  = Charset.forName("US-ASCII").newEncoder();
+    public static CharsetEncoder ENCODER = Charset.forName("US-ASCII").newEncoder();
 
     /**
@@ -59 +59 @@
      * Construct the server session handler
      *
-     * @param cache the cache to use
+     * @param cache            the cache to use
      * @param memcachedVersion the version string to return to clients
-     * @param verbosity verbosity level for debugging
-     * @param idle how long sessions can be idle for
+     * @param verbosity        verbosity level for debugging
+     * @param idle             how long sessions can be idle for
      */
     public ServerSessionHandler(Cache cache, String memcachedVersion, boolean verbosity, int idle) {
@@ -68 +68 @@
 
         this.cache = cache;
-        
+
         started = Now();
         version = memcachedVersion;
@@ -141 +141 @@
 
         ResponseMessage r = new ResponseMessage();
-        if (cmd == Commands.GET) {
+        if (cmd == Commands.GET || cmd == Commands.GETS) {
             for (int i = 0; i < cmdKeysSize; i++) {
                 MCElement result = get(command.keys.get(i));
                 if (result != null) {
-                    r.out.putString("VALUE " + result.keystring + " " + result.flags + " " + result.data_length + "\r\n", ENCODER);
+                    r.out.putString("VALUE " + result.keystring + " " + result.flags + " " + result.data_length + (cmd == Commands.GETS ? " " + result.cas_unique : "") + "\r\n", ENCODER);
                     r.out.put(result.data, 0, result.data_length);
                     r.out.putString("\r\n", ENCODER);
@@ -153 +153 @@
             r.out.putString("END\r\n", ENCODER);
         } else if (cmd == Commands.SET) {
-            r.out.putString(set(command.element), ENCODER);
+            String ret = set(command.element);
+            if (!command.noreply)
+                r.out.putString(ret, ENCODER);
+        } else if (cmd == Commands.CAS) {
+            String ret = cas(command.cas_key, command.element);
+            if (!command.noreply)
+                r.out.putString(ret, ENCODER);
         } else if (cmd == Commands.ADD) {
-            r.out.putString(add(command.element), ENCODER);
+            String ret = add(command.element);
+            if (!command.noreply)
+                r.out.putString(ret, ENCODER);
         } else if (cmd == Commands.REPLACE) {
-            r.out.putString(replace(command.element), ENCODER);
+            String ret = replace(command.element);
+            if (!command.noreply)
+                r.out.putString(ret, ENCODER);
         } else if (cmd == Commands.INCR) {
             r.out.putString(get_add(command.keys.get(0), parseInt(command.keys.get(1))), ENCODER);
@@ -207 +217 @@
     /**
      * Triggered when a session has gone idle.
+     *
      * @param session the MINA session
-     * @param status the idle status
+     * @param status  the idle status
      */
     public void sessionIdle(IoSession session, IdleStatus status) {
@@ -217 +228 @@
     /**
      * Triggered when an exception is caught by the protocol handler
+     *
      * @param session the MINA session
-     * @param cause the exception
+     * @param cause   the exception
      */
     public void exceptionCaught(IoSession session, Throwable cause) {
@@ -230 +242 @@
      * Handle the deletion of an item from the cache.
      *
-     * @param key the key for the item
+     * @param key  the key for the item
      * @param time only delete the element if time (time in seconds)
      * @return the message response
@@ -246 +258 @@
      */
     protected String add(MCElement e) {
-        if (cache.add(e)) return "STORED\r\n";
+        if (cache.add(e) == Cache.StoreResponse.STORED) return "STORED\r\n";
         else return "NOT_STORED\r\n";
+    }
+
+    protected String getStoreResponeString(Cache.StoreResponse storeResponse) {
+        switch (storeResponse) {
+            case EXISTS:
+                return "EXISTS\r\n";
+            case NOT_FOUND:
+                return "NOT_FOUND\r\n";
+            case NOT_STORED:
+                return "NOT_STORED\r\n";
+            case STORED:
+                return "STORED\r\n";
+        }
+        return null;
     }
 
@@ -257 +283 @@
      */
     protected String replace(MCElement e) {
-        if (cache.replace(e)) return "STORED\r\n";
-        else return "NOT_STORED\r\n";
+        return getStoreResponeString(cache.replace(e));
     }
 
@@ -268 +293 @@
      */
     protected String set(MCElement e) {
-        if (cache.set(e)) return "STORED\r\n";
-        else return "NOT_STORED\r\n";
-    }
-
-    /**
-     * Increment an (integer) element inthe cache
+        return getStoreResponeString(cache.set(e));
+    }
+
+    /**
+     * Check and set an element in the cache
+     *
+     * @param cas_key
+     * @param e       the element to set @return the message response string
+     */
+    protected String cas(Long cas_key, MCElement e) {
+        return getStoreResponeString(cache.cas(cas_key, e));
+    }
+
+    /**
+     * Increment an (integer) element in the cache
+     *
      * @param key the key to increment
      * @param mod the amount to add to the value
@@ -289 +324 @@
     /**
      * Check whether an element is in the cache and non-expired
+     *
      * @param key the key for the element to lookup
      * @return whether the element is in the cache and is live
@@ -298 +334 @@
     /**
      * Get an element from the cache
+     *
      * @param key the key for the element to lookup
      * @return the element, or 'null' in case of cache miss.
@@ -326 +363 @@
     /**
      * Return runtime statistics
+     *
      * @param arg additional arguments to the stats command
      * @return the full command response
@@ -372 +410 @@
     /**
      * Flush all cache entries
+     *
      * @return command response
      */
@@ -380 +419 @@
     /**
      * Flush all cache entries with a timestamp after a given expiration time
+     *
      * @param expire the flush time in seconds
      * @return command response
@@ -388 +428 @@
 
 
-
-
-
 }