Python(2.7)で更新可能な優先度つきキューを作りたい

概要

要素を更新可能な優先度つきキューを作って遊びます。
正確には、同じ要素を異なる優先度でpushすると、優先度を上書きするような優先度付きキューを作ります。

あらすじ

アルゴリズムクイックリファレンスの関係で、優先度つきキューを使用する必要が出てきたのですが、"優先度つきキュー Python"で検索して一番上に出てくる
http://docs.python.jp/2/library/heapq.html
を見ると、何やら物々しいことが書いてありました。

優先度つきキュー は、ヒープの一般的な使い方で、実装にはいくつか困難な点があります:

• ソート安定性: 優先度が等しい二つのタスクが、もともと追加された順序で返されるためにはどうしたらいいでしょうか？
• 将来の Python 3 では、優先度が等しくてタスクにデフォルトの比較順序がないと、タプルの比較は (priority, task) 対に分けられます。
• あるタスクの優先度が変化したら、どうやってそれをヒープの新しい位置に移動させるのでしょうか？
• 未解決のタスクが削除される必要があるとき、どのようにそれをキューから探して削除するのでしょうか？

最初の二つの困難の解決策は、項目を優先度、項目番号、そしてタスクを含む 3 要素のリストとして保存することです。この項目番号は、同じ優先度の二つのタスクが、追加された順序で返されるようにするための同点決勝戦として働きます。そして二つの項目番号が等しくなることはありませんので、タプルの比較が二つのタスクを直接比べようとすることはありえません。
残りの困難は主に、未解決のタスクを探して、その優先度を変更したり、完全に削除することです。タスクを探すことは、キュー内の項目を指し示す辞書によってなされます。
項目を削除したり、優先度を変更することは、ヒープ構造の不変関係を壊すことになるので、もっと難しいです。ですから、可能な解決策は、その項目が無効であるものとしてマークし、必要なら変更された優先度の項目を加えることです:

なるほど、まあ確かに言っていることは分かります。heapqのところに、こんなことをわざわざ書いてあるから、デフォルトではPythonは優先度つきキューを提供していないのかと思いました。
しかし、そんなことはなく、ちゃんとQueue.PriorityQueue()というものがあります。ただ、あるのですが、このPriorityQueue()を使って直ちに更新処理を行える、ということではなさそうでした。
また、このページに書いてあることを考慮すると、どれぐらい計算速度が良くなったり悪くなったりするものなのか、というどうでもいい興味がわきました。

設計

手製優先度つきキュー(以下、「手キュー」)は、(sortkey,(value,generation))というタプルのタプルのリストと、value別のgenerationを一つずつ保持するディクショナリからなるものとします。
削除は、上に書いてある通りに実際には行わず、フラグ付けで行うイメージですが、最新の世代番号を辞書に保存することにして、最新でないデータは使わない、という方法での制御とします。
また、要素がたまると検索性能が劣化するので、ガベージコレクション的にデータ削除＋ヒープ再構成をするものとします。ここでいう「ガベージコレクション」のタイミングですが、適当な定数mとして保持しておくことにし、要素の数(タプルの一番外側のカッコを1と数えた数)がmを超えたら再構成、というようにします。

この「手キュー」の設計では、

• push → heapq.heappush()してディクショナリを更新、大きさがmを超えたら、最新世代のものだけでヒープやりなおし ※辞書の世代が更新されるので、結果的にchangeも兼ねる。
• pop → heapq.heappop()を最新の世代を取り出すまで取り続ける（valueをもとにgenerationをディクショナリから抽出して、一致していなければheappop()を繰り返し）
• ※delete → 辞書のみを更新

という風にします。キューの要素へのランダムアクセス性は、とりあえず気にしないことにします。辞書に値を持たせてしまえば、メモリ使用量は増えるものの、理屈の上ではランダムアクセスできます。

実装

TeQueue.py

import heapq
def tepush(l, d, m, sortkey, value):
    if not value in d:
        generation = 1
    else:
        generation = d[value] + 1
    d[value] = generation
    heapq.heappush(l, (sortkey, value, generation))
    length = len(l)
    if length > m: # ヒープの再構成
        lt = []
        for n in range(0,length):
            if l[n][2] == d[l[n][1]]:
                heapq.heappush(lt, (l[n][0], l[n][1], 1))
                d[l[n][1]] = 1
        return lt
    else:
        return l

def tepop(l, d):
    while len(l) > 0:
        (sortkey, value, generation) = heapq.heappop(l)
        if d[value] == generation:
            return (sortkey, value, generation) 

これによって、設計に記載したような考え方で同じvalueに対応する値を「上書き」できるようになりました。

他のキューとの比較

Queue.PriorityQueue()と比較してみます。

初期化

さしあたって、次の初期化を行います。

initialize.py

# Nの乱数列を生成
import random
N = 100000
M = 100000
target = range(0,N)
temp = range(0,N)
for n in range(N-1,-1,-1):
  target[n] = temp.pop(random.randint(0,n))

push

pushtest.py

import Queue
import heapq
import datetime

d = datetime.datetime(1,1,1)
time1 = d.now()
t = Queue.PriorityQueue(N)
for n in range(0,N):
    t.put((target[n], n))
time2 = d.now()
u = []
for n in range(0,N):
    heapq.heappush(u, (target[n], n))
time3 = d.now()
v = []
dd = {}
for n in range(0,N):
    v = tepush(v, dd, 2*M, target[n], n)
time4 = d.now()
print('push:Queue',(time2 - time1).__str__())
print('push:Heap',(time3 - time2).__str__())
print('push:TeQ',(time4 - time3).__str__())

push結果

('push:Queue', '0:00:00.350000')
('push:Heap', '0:00:00.086000')
('push:TeQ', '0:00:00.137000')

だいたいこんな感じになりました。比較が多い分TeQ>Heapになっていますが、Queueよりはだいぶ速いです。

pop

poptest.py

import Queue
import heapq
import datetime

time1 = d.now()
for n in range(0,M):
    t.get()
time2 = d.now()
for n in range(0,M):
    heapq.heappop(u)
time3 = d.now()
for n in range(0,M):
    tepop(v, dd)
time4 = d.now()
print('pop:Queue',(time2 - time1).__str__())
print('pop:Heap',(time3 - time2).__str__())
print('pop:TeQ',(time4 - time3).__str__())

pop結果

('pop:Queue', '0:00:00.484000')
('pop:Heap', '0:00:00.214000')
('pop:TeQ', '0:00:00.299000')

だいたいこんな感じになりました。pushよりは差が少ないですが、やはり同じような傾向です。
本当は、重複度を変えてMをいじって遊びたかったのですが、Queueやheapqに対してはあまり意味がないことを悟ってしまい、手キューの中での比較をすることにしました。
手キュー以外との比較をする場合は、
http://stackoverflow.com/questions/9969236/how-to-implement-priority-queues-in-python
なんかと比較するのが良さそうでしたが、劣後としました。

重複度を変えた場合の、Mの係数と速度の比較

実験A

初期化

initializeA.py

# Nの乱数列を生成
import random
N = 1000000
M = 40000
target = range(0,N)
for n in range(N-1,-1,-1):
  target[n] = random.randint(0,M-1)

実施結果

testA.py

d = datetime.datetime(1,1,1)
for m in [2,5,10,20]:
    time1 = d.now()
    v = []
    dd = {}
    for n in range(0,N):
        v = tepush(v, dd, m*M, n, target[n])
    time2 = d.now()
    for n in range(0,M):
        tepop(v, dd)
    time3 = d.now()
    print('push:TeQ_' + str(m),(time2 - time1).__str__())
    print('pop:TeQ_' + str(m),(time3 - time2).__str__())

結果A

('push:TeQ_2', '0:00:02.731000')
('pop:TeQ_2', '0:00:00.193000')
('push:TeQ_5', '0:00:01.794000')
('pop:TeQ_5', '0:00:00.527000')
('push:TeQ_10', '0:00:01.667000')
('pop:TeQ_10', '0:00:00.711000')
('push:TeQ_20', '0:00:01.605000')
('pop:TeQ_20', '0:00:00.581000')

20の方がpopが速いのは、20*40000と10*40000の二回目とが重なるから、結局同じ状態になっている、ということなんでしょうか。。。ふむ。

実験B

初期化

initializeB.py

# Nの乱数列を生成
import random
N = 1000000
M = 4000
target = range(0,N)
for n in range(N-1,-1,-1):
  target[n] = random.randint(0,M-1)

実施結果

testB.py

d = datetime.datetime(1,1,1)
for m in [2,5,10,20]:
    time1 = d.now()
    v = []
    dd = {}
    for n in range(0,N):
        v = tepush(v, dd, m*M, n, target[n])
    time2 = d.now()
    for n in range(0,M):
        tepop(v, dd)
    time3 = d.now()
    print('push:TeQ_' + str(m),(time2 - time1).__str__())
    print('pop:TeQ_' + str(m),(time3 - time2).__str__())

結果B

('push:TeQ_2', '0:00:02.927000')
('pop:TeQ_2', '0:00:00.013000')
('push:TeQ_5', '0:00:02.035000')
('pop:TeQ_5', '0:00:00.018000')
('push:TeQ_10', '0:00:01.929000')
('pop:TeQ_10', '0:00:00.093000')
('push:TeQ_20', '0:00:01.846000')
('pop:TeQ_20', '0:00:00.026000')

何回か試しても、この傾向はあまり変わらない。なぜだろう。順番変えても変わらず、20が10より圧倒的に速いまま。ふむ。。。

まとめ

ガベージコレクションとかインデックスの貼り直しに相当することを、しなくても良いような気がしてしまいます。