[actor shutdown] Context method + no-deliver state (#71)

### Summary

Adds an initial `shutdown` message to the actor context. This allows an
actor to immediately shutdown (only callable from within the actor) and
sets some state on the actor so that no new messages are received.

### Motivation

Part of #29.

This is only a small subset of actor shutdown logic, see #29 for full
details.

### Test Plan

Still TBD. I'll see how this evolves as shutdown evolves over the course
of implementing #29.

src/actor/actor.rs

@@ -1,6 +1,8 @@
 use crate::actor::{ActorAddress, Letter, SenderType};
+use crate::executor::ExecutorCommands;
 use crate::message::{Message, ToMessage};
 use crate::system::RuntimeManagerRef;
+use crate::util::CommandChannel;
 use crossbeam_channel::Receiver;
 use log::{trace, warn};
 
@@ -54,6 +56,8 @@ pub struct ActorCell {
     pub(crate) address: ActorAddress,
     pub(crate) children: Vec<ActorAddress>,
     pub(crate) parent: Option<ActorAddress>,
+
+    state_flags: u8,
 }
 
 impl ActorCell {
@@ -69,8 +73,17 @@ impl ActorCell {
             address,
             children: Vec::new(),
             parent,
+            state_flags: 0,
         }
     }
+
+    pub(crate) fn set_shutdown(&mut self) {
+        self.state_flags |= 0b0000_0001;
+    }
+
+    pub(crate) fn is_shutdown(&self) -> bool {
+        self.state_flags & 0b0000_0001 != 0
+    }
 }
 
 /// Debug macro for serializing and deserializing a message. The goal is to reduce
@@ -95,6 +108,7 @@ macro_rules! debug_serialize_msg {
 pub struct Context<'a> {
     pub(crate) address: &'a ActorAddress,
     pub(crate) runtime_manager: &'a RuntimeManagerRef,
+    pub(crate) executor_command_channel: &'a CommandChannel<ExecutorCommands>,
     pub(crate) parent: &'a Option<ActorAddress>,
     pub(crate) children: &'a mut Vec<ActorAddress>,
     pub(crate) sender: &'a SenderType,
@@ -190,4 +204,11 @@ impl Context<'_> {
     pub fn parent(&self) -> Option<&ActorAddress> {
         self.parent.as_ref()
     }
+
+    /// Perform immediate shutdown for the current actor.
+    pub fn shutdown(&self) {
+        self.executor_command_channel
+            .send(ExecutorCommands::ShutdownActor(self.address.clone()))
+            .unwrap();
+    }
 }

src/executor/mod.rs

@@ -2,13 +2,14 @@
 
 pub(crate) mod thread_executor;
 
-use crate::actor::ActorCell;
+use crate::actor::{ActorAddress, ActorCell};
 use crate::config::ExecutorType;
 use crate::system::RuntimeManagerRef;
 use crate::util::CommandChannel;
 
 pub enum ExecutorCommands {
     AssignActor(ActorCell),
+    ShutdownActor(ActorAddress),
     Shutdown,
 }
 

src/executor/thread_executor.rs

@@ -78,12 +78,18 @@ impl Executor for ThreadExecutor {
                         cell.actor.before_start(Context {
                             address: &cell.address,
                             runtime_manager: &self.runtime_manager,
+                            executor_command_channel: &self.command_channel,
                             parent: &cell.parent,
                             children: &mut cell.children,
                             sender: &SenderType::System,
                         });
                         self.actor_cells.insert(cell.address.uri.clone(), cell);
                     }
+                    ExecutorCommands::ShutdownActor(address) => {
+                        let cell = self.actor_cells.get_mut(&address.uri).unwrap();
+                        cell.set_shutdown()
+                        // TODO: Call on_shutdown hook for actor
+                    }
                     ExecutorCommands::Shutdown => {
                         info!("received shutdown command");
                         break;
@@ -93,6 +99,10 @@ impl Executor for ThreadExecutor {
             // Iterate over the actor-cells and check if there are any non-empty mailboxes.
             // If one is found, process a message from it.
             for (_, cell) in self.actor_cells.iter_mut() {
+                if !cell.is_shutdown() {
+                    // TODO: Forward messages to dead-letter-queue
+                    continue;
+                }
                 if !cell.mailbox.is_empty() {
                     let result = cell.mailbox.try_recv();
                     if let Ok(letter) = result {
@@ -101,6 +111,7 @@ impl Executor for ThreadExecutor {
                             Context {
                                 address: &cell.address,
                                 runtime_manager: &self.runtime_manager,
+                                executor_command_channel: &self.command_channel,
                                 parent: &cell.parent,
                                 children: &mut cell.children,
                                 sender: &letter.sender,