fixed a problem with GC with no subscriptions

Author: kahing

clusterhq/queue/pom.xml

@@ -59,12 +59,6 @@
             <scope>test</scope>
             <version>3.2.0</version>
         </dependency>
-        <dependency>
-            <groupId>org.testng</groupId>
-            <artifactId>testng</artifactId>
-            <version>6.8.8</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

clusterhq/queue/queue.iml

@@ -44,9 +44,6 @@
     <orderEntry type="library" name="Maven: ch.qos.logback:logback-core:1.1.3" level="project" />
     <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.7.7" level="project" />
     <orderEntry type="library" scope="TEST" name="Maven: org.assertj:assertj-core:3.2.0" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.testng:testng:6.8.8" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.beanshell:bsh:2.0b4" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: com.beust:jcommander:1.27" level="project" />
     <orderEntry type="library" name="Maven: com.google.guava:guava:18.0" level="project" />
   </component>
 </module>

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

@@ -63,7 +63,7 @@ Path getChunkPath(long baseLSN) {
         return dir.resolve(name);
     }
 
-    public long getNextLSN() {
+    public synchronized long getNextLSN() {
         return nextLSN;
     }
 
@@ -111,10 +111,12 @@ public synchronized long post(InputStream data) throws IOException {
                 throw new IllegalArgumentException(nwritten + " != 8");
             }
             out.force(true);
-            nextLSN += total + 8;
-            if (getBaseLSN(nextLSN) != baseLSN) {
-                // we exceeded this chunk, round this up to the next chunk
-                nextLSN = getBaseLSN(nextLSN) + CHUNK_SIZE;
+            synchronized (this) {
+                nextLSN += total + 8;
+                if (getBaseLSN(nextLSN) != baseLSN) {
+                    // we exceeded this chunk, round this up to the next chunk
+                    nextLSN = getBaseLSN(nextLSN) + CHUNK_SIZE;
+                }
             }
         } catch (IOException e) {
             // truncate the file back to the original size
@@ -191,12 +193,12 @@ public Message get(long lsn) throws IOException {
                         String.format("%s/%s is %s bytes but chunk is %s bytes", relativeLSN, lsn, messageSize, chunkSize));
             }
 
-            long nextLSN = lsn + 8 + messageSize;
-            if (getBaseLSN(nextLSN) != baseLSN) {
+            long lsnAfter = lsn + 8 + messageSize;
+            if (getBaseLSN(lsnAfter) != baseLSN) {
                 // we exceeded this chunk, round this up to the next chunk
-                nextLSN = getBaseLSN(nextLSN) + CHUNK_SIZE;
+                lsnAfter = getBaseLSN(lsnAfter) + CHUNK_SIZE;
             }
-            return new Message(ByteStreams.limit(Channels.newInputStream(in), messageSize), nextLSN);
+            return new Message(ByteStreams.limit(Channels.newInputStream(in), messageSize), lsnAfter);
         } catch (BufferUnderflowException e) {
             throw new IOException(e);
         } catch (IOException e) {

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

@@ -1,20 +1,36 @@
 package us.hxbc.clusterhq.queue;
 
+import org.glassfish.grizzly.http.server.HttpServer;
 import org.glassfish.jersey.filter.LoggingFilter;
 import org.glassfish.jersey.grizzly2.httpserver.GrizzlyHttpServerFactory;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 
 public class Main {
     private static Logger logger = LoggerFactory.getLogger(Main.class);
+    final HttpServer server;
 
-    public static void main(String[] args) {
+    Main(int port, Path dir) {
+        ResourceConfig rc = new ResourceConfig();
+        rc.registerInstances(new Api(dir, 4096));
+        if (logger.isDebugEnabled()) {
+            rc.register(new LoggingFilter(java.util.logging.Logger.getGlobal(), false));
+        }
+        server = GrizzlyHttpServerFactory.createHttpServer(URI.create("http://0.0.0.0:" + port), rc, false);
+    }
+
+    void start() throws IOException {
+        server.start();
+    }
+
+    public static void main(String[] args) throws IOException {
         if (args.length != 2) {
             System.err.println("Usage: Main <port> <dir>");
             System.exit(1);
@@ -27,11 +43,6 @@ public static void main(String[] args) {
             System.exit(1);
         }
 
-        ResourceConfig rc = new ResourceConfig();
-        rc.registerInstances(new Api(dir, 4096));
-        if (logger.isDebugEnabled()) {
-            rc.register(new LoggingFilter(java.util.logging.Logger.getGlobal(), false));
-        }
-        GrizzlyHttpServerFactory.createHttpServer(URI.create("http://0.0.0.0:" + port), rc);
+        new Main(port, dir).start();
     }
 }

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

@@ -1,5 +1,8 @@
 package us.hxbc.clusterhq.queue;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.core.Response;
 import java.io.IOException;
@@ -24,6 +27,7 @@
  */
 public class Queue {
     private static final long CHUNK_SIZE = 4 * 1024; // 4KB
+    private final Logger logger = LoggerFactory.getLogger(getClass());
     private final Path dataDir, subscriptionDir;
     private final Map<String, Subscriber> subscriptions = new HashMap<>();
     private final DataStore dataStore;
@@ -124,7 +128,7 @@ void spawnGCThread() {
     }
 
     synchronized void gcNow() {
-        long curMinLSN = Long.MAX_VALUE;
+        long curMinLSN = dataStore.getNextLSN();
 
         synchronized (subscriptions) {
             for (Subscriber s : subscriptions.values()) {
@@ -136,6 +140,7 @@ synchronized void gcNow() {
             }
         }
 
+        logger.info("GC up to {}", curMinLSN);
         if (curMinLSN > minLSN) {
             dataStore.gc(curMinLSN);
             minLSN = curMinLSN;

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

@@ -173,4 +173,14 @@ public void testGcNone2Users() throws Exception {
         queue.gcNow();
         assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
     }
+
+    @Test
+    public void testGcAgain() throws Exception {
+        queue.gcNow();
+        testGcNone2Users();
+        assertThat(stream2String(queue.get("bar").in)).isEqualTo("world");
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(1);
+        queue.gcNow();
+        assertThat(Files.list(dir.resolve("data")).count()).isEqualTo(0);
+    }
 }