はじめに
どうも、y-tetsuです。
以前、Pythonでライフゲームを作った話を記事にした者です。
あれから1年半ほど経ちましたが、最近またライフゲームの創作欲が高まりまして、コードを少し触りました。折角なので、その内容を記事に残しておこうと思います。
今回の記事のテーマは、パフォーマンスの改善です。cProfileというPythonの標準モジュールを使って、足を引っ張っているボトルネックの部分を見つけ、そこに対策を打ち、コードをより高速化していく過程を書いてみようと思います。
できたもの
最終的に、以前のものより約5倍近く速く動作するものになりました(細かくは4.85倍)。前より大きな画面サイズで、わりとヌルヌル動くまでになったかなぁと思います。
以下は、長寿型の"どんぐり"を動かした様子です。なんとこの先、"5206世代"まで変化を続けるそうです。(なぜに"どんぐり"…)
完成形のコードは、以下のGitHubのリポジトリから取得できます。
ヨシ、いっちょ動かしてやろう！という奇特な御仁のために。
Python 3.9以上のWindows10環境で動作確認しております。
上記のGitHubリポジトリより、任意のフォルダに以下をダウンロードして下さい。
- game_of_life.py
- samples.json
ダウンロードしたフォルダで下記コマンドを実行すれば動くと思います。
> python game_of_life.py [オプション]
パフォーマンスを改善する
前提条件
「パフォーマンス改善」と一口に言っても、様々な手段が考えられると思います。今回は筆者の個人的な思いにより、以下の範囲で改善を試みたいと思います。
- 使う言語はPythonのみ (CPython)
- 画面表示は、もちろんCLI上で (CLIはコンソールと呼んだり、Windowsではコマンドプロンプトとも呼んだりします)
- 標準モジュール以外は使わない
他の言語に比べ、処理速度の面で非力なPythonのみのロジックで、無骨なコンソール上に、どこまで鮮やかなライフゲームの世界を表現できるのか？今回はその部分を追求してみます。
動作確認に使用したPC環境のスペックは以下の通りです。
| 環境 | スペック |
|---|---|
| ディスプレイサイズ | 1366x768 |
| OS | Windows |
| プロセッサ | 1.6GHz |
| メモリ | 4.00GB |
| 物理コア数 | 2 |
皆様のお使いのPCですと、性能が良すぎて、改善の効果があまり伝わらないかもしれませんが…。
元となるコード
以前の記事から、色々と手が入って形は変わっていますが、処理速度面は特に意識せずの作りっぱなしの状態です。
閉じておりますので、適宜開いてご確認下さい。Pythonに覚えのある方は、是非一緒に改善案を考えて頂けると、小躍りして喜びます。
★クリックで展開★ 改善前のコード
import time
from platform import system
from ctypes import windll
from random import random
from datetime import datetime
import json
import pprint
class GameOfLife:
def __init__(self, sample=None, name='game_of_life', x=30, y=15,
world=None, max_step=None, wait=0.03, delay=0.0,
alive='■', ratio=0.5, loop=False, torus=False, mortal=False,
color=False, json_file=None):
self.sample = sample
self.name = name
samples, colors = self._load_samples('samples.json'), None
if sample in samples:
self.name = sample
if 'world' in samples[sample]:
world = samples[sample]['world']
if 'max_step' in samples[sample]:
if max_step is None:
max_step = samples[sample]['max_step']
if 'colors' in samples[sample]:
colors = samples[sample]['colors']
if 'wait' in samples[sample]:
wait = samples[sample]['wait']
self.x, self.y = x, y
rand = False
if world is not None:
self.x, self.y = len(world[0]), len(world)
self.world = [[x for x in row] for row in world]
else:
rand = True
self.world = [
[random() < ratio for _ in range(x)] for _ in range(y)
]
self.max_step = max_step if max_step is not None else 100
self.wait = wait
self.delay = delay
self.alive = alive
self.ratio = ratio
self.loop = loop
self.torus = torus
self.mortal = mortal
self.color = color
if json_file is not None:
rand = False
self._load(json_file)
self.dirs = [
(-1, -1), (0, -1), (1, -1),
(-1, 0), (1, 0),
(-1, 1), (0, 1), (1, 1),
]
self.step = 1
self.ages = [[x for x in row] for row in self.world]
self.colors = [[0 for _ in row] for row in self.world]
if colors:
self.colors = colors
self.alives = ((' ', 0), (self.alive, 10), ('□', 30), ('・', 60))
self.lifespans = [alive[1] for alive in self.alives]
self.marks = [alive[0] for alive in self.alives]
self.console = Console(self.x, self.y, self.name, self.marks)
if rand:
self._dump()
def start(self):
try:
self.console.setup()
while True:
self.console.display(self.world, self.step, self.colors)
if self.step == self.max_step:
if not self.loop:
# if loop option is enabled, game_of_life is never end.
break
if self.step == 1:
time.sleep(self.delay)
self._update()
self._wait()
except KeyboardInterrupt:
return
finally:
self.console.teardown()
def _load_samples(self, samples_file):
samples = {}
try:
with open(samples_file, 'r') as f:
samples = json.load(f)
except FileNotFoundError:
pass
return samples
def _load(self, json_file):
try:
with open(json_file, 'r') as f:
settings = json.load(f)
self.name = settings['name']
self.x = settings['x']
self.y = settings['y']
self.world = settings['world']
self.max_step = settings['step']
self.wait = settings['wait']
self.delay = settings['delay']
self.alive = settings['alive']
self.ratio = settings['ratio']
self.loop = settings['loop']
self.torus = settings['torus']
self.mortal = settings['mortal']
self.color = settings['color']
except FileNotFoundError:
pass
def _update(self):
max_y, max_x = self.y, self.x
world, colors = self.world, self.colors
mortal, ages = self.mortal, self.ages
new_world = []
for y in range(max_y):
next_cells = []
for x in range(max_x):
if next_cell := self._next_cell(x, y):
if mortal:
# if mortal option is enabled, living cells are ageing.
if world[y][x]:
next_cell = self._ageing(x, y, next_cell)
else:
ages[y][x] = 1
else:
ages[y][x], colors[y][x] = 0, 0
next_cells += [next_cell]
new_world += [next_cells]
self.world = [[x for x in row] for row in new_world]
self.step += 1
def _ageing(self, x, y, cell):
self.ages[y][x] += 1
for index, lifespan in enumerate(self.lifespans):
if self.ages[y][x] < lifespan:
cell = index
break
else:
# lifespan has expired.
self.ages[y][x] = 0
cell = 0
return cell
def _next_cell(self, x, y):
alive, color = self._get_around(x, y)
if self.world[y][x]:
return 1 if alive == 2 or alive == 3 else 0
if alive == 3:
if self.color:
# if color option is enabled,
# the new cell evolves from the most evolved cell around it.
# immediate reflection due to diversity.
self.colors[y][x] = color + 1
return 1
return 0
def _get_around(self, x, y):
max_x, max_y = self.x, self.y
world, colors, torus = self.world, self.colors, self.torus
alive, color = 0, 0
for dx, dy in self.dirs:
next_x, next_y = x + dx, y + dy
if torus:
# if torus option is enabled,
# top and bottom, left and right are connected.
next_x, next_y = next_x % max_x, next_y % max_y
if (0 <= next_x < max_x) and (0 <= next_y < max_y):
if self.color:
# if color option is enabled,
# the color of most evolved cell around it is inherited.
color = max(color, colors[next_y][next_x])
if world[next_y][next_x]:
alive += 1
return alive, color
def _wait(self):
time.sleep(self.wait)
def _dump(self):
now = datetime.now().strftime('%Y%m%d%H%M%S')
json_file = 'world' + now + '.json'
settings = {
'name': self.console.name,
'x': self.x,
'y': self.y,
'world': self.world,
'step': self.max_step,
'wait': self.wait,
'delay': self.delay,
'alive': self.alive,
'ratio': self.ratio,
'loop': self.loop,
'torus': self.torus,
'mortal': self.mortal,
'color': self.color,
}
with open(json_file, 'w') as f:
output = pprint.pformat(settings, indent=4,
width=1000, sort_dicts=False)
output = output.replace("'", '"')
output = output.replace('True', 'true')
output = output.replace('False', 'false')
f.write(output)
class Console:
def __init__(self, x, y, name, marks):
self.x = x
self.y = y
self.name = name
self.marks = marks
if 'win' in system().lower():
self._enable_win_escape_code()
self.color_list = [
'\x1b[39m', # 0: default
'\033[38;2;255;255;255m', # 1: white
'\033[38;2;0;153;68m', # 2: green
'\033[38;2;230;0;18m', # 3: red
'\033[38;2;235;202;27m', # 4: yellow
'\033[38;2;145;0;0m', # 5: brown
'\033[38;2;0;160;233m', # 6: cyan
'\033[38;2;241;158;194m', # 7: pink
'\033[38;2;240;131;0m', # 8: orange
'\033[38;2;146;7;131m', # 9: purple
]
self.title = self._setup_title()
def setup(self):
self._clear_screen()
self._cursor_hyde()
def teardown(self):
self._cursor_show()
def display(self, world, step, colors):
if step > 1:
self._cursor_up(self.y + 4)
self._display_title()
self._display_world(world, colors)
self._display_step(step)
def _enable_win_escape_code(self):
kernel = windll.kernel32
kernel.SetConsoleMode(kernel.GetStdHandle(-11), 7)
def _clear_screen(self):
print("\033[;H\033[2J")
def _cursor_hyde(self):
print('\033[?25l', end='')
def _cursor_show(self):
print('\033[?25h', end='')
def _cursor_up(self, n):
print(f'\033[{n}A', end='')
def _setup_title(self):
name, x, y = self.name, self.x, self.y
count = x * y
max_cell_size = 3025
title = f'{name} ({x} x {y})'
if count > max_cell_size:
title += f' * warning : max_cell_size({max_cell_size}) is over! *'
return title
def _display_title(self):
print(f"{self.title}")
def _display_world(self, world, colors):
color_list = self.color_list
marks = self.marks
max_y, max_x = self.y, self.x
max_color = len(color_list)
line = []
# display world on cli
print('┌' + '─' * (max_x * 2) + '┐')
for y in range(max_y):
cells = ''
for x in range(max_x):
color = colors[y][x] % max_color
mark = marks[world[y][x]]
cells += self.color_list[color] + mark + self.color_list[0]
line += ['│' + cells + '│']
print('\n'.join(line))
print('└' + '─' * (max_x * 2) + '┘')
def _display_step(self, step):
print(f'step = {step}')
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(
description='A game of life simulator on CLI')
parser.add_argument('sample', nargs='?')
parser.add_argument('-n', '--name')
parser.add_argument('-x', type=int)
parser.add_argument('-y', type=int)
parser.add_argument('-j', '--json')
parser.add_argument('-s', '--step', type=int)
parser.add_argument('-w', '--wait', type=float)
parser.add_argument('-d', '--delay', type=float)
parser.add_argument('-a', '--alive')
parser.add_argument('-r', '--ratio', type=float)
parser.add_argument('-l', '--loop', action="store_true")
parser.add_argument('-t', '--torus', action="store_true")
parser.add_argument('-m', '--mortal', action="store_true")
parser.add_argument('-c', '--color', action="store_true")
args = parser.parse_args()
setting = {}
if args.sample:
setting['sample'] = args.sample
if args.name:
setting['name'] = args.name
if args.x:
setting['x'] = args.x
if args.y:
setting['y'] = args.y
if args.json:
setting['json_file'] = args.json
if args.step:
setting['max_step'] = args.step
if args.wait is not None:
setting['wait'] = args.wait
if args.delay:
setting['delay'] = args.delay
if args.alive:
setting['alive'] = args.alive
if args.ratio:
setting['ratio'] = args.ratio
setting['loop'] = args.loop
setting['torus'] = args.torus
setting['mortal'] = args.mortal
setting['color'] = args.color
GameOfLife(**setting).start()
テストを用意する
パフォーマンスを改善するにあたり、コードを改良していくわけですが、新たに加えた変更にどの程度の効果があるのかを、はっきり確認できるよう前もって準備しておく必要があります。
私の目指すところとしては、なるべく大画面のコンソール上で、最も負荷のかかるフルオプションの状況において、ヌルヌル＆サクサク動くライフゲームの世界を実現したいです。(もちろん、独自の機能は維持したままです)
今回は、47x47サイズで作った長寿型の"ノアの方舟"を使ってみようと思います。用意したサンプルの中ではそこそこ大きいですし、動きもありますので。
ちなみに、ウェイトを入れた素の"ノアの方舟"の様子は以下です。(ネーミング、理由は分かりませんがセンスを感じます)
この"ノアの方舟"に対して、本コードで用意している以下のオプションを設定し、最も負荷のかかる状態とします。(オプションの内容については前回の記事を参照して下さい)
| オプション名 | 設定内容 |
|---|---|
| トーラス世界 | 有効 |
| 寿命 | 有効 |
| 色進化 | 有効 |
| ウェイト時間 | 0.0秒(なし) |
そして、これを3回実行した際の「処理時間」を一つの指標とします。
これから、この指標をより短縮するよう、コードを改変していきます。
指標となる処理を実行するために、testsフォルダを用意してtest_game_of_life.pyファイルを配置します。そして、以下のテストコードを追加します。
import unittest
from game_of_life import GameOfLife
class TestGameOfLife(unittest.TestCase):
def test_noahs_ark_elp(self):
setting = {}
setting['sample'] = 'noahs-ark'
setting['torus'] = True
setting['mortal'] = True
setting['color'] = True
setting['wait'] = 0.0
for _ in range(3):
GameOfLife(**setting).start()
この時、test_game_of_life.pyにはライフゲームの基本機能のユニットテストも同時に入れておりますが、ここでは割愛いたします。
本記事では、コードを書き変える際、あらかじめ追加しておいたユニットテストを全てパスする前提で、話を進めます。呼吸するように、そうできたら良いなといつも思います。(コードを変えていくら速くなっても、元の動作が狂ってしまっては、意味がありませんからね)
改善のための4つのStep
元となるコードと、その性能を示す指標が用意出来ました。これからココを起点に、次の4つのStepを通してコードの処理速度を改善していきたいと思います。
Step1 : 処理時間を計測する
冒頭でご紹介したcProfileが、ここでいよいよ登場します。コイツを使うと何がいいのかというと、一口に、どの処理にどれだけ時間がかかっているのかを、きっちり計って示してくれるところです。
「cProfileとは」というところは一旦後回しにして、兎にも角にも動かして、現状のコードの実態を早急(さっきゅう)に見てみましょう。
以下を実行します。
> python -m cProfile -s cumtime -m unittest tests.test_game_of_life.TestGameOfLife.test_noahs_ark_elp
これは、先ほど用意したテストコードを走らせて、処理時間の計測結果を出力するコマンドです。
ちなみに現状のコードは、コンソール上で以下のような動作をしています。
ウェイトなしのはずが、随分とモッサリした印象です。
そしてお待ちかね、cProfileの計測結果が出ました！(長いので一部抜粋しております)
9940406 function calls (9939627 primitive calls) in 53.831 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
39/1 0.000 0.000 53.904 53.904 {built-in method builtins.exec}
1 0.000 0.000 53.903 53.903 runpy.py:200(run_module)
1 0.000 0.000 53.836 53.836 runpy.py:64(_run_code)
1 0.000 0.000 53.836 53.836 __main__.py:1(<module>)
1 0.000 0.000 53.836 53.836 main.py:65(__init__)
1 0.000 0.000 53.673 53.673 main.py:246(runTests)
1 0.000 0.000 53.673 53.673 runner.py:151(run)
2/1 0.000 0.000 53.662 53.662 suite.py:83(__call__)
2/1 0.000 0.000 53.662 53.662 suite.py:102(run)
1 0.000 0.000 53.662 53.662 case.py:652(__call__)
1 0.000 0.000 53.662 53.662 case.py:558(run)
1 0.000 0.000 53.659 53.659 case.py:549(_callTestMethod)
1 0.001 0.001 53.659 53.659 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.023 0.008 53.544 17.848 game_of_life.py:75(start)
450 0.050 0.000 32.899 0.073 game_of_life.py:247(display)
450 1.748 0.004 32.430 0.072 game_of_life.py:282(_display_world)
2707 31.074 0.011 31.074 0.011 {built-in method builtins.print}
447 1.544 0.003 20.598 0.046 game_of_life.py:122(_update)
987423 1.378 0.000 18.895 0.000 game_of_life.py:156(_next_cell)
987423 13.281 0.000 17.517 0.000 game_of_life.py:169(_get_around)
7899487 4.236 0.000 4.237 0.000 {built-in method builtins.max}
最初に見る箇所は冒頭の下記記載になります。全体の処理時間が53.831秒だったことが分かります。これが筆者のスタート地点です。
9940406 function calls (9939627 primitive calls) in 53.831 seconds
cProfileについては、以下の記事などを参考にしました。
https://qiita.com/meshidenn/items/4dbde22d1e7a13a255bb
本家のドキュメントを押さえておくなら以下から。
https://docs.python.org/3/library/profile.html
Step2 : ボトルネックを探す
コードのパフォーマンスを上げるためのセオリーは、そのコードが最も時間を要する箇所──つまり、ボトルネック──を見つけて、そこを改善することです。"勘"で決めた(残念ながらボトルネックじゃなかった)箇所を、物凄く頑張って10倍速くしても、全体から見ればほぼ変わらなかった…、なんてこともあり得ますからね。最も改善し甲斐のあるところに優先して手を打ちましょう。
それでは、現状のコードのボトルネックを探していきます。
cProfileが出力したログの内容で、次に見るべきポイントはざっくり以下です。
| カラム名 | 内容 |
|---|---|
| filename:lineno(function) | ファイル名:行番号(実行された関数/メソッド) |
| ncalls | コールされた回数 |
| cumtime | 関数の累積実行時間(別の関数コールも含む) |
| tottime | その関数自体の累積実行時間 |
ライフゲーム全体を実行するstartメソッドを除き、個別のメソッドで最も時間のかかっている箇所は、displayメソッドのようです。計450回呼ばれ、累積で約32秒かかっています。
450 0.050 0.000 32.899 0.073 game_of_life.py:247(display)
ここまでから、ライフゲーム全体の処理時間の53秒に対して、32秒もの大半(約6割)をこの処理が占めていると分かりました。
さらにもう少しよく見てみると、displayからは、_display_worldを経由して、print関数がたくさん呼ばれていているのですが、displayのほぼすべて(32秒中の31秒)は、print関数で占められている事も読み取れます。
450 0.050 0.000 32.899 0.073 game_of_life.py:247(display)
450 1.748 0.004 32.430 0.072 game_of_life.py:282(_display_world)
2707 31.074 0.011 31.074 0.011 {built-in method builtins.print}
おそらく、今倒すべき"敵の尻尾"を掴んだかもしれません。このprint関数を呼ぶ処理に最も時間がかかっており、ボトルネックとなっていることが判明しました。
ゲームの状況を逐次描画しなければならない性質上、妥当なものである可能性も十分あります。ですが、ここを改善することが出来れば、コード全体の性能を改善する上で、最も効果の見込めるポイントであることに違いはありません！
Step3 : コードを直す
ボトルネックとなっている箇所は見つかりました。それでは速度が改善するようにコードを直しましょう。…といっても、何をどう改善すればよいのでしょうか。
cProfileの結果から、print関数に大半の時間を費やしているという事でした。それではここで、画面描画のためにprint関数を呼んでいる箇所を全て見てみましょう。
class Console:
def display(self, world, step, colors):
if step > 1:
self._cursor_up(self.y + 4)
self._display_title()
self._display_world(world, colors)
self._display_step(step)
def _cursor_up(self, n):
print(f'\033[{n}A', end='')
def _display_title(self):
print(f"{self.title}")
def _display_world(self, world, colors):
color_list = self.color_list
marks = self.marks
max_y, max_x = self.y, self.x
max_color = len(color_list)
line = []
# display world on cli
print('┌' + '─' * (max_x * 2) + '┐')
for y in range(max_y):
cells = ''
for x in range(max_x):
color = colors[y][x] % max_color
mark = marks[world[y][x]]
cells += self.color_list[color] + mark + self.color_list[0]
line += ['│' + cells + '│']
print('\n'.join(line))
print('└' + '─' * (max_x * 2) + '┘')
def _display_step(self, step):
print(f'step = {step}')
全部で6回のprint関数を呼ぶコードになっています。…ひょっとして、何回もprint関数を呼ぶのがまずい…、とか？
── ChatGPT曰く ──
大量のデータを出力する場合や、パフォーマンスが重要なシステムでは、print文をできるだけ1回にまとめることが推奨されます。これにより、I/O操作のオーバーヘッドを削減し、バッファリングの効率を最大化することで、全体のパフォーマンスが向上します。
な、なるほど。良い後押しを得ました。とりあえず、ほんとかどうかは分かりませんが、print関数を1回にまとめるよう修正してみますね、ChatGPT先生！
変更後のコードは以下のようになりました。
class Console:
def display(self, world, step, colors):
screen = ''
if step > 1:
screen += self._cursor_up(self.y + 4)
screen += self._get_title()
screen += self._get_world(world, colors)
screen += self._get_step(step)
print(screen, end='')
def _cursor_up(self, n):
return f'\033[{n}A'
def _setup_title(self):
name, x, y = self.name, self.x, self.y
count = x * y
max_cell_size = 3025
title = f'{name} ({x} x {y})'
if count > max_cell_size:
title += f' * warning : max_cell_size({max_cell_size}) is over! *'
return title
def _get_title(self):
return f"{self.title}\n"
def _get_world(self, world, colors):
color_list = self.color_list
marks = self.marks
max_y, max_x = self.y, self.x
max_color = len(color_list)
line = []
# setup screen for display world on cli
screen = '┌' + ('─' * (max_x * 2)) + '┐\n'
for y in range(max_y):
cells = ''
for x in range(max_x):
color = colors[y][x] % max_color
mark = marks[world[y][x]]
cells += self.color_list[color] + mark + self.color_list[0]
line += ['│' + cells + '│']
screen += '\n'.join(line) + '\n'
screen += '└' + ('─' * (max_x * 2)) + '┘\n'
return screen
def _get_step(self, step):
return f'step = {step}\n'
print関数を呼ぶのは1回だけにしています。print関数を呼ぶまでは、表示させたい内容を、どんどん連結していく作りにしてみました。
Step4 : 効果を確認する
コードのボトルネックを見つけ、改善のための予想を立てて、それに基づきコードを変更しました。その次にする事は、効果の測定です。
今回のライフゲームのような目で楽しむようなものの場合、ついつい見た目の感覚で「あ、速くなってそう！」とやってしまいそうですね。ですが、そこはグッと堪えます。最初に決めた指標に対して、今回の結果がどうだったのかを冷静に見る必要があります。もう一度cProfileを実行し、はっきりとしたボトルネックの実行時間の変化を見定めましょう。
以下が結果です。
9938154 function calls (9937375 primitive calls) in 49.655 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
39/1 0.001 0.000 49.727 49.727 {built-in method builtins.exec}
1 0.000 0.000 49.726 49.726 runpy.py:200(run_module)
1 0.000 0.000 49.657 49.657 runpy.py:64(_run_code)
1 0.000 0.000 49.657 49.657 __main__.py:1(<module>)
1 0.000 0.000 49.657 49.657 main.py:65(__init__)
1 0.000 0.000 49.504 49.504 main.py:246(runTests)
1 0.000 0.000 49.504 49.504 runner.py:151(run)
2/1 0.000 0.000 49.488 49.488 suite.py:83(__call__)
2/1 0.000 0.000 49.488 49.488 suite.py:102(run)
1 0.000 0.000 49.487 49.487 case.py:652(__call__)
1 0.000 0.000 49.487 49.487 case.py:558(run)
1 0.000 0.000 49.484 49.484 case.py:549(_callTestMethod)
1 0.001 0.001 49.484 49.484 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.031 0.010 49.358 16.453 game_of_life.py:75(start)
450 0.027 0.000 29.118 0.065 game_of_life.py:247(display)
460 27.458 0.060 27.458 0.060 {built-in method builtins.print}
447 1.517 0.003 20.184 0.045 game_of_life.py:122(_update)
987423 1.345 0.000 18.514 0.000 game_of_life.py:156(_next_cell)
987423 12.963 0.000 17.169 0.000 game_of_life.py:169(_get_around)
7899487 4.206 0.000 4.207 0.000 {built-in method builtins.max}
450 1.632 0.004 1.651 0.004 game_of_life.py:284(_get_world)
変更前から変更後で、全体の処理時間が53.831秒から49.655秒へと、約"4.1秒"短縮されました。
(変更前)
9940406 function calls (9939627 primitive calls) in 53.831 seconds
(変更後)
9938154 function calls (9937375 primitive calls) in 49.655 seconds
そして、効果を狙ったprint関数の実行時間も以下の通り、31.074秒から27.5458と"3.6秒"の短縮が見られます。ボトルネックへの改善が、処理全体への改善に繋がっている事が見て取れました。(print関数のコール回数も減ってます)
(変更前)
2707 31.074 0.011 31.074 0.011 {built-in method builtins.print}
(変更後)
460 27.458 0.060 27.458 0.060 {built-in method builtins.print}
これにて、改善のための4Stepは無事完了です。幸い、速度向上に繋がる良い結果が出ました！
このように、狙った部分に狙った改善効果が見られて初めて、「コードが良くなった」と言えるのように思います(理由が分からず速い場合、それ以上コードを触れませんからね…)。もちろん、毎回狙った通り改善することも稀だったりしますが…。
お疲れ様でした。改善のためのStepを1サイクル回し終えました。
ですが、まだまだ私の求める速度には満たないようです。それでは次の敵を倒しに、再び最初のStepに戻りましょう。気力の続く限り、このサイクルを続けることが、求めるモノに近づく最短の道だと思います。
次なるボトルネックをつぶす！
サイクルをひと回しして改善はできたものの、print関数の実行にはまだまだ時間がかかっているようです。他に出来ることはないか？と考えたところ、1つ案が浮かんできました。
「表示させる文字列を減らせないか」
そうすれば当然print関数の実行時間は減らせるはず。ということで、変更したコードが以下になります。
class Console:
def _get_world(self, world, colors):
color_list = self.color_list
marks = self.marks
max_y, max_x = self.y, self.x
max_color = len(color_list)
line = []
# setup screen for display world on cli
screen = '┌' + ('─' * (max_x * 2)) + '┐\n'
for y in range(max_y):
cells, pre_color = '', 0
for x in range(max_x):
mark = marks[world[y][x]]
if mark == ' ':
# if no mark, color must be not care
cells += mark
else:
# change color if it is different from previous
if color := colors[y][x] % max_color:
if color == pre_color:
cells += mark
else:
cells += color_list[color] + mark
else:
if pre_color:
cells += color_list[0] + mark
else:
cells += mark
pre_color = color
if pre_color:
line += ['│' + cells + color_list[0] + '│']
else:
line += ['│' + cells + '│']
screen += '\n'.join(line) + '\n'
screen += '└' + ('─' * (max_x * 2)) + '┘\n'
return screen
変更したのは、表示文字列に色を付ける際のエスケープシーケンスを、なるべく減らすという部分です。
以前は、1セルずつ以下のようにして色付けしておりました。
[色付けのエスケープシーケンス] + [セル表示文字] + [デフォルトの色に戻すエスケープシーケンス]
これは、冗長なので、以下の対策を入れて文字数の多さを緩和しました。
- 前回表示した色を覚えておいて、今回表示させる色と異なる場合のみ、色付けする
- セルがない部分は空白文字なので、色付けはしない
この変更が、どの程度の効果なのか、結果は以下となりました。
9938154 function calls (9937375 primitive calls) in 23.952 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
39/1 0.001 0.000 24.024 24.024 {built-in method builtins.exec}
1 0.000 0.000 24.024 24.024 runpy.py:200(run_module)
1 0.000 0.000 23.956 23.956 runpy.py:64(_run_code)
1 0.000 0.000 23.956 23.956 __main__.py:1(<module>)
1 0.000 0.000 23.956 23.956 main.py:65(__init__)
1 0.000 0.000 23.800 23.800 main.py:246(runTests)
1 0.000 0.000 23.799 23.799 runner.py:151(run)
2/1 0.000 0.000 23.787 23.787 suite.py:83(__call__)
2/1 0.000 0.000 23.787 23.787 suite.py:102(run)
1 0.000 0.000 23.787 23.787 case.py:652(__call__)
1 0.000 0.000 23.787 23.787 case.py:558(run)
1 0.000 0.000 23.784 23.784 case.py:549(_callTestMethod)
1 0.001 0.001 23.784 23.784 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.023 0.008 23.646 7.882 game_of_life.py:75(start)
447 1.472 0.003 19.898 0.045 game_of_life.py:122(_update)
987423 1.321 0.000 18.271 0.000 game_of_life.py:156(_next_cell)
987423 12.825 0.000 16.951 0.000 game_of_life.py:169(_get_around)
7899487 4.126 0.000 4.127 0.000 {built-in method builtins.max}
450 0.016 0.000 3.706 0.008 game_of_life.py:247(display)
460 3.188 0.007 3.188 0.007 {built-in method builtins.print}
450 0.513 0.001 0.520 0.001 game_of_life.py:284(_get_world)
なんと、トータルの処理時間が、49.655秒から23.952秒と、25.7秒(約2倍)ほど速くなりました。
(変更前)
9938154 function calls (9937375 primitive calls) in 49.655 seconds
(変更後)
9938154 function calls (9937375 primitive calls) in 23.952 seconds
改善を狙ったprint関数の処理時間も、同じく24秒ほどの高速化が達成できています。
(変更前)
460 27.458 0.060 27.458 0.060 {built-in method builtins.print}
(変更後)
460 3.188 0.007 3.188 0.007 {built-in method builtins.print}
表示させていた文字列の長さも合わせてまとめると、以下となりました。
| 指標 | 変更前 | 変更後 | 差 | 削減率 |
|---|---|---|---|---|
| 全体の処理時間 | 49.655秒 | 23.952秒 | -25.703秒 | -51.7% |
| print関数の処理時間 | 27.458秒 | 3.188秒 | -24.270秒 | -88.3% |
| トータルの表示文字数 | 12,466,350字 | 2,718,894字 | -9,747,456字 | -78.1% |
ある程度、表示文字数の削減率がprint関数の処理時間の削減率とリンクしてそうです。今回は思った以上の効果が得られました。
さらなるボトルネックをつぶす！
この調子で次のボトルネックに向かいます。先ほどの改善で現在のボトルネックが変わっています。
1 0.001 0.001 23.784 23.784 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.023 0.008 23.646 7.882 game_of_life.py:75(start)
447 1.472 0.003 19.898 0.045 game_of_life.py:122(_update)
987423 1.321 0.000 18.271 0.000 game_of_life.py:156(_next_cell)
987423 12.825 0.000 16.951 0.000 game_of_life.py:169(_get_around)
7899487 4.126 0.000 4.127 0.000 {built-in method builtins.max}
450 0.016 0.000 3.706 0.008 game_of_life.py:247(display)
_updateメソッドから紐づく、次の世代のセルの計算に時間を要しているようです。特に、セルの周囲をチェックする_get_aroundメソッドが今回のターゲットのようです。このメソッドは987423回と大量に呼ばれています。そしてさらに、その8倍ほどmax関数がコールされているようです。
一度、該当箇所を見てます。
class GameOfLife:
def _get_around(self, x, y):
max_x, max_y = self.x, self.y
world, colors, torus = self.world, self.colors, self.torus
alive, color = 0, 0
for dx, dy in self.dirs:
next_x, next_y = x + dx, y + dy
if torus:
# if torus option is enabled,
# top and bottom, left and right are connected.
next_x, next_y = next_x % max_x, next_y % max_y
if (0 <= next_x < max_x) and (0 <= next_y < max_y):
if self.color:
# if color option is enabled,
# the color of most evolved cell around it is inherited.
color = max(color, colors[next_y][next_x])
if world[next_y][next_x]:
alive += 1
return alive, color
色進化のオプション有効時、周囲の色番号の最大値を取るために8方向をチェックした際、max関数を使っています。ここでやりたいことは、最大値の取得なので、必ずしもmax関数を毎回呼ぶ必要はなさそうです。
あと他に処理を軽くできそうなものとして、以下を思いつきました。
- トーラス世界有効時、範囲チェックはなくす
- 色進化を有効にする箇所は、別にもあるので条件判定を無くす(色進化なし時のパフォーマンス低下は目をつむる)
全部の変更を織り込んだコードが以下になります。(本来は1変更ずつ効果を見るべきですが、紙面の都合で割愛します)
class GameOfLife:
def _get_around(self, x, y):
max_x, max_y = self.x, self.y
world, colors, torus = self.world, self.colors, self.torus
alive, color = 0, 0
for dx, dy in self.dirs:
next_x, next_y = x + dx, y + dy
calc = False
if torus:
# if torus option is enabled,
# top and bottom, left and right are connected.
next_x, next_y = next_x % max_x, next_y % max_y
calc = True
else:
calc = (0 <= next_x < max_x) and (0 <= next_y < max_y)
if calc:
next_color = colors[next_y][next_x]
if next_color > color:
color = next_color
if world[next_y][next_x]:
alive += 1
return alive, color
続いて、計測結果は以下のとおりでした。
2038644 function calls (2037865 primitive calls) in 14.862 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
39/1 0.001 0.000 14.936 14.936 {built-in method builtins.exec}
1 0.000 0.000 14.936 14.936 runpy.py:200(run_module)
1 0.000 0.000 14.867 14.867 runpy.py:64(_run_code)
1 0.000 0.000 14.867 14.867 __main__.py:1(<module>)
1 0.000 0.000 14.867 14.867 main.py:65(__init__)
1 0.000 0.000 14.712 14.712 main.py:246(runTests)
1 0.000 0.000 14.712 14.712 runner.py:151(run)
2/1 0.000 0.000 14.709 14.709 suite.py:83(__call__)
2/1 0.000 0.000 14.709 14.709 suite.py:102(run)
1 0.000 0.000 14.709 14.709 case.py:652(__call__)
1 0.000 0.000 14.709 14.709 case.py:558(run)
1 0.000 0.000 14.708 14.708 case.py:549(_callTestMethod)
1 0.001 0.001 14.708 14.708 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.024 0.008 14.574 4.858 game_of_life.py:75(start)
447 1.420 0.003 10.768 0.024 game_of_life.py:122(_update)
987423 1.333 0.000 9.194 0.000 game_of_life.py:156(_next_cell)
987423 7.861 0.000 7.861 0.000 game_of_life.py:169(_get_around)
450 0.016 0.000 3.769 0.008 game_of_life.py:250(display)
460 3.242 0.007 3.242 0.007 {built-in method builtins.print}
450 0.514 0.001 0.522 0.001 game_of_life.py:287(_get_world)
今回もうまく速度UP達成できたようです。
(変更前)
9938154 function calls (9937375 primitive calls) in 23.952 seconds
(変更後)
2038644 function calls (2037865 primitive calls) in 14.862 seconds
トータル処理時間が、23.952秒から14.862秒と、9.09秒(約1.6倍)の高速化でした。割愛しますが、_get_aroundメソッドも同様に高速化できてそうです。
参考のため、各変更の高速化の内訳を以下に残しておきます。
| 変更内容 | 変更前 | 変更後 | 差 | 高速化 |
|---|---|---|---|---|
| max関数なくす | 23.952秒 | 17.953秒 | -5.999秒 | 1.3倍 |
| トーラス世界の時は範囲チェックしない | 17.953秒 | 15.565秒 | -2.388秒 | 1.15倍 |
| 色進化の条件判定なくす | 15.565秒 | 14.862秒 | -0.703秒 | 1.04倍 |
ここまで順調に来ましたが、実はもう筆者的にはこれ以上のロジックの改善は厳しそうです。…が、最後の一滴を無理やり絞ってみます。
最後の一押し！
ボトルネックとなるメソッドの改善は、もう十分ってくらい考え尽くしました。悲しいですがココが筆者の限界。他に改善しても効果のある箇所はなさそうにしか見えません。…ところが、一つだけ最後の切り札を残している筆者でした。それは処理のスレッドを分けて並列処理させるという案です。
具体的には、世代交代のセルを計算する部分と、コンソールへの描画を別のスレッドに分けるというものです。(画面を分割することも考えましたが、描画は1つの方が良さそうとの事で今回はパスです)
ただこれをしてしまうと、今後のメンテや改善効果の確認に色々と支障をきたすことになってしまいます。ですので最後まで取っておきました。
いよいよ、この案を使うときが来たのかもしれません。(できればまだ取っておきたい)
以下が、コード変更した結果です。
import queue
import threading
class GameOfLife:
def start(self):
try:
# 1st screen
self.console.setup()
world, step, colors = self.world, self.step, self.colors
screen = self.console._get_screen(world, step, colors)
print(screen, end='')
if self.delay:
time.sleep(self.delay)
# thread setup
q = queue.Queue(1)
thread = threading.Thread(
target=self._update, args=(q,), daemon=True)
thread.start()
# update screen
max_step, loop = self.max_step, self.loop
while True:
self._wait()
screen, step = q.get(timeout=0.5)
print(str(screen), end='')
if step >= max_step:
if not loop:
# if loop option is enabled, game_of_life is never end.
break
except KeyboardInterrupt:
return
finally:
self.console.teardown()
def _update(self, q):
while True:
self._next_step()
world, step, colors = self.world, self.step, self.colors
screen = self.console._get_screen(world, step, colors)
q.put((screen, step))
def _next_step(self):
max_y, max_x = self.y, self.x
world, colors = self.world, self.colors
mortal, ages = self.mortal, self.ages
new_world = []
for y in range(max_y):
next_cells = []
for x in range(max_x):
if next_cell := self._next_cell(x, y):
if mortal:
# if mortal option is enabled, living cells are ageing.
if world[y][x]:
next_cell = self._ageing(x, y, next_cell)
else:
ages[y][x] = 1
else:
ages[y][x], colors[y][x] = 0, 0
next_cells += [next_cell]
new_world += [next_cells]
self.world = [[x for x in row] for row in new_world]
self.step += 1
class Console:
def _get_screen(self, world, step, colors):
screen = ''
if step > 1:
screen += self._cursor_up(self.y + 4)
screen += self._get_title()
screen += self._get_world(world, colors)
screen += self._get_step(step)
return screen
コードの概要としては、以下の役割でスレッドを分けて並列処理するよう作り変えました。
| 役割 | メソッド名 |
|---|---|
| 世代(画面)計算 | self._update() |
| 画面表示 | self.start()の無限ループ部分 |
世代計算のスレッドは、スクリーンに表示させるテキストを作成し、Queueに格納します。画面表示スレッドはQueueに格納されたテキストを取り出し、print関数で表示させます。
画面表示のIO待ちの際に、次の画面計算に移れる分、速くなるだろうという算段です。
45981 function calls (45191 primitive calls) in 11.081 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
40/1 0.001 0.000 11.083 11.083 {built-in method builtins.exec}
1 0.000 0.000 11.083 11.083 runpy.py:200(run_module)
1 0.000 0.000 11.014 11.014 runpy.py:64(_run_code)
1 0.000 0.000 11.014 11.014 __main__.py:1(<module>)
1 0.000 0.000 11.014 11.014 main.py:65(__init__)
1 0.000 0.000 10.853 10.853 main.py:246(runTests)
1 0.000 0.000 10.853 10.853 runner.py:151(run)
2/1 0.000 0.000 10.850 10.850 suite.py:83(__call__)
2/1 0.000 0.000 10.850 10.850 suite.py:102(run)
1 0.000 0.000 10.849 10.849 case.py:652(__call__)
1 0.000 0.000 10.849 10.849 case.py:558(run)
1 0.000 0.000 10.849 10.849 case.py:549(_callTestMethod)
1 0.001 0.001 10.849 10.849 test_game_of_life.py:275(test_noahs_ark_elp)
3 0.015 0.005 10.731 3.577 game_of_life.py:77(start)
460 9.521 0.021 9.521 0.021 {built-in method builtins.print}
447 0.004 0.000 0.964 0.002 game_of_life.py:213(_wait)
447 0.960 0.002 0.960 0.002 {built-in method time.sleep}
やはり、速くなりました。全体で約3.8秒(1.34倍)ほどの改善が見られました。
(変更前)
2038644 function calls (2037865 primitive calls) in 14.862 seconds
(変更後)
45981 function calls (45191 primitive calls) in 11.081 seconds
しかし残念なことに、スレッドを分けた世代計算側の内訳は表示されなくなっています。これが筆者的に、最後まで変更せず取っておきたかったポイントです。(調べてみると、メインスレッド以外もちゃんと計測する方法はあるようですが、今回はトライしきれず…)
threadingの場合、厳密には1CPUしか使わないため並行処理と呼ばれるそうです。複数CPUで並列処理を行う場合はmultiprosessingが使えるそうですが、こちらを試したところ起動までの待ち時間がネックで採用しませんでした。
スタート地点の53.831秒から始めて、11.081秒にまで到達しました。4.85倍の高速化です。ここが筆者の終着点と相成りました。
ということで、最終形の動く様子を見て下さい！
体感では随分速くなりました(gifファイルだと伝わりづらい部分がありますが)。ただ、私の目指す全画面サイズ(100x50程度)での動作には及ばなかったようです。また妙案が浮かぶまで、ここで一旦手を止めようと思います。
とはいえ、ここまで、コードを改善し効果を確認するサイクルを通して、自分のコードが良くなっていくのを見るのは実にワクワクしました。実際のところ、コードの改善案を思いつくためには、"理屈"なのか"閃き"なのかどちらが強いんでしょう…。
おわりに
ちょっとやれば誰でもできる拙い内容でしたが、ここまでお読みいただきありがとうございました。処理の速さだけを追い求めるなら、Pythonである必要もコンソール表示である必要も無くなります。ただそうすると、私が作ろうとする必要性も無くなってしまうように感じます。世の中には、私より遥かに優れたライフゲームのプログラムが、既に存在してますのでね。
それだと、とてもさみしいよなぁ…という気持ちが、今回あえての制約条件のもと、コードや本記事を書く原動力になったように思います。Pythonはシンプルさが好きで、ただなんとなくでもいいので書いていたいです。
そしてやはり、ライフゲームは面白いなと思います。代表的なパターンの1つである"グライダー"も、ぶつかり方次第で思いもよらぬ変化が生まれる、なんてことにも気付けて、益々ライフゲームの奥深さを再認識しました。(とはいえ、ガチガチのコアな内容には足元にも及ばないですがね…)
それでは最後に、"グライダー"の衝突を利用し、かつ、独自のルールを織り込んだ自作のライフゲーム、"hanabi"で締めくくります。どうもありがとうございました！！
つづき
その後、続きの記事を書きましたので、良かったら読んでください。
外部ライブラリのNumPyとOpenCVを活用して、さらなる高速化を目指した内容となっております。