Struct StreamActor

pub struct StreamActor { /* private fields */ }

A stream represents a linear sequence of execution. Operations on different streams can execute concurrently.

For CUDA operators, streams will invoke the corresponding stream management APIs to perform synchronization.

For CPU operators, streams will just execute synchronously on their own OS thread.

Trait Implementations

impl Actor for StreamActor

fn spawn_server_task<F>(future: F) -> JoinHandle<F::Output>

where F: Future + Send + 'static, F::Output: Send + 'static,

Specialize spawn_server_task for StreamActor, because we want to run the stream on a dedicated OS thread. This is because:

• Streams do expensive blocking CPU operations (like calling CPU kernels).
• Torch/CUDA make use of thread-local state, so moving tasks across threads is problematic.

type Params = StreamParams

The type of initialization parameters accepted by this actor.

fn new<'async_trait>( __arg0: Self::Params, ) -> Pin<Box<dyn Future<Output = Result<Self>> + Send + 'async_trait>>

where Self: 'async_trait,

Creates a new actor instance given its instantiation parameters.

fn init<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Instance<Self>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Initialize the actor, after the runtime has been fully initialized. Init thus provides a mechanism by which an actor can reliably and always receive some initial event that can be used to kick off further (potentially delayed) processing.

fn spawn<'life0, 'async_trait>( cap: &'life0 (impl CanSpawn + 'async_trait), params: Self::Params, ) -> Pin<Box<dyn Future<Output = Result<ActorHandle<Self>, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Spawn a child actor, given a spawning capability (usually given by Instance). The spawned actor will be supervised by the parent (spawning) actor.

fn spawn_detached<'async_trait>( params: Self::Params, ) -> Pin<Box<dyn Future<Output = Result<ActorHandle<Self>, Error>> + Send + 'async_trait>>

where Self: 'async_trait,

Spawns this actor in a detached state, handling its messages in a background task. The returned handle is used to control the actor’s lifecycle and to interact with it.

fn handle_supervision_event<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _this: &'life1 Instance<Self>, _event: &'life2 ActorSupervisionEvent, ) -> Pin<Box<dyn Future<Output = Result<bool, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: 'async_trait,

Handle actor supervision event. Return `Ok(true)`` if the event is handled here.

fn handle_undeliverable_message<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Instance<Self>, __arg2: Undeliverable<MessageEnvelope>, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Default undeliverable message handling behavior.

impl Debug for StreamActor

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

Formats the value using the given formatter.

impl Handler<StreamMessage> for StreamActor

fn handle<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, message: StreamMessage, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Handle the next M-typed message.

impl StreamMessageHandler for StreamActor

fn call_function<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, params: CallFunctionParams, device_meshes: HashMap<Ref, DeviceMesh>, remote_process_groups: HashMap<Ref, (DeviceMesh, Vec<String>, Arc<ActorHandle<NcclCommActor>>)>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn borrow_create<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, borrow: u64, tensor: Ref, first_use_sender: PortHandle<(Option<Event>, TensorCellResult)>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn borrow_first_use<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, borrow: u64, result: Ref, first_use_receiver: Arc<Mutex<PortReceiver<(Option<Event>, TensorCellResult)>>>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn borrow_last_use<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, borrow: u64, result: Ref, last_use_sender: PortHandle<(Option<Event>, TensorCellResult)>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn borrow_drop<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, borrow: u64, last_use_receiver: Arc<Mutex<PortReceiver<(Option<Event>, TensorCellResult)>>>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn delete_refs<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, refs: Vec<Ref>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn request_status<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn init_comm<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, comm: ActorHandle<NcclCommActor>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn reduce<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, comm: Arc<ActorHandle<NcclCommActor>>, dim_size: i64, result: Ref, local_tensor: Ref, factory: Factory, reduction: Reduction, scatter: bool, in_place: bool, out: Option<Ref>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn send_tensor<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, result: Ref, from_rank: Option<usize>, to_rank: Option<usize>, tensor: Ref, factory: Factory, comm: Arc<ActorHandle<NcclCommActor>>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn send_value<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, seq: Seq, worker_actor_id: ActorId, mutates: Vec<Ref>, function: Option<ResolvableFunction>, args: Vec<WireValue>, kwargs: HashMap<String, WireValue>, device_meshes: HashMap<Ref, DeviceMesh>, pipe: Option<PortHandle<PipeMessage>>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn send_result_of_actor_call<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, worker_actor_id: ActorId, params: ActorCallParams, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn call_actor_method<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, params: ActorMethodParams, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn set_value<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, seq: Seq, results: Vec<Option<Ref>>, pipe: PortHandle<PipeMessage>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn define_recording<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, recording: Ref, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn finalize_recording<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, recording: Ref, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn recording_formal<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, result: Ref, argument_index: usize, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn recording_result<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, result: Ref, output_index: usize, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn call_recording<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, seq: Seq, recording: Ref, results: Vec<Ref>, actuals: Vec<Ref>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn set_ref_unit_tests_only<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, reference: Ref, value: WireValue, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn set_tensor_ref_unit_tests_only<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, reference: Ref, tensor_result: TensorCellResult, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn get_ref_unit_tests_only<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, reference: Ref, ) -> Pin<Box<dyn Future<Output = Result<Option<Result<WireValue, String>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn get_tensor_ref_unit_tests_only<'life0, 'life1, 'async_trait>( &'life0 mut self, _cx: &'life1 Context<'_, Self>, reference: Ref, ) -> Pin<Box<dyn Future<Output = Result<Option<TensorCellResult>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The generated handler method for this enum variant.

fn handle<'life0, 'life1, 'async_trait>( &'life0 mut self, cx: &'life1 Context<'_, Self>, message: StreamMessage, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Handle the next message.