Pythonのリストでキューを実現する方法

画像出典：https://1000ya.isis.ne.jp/1532.html

単純な方法（appendとpop）

Pythonのリストを使ってキューを実現するにはどうすれば良いでしょうか？ 単純な方法として、リストのappendとpopを使う方法が考えられます。

append(x)でリストの末尾にxを追加、pop(0)でリストの先頭を取得できます。

>>> q = []
>>> q.append(1)
>>> q
[1]
>>> q.append(2)
>>> q
[1, 2]
>>> x = q.pop(0)
>>> x
1
>>> q
[2]
>>> x = q.pop(0)
>>> x
2
>>> q
[]

しかし、実はこの pop(0) はメチャメチャ遅いのです。 先頭の要素をリストから消すとき、先頭から2個目以降の要素を1個ずつ前にズラさなくてはならないためです。 大きなリストを扱うときは特に処理時間に大きく影響します。

より望ましい方法 （dequeの利用）

では、どうすれば良いでしょうか？ 実は、キューを使うには、リストの代わりにdequeというオブジェクトを利用したほうがベターです。これは、上述のpopの問題が回避された実装になっています。

>>> from collections import deque
>>> q = deque([])
>>> q.append(1)
>>> q
deque([1])
>>> q.append(2)
>>> q
deque([1, 2])
>>> x = q.popleft()
>>> x
1
>>> q
deque([2])
>>>
>>> x = q.popleft()
>>> x
2
>>> q
deque([])

大きなリストで実行時間を比較！

では大きなリストを扱うタスクで速度を比較してみます。「ラウンドロビンスケジューリング」という問題について考えます。

ラウンドロビンスケジューリングとは以下のような問題です。

• 1度に同じ仕事を10個こなせる人（Xさん）がいます。
• 今ここに仕事Aが20個、仕事Bが5個、仕事Cが15個あります。
• Xさんは、並んでいる順に仕事をこなしていくのですが、1度にこなせなかった分は後回しにします。

以上の条件で、Xさんはどのような順番で仕事をこなしていくことになるでしょうか。

順番に考えていきます。

1. 仕事が A:20, B:5, C:15 と並んでいる
2. Xさんは仕事A 20個を受け取るが、そのうち10個だけこなして残りの10個は後回し。（残り B:5, C:15, A:10）
3. Xさんは仕事B 5個を受け取り、全てをこなす。（残り C:15, A:10）
4. Xさんは仕事C 15個を受け取るが、そのうち10個だけこなして残りの5個は後回し。（残り A:10, C:5）
5. Xさんは仕事A 10個を受け取り、全てをこなす。（残り C:5）
6. Xさんは仕事C 5個を受け取り、全てをこなす。（残り なし）

というわけで、A10→B5→C10→A10→C5 というのが答えとなります。

このように、こなせなかった仕事を後回しにしつつ、仕事を順番にこなしていく方法をラウンドロビンスケジューリングといいます。 これをキューを使ってシミュレートしてみます。

以下では、1度にこなせる仕事量(quantum)を30、仕事の種類が100000個、各仕事の個数が1個〜1000個ランダムにあるとして、リストを使ったキューと、dequeを使ったキューで速度を比較してみました。

# リストを使ったキュー。pop(0)とappend()で、キューに値を出し入れする。
def rrs_simple(tasks, quantum):
    elapsed = 0
    while 1:
        task = tasks.pop(0)
        if task['cost'] <= quantum:
            elapsed += task['cost']
            print(task['name'], elapsed)
        else:
            elapsed += quantum
            task['cost'] -= quantum
            tasks.append(task)
        if len(tasks) == 0:
            return None

# dequeを使ったキュー。popleft()とappend()で、キューに値を出し入れする。
def rrs_using_lib(tasks, quantum):
    from collections import deque
    tasks = deque(tasks)
    elapsed = 0
    while 1:
        task = tasks.popleft()
        if task['cost'] <= quantum:
            elapsed += task['cost']
            print(task['name'], elapsed)
        else:
            elapsed += quantum
            task['cost'] -= quantum
            tasks.append(task)
        if len(tasks) == 0:
            return None


if __name__ == '__main__':
    import time
    import random
    import copy

    # 1度に仕事を30個処理できる。
    quantum = 30
    tasks = []
    # 仕事の種類は100000種類ある。
    for i in range(100000):
        # 各仕事は1個〜1000個ある。（ランダム）
        cost = random.randint(1,1000)
        task = {'name': 'task' + str(i), 'cost': cost}
        tasks.append(task)

    # 前者の方法でラウンドロビンスケジューリング問題を解いてみて、実行時間を表示する
    copied_tasks = copy.deepcopy(tasks)
    start = time.time()
    rrs_simple(copied_tasks, quantum)
    elapsed_time = time.time() - start
    print('-------')
    print('simple:', elapsed_time)
    print('-------')

    # 後者の方法でラウンドロビンスケジューリング問題を解いてみて、実行時間を表示する
    copied_tasks = copy.deepcopy(tasks)
    start = time.time()
    rrs_using_lib(copied_tasks, quantum)
    elapsed_time = time.time() - start
    print('-------')
    print('library:', elapsed_time)
    print('-------')

実行してみます。

python round_robin_scheduling.py | grep ":"

simple: 32.75466179847717
library: 0.7376770973205566

32秒の差が出ました。 前者の方法はかなり遅いことがわかります。

スクラッチ実装でdequeと戦ってみた

pop(0)を回避しつつ、リストを使った方法でdequeと戦ってみました。実装のポイントはリストで「リングバッファ」を作るという点です。リングバッファとは、固定長の1次元リストのお尻と頭が繋がったリストです。

リングバッファの概念図

headとtailという、リストのインデックスを指すポインタを2つ用意し、キューに値を入れるときはheadが指している場所に値を代入し、キューから値を取り出すときはtailが指している場所を参照します。値を代入したときはheadを、値を参照した時はtailを、それぞれ1だけインクリメントします。ただし、headもしくはtailがリストのお尻を指している時に限り、インクリメントするのではなくリストの頭を指すようにします。こうすることで、pop(0)のように要素の並び替えをすることなくキューを実現できます。

では、先程と同様に速度比較してみましょう。

# スクラッチ実装
def rrs_from_scratch(tasks, quantum):
    tasks.append(None)
    elapsed = 0
    head = len(tasks) -1
    tail = 0
    max_idx = len(tasks) -1

    while 1:
        task = tasks[tail]
        tasks[tail] = None
        tail += 1
        if tail > max_idx:
            tail = 0

        if task['cost'] <= quantum:
            elapsed += task['cost']
            print(task['name'], elapsed)
        else:
            elapsed += quantum
            task['cost'] -= quantum
            tasks[head] = task
            head += 1
            if head > max_idx:
                head = 0

        if head == tail:
            return None

if __name__ == '__main__':
    import time
    import random
    import copy

    # 1度に仕事を30個処理できる。
    quantum = 30
    tasks = []
    # 仕事の種類は100000種類ある。
    for i in range(100000):
        # 各仕事は1個〜1000個ある。（ランダム）
        cost = random.randint(1,1000)
        task = {'name': 'task' + str(i), 'cost': cost}
        tasks.append(task)

    # 単純なリスト方式
    copied_tasks = copy.deepcopy(tasks)
    start = time.time()
    rrs_simple(copied_tasks, quantum)
    elapsed_time = time.time() - start
    print('-------')
    print('simple:', elapsed_time)
    print('-------')

    # dequeを使用
    copied_tasks = copy.deepcopy(tasks)
    start = time.time()
    rrs_using_lib(copied_tasks, quantum)
    elapsed_time = time.time() - start
    print('-------')
    print('library:', elapsed_time)
    print('-------')

    # スクラッチ実装（リングバッファ）
    copied_tasks = copy.deepcopy(tasks)
    start = time.time()
    rrs_from_scratch(copied_tasks, quantum)
    elapsed_time = time.time() - start
    print('-------')
    print('scratch:', elapsed_time)
    print('-------')

実行してみます。

python round_robin_scheduling.py | grep ":"

simple: 40.73896789550781
library: 0.7772879600524902
scratch: 0.8062808513641357

アルゴリズム	時間(秒)
単純リスト	40.73
deque	0.77
スクラッチ（リングバッファ）	0.80

という結果となりました。やはり単純なリストは遅いです。スクラッチは、dequeには勝てませんでしたがほぼ同速となりました。

以上です。

本日は「Pythonのリストでキューを実現する方法」をご紹介しました。pop(0)は遅いため、代わりにdequeやスクラッチで実装するほうがベターです。良い記事だと思っていただいた方は、以下の「★＋」ボタンのクリック、SNSでのシェア、「読者になる」ボタンのクリックをお願いします。　それではまたー！