改进 Numpy 中的矩阵乘法

Question

我和一些朋友正在做一个小的语言比赛来计算一些神经网络。一些在 C 中做其他在 fortran 中，而我：Python。

代码很简单，就是一堆向量点运算和一个求和，然后应用一个信号函数并返回 -1 或 1（激活或未激活）。

我们发送一堆随机数并检查（目前只有单个进程）哪种语言处理得更快。

我的代码很简单：

def sgn(h):
    """Signal function"""
    return -1 if h < 0 else 1

def lincomb(A, B):
    """Linear combinator between two matrices"""
    return np.einsum('ji,ij->', A, B)

def lincombrav(A, B):
return A.ravel().dot(B.ravel('F'))

def functional_test():
    w1 = np.random.random(50**2).reshape(50,50)
    w2 = np.random.random(50**2).reshape(50,50)
    return sgn(lincombrav(w1, w2))

其中 A 和 B 是表示神经网络中每一层的矩阵。然后我们将第一个矩阵的第 i 列与第二个矩阵的第 i 行点在一起，对所有结果求和并发送到信号函数。比如：

w1 = 2*np.random.random(100**2).reshape(100,100)-1
w2 = 2*np.random.random(100**2).reshape(100,100)-1

然后我们计时

%timeit sgn(lincomb(w1, w2))

Python 以 38 倍的速度输给 Fortran :-(

有没有办法改进那个 Python“代码”。

编辑：添加 timeit 结果：

Python 版本（已经有ravel 模式）

In [10]: %timeit functional_test()
8.72 µs ± 406 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

Python 版本（带有einsum）

In [16]: %timeit functional_test()
10.27 µs ± 490 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

Fortran 版本

In [13]: %timeit fort.test()
235 ns ± 12.7 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

Fortran 版本是使用“f2py”程序创建的，用于从 fortran 代码生成 python 可加载模块。

测试函数执行以下操作（在每种语言中）：

1. 创建矩阵 A
2. 创建矩阵 B
3. 调用 sgn(lincomb(A,B)) # 从每个相应的语言实现

我还将矩阵创建移到了外部，只运行数学运算而不是处理内存。尽管如此，python 还是落后了。

EDIT2：好蟒蛇消息。 Python 在除小矩阵测试之外的所有测试中都获胜。这里将遵循整个代码：

Python 函数 (bla.py)

import numpy as np
from numba import jit
import timeit
import matplotlib.pyplot as plt

def sgn(h):
    """Signal function"""
    return -1 if h < 0 else 1

def lincomb(A, B):
    """Linear combinator between two matrices"""
    return np.einsum('ji,ij->', A, B)

def lincombrav(A, B):
    return A.ravel().dot(B.ravel('F'))

def functional_test_ravel(n):
    """Functional tests (Victor experiment)"""

    w = 2*np.random.random(n**2).reshape(n,n)-1
    x = 2*np.random.random(n**2).reshape(n,n)-1

    return sgn(lincombrav(w, x))

def functional_test_einsum(n):
    """Functional tests (Victor experiment)"""

    w = 2*np.random.random(n**2).reshape(n,n)-1
    x = 2*np.random.random(n**2).reshape(n,n)-1

    return  sgn(lincomb(w, x))

@jit()
def functional_test_numbaein(n):
    """Functional tests (Victor experiment)"""

    w = 2*np.random.random(n**2).reshape(n,n)-1
    x = 2*np.random.random(n**2).reshape(n,n)-1

    return sgn(lincomb(w, x))


@jit()
def functional_test_numbarav(n):
    """Functional tests (Victor experiment)"""

    w = 2*np.random.random(n**2).reshape(n,n)-1
    x = 2*np.random.random(n**2).reshape(n,n)-1

    return sgn(lincombrav(w, x))

Fortran 函数 (fbla.f95)

module fbla
    implicit none
    integer, parameter::dp = selected_real_kind(12,100)
    public

contains

    real(kind=dp) function sgn(x)
        integer, parameter::dp = selected_real_kind(12,100)
        real(kind=dp), intent(in):: x

        if(x >= 0.0 ) then
            sgn = +1.0   
        else if (x < 0.0) then
            sgn = -1.0 
        end if
    end function sgn

    real(kind=dp) function lincomb(A, B, n)
        integer, parameter :: sp = selected_int_kind(r=8)
        integer, parameter :: dp = selected_real_kind(12,100)

        integer(kind=sp) :: i
        integer(kind=sp), intent(in):: n
        real(kind=DP), intent(in) :: A(n,n)
        real(kind=DP), intent(in) :: B(n,n)

        lincomb = 0
        do i=1,n
            lincomb = lincomb + dot_product(A(:,i),B(i,:))
        end do
    end function lincomb

    real(kind=dp) function functional_test(n)
        integer, parameter::dp = selected_real_kind(12,100)
        integer, parameter::sp = selected_int_kind(r=8)

        integer(kind=sp), intent(in):: n
        integer(kind=sp):: i, j
        real(kind=dp), allocatable, dimension(:,:):: x, w, wt   

        ALLOCATE(wt(n,n),w(n,n),x(n,n))

        do i=1,n
            do j=1,n
                w(i,j) = 2*rand(0)-1
                x(i,j) = 2*rand(0)-1
            end do
        end do

        wt = transpose(w)
        functional_test = sgn(lincomb(wt, x, n))
    end function functional_test

end module fbla

测试执行函数（tests.py）

import numpy as np
import timeit
import matplotlib.pyplot as plt
import bla
from fbla import fbla

def run_test(test_functions, N, runs=1000):
    results = []
    global rank
    for n in N:
        rank = n
        for t in test_functions:
            # print(f'Rank {globals()["rank"]}')
            print(f'Running {t} to matrix size {rank}', end='')
            r = min(timeit.Timer(t , globals=globals()).repeat(repeat=5, number=runs))
            print(f' total time {r} per run {r/runs}')
            results.append((t, n, r, r/runs))

    return results


def plotbars(results, test_functions, N):
    Nsz = len(N)
    M = len(test_functions)

    fig, ax = plt.subplots()

    ind = np.arange(int(Nsz))
    width = 1/(M+1)

    p = []
    for n in range(M):
        g = [ w*1000 for (x,y,z,w) in results if x==test_functions[n]]
        p.append(ax.bar(ind+n*width, g, width, bottom=0))

    ax.legend([ l[0] for l in p ], test_functions)
    ax.set_xticks(ind-width/2+((M/2) * width))
    ax.set_xticklabels(np.array(N).astype(str))
    ax.set_xlabel('Rank of square random matrix')
    ax.set_ylabel('Average time(ms) per run')
    ax.set_yscale('log')

    return fig

N = (10, 50, 100, 1000)
test_functions = [ 
        'bla.functional_test_einsum(rank)', 
        'fbla.functional_test(rank)'
]

results = run_test(test_functions, N)
plot = plotbars(results, test_functions, N)
plot.show()

结果是：

[('bla.functional_test_einsum(rank)', 10, 0.023221354000270367, 2.3221354000270368e-05),
 ('fbla.functional_test(rank)', 10, 0.005375514010665938, 5.375514010665938e-06),
 ('bla.functional_test_einsum(rank)', 50, 0.07035048000398092, 7.035048000398091e-05),
 ('fbla.functional_test(rank)', 50, 0.1242617039824836, 0.0001242617039824836),
 ('bla.functional_test_einsum(rank)', 100, 0.22694124400732107, 0.00022694124400732108),
 ('fbla.functional_test(rank)', 100, 0.5518505079962779, 0.0005518505079962779),
 ('bla.functional_test_einsum(rank)', 1000, 37.88827919398318, 0.03788827919398318),
 ('fbla.functional_test(rank)', 1000, 74.09929457501858, 0.07409929457501857)]

来自ipython3 会话的一些标准timeit 输出。 fbla 是 fortran 库，bla 是标准 python 库。

In : n=1000
In : w1 = 2*np.random.random(n**2).reshape(n,n)-1
In : w2 = 2*np.random.random(n**2).reshape(n,n)-1

In : bla.sgn(bla.lincomb(w1,w2))
Out: -1

In : fbla.sgn(fbla.lincomb(w1,w2))
Out: -1.0

In : %timeit fbla.sgn(fbla.lincomb(w1,w2))
11.3 ms ± 430 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

In : %timeit bla.sgn(bla.lincomb(w1,w2))
3.81 ms ± 573 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

Answer 1

我们可以通过matrix-multiplication 改进一点-

sgn(w1.ravel().dot(w2.ravel('F')))

Answer 2

如果您希望 Numpy 更快，请使用更快的 Numpy。尝试卸载 Numpy 并安装英特尔优化版本的 Numpy。英特尔优化的 Numpy 版本包括许多 CPU 级别的优化，这些优化应该会显着提高运算性能，例如使用英特尔 CPU 的机器上的矩阵乘法。

pip uninstall numpy
pip install intel-numpy