Improve time/scheduling in state machine executor.

atomix · Sep 2, 2015 · 60cf261 · 60cf261
1 parent e3e9293
commit 60cf261
Show file tree

Hide file tree

Showing 12 changed files with 166 additions and 236 deletions.
diff --git a/atomic/src/main/java/net/kuujo/copycat/atomic/state/AtomicValueState.java b/atomic/src/main/java/net/kuujo/copycat/atomic/state/AtomicValueState.java
@@ -16,10 +16,10 @@
 package net.kuujo.copycat.atomic.state;
 
 import net.kuujo.copycat.PersistenceMode;
-import net.kuujo.copycat.raft.session.Session;
 import net.kuujo.copycat.raft.Commit;
 import net.kuujo.copycat.raft.StateMachine;
 import net.kuujo.copycat.raft.StateMachineExecutor;
+import net.kuujo.copycat.raft.session.Session;
 
 import java.time.Instant;
 import java.util.HashMap;
@@ -132,7 +132,7 @@ protected Object get(Commit<AtomicValueCommands.Get> commit) {
    * Applies a set commit.
    */
   protected void set(Commit<AtomicValueCommands.Set> commit) {
-    if (!isActive(commit, context().time().instant())) {
+    if (!isActive(commit, now())) {
       commit.clean();
     } else {
       if (current != null) {
@@ -148,7 +148,7 @@ protected void set(Commit<AtomicValueCommands.Set> commit) {
    * Handles a compare and set commit.
    */
   protected boolean compareAndSet(Commit<AtomicValueCommands.CompareAndSet> commit) {
-    if (!isActive(commit, context().time().instant())) {
+    if (!isActive(commit, now())) {
       commit.clean();
       return false;
     } else if (isActive(current, commit.time())) {
@@ -178,9 +178,8 @@ protected boolean compareAndSet(Commit<AtomicValueCommands.CompareAndSet> commit
    * Handles a get and set commit.
    */
   protected Object getAndSet(Commit<AtomicValueCommands.GetAndSet> commit) {
-    if (!isActive(commit, context().time().instant())) {
+    if (!isActive(commit, now())) {
       commit.clean();
-
     }
 
     if (isActive(current, commit.time())) {

diff --git a/collections/pom.xml b/collections/pom.xml
@@ -32,11 +32,6 @@
       <artifactId>copycat</artifactId>
       <version>${project.version}</version>
     </dependency>
-    <dependency>
-      <groupId>net.kuujo.copycat</groupId>
-      <artifactId>copycat-netty</artifactId>
-      <version>${project.version}</version>
-    </dependency>
     <dependency>
       <groupId>net.kuujo.copycat</groupId>
       <artifactId>copycat-test</artifactId>

diff --git a/collections/src/main/java/net/kuujo/copycat/collections/state/MapState.java b/collections/src/main/java/net/kuujo/copycat/collections/state/MapState.java
@@ -16,28 +16,32 @@
 package net.kuujo.copycat.collections.state;
 
 import net.kuujo.copycat.PersistenceMode;
-import net.kuujo.copycat.raft.session.Session;
 import net.kuujo.copycat.raft.Commit;
 import net.kuujo.copycat.raft.StateMachine;
 import net.kuujo.copycat.raft.StateMachineExecutor;
+import net.kuujo.copycat.raft.session.Session;
+import net.kuujo.copycat.util.Listener;
+import net.kuujo.copycat.util.concurrent.Scheduled;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Map;
-import java.util.Set;
 
 /**
  * Map state machine.
  *
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 public class MapState extends StateMachine {
+  private StateMachineExecutor executor;
   private Map<Object, Commit<? extends MapCommands.TtlCommand>> map;
-  private final Set<Long> sessions = new HashSet<>();
+  private final Map<Object, Scheduled> timers = new HashMap<>();
+  private final Map<Object, Listener<Session>> listeners = new HashMap<>();
 
   @Override
   public void configure(StateMachineExecutor executor) {
+    this.executor = executor;
     executor.register(MapCommands.ContainsKey.class, this::containsKey);
     executor.register(MapCommands.Get.class, this::get);
     executor.register(MapCommands.GetOrDefault.class, this::getOrDefault);
@@ -49,28 +53,13 @@ public void configure(StateMachineExecutor executor) {
     executor.register(MapCommands.Clear.class, this::clear);
   }
 
-  @Override
-  public void register(Session session) {
-    sessions.add(session.id());
-  }
-
-  @Override
-  public void expire(Session session) {
-    sessions.remove(session.id());
-  }
-
-  @Override
-  public void close(Session session) {
-    sessions.remove(session.id());
-  }
-
   /**
    * Returns a boolean value indicating whether the given commit is active.
    */
   private boolean isActive(Commit<? extends MapCommands.TtlCommand> commit, Instant instant) {
     if (commit == null) {
       return false;
-    } else if (commit.operation().mode() == PersistenceMode.EPHEMERAL && !sessions.contains(commit.session().id())) {
+    } else if (commit.operation().mode() == PersistenceMode.EPHEMERAL && sessions().session(commit.session().id()) == null) {
       return false;
     } else if (commit.operation().ttl() != 0 && commit.operation().ttl() < instant.toEpochMilli() - commit.time().toEpochMilli()) {
       return false;
@@ -88,11 +77,7 @@ protected boolean containsKey(Commit<MapCommands.ContainsKey> commit) {
       }
 
       Commit<? extends MapCommands.TtlCommand> command = map.get(commit.operation().key());
-      if (!isActive(command, context().time().instant())) {
+      return isActive(command, commit.time());
-        map.remove(commit.operation().key());
-        return false;
-      }
-      return true;
     } finally {
       commit.close();
     }
@@ -108,14 +93,7 @@ protected Object get(Commit<MapCommands.Get> commit) {
 
     try {
       Commit<? extends MapCommands.TtlCommand> command = map.get(commit.operation().key());
-      if (command != null) {
+      return isActive(command, commit.time()) ? command.operation().value() : null;
-        if (!isActive(command, context().time().instant())) {
-          map.remove(commit.operation().key());
-        } else {
-          return command.operation().value();
-        }
-      }
-      return null;
     } finally {
       commit.close();
     }
@@ -131,12 +109,7 @@ protected Object getOrDefault(Commit<MapCommands.GetOrDefault> commit) {
 
     try {
       Commit<? extends MapCommands.TtlCommand> previous = map.get(commit.operation().key());
-      if (previous == null) {
+      return isActive(previous, commit.time()) ? previous.operation().value() : commit.operation().defaultValue();
-        return commit.operation().defaultValue();
-      } else if (isActive(previous, context().time().instant())) {
-        return previous.operation().value();
-      }
-      return commit.operation().defaultValue();
     } finally {
       commit.close();
     }
@@ -152,17 +125,51 @@ protected Object put(Commit<MapCommands.Put> commit) {
 
     Commit<? extends MapCommands.TtlCommand> previous = map.get(commit.operation().key());
     if (previous == null) {
-      if (!isActive(commit, context().time().instant())) {
+      if (commit.operation().ttl() > 0) {
-        commit.clean();
+        timers.put(commit.operation().key(), executor.schedule(() -> {
-      } else {
+          map.remove(commit.operation().key());
-        map.put(commit.operation().key(), commit);
+          Listener<Session> listener = listeners.remove(commit.operation().key());
+          if (listener != null)
+            listener.close();
+          commit.clean();
+        }, Duration.ofMillis(commit.operation().ttl())));
+      }
+
+      if (commit.operation().mode() == PersistenceMode.EPHEMERAL) {
+        listeners.put(commit.operation().key(), commit.session().onClose(s -> {
+          map.remove(commit.operation().key());
+          Scheduled scheduled = timers.remove(commit.operation().key());
+          if (scheduled != null)
+            scheduled.cancel();
+          commit.clean();
+        }));
       }
-      return null;
     } else {
-      map.put(commit.operation().key(), commit);
       previous.clean();
-      return isActive(previous, commit.time()) ? previous.operation().value() : null;
+
+      if (commit.operation().ttl() > 0) {
+        timers.put(commit.operation().key(), executor.schedule(() -> {
+          map.remove(commit.operation().key());
+          Listener<Session> listener = listeners.remove(commit.operation().key());
+          if (listener != null)
+            listener.close();
+          commit.close();
+        }, Duration.ofMillis(commit.operation().ttl())));
+      }
+
+      if (commit.operation().mode() == PersistenceMode.EPHEMERAL) {
+        listeners.put(commit.operation().key(), commit.session().onClose(s -> {
+          map.remove(commit.operation().key());
+          Scheduled scheduled = timers.remove(commit.operation().key());
+          if (scheduled != null)
+            scheduled.cancel();
+          commit.close();
+        }));
+      }
     }
+
+    map.put(commit.operation().key(), commit);
+    return isActive(previous, commit.time()) ? previous.operation().value() : null;
   }
 
   /**
@@ -174,22 +181,33 @@ protected Object putIfAbsent(Commit<MapCommands.PutIfAbsent> commit) {
     }
 
     Commit<? extends MapCommands.TtlCommand> previous = map.get(commit.operation().key());
-    if (previous == null) {
+    if (!isActive(previous, commit.time())) {
-      if (!isActive(commit, context().time().instant())) {
+      if (previous == null) {
-        commit.clean();
+        if (commit.operation().ttl() > 0) {
-      } else {
+          timers.put(commit.operation().key(), executor.schedule(() -> {
-        map.put(commit.operation().key(), commit);
+            map.remove(commit.operation().key());
+            Listener<Session> listener = listeners.remove(commit.operation().key());
+            if (listener != null)
+              listener.close();
+            commit.clean();
+          }, Duration.ofMillis(commit.operation().ttl())));
+        }
+
+        if (commit.operation().mode() == PersistenceMode.EPHEMERAL) {
+          listeners.put(commit.operation().key(), commit.session().onClose(s -> {
+            map.remove(commit.operation().key());
+            Scheduled scheduled = timers.remove(commit.operation().key());
+            if (scheduled != null)
+              scheduled.cancel();
+            commit.clean();
+          }));
+        }
       }
+
+      map.put(commit.operation().key(), commit);
       return null;
-    } else {
-      if (!isActive(previous, commit.time())) {
-        map.put(commit.operation().key(), commit);
-        previous.clean();
-        return null;
-      } else {
-        return previous.operation().value();
-      }
     }
+    return previous.operation().value();
   }
 
   /**
@@ -207,13 +225,16 @@ protected Object remove(Commit<MapCommands.Remove> commit) {
       } else if (!isActive(previous, commit.time())) {
         map.remove(commit.operation().key());
         previous.clean();
+        commit.close();
       } else {
         Object value = previous.operation().value();
         if ((value == null && commit.operation().value() == null) || (value != null && commit.operation().value() != null && value.equals(commit.operation().value()))) {
           map.remove(commit.operation().key());
           previous.clean();
+          commit.close();
           return true;
         }
+        commit.clean();
         return false;
       }
       return false;
@@ -224,6 +245,7 @@ protected Object remove(Commit<MapCommands.Remove> commit) {
         return true;
       } else {
         previous.clean();
+        commit.close();
         return isActive(previous, commit.time()) ? previous.operation().value() : null;
       }
     }

diff --git a/collections/src/main/java/net/kuujo/copycat/collections/state/SetState.java b/collections/src/main/java/net/kuujo/copycat/collections/state/SetState.java
@@ -16,10 +16,10 @@
 package net.kuujo.copycat.collections.state;
 
 import net.kuujo.copycat.PersistenceMode;
-import net.kuujo.copycat.raft.session.Session;
 import net.kuujo.copycat.raft.Commit;
 import net.kuujo.copycat.raft.StateMachine;
 import net.kuujo.copycat.raft.StateMachineExecutor;
+import net.kuujo.copycat.raft.session.Session;
 
 import java.time.Instant;
 import java.util.HashMap;
@@ -104,7 +104,7 @@ protected boolean add(Commit<SetCommands.Add> commit) {
     }
 
     Commit<? extends SetCommands.TtlCommand> previous = map.get(commit.operation().value());
-    if (!isActive(commit, context().time().instant())) {
+    if (!isActive(commit, now())) {
       commit.clean();
       return false;
     } else if (!isActive(previous, commit.time())) {