Package io.deephaven.server.barrage

Class BarrageMessageProducer

java.lang.Object
io.deephaven.util.referencecounting.ReferenceCounted
io.deephaven.engine.liveness.ReferenceCountedLivenessReferent
io.deephaven.engine.liveness.ReferenceCountedLivenessNode
io.deephaven.engine.liveness.LivenessArtifact
io.deephaven.server.barrage.BarrageMessageProducer
All Implemented Interfaces:
LogOutputAppendable, LivenessManager, LivenessNode, LivenessReferent, NotificationStepReceiver, DynamicNode, Serializable
public class BarrageMessageProducer extends LivenessArtifact implements DynamicNode, NotificationStepReceiver
The server-side implementation of a Barrage replication source.

When a client subscribes initially, a snapshot of the table is sent. The snapshot is obtained using either get() or getPrev() based on the state of the LogicalClock. On each subsequent update, the client is given the deltas between the last update propagation and the next.

When a client changes its subscription it will be sent a snapshot of only the data that the server believes it needs assuming that the client has been respecting the existing subscription. Practically, this means that the server may omit some data if the client's viewport change overlaps the currently recognized viewport.

It is possible to use this replication source to create subscriptions that propagate changes from one UGP to another inside the same JVM.

The client-side counterpart of this is the BarrageMessageReader.

See Also:

  • Field Details

    • SUBSCRIPTION_GROWTH_ENABLED

      public static final boolean SUBSCRIPTION_GROWTH_ENABLED

    • DELTA_CHUNK_SIZE

      public static final int DELTA_CHUNK_SIZE
      The number of rows in every chunk a delta records, except the last of a column. A configured deltaChunkSize that is not a power of two is rounded up to the next one, because that is what lets the copy kernel turn an encoded position into a chunk index and an offset within that chunk with a shift and a mask rather than a division per row; BarrageCopyKernel.copy(io.deephaven.extensions.barrage.chunk.BarrageCopyKernel.Runs, io.deephaven.chunk.WritableChunk<io.deephaven.chunk.attributes.Values>[], io.deephaven.chunk.WritableChunk<io.deephaven.chunk.attributes.Values>[][], io.deephaven.chunk.WritableChunk<io.deephaven.chunk.attributes.Values>[][], int) asserts it.

    • COMPACTION_ENABLED

      public static final boolean COMPACTION_ENABLED
      Whether a producer compacts its queue of pending deltas before the subscribers' update interval elapses. See shouldCompact() for the policy, and COMPACTION_MIN_FREED_FRACTION and COMPACTION_FLOOR_BYTES for its parameters.

    • COMPACTION_FLOOR_BYTES

      public static final long COMPACTION_FLOOR_BYTES
      A producer does not compact unless doing so would free at least this many bytes of chunk storage, however wasteful its queue looks as a fraction.

      This is what stops a stream of tiny deltas from compacting on every cycle. A delta that records a single row still occupies the pool's smallest chunk in each subscribed column, so from the second such delta on the freed fraction sits above 90% forever. That waste is real -- coalescedChunkBytes() measures allocated capacity, so the fraction is reporting chunks a compaction would genuinely release -- but what a fraction cannot say is how little it amounts to: such a queue grows by one smallest chunk per column per cycle, and the interval flushes it before it can grow far. Compacting there copies the whole queue to release one chunk per column, and the job's fixed costs -- scheduling, the run lock, the row set passes, the splice -- are the same as for a compaction that frees gigabytes. The floor makes such a producer wait until the saving is worth a job: at 4 MiB, a 20-byte row compacts about every 7,000 single-row appends and a 300-byte row about every 450.

    • COMPACTION_MIN_FREED_FRACTION

      public static final double COMPACTION_MIN_FREED_FRACTION
      A producer compacts once doing so would free this fraction of the chunk storage its pending deltas hold, *and* at least COMPACTION_FLOOR_BYTES. A fraction from 0.0 to 1.0; at the default a compaction has to release half the queue to be worth running.

      The setting trades processor time against heap. Total copying is at most (1 - f) / f times the data recorded, and the queue reaches about 1 / (1 - f) times the footprint it would have if it were coalesced: 0.5 copies once per byte recorded and holds twice that footprint, while 0.9 copies about a ninth as much and holds ten times it. Since reclaiming that heap is the point of compacting at all, raise it only where processor time is the scarcer resource. At 0.0 a producer compacts whenever the floor alone is met; at 1.0 it never compacts, since a queue that would coalesce to nothing still has a delta to coalesce.

      The value is trusted as configured, not validated. Outside 0.0 to 1.0 it does what the formula in shouldCompact() says: above one, or NaN, the threshold is unreachable and the producer never compacts; below zero the floor alone decides.

    • PROPAGATION_THREADS

      public static final int PROPAGATION_THREADS
      The most threads that write one propagation phase's messages to its subscribers at once, counting the thread that runs the propagation job. Writing to a subscriber serializes the message into the gRPC transport's buffers, which accept it without waiting for the client, so each write is independent work bounded by the processor; with one thread a phase takes as long as all of its subscribers' writes put together. The threads beyond the job's own come from a pool that every producer in the process shares, and that pool never holds more than this number less one, whatever the number of producers. One or less writes to every subscriber in turn on the job's thread.
      See Also:

    • PROPAGATION_JOB_SCHEDULER

      public static final String PROPAGATION_JOB_SCHEDULER
      The name under which the server binds the Supplier<JobScheduler> that producers write to their subscribers through. Each propagation phase takes a JobScheduler from it and runs its writes as one invokeParallel on it, so the schedulers it supplies must be usable from any thread. The server's are backed by a pool sized by PROPAGATION_THREADS.
      See Also:

    • DEDUP_IDENTICAL_SUBSCRIBERS

      public static final boolean DEDUP_IDENTICAL_SUBSCRIBERS
      Whether a propagation phase writes the propagated message once for each group of full subscriptions that would be sent identical bytes, handing each stream of the message to every member of the group as the stream is written, instead of writing the message once per subscription. A full subscription joins a group once the subscription has the whole table, and a group's members share their subscription options and columns. Viewport subscriptions, and full subscriptions still growing toward the whole table, are always written alone.

    • PARALLEL_COPY_MIN_BYTES

      public static final long PARALLEL_COPY_MIN_BYTES
      A group of subscriptions sharing one write copies each batch it captures to its members in parallel, on the propagation job scheduler, when the batch's size multiplied by one less than its members reaches this many bytes: the bytes the copies would move off the group's thread. Smaller rounds are copied one after another on the group's thread, where handing the copies out would cost more than it saves. With PROPAGATION_THREADS of one or less, every round is copied on the group's thread.

  • Constructor Details

  • Method Details

    • getRowSet

      public RowSet getRowSet()

    • getTableDefinition

      public TableDefinition getTableDefinition()

    • setOnGetSnapshot

      public void setOnGetSnapshot(Runnable onGetSnapshot, boolean isPreSnap)

    • addSubscription

      public void addSubscription(io.grpc.stub.StreamObserver<BarrageMessageWriter.MessageView> listener, BarrageSubscriptionOptions options, @Nullable @Nullable BitSet columnsToSubscribe, @Nullable @Nullable RowSet initialViewport, boolean reverseViewport)
      Add a subscription to this BarrageMessageProducer.
      Parameters:
      listener - The listener for this subscription
      options - The subscription options
      columnsToSubscribe - The initial columns to subscribe to
      initialViewport - Initial viewport, to be owned by the subscription

    • updateSubscription

      public boolean updateSubscription(io.grpc.stub.StreamObserver<BarrageMessageWriter.MessageView> listener, @Nullable @Nullable RowSet newViewport, @Nullable @Nullable BitSet columnsToSubscribe)

    • updateSubscription

      public boolean updateSubscription(io.grpc.stub.StreamObserver<BarrageMessageWriter.MessageView> listener, @Nullable @Nullable RowSet newViewport, @Nullable @Nullable BitSet columnsToSubscribe, boolean newReverseViewport)

    • removeSubscription

      public void removeSubscription(io.grpc.stub.StreamObserver<BarrageMessageWriter.MessageView> listener)

    • constructListener

      public InstrumentedTableUpdateListener constructListener()

    • destroy

      @OverridingMethodsMustInvokeSuper protected void destroy()
      Description copied from class: ReferenceCountedLivenessReferent
      Attempt to release (destructively when necessary) resources held by this object. This may render the object unusable for subsequent operations. Implementations should be sure to call super.destroy().

      This is intended to only ever be used as a side effect of decreasing the reference count to 0.

      Overrides:
      destroy in class ReferenceCountedLivenessReferent

    • isRefreshing

      public boolean isRefreshing()
      Description copied from interface: DynamicNode
      Is the node updating?
      Specified by:
      isRefreshing in interface DynamicNode
      Returns:
      true if the node is updating; false otherwise.

    • setRefreshing

      public boolean setRefreshing(boolean refreshing)
      Description copied from interface: DynamicNode
      Change the node's run mode.
      Specified by:
      setRefreshing in interface DynamicNode
      Parameters:
      refreshing - true to cause the node to update; false otherwise.
      Returns:
      new refreshing state

    • addParentReference

      public void addParentReference(Object parent)
      Description copied from interface: DynamicNode
      Called on a dependent node to ensure that a strong reference is maintained to any parent object that is required for the proper maintenance and functioning of the dependent. In the most common case, the parent object is a child listener to a parent node. The parent node only keeps a weak reference to its child listener, but the listener maintains a strong reference to the parent node. In this scenario, the only strong reference to the listener (and thus indirectly to the parent node itself) is the reference kept by the dependent node.
      Specified by:
      addParentReference in interface DynamicNode
      Parameters:
      parent - A parent of this node

    • setLastNotificationStep

      public void setLastNotificationStep(long lastNotificationStep)
      Description copied from interface: NotificationStepReceiver
      Deliver a last notification step to this receiver.
      Specified by:
      setLastNotificationStep in interface NotificationStepReceiver
      Parameters:
      lastNotificationStep - The last notification step to be delivered