uses FileSystem sync for AOF (#91)

Author: the123saurav

core/src/main/java/dev/keva/core/aof/AOFContainer.java

@@ -18,20 +18,21 @@
 @Component
 public class AOFContainer {
     private ReentrantLock bufferLock;
-    private List<Command> buffer;
     private ObjectOutputStream output;
+    private FileDescriptor fd;
+    private boolean alwaysFlush;
 
     @Autowired
     private KevaConfig kevaConfig;
 
     public void init() {
+        alwaysFlush = kevaConfig.getAofInterval() == 0;
         bufferLock = new ReentrantLock();
-        buffer = new ArrayList<>(64);
 
         try {
-            boolean isExists = new File(getWorkingDir() + "keva.aof").exists();
             FileOutputStream fos = new FileOutputStream(getWorkingDir() + "keva.aof", true);
-            output = isExists ? new AppendOnlyObjectOutputStream(fos) : new ObjectOutputStream(fos);
+            fd = fos.getFD();
+            output = new ObjectOutputStream(fos);
         } catch (IOException e) {
             if (e instanceof FileNotFoundException) {
                 log.info("AOF file not found, creating new file...");
@@ -45,11 +46,11 @@ public void init() {
             }
         }
 
-        if (kevaConfig.getAofInterval() != 0) {
+        if (!alwaysFlush) {
             ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
             executorService.scheduleAtFixedRate(() -> {
                 try {
-                    sync();
+                    flush();
                 } catch (IOException e) {
                     log.error("Error writing AOF file", e);
                 }
@@ -61,51 +62,28 @@ public void init() {
     }
 
     public void write(Command command) {
-        if (kevaConfig.getAofInterval() == 0) {
-            syncPerMutation(command);
-            return;
-        }
-
-        bufferLock.lock();
-        try {
-            buffer.add(command);
-        } finally {
-            bufferLock.unlock();
-        }
-    }
-
-    public void sync() throws IOException {
-        if (buffer.isEmpty()) {
-            return;
-        }
         bufferLock.lock();
         try {
-            for (Command command : buffer) {
-                output.writeObject(command.getObjects());
+            output.writeObject(command.getObjects());
+            if (alwaysFlush) {
+                flush();
             }
+        } catch (IOException e) {
+            log.error("Error writing AOF file", e);
         } finally {
-            output.flush();
-            for (Command command : buffer) {
-                command.recycle();
-            }
-            buffer.clear();
             bufferLock.unlock();
         }
     }
 
-    public void syncPerMutation(Command command) {
-        try {
-            output.writeObject(command.getObjects());
-            output.flush();
-        } catch (IOException e) {
-            log.error("Error writing AOF file", e);
-        }
+    private void flush() throws IOException {
+        fd.sync();
     }
 
     public List<Command> read() throws IOException {
         try {
             List<Command> commands = new ArrayList<>(100);
             FileInputStream fis = new FileInputStream(getWorkingDir() + "keva.aof");
+            log.info("AOF size is: {}", fis.getChannel().size());
             ObjectInputStream input = new ObjectInputStream(fis);
             while (true) {
                 try {
@@ -129,15 +107,4 @@ private String getWorkingDir() {
         String workingDir = kevaConfig.getWorkDirectory();
         return workingDir.equals("./") ? "" : workingDir + "/";
     }
-
-    private static class AppendOnlyObjectOutputStream extends ObjectOutputStream {
-        public AppendOnlyObjectOutputStream(OutputStream out) throws IOException {
-            super(out);
-        }
-
-        @Override
-        protected void writeStreamHeader() throws IOException {
-            reset();
-        }
-    }
 }

core/src/main/java/dev/keva/core/aof/AOFManager.java

@@ -34,7 +34,7 @@ public void init() {
         try {
             List<Command> commands = aof.read();
             if (commands != null) {
-                log.info("Loading AOF file");
+                log.info("Processing {} commands from AOF file", commands.size());
                 for (Command command : commands) {
                     val name = command.getName();
                     val commandWrapper = commandMapper.getMethods().get(new BytesKey(name));

core/src/main/java/dev/keva/core/command/mapping/CommandMapper.java

@@ -109,11 +109,7 @@ public void init() {
                                 }
                                 Object[] objects = new Object[types.length];
                                 command.toArguments(objects, types, ctx);
-                                // If not in AOF mode, then recycle(),
-                                // else, the command will be recycled in the AOF dump
-                                if (!kevaConfig.getAof() || kevaConfig.getAofInterval() == 0) {
-                                    command.recycle();
-                                }
+                                command.recycle();
                                 return (Reply<?>) method.invoke(instance, objects);
                             } finally {
                                 lock.unlock();

core/src/test/java/dev/keva/core/server/AOFTest.java

@@ -5,8 +5,11 @@
 import lombok.val;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
 import redis.clients.jedis.Jedis;
 
+import java.io.IOException;
+import java.nio.file.Path;
 import java.util.concurrent.TimeUnit;
 
 import static dev.keva.core.utils.PortUtil.getAvailablePort;
@@ -17,14 +20,17 @@
 public class AOFTest {
     static String host = "localhost";
 
+    @TempDir
+    static Path tempDir;
+
     Server startServer(int port) throws Exception {
         val config = KevaConfig.builder()
                 .hostname(host)
                 .port(port)
                 .persistence(false)
                 .aof(true)
                 .aofInterval(1000)
-                .workDirectory(System.getProperty("java.io.tmpdir"))
+                .workDirectory(tempDir.toString())
                 .build();
         val server = KevaServer.of(config);
         new Thread(() -> {