(For discussion) Support dynamic cluster reconfiguration

barge-core/src/main/java/org/robotninjas/barge/BargeThreadPools.java

@@ -0,0 +1,91 @@
+package org.robotninjas.barge;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+
+public class BargeThreadPools implements Closeable {
+
+  private static final int NUM_THREADS = 1;
+
+  boolean closed;
+  final NioEventLoopGroup raftEventLoop;
+  final boolean closeEventLoop;
+  final ListeningExecutorService stateMachineExecutor;
+  final boolean closeStateMachineExecutor;
+  final ListeningScheduledExecutorService raftExecutor;
+
+  public BargeThreadPools(Optional<NioEventLoopGroup> eventLoopGroup,
+      Optional<ListeningExecutorService> stateMachineExecutor) {
+    if (eventLoopGroup.isPresent()) {
+      this.raftEventLoop = eventLoopGroup.get();
+      this.closeEventLoop = false;
+    } else {
+      this.raftEventLoop = new NioEventLoopGroup(1, new DefaultThreadFactory("pool-raft"));
+      this.closeEventLoop = true;
+      // Runtime.getRuntime().addShutdownHook(new Thread() {
+      // public void run() {
+      // eventLoop.shutdownGracefully();
+      // }
+      // });
+    }
+
+    this.raftExecutor = MoreExecutors.listeningDecorator(this.raftEventLoop);
+
+    if (stateMachineExecutor.isPresent()) {
+      this.stateMachineExecutor = stateMachineExecutor.get();
+      this.closeStateMachineExecutor = false;
+    } else {
+      this.stateMachineExecutor = MoreExecutors.listeningDecorator(
+          Executors.newSingleThreadExecutor(new DefaultThreadFactory("pool-raft-worker")));
+      this.closeStateMachineExecutor = true;
+      //
+      // Runtime.getRuntime().addShutdownHook(new Thread() {
+      // public void run() {
+      // executor.shutdownNow();
+      // }
+      // });
+    }
+
+  }
+
+  public NioEventLoopGroup getEventLoopGroup() {
+    Preconditions.checkState(!closed);
+    return raftEventLoop;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (closeEventLoop) {
+      raftEventLoop.shutdownGracefully();
+    }
+    if (closeStateMachineExecutor) {
+      stateMachineExecutor.shutdown();
+    }
+    closed = true;
+  }
+
+  public ListeningExecutorService getStateMachineExecutor() {
+    Preconditions.checkState(!closed);
+    return stateMachineExecutor;
+  }
+
+  public ListeningExecutorService getRaftExecutor() {
+    return raftExecutor;
+  }
+
+  public ScheduledExecutorService getRaftScheduler() {
+    return raftExecutor;
+  }
+
+}

barge-core/src/main/java/org/robotninjas/barge/RaftClusterHealth.java

@@ -0,0 +1,23 @@
+package org.robotninjas.barge;
+
+import java.util.List;
+
+public class RaftClusterHealth {
+
+  final RaftMembership membership;
+  final List<Replica> deadPeers;
+
+  public RaftClusterHealth(RaftMembership membership, List<Replica> deadPeers) {
+    this.membership = membership;
+    this.deadPeers = deadPeers;
+  }
+
+  public RaftMembership getClusterMembership() {
+    return membership;
+  }
+
+  public List<Replica> getDeadPeers() {
+    return deadPeers;
+  }
+
+}

barge-core/src/main/java/org/robotninjas/barge/RaftMembership.java

@@ -0,0 +1,40 @@
+package org.robotninjas.barge;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+public class RaftMembership {
+
+  final long id;
+  final ImmutableList<String> members;
+
+  public RaftMembership(long id, Collection<String> members) {
+    this.id = id;
+    List<String> sorted = Lists.newArrayList(members);
+    Collections.sort(sorted);
+    this.members = ImmutableList.copyOf(sorted);
+  }
+
+  public long getId() {
+    return id;
+  }
+
+  public ImmutableList<String> getMembers() {
+    return members;
+  }
+
+  public RaftMembership merge(RaftMembership other) {
+    Set<String> merged = Sets.newHashSet();
+    merged.addAll(members);
+    merged.addAll(other.members);
+
+    return new RaftMembership(-1L, merged);
+  }
+
+}

barge-core/src/main/java/org/robotninjas/barge/RaftModule.java

@@ -18,7 +18,6 @@
 
 import com.google.common.base.Optional;
 import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.inject.PrivateModule;
 import io.netty.channel.nio.NioEventLoopGroup;
 import org.robotninjas.barge.log.LogModule;
@@ -28,7 +27,6 @@
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.Immutable;
 import java.io.File;
-import java.util.concurrent.Executors;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -67,34 +65,13 @@ public void setTimeout(long timeout) {
   @Override
   protected void configure() {
 
-    install(new StateModule(timeout));
     Replica local = config.local();
+    install(new StateModule(local, timeout));
 
-    final NioEventLoopGroup eventLoop;
-    if (eventLoopGroup.isPresent()) {
-      eventLoop = eventLoopGroup.get();
-    } else {
-      eventLoop = new NioEventLoopGroup();
-      Runtime.getRuntime().addShutdownHook(new Thread() {
-        public void run() {
-          eventLoop.shutdownGracefully();
-        }
-      });
-    }
-    install(new RpcModule(local.address(), eventLoop));
-
-    final ListeningExecutorService executor;
-    if (stateMachineExecutor.isPresent()) {
-      executor = stateMachineExecutor.get();
-    } else {
-      executor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
-      Runtime.getRuntime().addShutdownHook(new Thread() {
-        public void run() {
-          executor.shutdownNow();
-        }
-      });
-    }
-    install(new LogModule(logDir, stateMachine, executor));
+    BargeThreadPools bargeThreadPools = new BargeThreadPools(eventLoopGroup, stateMachineExecutor);
+    install(new RpcModule(local.address(), bargeThreadPools));
+
+    install(new LogModule(logDir, stateMachine));
 
     bind(ClusterConfig.class)
       .toInstance(config);

barge-core/src/main/java/org/robotninjas/barge/RaftService.java

@@ -23,9 +23,11 @@
 import com.google.inject.Injector;
 import com.google.protobuf.Service;
 import io.netty.channel.nio.NioEventLoopGroup;
+import org.robotninjas.barge.log.RaftLog;
+import org.robotninjas.barge.proto.RaftEntry.Membership;
 import org.robotninjas.barge.proto.RaftProto;
-import org.robotninjas.barge.rpc.RaftExecutor;
 import org.robotninjas.barge.state.RaftStateContext;
+import org.robotninjas.barge.state.RaftStateContext.StateType;
 import org.robotninjas.protobuf.netty.server.RpcServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,16 +55,31 @@ public class RaftService extends AbstractService {
   private final ListeningExecutorService executor;
   private final RpcServer rpcServer;
   private final RaftStateContext ctx;
+  private final RaftLog raftLog;
+
+  private final BargeThreadPools bargeThreadPools;
 
   @Inject
-  RaftService(@Nonnull RpcServer rpcServer, @RaftExecutor ListeningExecutorService executor, @Nonnull RaftStateContext ctx) {
+  RaftService(@Nonnull RpcServer rpcServer, @Nonnull BargeThreadPools bargeThreadPools, @Nonnull RaftStateContext ctx, @Nonnull RaftLog raftLog) {
 
-    this.executor = checkNotNull(executor);
+    this.bargeThreadPools = checkNotNull(bargeThreadPools);
+    this.executor = checkNotNull(bargeThreadPools.getRaftExecutor());
     this.rpcServer = checkNotNull(rpcServer);
     this.ctx = checkNotNull(ctx);
+    this.raftLog = raftLog;
 
   }
 
+  public void bootstrap(Membership membership) {
+
+    LOGGER.info("Bootstrapping log with {}", membership);
+    if (!raftLog.isEmpty()) {
+      LOGGER.warn("Cannot bootstrap, as raft log already contains data");
+      throw new IllegalStateException();
+    }
+    raftLog.append(null, membership);
+  }
+
   @Override
   protected void doStart() {
 
@@ -89,6 +106,13 @@ protected void doStop() {
 
     try {
       rpcServer.stopAsync().awaitTerminated();
+      ctx.stop();
+      while (!ctx.isStopped()) {
+        Thread.sleep(10);
+      }
+      raftLog.close();
+
+      bargeThreadPools.close();
       notifyStopped();
     } catch (Exception e) {
       notifyFailed(e);
@@ -176,5 +200,49 @@ public RaftService build(StateMachine stateMachine) {
 
   }
 
+  public ListenableFuture<Boolean> setConfiguration(final RaftMembership oldMembership, final RaftMembership newMembership) {
+
+    // Make sure this happens on the Barge thread
+    ListenableFuture<ListenableFuture<Boolean>> response =
+      executor.submit(new Callable<ListenableFuture<Boolean>>() {
+        @Override
+        public ListenableFuture<Boolean> call() throws Exception {
+          return ctx.setConfiguration(oldMembership, newMembership);
+        }
+      });
+
+    return Futures.dereference(response);
+
+  }
+
+  public RaftMembership getClusterMembership() {
+    return ctx.getConfigurationState().getClusterMembership();
+  }
+
+  public String getServerKey() {
+    return ctx.getConfigurationState().self().getKey();
+  }
+
+  public boolean isLeader() {
+    return ctx.getState() == StateType.LEADER;
+  }
+
+  public RaftClusterHealth getClusterHealth() throws RaftException {
+
+    // Make sure this happens on the Barge thread
+    ListenableFuture<RaftClusterHealth> response = executor.submit(new Callable<RaftClusterHealth>() {
+      @Override
+      public RaftClusterHealth call() throws Exception {
+        return ctx.getClusterHealth();
+      }
+    });
+
+    return Futures.get(response, RaftException.class);
+
+  }
+
+  public boolean isUninitialized() {
+    return raftLog.isEmpty();
+  }
 
 }

barge-core/src/main/java/org/robotninjas/barge/Replica.java

@@ -35,19 +35,21 @@
 public class Replica {
 
   private final InetSocketAddress address;
+  private final String key;
 
-  Replica(@Nonnull InetSocketAddress address) {
+  private Replica(@Nonnull String key, @Nonnull InetSocketAddress address) {
+    this.key = checkNotNull(key);
     this.address = checkNotNull(address);
   }
 
   @Nonnull
-  public static Replica fromString(@Nonnull String info) {
+  public static Replica fromString(@Nonnull String key) {
     try {
-      checkNotNull(info);
-      HostAndPort hostAndPort = HostAndPort.fromString(info);
+      checkNotNull(key);
+      HostAndPort hostAndPort = HostAndPort.fromString(key);
       InetAddress addr = InetAddress.getByName(hostAndPort.getHostText());
       InetSocketAddress saddr = new InetSocketAddress(addr, hostAndPort.getPort());
-      return new Replica(saddr);
+      return new Replica(key, saddr);
     } catch (UnknownHostException e) {
       throw Throwables.propagate(e);
     }
@@ -83,4 +85,8 @@ public boolean equals(Object o) {
   public String toString() {
     return address.getAddress().getHostName() + ":" + address.getPort();
   }
+
+  public String getKey() {
+    return key;
+  }
 }

barge-core/src/main/java/org/robotninjas/barge/log/LogModule.java

@@ -16,7 +16,6 @@
 
 package org.robotninjas.barge.log;
 
-import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.inject.PrivateModule;
 import com.google.inject.Provides;
 import com.google.inject.Singleton;
@@ -35,12 +34,10 @@ public class LogModule extends PrivateModule {
 
   private final File logDirectory;
   private final StateMachine stateMachine;
-  private final ListeningExecutorService executor;
 
-  public LogModule(@Nonnull File logDirectory, @Nonnull StateMachine stateMachine, @Nonnull ListeningExecutorService executor) {
+  public LogModule(@Nonnull File logDirectory, @Nonnull StateMachine stateMachine) {
     this.logDirectory = checkNotNull(logDirectory);
     this.stateMachine = checkNotNull(stateMachine);
-    this.executor = checkNotNull(executor);
   }
 
   @Override
@@ -49,9 +46,9 @@ protected void configure() {
     bind(StateMachine.class).toInstance(stateMachine);
     bind(StateMachineProxy.class);
     bind(RaftLog.class).asEagerSingleton();
-    bind(ListeningExecutorService.class)
-      .annotatedWith(StateMachineExecutor.class)
-      .toInstance(executor);
+//    bind(ListeningExecutorService.class)
+//      .annotatedWith(StateMachineExecutor.class)
+//      .toInstance(executor);
     expose(RaftLog.class);
 
   }