plumbed GC through Queue

Queue will GC with the min LSN for all subscribers of a topic

Author: kahing

clusterhq/queue/src/main/java/us/hxbc/clusterhq/queue/Api.java

@@ -2,7 +2,6 @@
 
 import org.glassfish.grizzly.http.server.Request;
 
-import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.GET;
 import javax.ws.rs.POST;
@@ -45,6 +44,7 @@ private Queue ensureTopic(@PathParam("topic") String topic) throws IOException {
                 Files.createDirectory(p);
                 q = new Queue(p, CHUNK_SIZE);
                 topics.put(topic, q);
+                q.spawnGCThread();
             }
         }
         return q;

clusterhq/queue/src/main/java/us/hxbc/clusterhq/queue/Queue.java

@@ -5,6 +5,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.StreamCorruptedException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
@@ -26,6 +27,8 @@ public class Queue {
     private final Path dataDir, subscriptionDir;
     private final Map<String, Subscriber> subscriptions = new HashMap<>();
     private final DataStore dataStore;
+    private final Thread gcThread;
+    private long minLSN = 0;
 
     public Queue(Path dir, long chunkSize) throws IOException {
         requireNonNull(dir);
@@ -38,6 +41,16 @@ public Queue(Path dir, long chunkSize) throws IOException {
         subscriptionDir = dir.resolve("subscriptions");
         Files.createDirectory(subscriptionDir);
         dataStore = new DataStore(dataDir, chunkSize);
+
+        gcThread = new Thread(() -> {
+            while (true) {
+                gcNow();
+                try {
+                    Thread.sleep(10000);
+                } catch (InterruptedException e) {
+                }
+            }
+        });
     }
 
     public void subscribe(String user) throws IOException {
@@ -73,9 +86,13 @@ public DataStore.Message get(String user) throws IOException {
             }
         }
 
-        synchronized (subscriber) {
-            DataStore.Message m = dataStore.get(subscriber.nextLSN);
-            if (m.in != null) {
+        DataStore.Message m = dataStore.get(subscriber.nextLSN);
+        if (m.in != null) {
+            if (m.nextLSN <= subscriber.nextLSN) {
+                throw new StreamCorruptedException(m.nextLSN + " <= " + subscriber.nextLSN);
+            }
+
+            synchronized (subscriber) {
                 Path p = subscriptionDir.resolve(user);
                 try (FileChannel out = FileChannel.open(p,
                         StandardOpenOption.WRITE,
@@ -87,8 +104,9 @@ public DataStore.Message get(String user) throws IOException {
                 }
                 subscriber.nextLSN = m.nextLSN;
             }
-            return m;
         }
+
+        return m;
     }
 
     public void unsubscribe(String user) throws IOException {
@@ -101,6 +119,29 @@ public void unsubscribe(String user) throws IOException {
         }
     }
 
+    void spawnGCThread() {
+        gcThread.start();
+    }
+
+    synchronized void gcNow() {
+        long curMinLSN = Long.MAX_VALUE;
+
+        synchronized (subscriptions) {
+            for (Subscriber s : subscriptions.values()) {
+                synchronized (s) {
+                    if (s.nextLSN < curMinLSN) {
+                        curMinLSN = s.nextLSN;
+                    }
+                }
+            }
+        }
+
+        if (curMinLSN > minLSN) {
+            dataStore.gc(curMinLSN);
+            minLSN = curMinLSN;
+        }
+    }
+
     static class Subscriber {
         final String name;
         long nextLSN;

clusterhq/queue/src/test/java/us/hxbc/clusterhq/queue/QueueTest.java

@@ -128,4 +128,49 @@ public void testUnsubscribe() throws Exception {
         Path p = dir.resolve("subscriptions").resolve("foo");
         assertThat(Files.exists(p)).isFalse();
     }
+
+    @Test
+    public void testGCNone() throws Exception {
+        queue.subscribe("foo");
+        queue.post(string2Stream("hello"));
+        queue.gcNow();
+        assertThat(stream2String(queue.get("foo").in)).isEqualTo("hello");
+    }
+
+    @Test
+    public void testGCUnsubscribed() throws Exception {
+        queue.subscribe("foo");
+        queue.post(string2Stream("hello"));
+        queue.post(string2Stream("world"));
+        queue.unsubscribe("foo");
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
+        queue.gcNow();
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(0);
+    }
+
+    @Test
+    public void testGCMsgGot() throws Exception {
+        queue.subscribe("foo");
+        queue.post(string2Stream("hello"));
+        queue.post(string2Stream("world"));
+        assertThat(stream2String(queue.get("foo").in)).isEqualTo("hello");
+        assertThat(stream2String(queue.get("foo").in)).isEqualTo("world");
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
+        queue.gcNow();
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(0);
+    }
+
+    @Test
+    public void testGcNone2Users() throws Exception {
+        queue.subscribe("foo");
+        queue.subscribe("bar");
+        queue.post(string2Stream("hello"));
+        queue.post(string2Stream("world"));
+        assertThat(stream2String(queue.get("foo").in)).isEqualTo("hello");
+        assertThat(stream2String(queue.get("foo").in)).isEqualTo("world");
+        assertThat(stream2String(queue.get("bar").in)).isEqualTo("hello");
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
+        queue.gcNow();
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
+    }
 }