PythonでO(1)ランダムアクセス可能なdequeの新規提案

はじめましての人ははじめまして。[alumi](https://x.com/alumi_02)と申します。
まず本編の前に先駆者であるごりちゃんさんへ感謝とリスペクトを。

Pythonの標準ライブラリのdequeは、内部実装の都合で任意要素へindexを指定して参照するランダムアクセスに$O(N)$かかります。
C++のdequeは$O(1)$でランダムアクセス可能なため、これを前提とした問題が出されるとPython勢は慌てふためくことになります。
しかし偉大な先駆者であるごりちゃんさんが[リングバッファで実装したdeque](https://prd-xxx.hateblo.jp/entry/2020/02/07/114818)を公開してくださっているので、Python勢がこのケースで慌てることはなくなりました。良かった。

この記事ではごりちゃんさんの実装よりもappend, pop操作が若干早い実装ができたので、実装とその説明をしたいと思います[^1]。

[^1]: 実用上はごりちゃんさんの実装で困ることはないので、そんな実装もあるんだな～という気持ちで読んでください。類似の実装に関する実装や計算量解析については[Open Data Structures](https://jinnaiyuu.github.io/pdf/ods-cpp.pdf#page=60)などの資料をご参考ください

## 実装

### 最小構成

```python
class TwoStackDeque:
    __slots__ = ("front", "back")

    def __init__(self, iterable=None) -> None:
        init_arr = list(iterable) if iterable else []
        mid = len(init_arr) >> 1
        self.front = init_arr[:mid][::-1]
        self.back = init_arr[mid:]

    def _balance(self) -> None:
        source, target = (
            (self.front, self.back) if not self.back else (self.back, self.front)
        )
        mid = len(source) >> 1
        target.extend(source[: mid + 1][::-1])
        del source[: mid + 1]

    def append(self, item) -> None:
        self.back.append(item)

    def appendleft(self, item) -> None:
        self.front.append(item)

    def pop(self):
        if not self:
            raise IndexError("pop from empty deque")
        if not self.back:
            self._balance()
        return self.back.pop()

    def popleft(self):
        if not self:
            raise IndexError("popleft from empty deque")
        if not self.front:
            self._balance()
        return self.front.pop()

    def __getitem__(self, i: int):
        l = len(self)
        if i < -l or l <= i:
            raise IndexError("deque index out of range")
        i = i if i >= 0 else i + l
        if i < len(self.front):
            return self.front[~i]
        else:
            return self.back[i - len(self.front)]

    def __len__(self) -> int:
        return len(self.front) + len(self.back)
```

<details><summary>全部のせ</summary>

```python
class TwoStackDeque:
    __slots__ = ("front", "back")

    def __init__(self, iterable=None) -> None:
        init_arr = list(iterable) if iterable else []
        mid = len(init_arr) >> 1
        self.front = init_arr[:mid][::-1]
        self.back = init_arr[mid:]

    def __len__(self) -> int:
        return len(self.front) + len(self.back)

    def __str__(self) -> str:
        return f"{self.__class__.__name__}({list(self)})"

    __repr__ = __str__

    def __iter__(self):
        yield from reversed(self.front)
        yield from self.back

    def __reversed__(self):
        yield from reversed(self.back)
        yield from self.front

    def __contains__(self, item) -> bool:
        return item in self.front or item in self.back

    def _normalize_index(self, i: int) -> int:
        length = len(self)
        if i < -length or length <= i:
            raise IndexError("deque index out of range")
        return i if i >= 0 else i + length

    def __getitem__(self, i: int):
        i = self._normalize_index(i)
        if i < len(self.front):
            return self.front[~i]
        else:
            return self.back[i - len(self.front)]

    def __setitem__(self, i: int, item) -> None:
        i = self._normalize_index(i)
        if i < len(self.front):
            self.front[~i] = item
        else:
            self.back[i - len(self.front)] = item

    def __delitem__(self, i: int) -> None:
        i = self._normalize_index(i)
        if i < len(self.front):
            del self.front[~i]
        else:
            del self.back[i - len(self.front)]

    def _balance(self) -> None:
        source, target = (
            (self.front, self.back) if not self.back else (self.back, self.front)
        )
        mid = len(source) >> 1
        target.extend(source[: mid + 1][::-1])
        del source[: mid + 1]

    def append(self, item) -> None:
        self.back.append(item)

    def appendleft(self, item) -> None:
        self.front.append(item)

    def extend(self, iterable) -> None:
        self.back.extend(iterable)

    def extendleft(self, iterable) -> None:
        self.front.extend(iterable)

    def pop(self):
        if not self:
            raise IndexError("pop from empty deque")
        if not self.back:
            self._balance()
        return self.back.pop()

    def popleft(self):
        if not self:
            raise IndexError("popleft from empty deque")
        if not self.front:
            self._balance()
        return self.front.pop()

    def rotate(self, n: int = 1) -> None:
        if not self:
            return
        length = len(self)
        n %= length
        if n == 0:
            return

        if n <= length // 2:
            for _ in range(n):
                self.appendleft(self.pop())
        else:
            for _ in range(length - n):
                self.append(self.popleft())

    def count(self, item) -> int:
        return sum(1 for x in self if x == item)

    def clear(self) -> None:
        self.front.clear()
        self.back.clear()

    def remove(self, item) -> None:
        for i, x in enumerate(self):
            if x == item:
                del self[i]
                return
        raise ValueError(f"{item} is not in deque")

    def insert(self, i: int, item) -> None:
        i = i if i >= 0 else i + len(self)
        if i <= len(self.front):
            if i == 0:
                self.front.append(item)
            else:
                self.front.insert(~i + 1, item)
        else:
            self.back.insert(i - len(self.front), item)

    def index(self, item, start: int = 0, stop=None) -> int:
        if stop is None:
            stop = len(self)
        start = start if start >= 0 else start + len(self)
        stop = stop if stop >= 0 else stop + len(self)
        for i in range(max(start, 0), min(stop, len(self))):
            if self[i] == item:
                return i
        raise ValueError(f"{item} is not in deque")

    def deepcopy(self) -> "TwoStackDeque":
        new_ins = TwoStackDeque()
        new_ins.front = self.front[:]
        new_ins.back = self.back[:]
        return new_ins

```

</details>

2つのStackを使っているので、説明上「2スタックdeque」と呼びます。
やっていることは非常に単純で、

- appendleft, popleftを受け付ける可変長配列`front`
- append, popを受け付ける可変長配列`back`

を持って、`appendleft`・`popleft`を`front`への`append`・`pop`で代替し、ランダムアクセス時はindexの大きさによって`front`, `back`のどっちを参照するかを判断しています。
また`popleft`時に`front`が空、もしくは`pop`時に`back`が空のときは、`_balance`を走らせてもう一方の配列から半分要素をもらいます[^2]。`front`側は要素が逆向きに入っているので、振り分け時に反転しています（下図参照）。要素が空の方を`target`、要素を与える方を`source`とすることで、どちらの場合でも同じように振り分けができます。

[^2]: このような操作を行う先行例として、[永続データ構造 - Wikipedia](https://ja.wikipedia.org/wiki/%E6%B0%B8%E7%B6%9A%E3%83%87%E3%83%BC%E3%82%BF%E6%A7%8B%E9%80%A0)内に「銀行家のキュー（banker's queue）」が記載されています。しかしこれが単なるqueueの実装なのか、永続queueの実装の呼び名なのかは不明瞭です（「銀行家」は償却分析のbanker法から採っている？：[9.3.3. Bankers and Physicists · Functional Programming in OCaml](https://courses.cs.cornell.edu/cs3110/2021sp/textbook/eff/amortized_methods.html)）。

![balance01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3665424/825bef24-240e-6548-8c33-113f6acce16b.png)

![balance02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3665424/502b1177-2f4b-59dd-1c7f-95c8e709bf5d.png)

一応`insert`、`count`なども実装してありますが$O(N)$なので注意です（`count`はdeque内の要素とその要素数を辞書で持って、`append`時に+1、`pop`時に-1すればメモリとトレードオフで$O(1)$実装できます。使い道なさそうだからやらんけど）。

## 速度テスト

標準ライブラリのdeque・今回の2スタックdeque・ごりちゃんさんdeque（コードでは`RingBufferDeque`）の`append`, `appendleft`, `pop`, `popleft`, random accessの速度比較を行います。

<details><summary>テストコード</summary>

```python
class TwoStackDeque:
    __slots__ = ("front", "back")

    def __init__(self, iterable=None) -> None:
        init_arr = list(iterable) if iterable else []
        mid = len(init_arr) >> 1
        self.front = init_arr[:mid][::-1]
        self.back = init_arr[mid:]

    def _balance(self) -> None:
        source, target = (
            (self.front, self.back) if not self.back else (self.back, self.front)
        )
        mid = len(source) >> 1
        target.extend(source[: mid + 1][::-1])
        del source[: mid + 1]

    def append(self, item) -> None:
        self.back.append(item)

    def appendleft(self, item) -> None:
        self.front.append(item)

    def pop(self):
        if not self:
            raise IndexError("pop from empty deque")
        if not self.back:
            self._balance()
        return self.back.pop()

    def popleft(self):
        if not self:
            raise IndexError("popleft from empty deque")
        if not self.front:
            self._balance()
        return self.front.pop()

    def __getitem__(self, i: int):
        l = len(self)
        if i < -l or l <= i:
            raise IndexError("deque index out of range")
        i = i if i >= 0 else i + l
        if i < len(self.front):
            return self.front[~i]
        else:
            return self.back[i - len(self.front)]

    def __len__(self) -> int:
        return len(self.front) + len(self.back)


# ごりちゃんさんのDeque
class RingBufferDeque:
    def __init__(self, src_arr=[], max_size=300000):
        self.N = max(max_size, len(src_arr)) + 1
        self.buf = list(src_arr) + [None] * (self.N - len(src_arr))
        self.head = 0
        self.tail = len(src_arr)

    def __index(self, i):
        l = len(self)
        if not -l <= i < l:
            raise IndexError("index out of range: " + str(i))
        if i < 0:
            i += l
        return (self.head + i) % self.N

    def __extend(self):
        ex = self.N - 1
        self.buf[self.tail + 1 : self.tail + 1] = [None] * ex
        self.N = len(self.buf)
        if self.head > 0:
            self.head += ex

    def is_full(self):
        return len(self) >= self.N - 1

    def is_empty(self):
        return len(self) == 0

    def append(self, x):
        if self.is_full():
            self.__extend()
        self.buf[self.tail] = x
        self.tail += 1
        self.tail %= self.N

    def appendleft(self, x):
        if self.is_full():
            self.__extend()
        self.buf[(self.head - 1) % self.N] = x
        self.head -= 1
        self.head %= self.N

    def pop(self):
        if self.is_empty():
            raise IndexError("pop() when buffer is empty")
        ret = self.buf[(self.tail - 1) % self.N]
        self.tail -= 1
        self.tail %= self.N
        return ret

    def popleft(self):
        if self.is_empty():
            raise IndexError("popleft() when buffer is empty")
        ret = self.buf[self.head]
        self.head += 1
        self.head %= self.N
        return ret

    def __len__(self):
        return (self.tail - self.head) % self.N

    def __getitem__(self, key):
        return self.buf[self.__index(key)]

    def __setitem__(self, key, value):
        self.buf[self.__index(key)] = value

    def __str__(self):
        return "Deque({0})".format(str(list(self)))


import time
from collections import deque
import random
import matplotlib.pyplot as plt
import numpy as np
import statistics


# 特定操作の時間計測
def test_operation(deque_obj, operation, n):
    start_time = time.perf_counter()
    for _ in range(n):
        if operation == "append":
            deque_obj.append(random.randint(-100, 100))
        elif operation == "appendleft":
            deque_obj.appendleft(random.randint(-100, 100))
        elif operation == "pop":
            if len(deque_obj) > 0:
                deque_obj.pop()
        elif operation == "popleft":
            if len(deque_obj) > 0:
                deque_obj.popleft()
        elif operation == "random_access":
            if len(deque_obj) > 0:
                index = random.randrange(len(deque_obj))
                _ = deque_obj[index]
    return time.perf_counter() - start_time


def run_tests(deque_class, size, operations, repetitions, num_trials=5):
    results = {}
    for op in operations:
        op_times = []
        for _ in range(num_trials):
            deque_obj = deque_class(list(range(size)))
            op_times.append(test_operation(deque_obj, op, repetitions))
        results[op] = {
            "mean": statistics.mean(op_times),
            "stdev": statistics.stdev(op_times) if len(op_times) > 1 else 0,
        }
    return results


# テスト結果の描画関数
def plot_results(sizes, std_results, rb_results, ts_results, operations):
    fig, axs = plt.subplots(
        len(operations), 1, figsize=(10, 4 * len(operations)), sharex=True
    )
    fig.suptitle("Deque Performance Comparison", y=1.02)

    x = np.arange(len(sizes))
    width = 0.25

    for i, op in enumerate(operations):
        std_times = [std_results[size][op]["mean"] for size in sizes]
        rb_times = [rb_results[size][op]["mean"] for size in sizes]
        ts_times = [ts_results[size][op]["mean"] for size in sizes]

        std_err = [std_results[size][op]["stdev"] for size in sizes]
        rb_err = [rb_results[size][op]["stdev"] for size in sizes]
        ts_err = [ts_results[size][op]["stdev"] for size in sizes]

        axs[i].bar(
            x - width,
            std_times,
            width,
            yerr=std_err,
            label="Std Deque",
            color="blue",
            capsize=5,
        )
        axs[i].bar(
            x,
            rb_times,
            width,
            yerr=rb_err,
            label="Ring buffer Deque",
            color="green",
            capsize=5,
        )
        axs[i].bar(
            x + width,
            ts_times,
            width,
            yerr=ts_err,
            label="Two Stack Deque",
            color="red",
            capsize=5,
        )

        axs[i].set_ylabel("Time (s)")
        axs[i].set_title(f"{op} Operation")
        axs[i].set_xticks(x)
        axs[i].set_xticklabels(sizes)
        axs[i].legend(loc="upper left")

        # random_accessの場合のみ対数スケールを使用
        if op == "random_access":
            axs[i].set_yscale("log")
        else:
            axs[i].set_yscale("linear")

    plt.xlabel("Number of Elements")
    plt.tight_layout()
    plt.subplots_adjust(top=0.95)
    plt.show()


# テスト実行
sizes = [1000, 10000, 100000, 1000000]
operations = ["append", "appendleft", "pop", "popleft", "random_access"]
repetitions = 10000
num_trials = 100  # 試行回数

std_results = {}
rb_results = {}
ts_results = {}

for size in sizes:
    print(f"Testing with {size} elements:")
    std_results[size] = run_tests(deque, size, operations, repetitions, num_trials)
    rb_results[size] = run_tests(
        RingBufferDeque, size, operations, repetitions, num_trials
    )
    ts_results[size] = run_tests(
        TwoStackDeque, size, operations, repetitions, num_trials
    )
    # print("Standard deque results:", std_results[size])
    # print("Ring Buffer deque results:", rb_results[size])
    # print("Two Stack deque results:", ts_results[size])
    # print()

# 結果を描画
plot_results(sizes, std_results, rb_results, ts_results, operations)

```
</details>

### テスト結果

![Figure_1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3665424/f59cec86-ce5e-01aa-f723-a4d302d074c4.png)

`append`, `appendleft`, `pop`, `popleft`においては標準ライブラリdequeが圧勝ですね。ランダムアクセスを犠牲にしてdequeインターフェースで要求される基本操作だけに特化したような感じなんでしょうか（参考：[python - Why is deque implemented as a linked list instead of a circular array? - Stack Overflow](https://stackoverflow.com/questions/45134139/why-is-deque-implemented-as-a-linked-list-instead-of-a-circular-array)）。
今回の2スタックdequeはごりちゃんさんdequeに比べて`append`, `appendleft`, `pop`, `popleft`で多少早くなっています。ただ `pop`, `popleft`は`_balance`（ごりちゃんさんdequeなら`__extend`）が生じやすい条件で結構変わってくると思います。
ランダムアクセスについては2スタックdequeとごりちゃんさんdequeとの差はほぼなさそうですね。まあindex計算と配列参照という同じことをしているのでそれはそう。

## 余談

今回の実装は[SWAG](https://judge.yosupo.jp/submission/230145)（Sliding Window Aggregation）の実装を見直しているときに思いついたものです（後でちゃんと調べたら車輪の再発明だったけど）。見比べていただければわかりますがSWAGの実装からfoldに必要な構成を取り除いただけです。`tolist`や`reverse`なども直感的に実装できるし汎用性も高そうで、なによりわかりやすくてソラでかけそう（個人の感想）と思ったのでちゃんと実装してみました。
ミスやバグなどあったら教えてくださると嬉しいです❗一応次の問題は通せてます。

- [ABC335C](https://atcoder.jp/contests/abc335/submissions/56992471)
- [典型90問044](https://atcoder.jp/contests/typical90/submissions/56992697)
- [典型90問061](https://atcoder.jp/contests/typical90/submissions/56993655)
- [ITP2_1_B | Aizu Online Judge](https://onlinejudge.u-aizu.ac.jp/solutions/problem/ITP2_1_B/review/9579833/Alumite14/PyPy3)
- [両端への挿入 | アルゴ式](https://algo-method.com/submissions/1479453)

今回は要素数が偏ることのデメリットよりも実装の簡潔さを優先したので、片方が空になったときだけ`_balance`を呼んでいますが、片方が所定の要素数（例えばもう一方の1/3以下など）になったら`_balance`の処理を行うとより要素数の平衡を保てメモリ効率がよくなると思います（毎回`_balance`呼ぶので速度は悪化しますが）。

こちらが把握できていないpros・consやメソッドの追加、他実装の提案などもお待ちしています。
皆さんの競プロライフの一助となれば幸いです。読んでくださりありがとうございました（いいね押してくれるとより嬉しいです🙏）

## 参考

- [Pythonで各要素にO(1)でランダムアクセスできるdeque(両端キュー)を書いてみた - ごりちゃんがゆく](https://prd-xxx.hateblo.jp/entry/2020/02/07/114818)
- [Python3のdequeの実装を調べる: #時間計算量 - Qiita](https://qiita.com/recuraki/items/352d99cb6322576fbfef)
- [python - Why is deque implemented as a linked list instead of a circular array? - Stack Overflow](https://stackoverflow.com/questions/45134139/why-is-deque-implemented-as-a-linked-list-instead-of-a-circular-array)