👻

[Rust] actixを使ってアクターモデルを実装してみた

2022/07/08に公開

Rust

Akka

actor

tech

はじめに

actixを使ってアクターモデリングを実装している。
アクターモデルのライブラリといえばakkaが有名だ。
実際にakkaを使ってアクターモデル実装をしたことがあるのでその内容をRustを使って書き直してakkaとactixでどのような差分があるのかなどをみてみる。

アクターモデルとは

こちらを参照

actix以外の候補

redditのスレッドにRust製アクターモデル実装ライブラリをまとめたものがあった。

github starが1番多かったのがactixなのでactixを使ってみた。

https://github.com/actix/actix

実行環境

macOS Big Sur
cargo 1.61.0 (a028ae42f 2022-04-29)
Rust edition 2021

前提

webframeworkとして、axumを使用している

アクターの全体構成

特にこの構成にした意味はなく、実装したい内容を組み込むとこの構成になった。

supervisor actor ----> search actor  --> search engine
              ↑
              ↑
              ↑
timer actor --↑

supervisor actorはsearch actorを作成したり等の監視を行う
timer actorは、search actorが処理するものがなくなったらストップさせるために監視目的で定期的にsupervisor actorにメッセージを送っている。akkaではwatchWithみたいな監視用の便利なメソッドがあったが、actixにはなさそうだったので自前で監視する

実装例

以下githubリポジトリに全体のコードを置いている。

まずsupervisor　actorとtimer actorを起動

// プロセス内で共有するモジュールを格納する
pub struct Modules {
    // 各ユースケースインスタンス
    pub project_usecase: ProjectUsecase<RepositoryImpl>,
    pub actor_usecase: ActorUsecase,
}

impl Modules {
    pub async fn new(settings: &Settings) -> Modules {
        let mysql_client: Arc<MysqlClient> = Arc::new(MysqlClient::new(settings));
        let repository_modules: RepositoryImpl = RepositoryImpl::new(mysql_client);
        let project_usecase: ProjectUsecase<RepositoryImpl> =
            ProjectUsecase::new(repository_modules);
        // ↓こいつ
        let mut supervisor_actor = SuperVisorActor::new();
        let message = supervisor_actor.initializing();
        let supervisor_actor = Arc::new(Supervisor::start(|_| supervisor_actor));
        // ↓こいつ
        let mut timer_actor = TimerActor::new();
        let timer_actor = Supervisor::start(|_| timer_actor);
        timer_actor
            .send(Message::CheckSearchActor {
                supervisor_actor_address: Arc::clone(&supervisor_actor),
            })
            .await;
        supervisor_actor
            .send(message)
            .await
            .expect("mailbox error")
            .expect("can not initialize supervisor actor");

        let actor_usecase: ActorUsecase = ActorUsecase::new(supervisor_actor);
        Self {
            project_usecase,
            actor_usecase,
        }
    }
}

supervisor actorの構成

FSMにしたいと思ったので、自分自身の状態を保持
子アクターであるsearch actorを管理するために作成したsearch actorを保持

pub struct SuperVisorActor {
    // 状態を持っておけば、finite state machineにできる
    state: State,
    child_actors: Arc<Mutex<HashMap<u64, Addr<SearchActor>>>>, // NOTE: Arcで包んだ参照を可変参照にするには、Mutexで包んであげる必要がある。lockして他のスレッドから参照されないようにする必要がある
}

timer actorの構成

定期的にsearch actorに生存確認メッセージを投げるために、intervalを保持

pub struct TimerActor {
    interval: Duration,
}

supervisor actorにメッセージを投げる

sendメソッドで先ほど作成したsupervisor actorにメッセージを投げている
非同期なので、async fnにする必要がある

pub struct ActorUsecase {
    supervisor_actor: Arc<Addr<SuperVisorActor>>,
}

impl ActorUsecase {
    // TODO: Resultが二重になっているのでflatにする
    pub async fn execute_actor(&self, project_id: u64) -> Result<Result<(), Error>, Error> {
        let res: Result<Result<(), Error>, Error> = self
            .supervisor_actor
            .send(Message::StartSearch { project_id })
            .await
            .map_err(|e| Error::SupervisorActorMailBoxError(e.to_string()));
        debug!("{:?}", res);
        res
    }

    pub fn new(supervisor_actor: Arc<Addr<SuperVisorActor>>) -> Self {
        Self { supervisor_actor }
    }
}

supervisor actorでメッセージを受け取る

actixでは、受け取ったメッセージ毎にレスポンス型を厳格に定義しないといけないので、別のメッセージを受け取りたい場合には別のHandlerを定義する必要がある。
FSMにしたいもがきが↓コードから見て取れる
cloneはしたくなかったので、Arcを駆使している

impl Handler<Message> for SuperVisorActor {
    // acto・コンテキスト内にアクセスしたい場合には、ResponseFutureではなくResponseActFutureを戻り方として定義する
    type Result = ResponseActFuture<Self, Result<(), Error>>;

    // SuperVisorActorが受け取ったメッセージ毎にこのhandleメソッドが呼ばれる
    fn handle(&mut self, msg: Message, ctx: &mut Self::Context) -> Self::Result {
        let ctx_address = Arc::new(ctx.address());
        let child_actors = Arc::clone(&self.child_actors);
        // actorの状態によって処理を変える
        match self.state {
            State::Idle => {
                self.state = State::Active;
                Box::pin(
                    async move { Ok(()) }
                        .into_actor(self)
                        .map(|res, _act, _ctx| {
                            // 自分自身に同じメッセージをもう一回投げている
                            // ResponseActFutureを使うことで、↓のようにcontextを利用できる
                            _ctx.notify(msg);
                            res
                        }),
                )
            }
            State::Active => {
                // https://users.rust-lang.org/t/actix-await-for-send-in-handle/47844
                // https://github.com/actix/actix/issues/438
                Box::pin(
                    async move {
                        // ここで何かしらの非同期処理を行う
                        SuperVisorActor::execute_message(msg, ctx_address, child_actors).await
                    }
                    .into_actor(self)
                    .map(|res, act, _ctx| res),
                )
            }
        }
    }
}

↑で呼んでいるexecute_messageメソッドがこれ

保持しているハッシュマップに対象search actorが存在しない場合には、search actorを新規で作成・起動している
ハッシュマップに存在していればハッシュマップからsearch actorのaddressを取得しメッセージを投げる

    async fn execute_message(
        msg: Message,
        reply_to: Arc<Addr<SuperVisorActor>>,
        child_actors: Arc<Mutex<HashMap<u64, Addr<SearchActor>>>>,
    ) -> Result<(), Error> {
        match msg {
            Message::StartSearch { project_id } => {
                // 子アクターをすでに保持していなかったら作成するロジック
                //   作成したらHashMapに格納する　＝> ここではできない（Arc<HashMap>になっているので）
                //   なのでこの関数の戻り値を作成したsearchActorのアドレスとidにする
                let mut locked_map = child_actors.lock().unwrap();
                match locked_map.get(&project_id) {
                    Some(search_actor) => {
                        search_actor
                            .send(search_actor::Message::Execute { project_id })
                            .await;
                        Ok(())
                    }
                    None => {
                        let mut search_actor = SearchActor::new(project_id, reply_to);
                        let message = search_actor.initializing();
                        // NOTE: cpu boundな計算をするときは、そのactorをSyncArbiter::startで別スレッドで実行するべき！
                        // 参考：https://actix.rs/book/actix/sec-5-arbiter.html#system-and-arbiter
                        // let search_actor = SyncArbiter::start(1, search_actor);
                        let search_actor = Supervisor::start(|_| search_actor);
                        let res: Result<(), Error> = search_actor
                            .send(message)
                            .await
                            .map_err(|e| Error::SearchActorMailBoxError(e.to_string()))?;
                        search_actor
                            .send(search_actor::Message::Execute { project_id })
                            .await;
                        locked_map.insert(project_id, search_actor);
                        res
                    }
                }
            }
            Message::CompletedSearch => {
                debug!("complete!");
                Ok(())
            }
        }
    }

timer actorでの監視

run_intervalメソッドで、定期的に引数に渡したクロージャを実行する
- supervisor actorにCheckSearchActorメッセージを送っている

impl Handler<Message> for TimerActor {
    type Result = Result<(), Error>;

    fn handle(&mut self, msg: Message, ctx: &mut Self::Context) -> Self::Result {
        match msg {
            Message::CheckSearchActor {
                supervisor_actor_address,
            } => {
                // run_intervalでasync functionを実行したいならarbiterスレッドを作成して、そいつにasync taskを投げるようにする
                ctx.run_interval(self.interval, move |timer_actor, context| {
                    // 参考：https://www.reddit.com/r/rust/comments/srfho0/help_with_actix_arbiter/
                    let supervisor_actor_address_clone = Arc::clone(&supervisor_actor_address);
                    context.spawn(
                        async move {
                            supervisor_actor_address_clone
                                .send(supervisor_actor::Message::CheckSearchActor)
                                .await;
                        }
                        .into_actor(timer_actor),
                    ); // futureをactorにしている
                });
            }
        }
        Ok(())
    }
}

CheckSearchActorメッセージを受け取ったsupervisor actorは？

自分が保持している子アクター（search actors）全てに生存確認メッセージを送る処理をしている。
ハッシュマップに存在するものをfor loopで処理しても良いが、スレッドを掴んだままにしたくないので再起的に自分自身にメッセージを投げている（LoopExeute）
- （別のメッセージがsupervisor actorに送られた時にfor loopが終わるまで処理されなくなりスループットが落ちる）

    async fn execute_message(
        msg: Message,
        reply_to: Arc<Addr<SuperVisorActor>>,
        child_actors: Arc<Mutex<HashMap<u64, Addr<SearchActor>>>>,
    ) -> Result<(), Error> {
        match msg {
            Message::CheckSearchActor => {
                // 保持しているsearchActorに生存確認messageを投げる
                // スレッドを占有しないようにfor文で投げるんじゃなくて、再帰で投げる
                // 自分自身にmessageを投げる
                let res: Result<(), Error> = reply_to
                    .send(Message::LoopExecute { child_actors })
                    .await
                    .map_err(|e| Error::SupervisorActorMailBoxError(e.to_string()))?;
                res
            }
            Message::LoopExecute { child_actors } => {
                let child_actors_clone = Arc::clone(&child_actors);
                let mut locked_map = child_actors_clone.lock().unwrap();
                if locked_map.len() != 0usize {
                    debug!("manage child actors: {:?}", locked_map);
                    let key = locked_map.keys().next().copied().unwrap_or(0); // copied()・・・Option<&A> -> Option<A>
                    let search_actor_address = locked_map.get(&key);
                    if let Some(search_actor) = search_actor_address {
                        let res: SearchActorStatus = search_actor
                            .send(search_actor::CheckStatusMessage::Check)
                            .await
                            .map_err(|e| Error::SupervisorActorMailBoxError(e.to_string()))??;
                        match res {
                            SearchActorStatus::Idle => locked_map.remove(&key),
                            _ => None,
                        };
                    }
                    drop(locked_map);
                    // NOTE: ↓でさらにLoopExecute messageを投げることで、locked_mapがdead lockを起こしていたので、スレッドが止まってしまっていた。
                    // 明示的にlockを解消するためにdropさせている。
                    reply_to.send(Message::LoopExecute { child_actors }).await;
                } else {
                    debug!("child actor count is zero");
                }
                Ok(())
            }
        }
    }

ざっとこれくらい

感想

FSM(Finite State Machine)での実装がよくわからず

akkaでは、受け取れないメッセージを格納するbuffer機構があるのだが、actixではないので自分で実装するしかなさそう、、、
また受け取るメッセージ毎にエラー型を定義しHandlerを実装しないといけないのでめんどくさい、、、

同じアクター内での非同期処理がアクターモデルの良さを潰している？？

supervisor actorでメッセージを受け取った後にsearch actorにメッセージを投げる際には非同期になるのだが、supervisor actor自身が保持するオブジェクトをArc<Mutex<>>にする必要があり、アクターモデルの良さである「マルチスレッド環境でよくある競合を気にしなくても良い」メリットがないのでは？？
lock取得・解除等を結局実装しないといけないので、これは自分の実装が悪いからか？

参考

actixの公式githubリポジトリでもいくつかサンプルコードがある
https://github.com/actix/actix/tree/master/actix