複数の関数で状態を共有する方法まとめ

プログラミングをしていると、「複数の関数で状態を共有」したくなることがよくある。

このとき、何も考えずにグローバル変数を使ったり、ポインタをバケツリレーしたりするよりも、いろんな解法からベストを選べた方がよい。

そこで、複数の関数で状態を共有する方法を、メリット＆デメリットとともに、思いつく限りまとめてみる。

サンプルコードとしてはPythonを使い、Pythonで書けないときにRacket（Schemeの一種）を使う。好きだから。

グローバル変数

state = "initial state"

def fuga0():
    # 状態を参照したのち変化させる
    global state
    print(f"fuga0: {state}")
    state = "another state"
    return "result of fuga0"

def fuga1():
    # 状態を参照する
    print(f"fuga1: {state}")

def hoge():
    print(fuga0())
    fuga1()
    print(f"last: {state}")

hoge()

結果（以下は省略）

fuga0: initial state
result of fuga0
fuga1: another state
last: another state

あまり推奨される書き方ではない。

• (-)変数がプログラム全域で参照される可能性があるので、どの処理が変数に依存するのかわかりにくくなる
• (-)変数がプログラム全域で変更される可能性があるので、状態の変化過程が追跡しきれず、バグの温床になる
  • ただし、定数の場合はこの問題はない

ポインタをバケツリレー

from dataclasses import dataclass
from typing import Any

# こいつがバケツリレーされるポインタ
@dataclass
class Box:
    value: Any

def fuga0(box):
    print(f"fuga0: {box.value}")
    box.value = "another state"
    return "result of fuga0"

def fuga1(box):
    print(f"fuga1: {box.value}")

def hoge(box):
    print(fuga0(box))
    fuga1(box)
    print(f"last: {box.value}")

b = Box("initial state")

hoge(b)

• (+)グローバル変数に比べ、引数にポインタをとった関数の中に、変化・参照の範囲を局限できる
• (-)引数でバケツリレーする煩わしさがある
  • オブジェクト間で状態を共有する場合、DIコンテナで煩わしさを軽減できる。

定数の場合はポインタではなく値をバケツリレーすればよい。

現在の状態を引数で受け、新しい状態を返り値で返す

いわゆる関数型な書き方で、コード上では状態は変化していない。

def fuga0(state):
    print(f"fuga0: {state}")
    return "result of fuga0", "another state"

def fuga1(state):
    print(f"fuga1: {state}")

def hoge(state):
    result, new_state = fuga0(state)
    print(result)
    fuga1(new_state)
    print(f"last: {new_state}")

hoge("initial")

• (+)関数的であり、コード上の状態変化はなしでいける
  • 状態が変化しているかどうかを、変数が同じかどうかで把握できる
  • テストがしやすい
• (-)バケツリレーが煩わしい
• (-)状態を変化させつつ結果も返す場合、複数の値を受け取るのが煩わしい

オブジェクト指向

オブジェクト指向言語ではよく使われるやり方。共有したい状態をメンバ変数にする。

class Hoge:
    def __init__(self):
        self._state = "initial state"

    def _fuga0(self):
        print(f"fuga0: {self._state}")
        self._state = "another state"
        return "result of fuga0"

    def _fuga1(self):
        print(f"fuga0: {self._state}")

    def __call__(self):
        print(self._fuga0())
        self._fuga1()
        print(f"last: {self._state}")

hoge = Hoge()
hoge()

• (+)バケツリレーはなしでいける
• (+)状態変化の影響範囲をクラス内のメソッドに局限可能
  • 状態変化を「隠蔽」できる
• (-)クラス定義構文が煩わしい（言語によって程度が違うが）
• (-)状態と関数（メソッド）が一つのクラス定義として強く結合している
  • 状態を共有する関数を追加・変更したい場合、わざわざクラス定義を書き換えるor継承する必要がある

クロージャ

オブジェクト指向に近い。

def create_hoge():
    state = "initial"

    def fuga0():
        nonlocal state
        print(f"fuga0: {state}")
        state = "another state"
        return "result of fuga0"

    def fuga1():
        print(f"fuga1: {state}")

    def hoge():
        print(fuga0())
        fuga1()
        print(f"last: {state}")

    return hoge

create_hoge()()

• (+)クラス定義よりも構文的煩雑さは小さい（言語によって程度が違うが）
  • SchemeやRacketならnonlocalもいらず、さらっと書ける。

Glocal Variable

ぼくがかんがえたさいきょうのパターン。withの中だけ使える変数を作る。詳しい内容はリンク先を参照。

param.py

_param = None
_initialized = False

@contextmanager
def parametrize(data):
    global _param
    global _initialized
    before = _param
    before_initialized = _initialized
    _param = data
    _initialized = True
    try:
        yield
    finally:
        _param = before
        _initialized = before_initialized

def set_param(data):
    if _initialized:
        global _param
        _param = data
    else:
        raise RuntimeError

def get_param():
    if _initialized:
        return _param
    else:
        raise RuntimeError

from param import get_param, set_param, parametrize

def fuga0():
    print(f"fuga0: {get_param()}")
    set_param("another state")
    return "result of fuga0"

def fuga1():
    print(f"fuga0: {get_param()}")

def hoge():
    print(fuga0())
    fuga1()
    print(f"last: {get_param()}")

with parametrize("initial state"):
    hoge()

• (+)バケツリレーは避けることができる
• (+)withの中でしか変数に触れないので、グローバル変数に比べると自由度に制限がある
• (+)withの前後で必ず値が初期化・巻き戻しされるので、わけのわからない値が残ってバグを起こす心配がない
• (-)withの外でgetter/setterを呼び出しても、静的解析ではエラーを拾えない

状態モナド

モナドって何？状態モナドって？みたいなのは適当にググってください。

Racketで書いてみる。applicativeの実装はちょっと自信なし……

; 状態モナドの定義ここから
(require (prefix-in base: racket/base))
(require data/functor)
(require data/applicative)
(require data/monad)

(struct result (value state) #:transparent)

(struct state (f)
  #:methods gen:functor
  [(define (map g x)
     (state (λ (s)
              (match-define (result v ss) (run x s))
              (result (g v) ss))))]
  #:methods gen:applicative
  [(define (pure _ x)
     (state (λ (s) (result x s))))
   (define (apply f xs)
     (define (get-args xs s)
       (match xs
         [(cons x rest)
          (match-define (result xv xs) (run x s))
          (match-define (result args argss) (get-args rest xs))
          (result (cons xv args) argss)]
         [_ (result `() s)]))
     (state (λ (s)
              (match-define (result fv fs) (run f s))
              (match-define (result args argss) (get-args xs fs))
              (result (base:apply fv args) argss))))]
  #:methods gen:monad
  [(define (chain f x)
     (state (λ (s)
              (match-define (result xv xs) (run x s))
              (match-define (result fv fs) (run (f xv) xs))
              (result fv fs))))])

(define (run m s) ((state-f m) s))

(define get
  (state (λ (s) (result s s))))

(define (set ns)
  (state (λ (s) (result s ns))))
; 定義ここまで

(define fuga0
  (do [x <- get]
      (pure (printf "fuga0: ~a\n" x))
      (set "another state")
      (pure "result of fuga0")))

(define fuga1
  (do [x <- get]
      (pure (printf "fuga1: ~a\n" x))))

(define hoge
  (do [x <- fuga0]
      (pure (displayln x))
      fuga1
      [y <- get]
      (pure (printf "last: ~a\n" y))))

(run hoge "initial state")

• (+)バケツリレーは避けることができる
• (-)関数的であり、コード上の状態変化は避けることができる
• (-)do記法ないし高階関数を使って書かないといけないのが面倒っちゃ面倒

定数ならReaderモナドを使えばよい。

Pythonではなかなか苦行だが頑張るとこんな感じ。

# 状態モナド定義ここから
from typing import Callable, TypeVar, Generic, Tuple


S = TypeVar("S") # 状態の型
R = TypeVar("R") # 返り値の型
A = TypeVar("A") # 新しい返り値の型

class State(Generic[S, R]):
    def __init__(self, f: Callable[[S], Tuple[S, R]]):
        self._f = f

    def run(self, state: S) -> Tuple[S, R]:
        return self._f(state)

    @staticmethod
    def of(value: R):
        return State(lambda s: (value, s))

    def flatmap(self, g: Callable[[R], State[S, A]]) -> State[S, A]:
        def _new(state):
            f_ret, f_state = self.run(state)
            return g(f_ret).run(f_state)
        return State(_new)

    def map(self, g: Callable[[R], A]) -> State[S, A]:
        return self.flatmap(lambda x: State.of(g(x)))

    def then(self, m: State[S, A]) -> State[S, A]:
        return self.flatmap(lambda _: m)

    def value(self, v: A) -> State[S, A]:
        return self.map(lambda _: v)

get_m = State(lambda s: (s, s))

def set_m(new_state):
    return State(lambda s: (s, new_state))
# 状態モナド定義ここまで

fuga0 = (get_m
    .map(lambda v: print(f"fuga0: {v}"))
    .then(set_m("another_state"))
    .value("result of fuga0"))

fuga1 = (get_m
    .map(lambda v: print(f"fuga1: {v}")))

hoge = (fuga0
    .map(print)
    .then(fuga1)
    .then(get_m)
    .map(lambda v: print(f"last: {v}")))

hoge.run("initial state")

限定継続

限定継続って何？それでなぜ状態変化が扱えるの？って人は浅井先生のチュートリアルを参照。

(require racket/control)

(define (get)
  (shift k
         (λ (x)
           ((k x) x))))

(define (set s)
  (shift k
         (λ (x)
           ((k x) s))))

(define (run m s)
  ((reset
    (let ([ret (m)])
      (λ (_) ret))) s))


(define (fuga0)
  (printf "fuga0: ~a\n" (get))
  (set "another state")
  "result of fuga0")

(define (fuga1)
  (printf "fuga1: ~a\n" (get)))

(define (hoge)
  (displayln (fuga0))
  (fuga1)
  (printf "last: ~a\n" (get)))

(run hoge "initial state")

• (+)バケツリレーは避けることができる
• (+)do記法に従って書かなくても普通にそのまま書ける
• (-)継続自体の問題として、慣れないと処理を追ったりデバッグしたりするのが大変
• いちおうコード上の状態変化はなしでいける
  • shift&resetに副作用があるので関数的とは言い切れない

まとめ

みんな違ってみんないい。

プログラマたるもの、知ってるやり方に固執するのではなく、道具を増やしてうまく使い分けましょう。

あと、Pythonの標準ライブラリにcontextvarsってのもある。非同期を念頭に置かれて作られたものっぽいので今回は言及しなかった。