Changeset 1:c8c4ca70ccb5

Timestamp:

03/29/08 16:09:11 (8 months ago)

Author:

ryan@…

Branch:

default

Message:

Some restructuring, cleanup, and documentation

Location:

src/main/java/com/jehiah/memcached

Files:

• Cache.java (added)
• CommandDecoder.java (modified) (8 diffs)
• CommandMessage.java (modified) (1 diff)
• LRUCache.java (modified) (9 diffs)
• LRUCacheDelegate.java (added)
• MCCache.java (deleted)
• MCElement.java (modified) (1 diff)
• Main.java (modified) (3 diffs)
• MemCacheD.java (deleted)
• MemCacheDaemon.java (added)
• ServerSessionHandler.java (modified) (9 diffs)

src/main/java/com/jehiah/memcached/CommandDecoder.java

@@ -15 +15 @@
 
 /**
+ * MINA MessageDecoderAdapter responsible for parsing inbound lines from the memcached protocol session.
  */
 public final class CommandDecoder extends MessageDecoderAdapter {
@@ -23 +24 @@
 
     private static final int WORD_BUFFER_INIT_SIZE = 16;
+
     private static final String SESSION_STATUS = "sessionStatus";
 
+    /**
+     * Possible states that the current session is in.
+     */
     enum SessionState {
         ERROR,
-        SET_WAIT_LINE,
+        WAITING_FOR_DATA,
         READY
     }
 
+    /**
+     * Object for holding the current session status.
+     */
     final class SessionStatus implements Serializable {
+        // the state the session is in
         public SessionState state;
+
+        // if we are waiting for more data, how much?
         public int bytesNeeded;
+
+        // the current working command
         public CommandMessage cmd;
 
@@ -47 +60 @@
     }
 
+    /**
+     * Checks the specified buffer is decodable by this decoder.
+     *
+     * In our case checks the session state to see if we are waiting for data.  If we are, make sure
+     * that we actually have all the data we need.
+     *
+     * @return {@link #OK} if this decoder can decode the specified buffer.
+     *         {@link #NOT_OK} if this decoder cannot decode the specified buffer.
+     *         {@link #NEED_DATA} if more data is required to determine if the
+     *         specified buffer is decodable ({@link #OK}) or not decodable
+     *         {@link #NOT_OK}.
+     */
     public final MessageDecoderResult decodable(IoSession session, ByteBuffer in) {
         // ask the session for its state,
         SessionStatus sessionStatus = (SessionStatus) session.getAttribute(SESSION_STATUS);
-        if (sessionStatus != null &&  sessionStatus.state == SET_WAIT_LINE) {
+        if (sessionStatus != null &&  sessionStatus.state == WAITING_FOR_DATA) {
             if (in.remaining() <= sessionStatus.bytesNeeded)
                 return MessageDecoderResult.NEED_DATA;
@@ -57 +82 @@
     }
 
+    /**
+     * Actually decodes inbound data from the memcached protocol session.
+     *
+     * MINA invokes {@link #decode(IoSession, ByteBuffer, ProtocolDecoderOutput)}
+     * method with read data, and then the decoder implementation puts decoded
+     * messages into {@link ProtocolDecoderOutput}.
+     *
+     * @return {@link #OK} if finished decoding messages successfully.
+     *         {@link #NEED_DATA} if you need more data to finish decoding current message.
+     *         {@link #NOT_OK} if you cannot decode current message due to protocol specification violation.
+     *
+     * @throws Exception if the read data violated protocol specification
+     */
     public final MessageDecoderResult decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception {
         SessionStatus sessionStatus = (SessionStatus) session.getAttribute(SESSION_STATUS);
         SessionStatus returnedSessionStatus;
-        if (sessionStatus != null && sessionStatus.state == SET_WAIT_LINE) {
+        if (sessionStatus != null && sessionStatus.state == WAITING_FOR_DATA) {
             if (in.remaining() < sessionStatus.bytesNeeded)
                 return MessageDecoderResult.NEED_DATA;
@@ -108 +146 @@
 
 
+    /**
+     * Process an individual completel protocol line and either passes the command for processing by the
+     * session handler, or (in the case of SET-type commands) partially parses the command and sets the session into
+     * a state to wait for additional data.
+     * @param parts the (originally space separated) parts of the command
+     * @param session the MINA IoSession
+     * @param out the MINA protocol decoder output to pass our command on to
+     * @return the session status we want to set the session to
+     */
     private SessionStatus processLine(List<String> parts, IoSession session, ProtocolDecoderOutput out) {
         if (parts.isEmpty())
@@ -134 +181 @@
             cmd.element.data_length = size;
 
-            return new SessionStatus(SET_WAIT_LINE, size, cmd);
+            return new SessionStatus(WAITING_FOR_DATA, size, cmd);
 
         } else if (cmd.cmd == Commands.GET ||
@@ -155 +202 @@
     }
 
+    /**
+     * Handles the continuation of a SET/ADD/REPLACE command with the data it was waiting for.
+     *
+     * @param session the MINA IoSession
+     * @param out the MINA protocol decoder output which we signal with the completed command
+     * @param state the current session status (unused)
+     * @param remainder the bytes picked up
+     * @return the new status to set the session to
+     */
     private SessionStatus continueSet(IoSession session, ProtocolDecoderOutput out, SessionStatus state, byte[] remainder) {
         state.cmd.element.data = remainder;
@@ -163 +219 @@
     }
 
+    /**
+     * @return the current time in seconds
+     */
     public final int Now() {
         return (int) (System.currentTimeMillis() / 1000);

src/main/java/com/jehiah/memcached/CommandMessage.java

@@ -13 +13 @@
 import java.util.ArrayList;
 
-public class CommandMessage implements Serializable {
-    // should be byte buffers?
+/**
+ * The payload object holding the parsed message.
+ */
+public final class CommandMessage implements Serializable {
     public String cmd;
     public MCElement element;

src/main/java/com/jehiah/memcached/LRUCache.java

@@ -31 +31 @@
  * <li> cache items are expelled when the total maximum size of the cache is
  * reached
+ * <li> a ceiling parameter is available to provide a certain amount of reserved room
+ * in the cache
  * <li> unusual: not built to handle the case of user input affecting the
  * data; only handles case of changes to the datastore eventually refreshing
  * the cache. Thus, this cache is not suitable for the case in which user
- * input must be reflected.
+ * input must be reflected.  That is: all changes to data must happen through "puts"
  * </ul>
  * <p/>
  * <p/>
- * Source: http://www.javapractices.com/Topic118.cjp
+ * Based on (heavily altered) code from original source: http://www.javapractices.com/Topic118.cjp
  */
 public final class LRUCache<ID_TYPE, ITEM_TYPE> {
 
-    private final Map<ID_TYPE, CacheEntry<ITEM_TYPE>> fItems;
+    /**
+     * Map containing the actual storage
+     */
+    private final Map<ID_TYPE, CacheEntry<ITEM_TYPE>> items;
 
     /**
@@ -78 +83 @@
         this.maximumSize = maximumSize;
         this.ceilingSize = ceilingSize;
-        fItems = new LinkedHashMap<ID_TYPE, CacheEntry<ITEM_TYPE>>(INITIAL_TABLE_SIZE) {
+
+        /**
+         * Creates a linked hash map which expels old elements on declared criterion
+         */
+        items = new LinkedHashMap<ID_TYPE, CacheEntry<ITEM_TYPE>>(INITIAL_TABLE_SIZE) {
             protected boolean removeEldestEntry(Map.Entry<ID_TYPE, CacheEntry<ITEM_TYPE>> eldest) {
                 if (size + ceilingSize > maximumSize || size() > maximumItems) {
@@ -104 +113 @@
         startRead();
         try {
-            return fItems.containsKey(aId);
+            return items.containsKey(aId);
         }
         finally {
@@ -125 +134 @@
         ITEM_TYPE result;
         try {
-            if (fItems.containsKey(aId)) {
-                result = fItems.get(aId).item;
+            if (items.containsKey(aId)) {
+                result = items.get(aId).item;
                 if (result == null) {
                     throw new IllegalStateException("Stored item should not be null. Id:" + aId);
@@ -156 +165 @@
         startWrite();
         try {
-            fItems.put(aId, new CacheEntry<ITEM_TYPE>(item_size, aItem));
+            items.put(aId, new CacheEntry<ITEM_TYPE>(item_size, aItem));
             size += item_size;
         }
@@ -165 +174 @@
 
     /**
-     * Start from beginning, and remove all items from the cache; if cache is
-     * disabled, do nothing.
-     * <p/>
-     * Forces a re-population of all items into the cache.
-     */
-    public void clear() {
+     * Remove an entry from the cache
+     * @param key the key for the entry
+     */
+    public void remove(ID_TYPE key) {
         startWrite();
         try {
-            fItems.clear();
-            size = 0;
-        }
-        finally {
-            finishWrite();
-        }
-    }
-
-    public Set<ID_TYPE> keys() {
-        startRead();
-        try {
-            return fItems.keySet();
-        } finally {
-            finishRead();
-        }
-    }
-
-    public void remove(ID_TYPE keystring) {
-        startWrite();
-        try {
-            CacheEntry<ITEM_TYPE> item = fItems.get(keystring);
+            CacheEntry<ITEM_TYPE> item = items.get(key);
             if (item != null) {
-                fItems.remove(keystring);
+                items.remove(key);
                 size -= item.size;
             }
@@ -204 +191 @@
 
 
+    /**
+     * Start from beginning, and remove all items from the cache; if cache is
+     * disabled, do nothing.
+     * <p/>
+     * Forces a re-population of all items into the cache.
+     */
+    public void clear() {
+        startWrite();
+        try {
+            items.clear();
+            size = 0;
+        }
+        finally {
+            finishWrite();
+        }
+    }
+
+    /**
+     * @return the set of all keys in the cache
+     */
+    public Set<ID_TYPE> keys() {
+        startRead();
+        try {
+            return items.keySet();
+        } finally {
+            finishRead();
+        }
+    }
+
+    /**
+     * @return the number of entries in the cache
+     */
+    public long count() {
+        startRead();
+        try {
+            return items.size();
+        } finally {
+            finishRead();
+        }
+    }
+
+    /**
+     * @return the maximum number of items in the cache
+     */
+    public int getMaximumItems() {
+        return maximumItems;
+    }
+
+    /**
+     * @return the size (in bytes) of all entries in the cache.
+     */
+    public long getSize() {
+        return size;
+    }
+
+    /**
+     * @return the maximum capacity (in bytes) of the cache
+     */
+    public long getMaximumSize() {
+        return maximumSize;
+    }
+
+    /**
+     * @return the reserved headroom/ceiling size (in bytes)
+     */
+    public long getCeilingSize() {
+        return ceilingSize;
+    }
+
 
     /**
@@ -218 +274 @@
     }
 
+    /**
+     * Blocks of code in which the contents of fItems and fTimePlacedIntoCache
+     * are examined in any way must be surrounded by calls to <code>startRead</code>
+     * and <code>finishRead</code>. See documentation for ReadWriteLock.
+     * <p/>
+     * Translates the InterruptedException into a generic DataAccessException, to
+     * protect the higher layers from implementation details.
+     */
     private void finishRead() {
         readWriteLock.readLock().unlock();
@@ -236 +300 @@
     }
 
+    /**
+     * Blocks of code in which the contents of fItems and fTimePlacedIntoCache
+     * are changed in any way must be surrounded by calls to <code>startWrite</code> and
+     * <code>finishWrite</code>. See documentation for ReadWriteLock.
+     * <p/>
+     * Translates the InterruptedException into a generic DataAccessException, to
+     * protect the higher layers from implementation details.
+     */
     private void finishWrite() {
         readWriteLock.writeLock().unlock();
     }
 
-    //
-    public long count() {
-        startRead();
-        try {
-            return fItems.size();
-        } finally {
-            finishRead();
-        }
-    }
-
-    public long getSize() {
-        return size;
-    }
-
-    public long getMaximumSize() {
-        return maximumSize;
-    }
-
-    public long getCeilingSize() {
-        return ceilingSize;
-    }
-
-    public int getMaximumItems() {
-        return maximumItems;
-    }
 
 }

src/main/java/com/jehiah/memcached/MCElement.java

@@ -18 +18 @@
     public byte[] data;
     public String keystring;
-
-    /*
-      * Each item sent by the server looks like this:
-      VALUE <key> <flags> <bytes>\r\n
-      <data block>\r\n
-      */
-    public String toString() {
-        return new StringBuffer().append("VALUE ").append(this.keystring).append(" ").append(this.flags).append(" ").append(this.data_length).append("\r\n").append(this.data).append("\r\n").toString();
-//              return "VALUE " + this.keystring + " " + this.flags + " " +String.valueOf(this.data_length)+ "\r\n"+ this.data + "\r\n";
-    }
-
 }

src/main/java/com/jehiah/memcached/Main.java

@@ -39 +39 @@
 
         if (cmdline.hasOption("help") || cmdline.hasOption("h")) {
-            System.out.println("Memcached Version " + MemCacheD.memcachedVersion);
+            System.out.println("Memcached Version " + MemCacheDaemon.memcachedVersion);
             System.out.println("http://jehiah.com/projects/memcached\n");
 
@@ -48 +48 @@
 
         if (cmdline.hasOption("V")) {
-            System.out.println("Memcached Version " + MemCacheD.memcachedVersion);
+            System.out.println("Memcached Version " + MemCacheDaemon.memcachedVersion);
             return;
         }
@@ -118 +118 @@
 
         // create daemon and start it
-        new MemCacheD(addr, port, max_size, maxBytes, idle, verbose).start();
+        MemCacheDaemon daemon = new MemCacheDaemon();
+        LRUCacheDelegate cacheDelegate = new LRUCacheDelegate(max_size, max_size, 1024000);
+        daemon.setCacheDelegate(cacheDelegate);
+        daemon.start();
     }
 

src/main/java/com/jehiah/memcached/ServerSessionHandler.java

@@ -38 +38 @@
     public int idle_limit;
     public boolean verbose;
-    protected MCCache data;
+    protected Cache cache;
 
     public static CharsetEncoder ENCODER  = Charset.forName("US-ASCII").newEncoder();
 
-    public ServerSessionHandler(MCCache cache, String memcachedVersion, boolean verbosity, int idle) {
+    public ServerSessionHandler(Cache cache, String memcachedVersion, boolean verbosity, int idle) {
         initStats();
-        this.data = cache;
+        this.cache = cache;
 
         this.started = Now();
@@ -175 +175 @@
         if (is_there(keystring)) {
             if (time != 0) {
-                MCElement el = this.data.get(keystring);
+                MCElement el = this.cache.get(keystring);
                 if (el.expire == 0 || el.expire > (Now() + time)) {
                     el.expire = Now() + time; // update the expire time
-                    this.data.put(keystring, el);
+                    this.cache.put(keystring, el);
                 }// else it expire before the time we were asked to expire it
             } else {
-                this.data.remove(keystring); // just remove it
+                this.cache.remove(keystring); // just remove it
             }
             return "DELETED\r\n";
@@ -209 +209 @@
     public String set(MCElement e) {
         set_cmds += 1;//update stats
-        this.data.put(e.keystring, e);
+        this.cache.put(e.keystring, e);
         return "STORED\r\n";
     }
 
     public String get_add(String keystring, int mod) {
-        MCElement e = this.data.get(keystring);
+        // TODO make this threadsafe by cooperating more directly with the cache
+        MCElement e = this.cache.get(keystring);
         if (e == null) {
             get_misses += 1;//update stats
@@ -230 +231 @@
         e.data = valueOf(old_val).getBytes(); // toString
         e.data_length = e.data.length;
-        this.data.put(e.keystring, e); // save new value
+        this.cache.put(e.keystring, e); // save new value
         return valueOf(old_val) + "\r\n"; // return new value
     }
 
     /*
-      * this.data.containsKey() would work except it doesn't check the expire time
+      * this.cache.containsKey() would work except it doesn't check the expire time
       */
     public boolean is_there(String keystring) {
-        MCElement e = this.data.get(keystring);
+        MCElement e = this.cache.get(keystring);
         return e != null && !(e.expire != 0 && e.expire < Now());
     }
@@ -244 +245 @@
     public MCElement get(String keystring) {
         get_cmds += 1;//updates stats
-        MCElement e = this.data.get(keystring);
+        MCElement e = this.cache.get(keystring);
         if (e == null) {
             get_misses += 1;//update stats
@@ -251 +252 @@
         if (e.expire != 0 && e.expire < Now()) {
             get_misses += 1;//update stats
+
+            // TODO shouldn't this actually remove the item from cache since it's expired?
             return null;
         }
@@ -273 +276 @@
 
         if (arg.equals("keys")) {
-            Iterator itr = this.data.keys();
+            Iterator itr = this.cache.keys();
             while (itr.hasNext()) {
                 builder.append("STAT key ").append(itr.next()).append("\r\n");
@@ -291 +294 @@
         builder.append("STAT time ").append(valueOf(Now())).append("\r\n");
         builder.append("STAT uptime ").append(valueOf(Now() - this.started)).append("\r\n");
-        builder.append("STAT cur_items ").append(valueOf(this.data.count())).append("\r\n");
-        builder.append("STAT limit_maxbytes ").append(valueOf(this.data.maxSize())).append("\r\n");
-        builder.append("STAT current_bytes ").append(valueOf(this.data.size())).append("\r\n");
+        builder.append("STAT cur_items ").append(valueOf(this.cache.count())).append("\r\n");
+        builder.append("STAT limit_maxbytes ").append(valueOf(this.cache.maxSize())).append("\r\n");
+        builder.append("STAT current_bytes ").append(valueOf(this.cache.size())).append("\r\n");
         builder.append("STAT free_bytes ").append(valueOf(Runtime.getRuntime().freeMemory())).append("\r\n");
 
@@ -313 +316 @@
 
     public String flush_all(int expire) {
-        this.data.flushAll();
+        this.cache.flushAll();
 
         return "OK\r\n";