Struct SharedContext

pub struct SharedContext {
    pub(crate) channel_map: Mutex<HashMap<UpDownActorIds, ConsumableChannelPair>>,
    pub(crate) actor_infos: RwLock<HashMap<ActorId, ActorInfo>>,
    pub(crate) addr: HostAddr,
    pub(crate) compute_client_pool: ComputeClientPoolRef,
    pub(crate) config: StreamingConfig,
    pub(crate) local_barrier_manager: LocalBarrierManager,
}

Stores the information which may be modified from the data plane.

The data structure is created in LocalBarrierWorker and is shared by actors created between two recoveries. In every recovery, the LocalBarrierWorker will create a new instance of SharedContext, and the original one becomes stale. The new one is shared by actors created after recovery.

Fields

channel_map: Mutex<HashMap<UpDownActorIds, ConsumableChannelPair>>

Stores the senders and receivers for later Processor’s usage.

Each actor has several senders and several receivers. Senders and receivers are created during update_actors and stored in a channel map. Upon build_actors, all these channels will be taken out and built into the executors and outputs. One sender or one receiver can be uniquely determined by the upstream and downstream actor id.

There are three cases when we need local channels to pass around messages:

1. pass Message between two local actors
2. The RPC client at the downstream actor forwards received Message to one channel in ReceiverExecutor or MergerExecutor.
3. The RPC Output at the upstream actor forwards received Message to ExchangeServiceImpl.

The channel serves as a buffer because ExchangeServiceImpl is on the server-side and we will also introduce backpressure.

actor_infos: RwLock<HashMap<ActorId, ActorInfo>>

Stores all actor information.

addr: HostAddr

Stores the local address.

It is used to test whether an actor is local or not, thus determining whether we should setup local channel only or remote rpc connection between two actors/actors.

compute_client_pool: ComputeClientPoolRef

Compute client pool for streaming gRPC exchange.

config: StreamingConfig

local_barrier_manager: LocalBarrierManager

Implementations

impl SharedContext

pub fn new( env: &StreamEnvironment, local_barrier_manager: LocalBarrierManager, ) -> Self

fn get_or_insert_channels( &self, ids: UpDownActorIds, ) -> MappedMutexGuard<'_, ConsumableChannelPair>

Get the channel pair for the given actor ids. If the channel pair does not exist, create one with the configured permits.

pub fn take_sender(&self, ids: &UpDownActorIds) -> StreamResult<Sender>

pub fn take_receiver(&self, ids: UpDownActorIds) -> StreamResult<Receiver>

pub fn get_actor_info(&self, actor_id: &ActorId) -> StreamResult<ActorInfo>

pub fn config(&self) -> &StreamingConfig

pub(crate) fn drop_actors(&self, actors: &HashSet<ActorId>)

Trait Implementations

impl Debug for SharedContext

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.