Class CounterService

java.lang.Object
org.jgroups.raft.blocks.CounterService
All Implemented Interfaces:
RAFT.RoleChange, StateMachine
public class CounterService extends Object implements StateMachine, RAFT.RoleChange
Provides a consensus based distributed counter (similar to AtomicLong) which can be atomically updated across a cluster.
Since:
0.2
Author:
Bela Ban

  • Field Details

    • raft

      protected RaftHandle raft

    • repl_timeout

      protected long repl_timeout

    • allow_dirty_reads

      protected boolean allow_dirty_reads
      If true, reads can return the local counter value directly. Else, reads have to go through the leader

    • counters

      protected final Map<String,Long> counters

  • Constructor Details

    • CounterService

      public CounterService(org.jgroups.JChannel ch)

  • Method Details

    • setChannel

      public void setChannel(org.jgroups.JChannel ch)

    • channel

      public org.jgroups.JChannel channel()

    • addRoleChangeListener

      public void addRoleChangeListener(RAFT.RoleChange listener)

    • replTimeout

      public long replTimeout()

    • replTimeout

      public CounterService replTimeout(long timeout)

    • allowDirtyReads

      public boolean allowDirtyReads()

    • allowDirtyReads

      public CounterService allowDirtyReads(boolean flag)

    • lastApplied

      public long lastApplied()

    • commitIndex

      public long commitIndex()

    • snapshot

      public void snapshot() throws Exception
      Throws:
      Exception

    • logSize

      public long logSize()

    • raftId

      public String raftId()

    • raftId

      public CounterService raftId(String id)

    • getOrCreateCounter

      public RaftSyncCounter getOrCreateCounter(String name, long initial_value) throws Exception
      Returns an existing counter, or creates a new one if none exists. This is a cluster-wide operation which would fail if no leader is elected.
      Parameters:
      name - Name of the counter, different counters have to have different names
      initial_value - The initial value of a new counter if there is no existing counter. Ignored if the counter already exists
      Returns:
      The counter implementation
      Throws:
      Exception

    • deleteCounter

      public void deleteCounter(String name) throws Exception
      Deletes a counter instance (on the coordinator)
      Parameters:
      name - The name of the counter. No-op if the counter doesn't exist
      Throws:
      Exception

    • deleteCounterAsync

      public CompletionStage<Void> deleteCounterAsync(String name)
      Deletes a counter instance.
      Parameters:
      name - The name of the counter. No-op if the counter doesn't exist
      Returns:
      Returns a CompletionStage which is completed when the majority reach consensus.

    • printCounters

      public String printCounters()

    • apply

      public byte[] apply(byte[] data, int offset, int length, boolean serialize_response)
      Description copied from interface: StateMachine
      Applies a command to the state machine.

      The contents of the byte[] buffer are interpreted by the state machine and serialized into the actual command. The serialization mechanism should be strict with the given intervals to avoid copying the byte buffer.

      This method should never throw an exception as it could leave the protocol in an undefined state. Instead, it should serialize any exception and return the buffer for the client application to handle the exception.

      Specified by:
      apply in interface StateMachine
      Parameters:
      data - The byte[] buffer
      offset - The offset at which the data starts
      length - The length of the data
      serialize_response - If true, serialize and return the response, else return null
      Returns:
      A serialized response value, or null (e.g. if the method returned void)

    • writeContentTo

      public void writeContentTo(DataOutput out) throws Exception
      Description copied from interface: StateMachine
      Writes the contents of the state machine to an output stream.

      This is typically called on the leader to provide state to a new node, or a node that's lagging far behind. All the data needed to completely recover the state machine should be written to the buffer.

      Synchronous Execution & Event Loop Blocking

      This method is invoked synchronously by the main Raft event loop. While this method is executing, the core Raft protocol is entirely blocked. Updates to the state machine may need to be put on hold while the state is written to the output stream. Otherwise, it is not possible to guarantee a consistent view of the state machine. For an asynchronous version, check AsyncSnapshot.

      Buffer Ownership & Determinism

      The framework owns the DataOutput stream. The implementation must deterministically and completely write its current state to the stream before returning, as the buffer is flushed and released immediately after the method finishes. Try to stick to the minimum data needed to restore the state machine.

      Optimization Recommendations

      Because this method blocks the main event loop until the snapshot is fully written, users are strongly encouraged to think carefully about optimizations. To minimize the blocking time, consider maintaining a fast serialization format or utilizing internal Copy-on-Write (COW) data structures so that the actual state iteration and serialization can be performed as quickly as possible.

      Failures

      If an exception is thrown while taking the snapshot, we do not submit the underlying buffer to remote nodes.

      Specified by:
      writeContentTo in interface StateMachine
      Parameters:
      out - The output stream
      Throws:
      Exception - Thrown on serialization or other failure.

    • readContentFrom

      public void readContentFrom(DataInput in)
      Description copied from interface: StateMachine
      Reads the contents of the state machine from an input stream.

      This can be the case when an InstallSnapshot RPC is used to bootstrap a new node, or a node that's lagging far behind. The parsing depends on the concrete state machine implementation, but the idea is that the stream is a sequence of commands, each of which can be passed to StateMachine.apply(byte[], int, int, boolean).

      Synchronous Execution & Event Loop Blocking

      This method is invoked synchronously by the main Raft event loop. While this method is executing, the core Raft protocol is entirely blocked. The node will not process incoming client requests until the restoration is fully complete. For an asynchronous version, check AsyncSnapshot.

      Buffer Ownership & Determinism

      The framework owns the provided DataInput buffer, and this ownership is released the moment this method finishes. The implementation must completely read the buffer and restore its internal state deterministically before returning. The state machine implementation may need to remove all contents before populating itself from the stream.

      This method is never called concurrently with invocations to StateMachine.apply(byte[], int, int, boolean). No locking mechanism is needed in the state machine.

      Failures

      Similar to StateMachine.apply(byte[], int, int, boolean), exceptions must be handled carefully when restoring the state machine. If an exception is thrown, the underlying system could be left in an undefined state. You should ensure the state machine remains consistent and deterministic.

      Specified by:
      readContentFrom in interface StateMachine
      Parameters:
      in - The input stream

    • readAndDumpSnapshot

      public static String readAndDumpSnapshot(DataInput in)

    • dumpLog

      public void dumpLog()

    • dumpLogEntry

      public static String dumpLogEntry(LogEntry e)

    • roleChanged

      public void roleChanged(Role role)
      Specified by:
      roleChanged in interface RAFT.RoleChange

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • asyncCounter

      public RaftAsyncCounter asyncCounter(String name)
      Returns an RaftAsyncCounter instance of the counter.

      This is local operation, and it does not create the counter in the raft log.

      Parameters:
      name - Name of the counter, different counters have to have different names.
      Returns:
      The RaftAsyncCounter instance

    • getOrCreateAsyncCounter

      public CompletionStage<RaftAsyncCounter> getOrCreateAsyncCounter(String name, long initialValue)
      Returns an existing counter, or creates a new one if none exists.

      This is a cluster-wide operation which would fail if no leader is elected.

      Parameters:
      name - Name of the counter, different counters have to have different names
      initialValue - The initial value of a new counter if there is no existing counter. Ignored if the counter already exists
      Returns:
      The RaftAsyncCounter implementation.

    • asyncGet

      protected CompletionStage<Long> asyncGet(org.jgroups.util.AsciiString name)

    • asyncSet

      protected CompletionStage<Void> asyncSet(org.jgroups.util.AsciiString name, long value)

    • asyncAddAndGet

      protected CompletionStage<Long> asyncAddAndGet(org.jgroups.util.AsciiString name, long delta, Options opts)

    • asyncCompareAndSwap

      protected CompletionStage<Long> asyncCompareAndSwap(org.jgroups.util.AsciiString name, long expected, long value, Options opts)

    • asyncUpdate

      protected <T extends org.jgroups.util.Streamable> CompletionStage<T> asyncUpdate(org.jgroups.util.AsciiString name, org.jgroups.blocks.atomic.CounterFunction<T> function, Options options)

    • invokeAsyncAndGet

      protected CompletionStage<Long> invokeAsyncAndGet(CounterService.Command command, org.jgroups.util.AsciiString name, Options opts)

    • invokeAsyncAddAndGet

      protected CompletionStage<Long> invokeAsyncAddAndGet(org.jgroups.util.AsciiString name, long arg, Options opts)

    • invokeAsync

      protected CompletionStage<Void> invokeAsync(CounterService.Command command, org.jgroups.util.AsciiString name, long arg)

    • print

      protected static String print(CounterService.Command command, String name, int num_args, DataInput in)

    • _create

      protected long _create(String name, long initial_value)

    • _delete

      protected void _delete(String name)

    • _get

      protected long _get(String name)

    • _set

      protected void _set(String name, long new_val)

    • _add

      protected long _add(String name, long delta)

    • _compareAndSwap

      protected long _compareAndSwap(String name, long expected, long value)

    • _update

      protected <T extends org.jgroups.util.Streamable> T _update(String name, org.jgroups.blocks.atomic.CounterFunction<T> function)