使 C# 矩阵代码更快

Question

在处理一些矩阵代码时，我担心性能问题。

它的工作原理如下：我有一个 IMatrix 抽象类（包含所有矩阵操作等），由 ColumnMatrix 类实现。

abstract class IMatrix
{
    public int Rows {get;set;}
    public int Columns {get;set;}
    public abstract float At(int row, int column);
}

class ColumnMatrix : IMatrix
{
    private data[];

    public override float At(int row, int column)
    {
        return data[row + columns * this.Rows];
    }
}

这个类在我的应用程序中被大量使用，但我担心性能问题。 测试仅针对相同大小的锯齿状数组读取 2000000x15 矩阵，数组访问时间为 1359 毫秒，矩阵访问时间为 9234 毫秒：

public void TestAccess()
{
    int iterations = 10;

    int rows = 2000000;
    int columns = 15;
    ColumnMatrix matrix = new ColumnMatrix(rows, columns);
    for (int i = 0; i < rows; i++)
        for (int j = 0; j < columns; j++)
            matrix[i, j] = i + j;

    float[][] equivalentArray = matrix.ToRowsArray();

    TimeSpan totalMatrix = new TimeSpan(0);
    TimeSpan totalArray = new TimeSpan(0);

    float total = 0f;
    for (int iteration = 0; iteration < iterations; iteration++)
    {
        total = 0f;
        DateTime start = DateTime.Now;
        for (int i = 0; i < rows; i++)
            for (int j = 0; j < columns; j++)
                total = matrix.At(i, j);
        totalMatrix += (DateTime.Now - start);

        total += 1f; //Ensure total is read at least once.
        total = total > 0 ? 0f : 0f;

        start = DateTime.Now;
        for (int i = 0; i < rows; i++)
            for (int j = 0; j < columns; j++)
                total = equivalentArray[i][j];
        totalArray += (DateTime.Now - start);
    }

    if (total < 0f)
        logger.Info("Nothing here, just make sure we read total at least once.");

    logger.InfoFormat("Average time for a {0}x{1} access, matrix : {2}ms", rows, columns, totalMatrix.TotalMilliseconds);
    logger.InfoFormat("Average time for a {0}x{1} access, array : {2}ms", rows, columns, totalArray.TotalMilliseconds);
    Assert.IsTrue(true);
}

所以我的问题：如何才能使这件事更快？有什么办法可以让我的 ColumnMatrix.At 更快吗？ 干杯！

Answer 1

1. 删除抽象类 IMatrix。这是错误的，因为它不是接口，并且调用重写方法比调用final（也称为非修饰符方法）慢。
2. 您可以使用不安全的代码（指针）来获取数组的元素，而无需进行数组边界检查（更快，但更多的工作和不安全）

Answer 2

您编写的数组代码可以很容易地进行优化，因为很明显您正在顺序访问内存。这意味着 JIT 编译器可能会更好地将其转换为本机代码，从而获得更好的性能。
您没有考虑的另一件事是，内联仍然会碰巧发生，所以如果您的 At 方法（为什么顺便说一下，不使用索引器属性？）没有内联，由于使用调用和堆栈操作，您将遭受巨大的性能损失。最后，您应该考虑密封 ColumnMatrix 类，因为这将使 JIT 编译器的优化变得更加容易（call 肯定比 callvirt 更好）。

Answer 3

如果二维数组的性能好得多，那么您不使用二维数组作为类的内部存储，而不是使用具有计算索引开销的一维数组吗？

Answer 4

当您使用 DateTime.Now 来测量性能时，结果是相当随机的。时钟的分辨率约为 1/20 秒，因此您不是测量实际时间，而是测量代码中时钟滴答作响的位置。

您应该使用 Stopwatch 类，它具有更高的分辨率。

Answer 5

对于元素的每次访问，您都会执行乘法：行 + 列 * this.Rows。
您可能会看到，如果在内部也可以使用二维数组，

您还会获得额外的开销，因为该事物被抽象到类中。每次访问矩阵中的元素时，您都会执行额外的方法调用

Answer 6

对此进行更改：

interface IMatrix
{
    int Rows {get;set;}
    int Columns {get;set;}
    float At(int row, int column);
}

class ColumnMatrix : IMatrix
{
    private data[,];

    public int Rows {get;set;}
    public int Columns {get;set;}

    public float At(int row, int column)
    {
        return data[row,column];
    }
}

使用接口比抽象类更好 - 如果您需要它的通用功能，请为接口添加扩展方法。

此外，二维矩阵比锯齿状矩阵或扁平矩阵更快。

Answer 7

您可以使用并行编程来加速算法。
您可以编译此代码，并比较普通矩阵方程（MultiplyMatricesSequential 函数）和并行矩阵方程（MultiplyMatricesParallel 函数）的性能。您已经实现了该方法的性能比较函数（在 Main 函数中）。

您可以在 Visual Studio 2010 (.NET 4.0) 下编译此代码

namespace MultiplyMatrices
{
    using System;
    using System.Collections.Generic;
    using System.Collections.Concurrent;
    using System.Diagnostics;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    class Program
    {
        #region Sequential_Loop
        static void MultiplyMatricesSequential(double[,] matA, double[,] matB,
                                                double[,] result)
        {
            int matACols = matA.GetLength(1);
            int matBCols = matB.GetLength(1);
            int matARows = matA.GetLength(0);

            for (int i = 0; i < matARows; i++)
            {
                for (int j = 0; j < matBCols; j++)
                {
                    for (int k = 0; k < matACols; k++)
                    {
                        result[i, j] += matA[i, k] * matB[k, j];
                    }
                }
            }
        }
        #endregion

        #region Parallel_Loop

        static void MultiplyMatricesParallel(double[,] matA, double[,] matB, double[,] result)
        {
            int matACols = matA.GetLength(1);
            int matBCols = matB.GetLength(1);
            int matARows = matA.GetLength(0);

            // A basic matrix multiplication.
            // Parallelize the outer loop to partition the source array by rows.
            Parallel.For(0, matARows, i =>
            {
                for (int j = 0; j < matBCols; j++)
                {
                    // Use a temporary to improve parallel performance.
                    double temp = 0;
                    for (int k = 0; k < matACols; k++)
                    {
                        temp += matA[i, k] * matB[k, j];
                    }
                    result[i, j] = temp;
                }
            }); // Parallel.For
        }

        #endregion


        #region Main
        static void Main(string[] args)
        {
            // Set up matrices. Use small values to better view 
            // result matrix. Increase the counts to see greater 
            // speedup in the parallel loop vs. the sequential loop.
            int colCount = 180;
            int rowCount = 2000;
            int colCount2 = 270;
            double[,] m1 = InitializeMatrix(rowCount, colCount);
            double[,] m2 = InitializeMatrix(colCount, colCount2);
            double[,] result = new double[rowCount, colCount2];

            // First do the sequential version.
            Console.WriteLine("Executing sequential loop...");
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();

            MultiplyMatricesSequential(m1, m2, result);
            stopwatch.Stop();
            Console.WriteLine("Sequential loop time in milliseconds: {0}", stopwatch.ElapsedMilliseconds);

            // For the skeptics.
            OfferToPrint(rowCount, colCount2, result);

            // Reset timer and results matrix. 
            stopwatch.Reset();
            result = new double[rowCount, colCount2];

            // Do the parallel loop.
            Console.WriteLine("Executing parallel loop...");
            stopwatch.Start();
            MultiplyMatricesParallel(m1, m2, result);
            stopwatch.Stop();
            Console.WriteLine("Parallel loop time in milliseconds: {0}", stopwatch.ElapsedMilliseconds);
            OfferToPrint(rowCount, colCount2, result);

            // Keep the console window open in debug mode.
            Console.WriteLine("Press any key to exit.");
            Console.ReadKey();
        }


        #endregion

        #region Helper_Methods

        static double[,] InitializeMatrix(int rows, int cols)
        {
            double[,] matrix = new double[rows, cols];

            Random r = new Random();
            for (int i = 0; i < rows; i++)
            {
                for (int j = 0; j < cols; j++)
                {
                    matrix[i, j] = r.Next(100);
                }
            }
            return matrix;
        }

        private static void OfferToPrint(int rowCount, int colCount, double[,] matrix)
        {
            Console.WriteLine("Computation complete. Print results? y/n");
            char c = Console.ReadKey().KeyChar;
            if (c == 'y' || c == 'Y')
            {
                Console.WindowWidth = 180;
                Console.WriteLine();
                for (int x = 0; x < rowCount; x++)
                {
                    Console.WriteLine("ROW {0}: ", x);
                    for (int y = 0; y < colCount; y++)
                    {
                        Console.Write("{0:#.##} ", matrix[x, y]);
                    }
                    Console.WriteLine();
                }

            }
        }

        #endregion
    }

}