import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
private CopycatServer buildServer() {
    CopycatServer server = CopycatServer.builder(localAddress)
            .withName("partition-" + partition.getId())
            .withSerializer(serializer.clone())
            .withTransport(transport.get())
            .withStateMachine(ResourceManagerState::new)
            .withStorage(Storage.builder()
                    .withStorageLevel(StorageLevel.DISK)
                    .withCompactionThreads(1)
                    .withDirectory(dataFolder)
                    .withMaxEntriesPerSegment(MAX_ENTRIES_PER_LOG_SEGMENT)
                    .build())
            .build();
    server.serializer().resolve(new ResourceManagerTypeResolver());
    return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Creates a Copycat server.
 */
protected CopycatServer createCopycatServer(Address address) {
    CopycatServer server = CopycatServer.builder(address)
            .withTransport(new LocalTransport(registry))
            .withStorage(Storage.builder()
                    .withStorageLevel(StorageLevel.DISK)
                    .withDirectory(TEST_DIR + "/" + address.port())
                    .build())
            .withStateMachine(ResourceManagerState::new)
            .withSerializer(serializer.clone())
            .withHeartbeatInterval(Duration.ofMillis(25))
            .withElectionTimeout(Duration.ofMillis(50))
            .withSessionTimeout(Duration.ofMillis(100))
            .build();
    copycatServers.add(server);
    return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@Override
protected CopycatServer createCopycatServer(Address address) {
    CopycatServer server = CopycatServer.builder(address)
            .withTransport(new LocalTransport(registry))
            .withStorage(Storage.builder()
                                 .withStorageLevel(StorageLevel.MEMORY)
                                 .withDirectory(testDir + "/" + address.port())
                                 .build())
            .withStateMachine(ResourceManagerState::new)
            .withSerializer(serializer.clone())
            .withHeartbeatInterval(Duration.ofMillis(25))
            .withElectionTimeout(Duration.ofMillis(50))
            .withSessionTimeout(Duration.ofMillis(100))
            .build();
    copycatServers.add(server);
    return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public ServerContext(String name, Member.Type type, Address serverAddress, Address clientAddress, Storage storage, Serializer serializer, Supplier<StateMachine> stateMachineFactory, ConnectionManager connections, ThreadContext threadContext) {
  this.name = Assert.notNull(name, "name");
  this.storage = Assert.notNull(storage, "storage");
  this.serializer = Assert.notNull(serializer, "serializer");
  this.threadContext = Assert.notNull(threadContext, "threadContext");
  this.connections = Assert.notNull(connections, "connections");
  this.stateMachineFactory = Assert.notNull(stateMachineFactory, "stateMachineFactory");
  this.stateContext = new SingleThreadContext(String.format("copycat-server-%s-%s-state", serverAddress, name), threadContext.serializer().clone());

  // Open the meta store.
  threadContext.execute(() -> this.meta = storage.openMetaStore(name)).join();

  // Load the current term and last vote from disk.
  this.term = meta.loadTerm();
  this.lastVotedFor = meta.loadVote();

  // Reset the state machine.
  threadContext.execute(this::reset).join();

  this.cluster = new ClusterState(type, serverAddress, clientAddress, this);
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Returns a list of compactable segments.
 *
 * @return A list of compactable segments.
 */
private Iterable<Segment> getCompactableSegments(Storage storage, SegmentManager manager) {
  List<Segment> segments = new ArrayList<>(manager.segments().size());
  Iterator<Segment> iterator = manager.segments().iterator();
  Segment segment = iterator.next();
  while (iterator.hasNext()) {
    Segment nextSegment = iterator.next();

    // Segments that have already been compacted are eligible for compaction. For uncompacted segments, the segment must be full, consist
    // of entries less than the minorIndex, and a later segment with at least one committed entry must exist in the log. This ensures that
    // a non-empty entry always remains at the end of the log.
    if (segment.isCompacted() || (segment.isFull() && segment.lastIndex() < compactor.minorIndex() && nextSegment.firstIndex() <= manager.commitIndex() && !nextSegment.isEmpty())) {
      // Calculate the percentage of entries that have been released in the segment.
      double compactablePercentage = segment.releaseCount() / (double) segment.count();

      // If the percentage of entries released times the segment version meets the compaction threshold,
      // add the segment to the segments list for compaction.
      if (compactablePercentage * segment.descriptor().version() >= storage.compactionThreshold()) {
        segments.add(segment);
      }
    }

    segment = nextSegment;
  }
  return segments;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@BeforeMethod
public void createStateMachine() throws Throwable {
  Serializer serializer = new Serializer().resolve(
    new ClientRequestTypeResolver(),
    new ClientResponseTypeResolver(),
    new ProtocolSerialization(),
    new ServerSerialization(),
    new StorageSerialization()
  ).disableWhitelist();

  callerContext = new SingleThreadContext("caller", serializer.clone());
  stateContext = new SingleThreadContext("state", serializer.clone());
  LocalServerRegistry registry = new LocalServerRegistry();
  transport = new LocalTransport(registry);
  Storage storage = new Storage(StorageLevel.MEMORY);
  ServerMember member = new ServerMember(Member.Type.ACTIVE, new Address("localhost", 5000), new Address("localhost", 6000), Instant.now());

  new SingleThreadContext("test", serializer.clone()).executor().execute(() -> {
    state = new ServerContext("test", member.type(), member.serverAddress(), member.clientAddress(), storage, serializer, TestStateMachine::new, new ConnectionManager(new LocalTransport(registry).client()), callerContext);
    resume();
  });
  await(1000);
  timestamp = System.currentTimeMillis();
  sequence = new AtomicLong();
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Creates a Copycat server.
 */
private CopycatServer createServer(Member member) {
  CopycatServer.Builder builder = CopycatServer.builder(member.clientAddress(), member.serverAddress())
    .withType(member.type())
    .withTransport(new NettyTransport())
    .withStorage(Storage.builder()
      .withStorageLevel(StorageLevel.DISK)
      .withDirectory(new File(String.format("target/fuzz-logs/%d", member.address().hashCode())))
      .withMaxSegmentSize(randomNumber(1024 * 1024 * 7) + (1024 * 1024))
      .withMaxEntriesPerSegment(randomNumber(10000) + 1000)
      .withCompactionThreads(randomNumber(4) + 1)
      .withCompactionThreshold(Math.random() / (double) 2)
      .withEntryBufferSize(randomNumber(10000) + 1)
      .withFlushOnCommit(randomBoolean())
      .withMinorCompactionInterval(Duration.ofSeconds(randomNumber(30) + 15))
      .withMajorCompactionInterval(Duration.ofSeconds(randomNumber(60) + 60))
      .build())
    .withStateMachine(FuzzStateMachine::new);

  CopycatServer server = builder.build();
  server.serializer().disableWhitelist();
  servers.add(server);
  return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Creates a Copycat server.
 */
private CopycatServer createServer(Member member) {
  CopycatServer.Builder builder = CopycatServer.builder(member.clientAddress(), member.serverAddress())
    .withType(member.type())
    .withTransport(new NettyTransport())
    .withStorage(Storage.builder()
      .withStorageLevel(StorageLevel.DISK)
      .withDirectory(new File(String.format("target/performance-logs/%d", member.address().hashCode())))
      .withCompactionThreads(1)
      .build())
    .withStateMachine(PerformanceStateMachine::new);

  CopycatServer server = builder.build();
  server.serializer().disableWhitelist();
  servers.add(server);
  return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Creates a Copycat server.
 */
private CopycatServer createServer(Member member) {
  CopycatServer.Builder builder = CopycatServer.builder(member.clientAddress(), member.serverAddress())
    .withType(member.type())
    .withTransport(new LocalTransport(registry))
    .withStorage(Storage.builder()
      .withStorageLevel(StorageLevel.MEMORY)
      .withMaxSegmentSize(1024 * 1024)
      .withCompactionThreads(1)
      .build())
    .withStateMachine(TestStateMachine::new);

  CopycatServer server = builder.build();
  server.serializer().disableWhitelist();
  servers.add(server);
  return server;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Starts the server.
 */
public static void main(String[] args) throws Exception {
  if (args.length < 2)
    throw new IllegalArgumentException("must supply a path and set of host:port tuples");

  // Parse the address to which to bind the server.
  String[] mainParts = args[1].split(":");
  Address address = new Address(mainParts[0], Integer.valueOf(mainParts[1]));

  // Build a list of all member addresses to which to connect.
  List<Address> members = new ArrayList<>();
  for (int i = 1; i < args.length; i++) {
    String[] parts = args[i].split(":");
    members.add(new Address(parts[0], Integer.valueOf(parts[1])));
  }

  CopycatServer server = CopycatServer.builder(address)
    .withStateMachine(ValueStateMachine::new)
    .withTransport(new NettyTransport())
    .withStorage(Storage.builder()
      .withDirectory(args[0])
      .withMaxSegmentSize(1024 * 1024 * 32)
      .withMinorCompactionInterval(Duration.ofMinutes(1))
      .withMajorCompactionInterval(Duration.ofMinutes(15))
      .build())
    .build();

  server.serializer().register(SetCommand.class, 1);
  server.serializer().register(GetQuery.class, 2);
  server.serializer().register(DeleteCommand.class, 3);

  server.bootstrap(members).join();

  while (server.isRunning()) {
    Thread.sleep(1000);
  }
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@Override
public List<CompactionTask> buildTasks(Storage storage, SegmentManager segments) {
  List<CompactionTask> tasks = new ArrayList<>(segments.segments().size());
  for (Segment segment : getCompactableSegments(storage, segments)) {
    tasks.add(new MinorCompactionTask(segments, segment, compactor.snapshotIndex(), compactor.majorIndex(), compactor.getDefaultCompactionMode()));
  }
  return tasks;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
public Compactor(Storage storage, SegmentManager segments, ScheduledExecutorService executor) {
  this.storage = Assert.notNull(storage, "storage");
  this.segments = Assert.notNull(segments, "segments");
  this.executor = Assert.notNull(executor, "executor");
  minor = executor.scheduleAtFixedRate(() -> compact(Compaction.MINOR), storage.minorCompactionInterval().toMillis(), storage.minorCompactionInterval().toMillis(), TimeUnit.MILLISECONDS);
  major = executor.scheduleAtFixedRate(() -> compact(Compaction.MAJOR), storage.majorCompactionInterval().toMillis(), storage.majorCompactionInterval().toMillis(), TimeUnit.MILLISECONDS);
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Returns a list of segments lists to compact, where segments are grouped according to how they will be merged during
 * compaction.
 */
public List<List<Segment>> getCompactableGroups(Storage storage, SegmentManager manager) {
  List<List<Segment>> compact = new ArrayList<>();
  List<Segment> segments = null;
  for (Segment segment : getCompactableSegments(manager)) {
    // If this is the first segment in a segments list, add the segment.
    if (segments == null) {
      segments = new ArrayList<>();
      segments.add(segment);
    }
    // If the total size of all segments is less than the maximum size of any segment, add the segment to the segments list.
    else if (segments.stream().mapToLong(Segment::size).sum() + segment.size() <= storage.maxSegmentSize()
      && segments.stream().mapToLong(Segment::count).sum() + segment.count() <= storage.maxEntriesPerSegment()) {
      segments.add(segment);
    }
    // If there's not enough room to combine segments, reset the segments list.
    else {
      compact.add(segments);
      segments = new ArrayList<>();
      segments.add(segment);
    }
  }

  // Ensure all compactable segments have been added to the compact segments list.
  if (segments != null) {
    compact.add(segments);
  }
  return compact;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
public MetaStore(String name, Storage storage, Serializer serializer) {
  this.storage = Assert.notNull(storage, "storage");
  this.serializer = Assert.notNull(serializer, "serializer");

  Assert.notNull(storage.directory(), "null storage directory");
  if (!(storage.directory().isDirectory() || storage.directory().mkdirs())) {
    throw new IllegalArgumentException(String.format("Can't create storage directory [%s].", storage.directory()));
  }

  // Note that for raft safety, irrespective of the storage level, <term, vote> metadata is always persisted on disk.
  File metaFile = new File(storage.directory(), String.format("%s.meta", name));
  FileBuffer tmpMetadataBuffer = FileBuffer.allocate(metaFile, 12);

  if (storage.level() == StorageLevel.MEMORY) {
    configurationBuffer = HeapBuffer.allocate(32);
  } else {
    File confFile = new File(storage.directory(), String.format("%s.conf", name));
    configurationBuffer = FileBuffer.allocate(confFile, 32);

    //Note: backward compatibility with pre 1.2.6 release. This can be removed in later releases.
    Configuration configuration = loadConfiguration();
    if (configuration == null) {
      configuration = loadConfigurationFromMetadataBuffer(tmpMetadataBuffer);
      if (configuration != null) {
        storeConfiguration(configuration);
        tmpMetadataBuffer = deleteConfigurationFromMetadataBuffer(metaFile, tmpMetadataBuffer);
      }
    }
  }

  metadataBuffer = tmpMetadataBuffer;
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
/**
 * Sets up a server state.
 */
@BeforeMethod
void beforeMethod() throws Throwable {
  serializer = new Serializer();
  serializer.resolve(
    new ClientRequestTypeResolver(),
    new ClientResponseTypeResolver(),
    new ProtocolSerialization(),
    new ServerSerialization(),
    new StorageSerialization()
  ).disableWhitelist();

  storageDir = Files.createTempDirectory("copycat-test");
  storage = Storage.builder().withStorageLevel(StorageLevel.MEMORY).withDirectory(storageDir.toFile()).build();

  members = createMembers(3);
  transport = new LocalTransport(new LocalServerRegistry());

  serverCtx = new SingleThreadContext("test-server", serializer);
  new SingleThreadContext("test", serializer.clone()).executor().execute(() -> {
    serverContext = new ServerContext("test", members.get(0).type(), members.get(0).serverAddress(), members.get(0).clientAddress(), storage, serializer, TestStateMachine::new, new ConnectionManager(transport.client()), serverCtx);
    serverContext.getThreadContext().executor().execute(() -> {
      serverContext.getClusterState().configure(new Configuration(0, 0, Instant.now().toEpochMilli(), members));
      resume();
    });
  });
  await(1000);
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
public static void configureMemoryOnlyStorage(ConsensusServer server) {
  server.setConsensusStorageProvider(new ConsensusStorageProvider(null) {
    @Override
    public Storage createStorage() {
      return Storage.builder().withStorageLevel(StorageLevel.MEMORY).build();
    }
  });
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@Override
public List<CompactionTask> buildTasks(Storage storage, SegmentManager segments) {
  List<List<Segment>> groups = getCompactableGroups(storage, segments);
  return !groups.isEmpty() ? Collections.singletonList(new MajorCompactionTask(segments, groups, compactor.snapshotIndex(), compactor.majorIndex(), compactor.getDefaultCompactionMode())) : Collections.emptyList();
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
public SnapshotStore(String name, Storage storage, Serializer serializer) {
  this.name = Assert.notNull(name, "name");
  this.storage = Assert.notNull(storage, "storage");
  this.serializer = Assert.notNull(serializer, "serializer");
  open();
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
@Override
public void clusterInitialized() {
  Address ourAddr = toCopycatAddress(ourNodeAddressProvider.getOurNodeAddress());
  List<Address> members = consensusClusterNodeAddressProvider.getClusterNodeAddressesForConsensus().stream()
      .map(addr -> toCopycatAddress(addr)).collect(Collectors.toList());

  if (members.isEmpty())
    // if there's no member we could connect to, use a single node cluster with only ourselves.
    members.add(ourAddr);

  File consensusDataDirFile = new File(consensusDataDir);

  if (!consensusDataDirFile.exists())
    if (!consensusDataDirFile.mkdirs())
      throw new RuntimeException("Could not create consenusDataDir at " + consensusDataDirFile.getAbsolutePath()
          + ". Restart diqube-server!");

  logger.info("Starting up consensus node with local data dir at '{}'.", consensusDataDirFile.getAbsolutePath());
  if (consensusStorageProvider == null)
    consensusStorageProvider = new ConsensusStorageProvider(consensusDataDirFile);
  Storage storage = consensusStorageProvider.createStorage();

  copycatServer = CopycatServer.builder(ourAddr, members). //
      withName(COPYCAT_SERVER_NAME). //
      withType(Member.Type.ACTIVE). //
      withTransport(transport). //
      withStorage(storage). //
      withSerializer(serializer). //
      withSessionTimeout(Duration.ofMillis(sessionTimeoutMs)). //
      withElectionTimeout(Duration.ofMillis(electionTimeoutMs)). //
      withHeartbeatInterval(Duration.ofMillis(keepAliveMs)). //
      // seems to not be used in rc5, nevertheless set to a meaningful value:
      withGlobalSuspendTimeout(Duration.ofMinutes(GLOBAL_SUSPEND_TIMEOUT_MIN)). //
      withStateMachine(() -> new DiqubeStateMachine()).build();

  CompletableFuture<?> serverOpenFuture = copycatServer.start().handle((result, error) -> {
    if (error != null)
      throw new RuntimeException("Could not start Consensus node. Restart diqube-server!", error);

    logger.info("Consensus node started successfully.");

    copycatServer.cluster().onLeaderElection(leaderMember -> {
      if (leaderMember != null) {
        lastKnownCopycatLeaderAddress = leaderMember.address();
        lastKnownCopycatLeaderTimestamp = Instant.now();
        logger.info("New consensus leader address: {}", lastKnownCopycatLeaderAddress);
      }
    });
    lastKnownCopycatLeaderAddress = copycatServer.cluster().leader().address();
    lastKnownCopycatLeaderTimestamp = Instant.now();
    logger.info("New consensus leader address: {}", lastKnownCopycatLeaderAddress);

    if (listeners != null)
      listeners.forEach(l -> l.consensusInitialized());

    return null;
  });
  // TODO #91 Workaround for unreliable copycat server startup: If listen is too quick, not all handlers might be
  // installed on a future, and then some stuff might get called on the wrong thread.
  catalystServer.allowCompletionOfListen();
  serverOpenFuture.join();
}
 

import io.atomix.copycat.server.storage.Storage; //导入依赖的package包/类
public Storage createStorage() {
  return Storage.builder().withStorageLevel(StorageLevel.DISK).withDirectory(consensusDataDirFile).build();
}