Ml-tips

PuLPの高速化:総和は lpSum で書く

はじめに

PuLPで大きな問題を組むと、solve()より前のモデル構築が思いのほか遅くなることがあります。よくある原因が、目的関数や制約の総和をPython組み込みのsum()で作っているケースです。ここをpulp.lpSumに替えるだけで速くなることがあります。今回の記事ではその差を実測します。pulp==3.3.2です。

なぜ sum() は遅いのか

まずはじめになぜsumだと遅くなるのかを簡単に説明します。 PuLPの変数を足し合わせるとLpAffineExpression(項の集まり)ができます。sum()0 + x0 + x1 + ...と左から順に足しますが、+のたびにそれまでの式を丸ごとコピーした新しい式を作ります。n項を足すと1+2+…+n回のコピーが起きるので、計算量はO(n²)です。

lpSumだと項を1つの式にまとめて追加するためコピーが積み上がらずO(n)となります。

import pulp

n = 8000
x = [pulp.LpVariable(f"x{i}", lowBound=0) for i in range(n)]
c = [(i % 9) + 1 for i in range(n)]

obj_slow = sum(c[i] * x[i] for i in range(n))       # O(n^2)
obj_fast = pulp.lpSum(c[i] * x[i] for i in range(n)) # O(n)

実測

項数nを変えて、目的関数を作る時間だけを測ります。変数の生成は計測の外に置き、総和を組む部分だけをtime.perf_counterで挟みます。ばらつきを見るため各nで9回測り、全部の時間を返しておきます。

import pulp, time, statistics

def build_times(n, use_lpsum, reps=9):
    x = [pulp.LpVariable(f"x{i}", lowBound=0) for i in range(n)]
    c = [(i % 9) + 1 for i in range(n)]
    ts = []
    for _ in range(reps):
        t = time.perf_counter()
        if use_lpsum:
            obj = pulp.lpSum(c[i] * x[i] for i in range(n))
        else:
            obj = sum(c[i] * x[i] for i in range(n))
        ts.append((time.perf_counter() - t) * 1000)  # ミリ秒
    return ts

for n in [100, 250, 500, 1000, 2000, 4000, 8000]:
    s = build_times(n, use_lpsum=False)
    l = build_times(n, use_lpsum=True)
    sm, lm = statistics.median(s), statistics.median(l)
    print(f"{n:>6} {sm:8.2f}ms {lm:8.2f}ms {sm/lm:4.0f}x")
     n     sum()   lpSum()   速度差
   100      0.62ms     0.21ms    3x
   250      2.79ms     0.53ms    5x
   500     10.02ms     1.11ms    9x
  1000     36.82ms     2.21ms   17x
  2000    139.41ms     4.40ms   32x
  4000    559.91ms     8.96ms   62x
  8000   2199.85ms    18.73ms  121x

下のグラフは、点が9回の中央値、エラーバーがその最小〜最大です。エラーバーが点とほぼ重なるくらい測定は安定しています。

n=100ではどちらも1ミリ秒未満で、体感差はありません。そこからnを増やすと、sum()は傾きの急な直線(nが2倍で時間は約4倍=O(n²))、lpSumは傾きの緩い直線(nが2倍で約2倍=O(n))に分かれていきます。最初はほぼ同じでも、項数が増えるほど差が開き、8000項では120倍近くになりました。数千項を超えるあたりから、体感できるレベルで効いてきます。

solve の時間は変わるか

sum()lpSumで組んだモデルは、出来上がりは同じ問題です。同じ問題を両方の書き方で組んで解き、solve()の時間と目的値を、構築と同じ100〜8000の範囲で比べます。

一つ注意があります。PuLPの既定ソルバー(CBC)は外部プログラムを起動して解くので、同じプロセス内で続けて測ると、先に測ったほうが起動コストを被って遅く出ます。そこで1回の計測ごとにプロセスを立ち上げ直し、どちらも同じcoldな状態で測ります。まず1回分を測るスクリプトを用意します。

# solve_once.py
import sys, time, pulp

use_lpsum = sys.argv[1] == "lpsum"
n = int(sys.argv[2])
x = [pulp.LpVariable(f"x{i}", lowBound=0, upBound=10) for i in range(n)]
c = [(i % 9) + 1 for i in range(n)]
prob = pulp.LpProblem("p", pulp.LpMaximize)
prob += (pulp.lpSum(c[i]*x[i] for i in range(n)) if use_lpsum
         else sum(c[i]*x[i] for i in range(n)))
prob += (pulp.lpSum(x) if use_lpsum else sum(x)) <= n * 3
t = time.perf_counter()
prob.solve(pulp.PULP_CBC_CMD(msg=0))
print((time.perf_counter() - t) * 1000, pulp.value(prob.objective))

これをsum / lpsumそれぞれ別プロセスで何度も呼び、中央値を取ります。

import subprocess, statistics, sys

def solve_med(method, n, samples=7):
    ts = []
    for _ in range(samples):
        out = subprocess.run([sys.executable, "solve_once.py", method, str(n)],
                             capture_output=True, text=True).stdout.split()
        ts.append(float(out[0]))
    return statistics.median(ts)

for n in [100, 250, 500, 1000, 2000, 4000, 8000]:
    print(f"{n:>6} sum={solve_med('sum', n):6.1f}ms lpSum={solve_med('lpsum', n):6.1f}ms")
   100 sum=  15.9ms lpSum=  14.1ms
   250 sum=  15.7ms lpSum=  15.7ms
   500 sum=  19.1ms lpSum=  18.2ms
  1000 sum=  20.3ms lpSum=  20.0ms
  2000 sum=  25.9ms lpSum=  25.5ms
  4000 sum=  40.5ms lpSum=  41.4ms
  8000 sum=  71.8ms lpSum=  69.6ms

solveの時間は両者ほぼ同じですね。

まとめ

  • PuLPの総和をsum()で書くとO(n²)、lpSumならO(n)
  • 8000項で約120倍、モデル構築が速くなった
  • solveの時間や解そのものは変わらない(速くなるのは構築だけ)

lpSumに替えるだけなので、まず入れておいて損はないやつ。

参考リンク