1

提前很抱歉这篇长篇文章，感谢任何愿意花时间完成并给我反馈的人。

我有关于列表和数组操作的性能相关问题。

我编写了一个软件来对从一组传感器收集的数据执行一些操作。为了让它运行得更快，我目前正在尝试编写一些优化。

收集的数据位于 N×M 的双精度数组中（实际上实现为类扩展List<List<double>>）。对于 i 的任何值，我们总是有this.Count() == N和this[i].Count() == M。基本上，它是一个矩形阵列。

该数组中的每个数据点都与 X-by-Y 地图上的一些点相关联。基本上，想象这是我从数据中制作的图像，以快速清晰的方式表示它。因此，对于每个数据点，都有一个List<int[]>与之相关的地图点。这个事实由 表示List<int[]>[,] pointsLocal。我还保存了一个静态文件List<int[]>[,]来存储相同的信息：这样我可以pointsLocal在一个细化循环中根据我的闲暇进行修改，并在下次调用这些方法时拥有一个新的副本。相同的传感器将始终与相同的点相关联，这就是我拥有该本地阵列的原因。有些点（实际上是大多数点）与多个传感器相关，因此出现在许多列表中。

在我的代码的其他部分，我能够正确识别出阵列的某些传感器存在问题，然后数据包含错误。这在 中表示private List<List<bool>> faultyData。如果传感器给出了错误的输出，那么我必须假设与之相关的所有点都可能受到故障的影响，因此我不关心这些地图点中进一步分析的结果。

我的代码的计算部分为每个地图点聚合来自阵列中所有传感器的数据。我想要做的是预先确定一个我不需要执行任何分析的地图点子集。

该类PointsEqualityComparer是 的自定义比较运算符int[]，我使用它是因为地图点由它们的 2D 坐标标识。

public class Sinogram : List<List<double>>
{
    //various enums
   private List<List<bool>> faultyData; //faultyData[i][j] is true if there is an error in the data
    //constructors
    //some methods
    public void dataAnalysis()
    {
        List<int[]>[,] pointsLocal = new List<int[]>[this.Count(), this[0].Count()];
        List<int[]> faultyPoints = new List<int[]>();
        //Fill pointsLocal with the correlated points from the static array
        PointsEqualityComparer myComparer = new PointsEqualityComparer();
        //Point selection parts (see later for the two different implementations)
        //Actual analysis parts (not here because it is not relevant to my question, but it works)
    }
}

比较器类如下。我已经发现该GetHashCode方法必须返回尽可能独特的结果以提高性能，因此我按照您在此代码段中看到的那样实现了它。

 public class PointsEqualityComparer : IEqualityComparer<int[]>
 {
    public bool Equals(int[] p1, int[] p2)
    {
        bool result = (p1.Count() == p2.Count()) && (p1.Count() == 2) && (p1[0] == p2[0]) && (p1[1] == p2[1]);
        return result;
    }

    public int GetHashCode(int[] obj)
    {
        return ((obj[0] + obj[1]) * (obj[0] + obj[1] + 1) + obj[1]);
    }
}

现在是棘手的部分。对于我实际选择有趣的地图点的代码部分，我有两种不同的实现。有趣的意思是我必须在其上聚合来自传感器的数据的地图点。我通过实际识别容易出错的点并将它们从列表中删除来选择它们。

在我的第一个实现中，我遍历所有地图点列表。如果相应的传感器有故障，我会将这些点添加到故障点列表中（避免重复）。一旦我遍历所有点并生成错误点的完整列表，我会allPairsLocal通过删除它们来更新。faultyPoints 的列表可能会变得相当大，特别是在某些情况下，当很多传感器报告错误时（最大理论大小超过 2000000 个元素，如果所有传感器都报告错误并且我正在尝试创建一个 1920*1080 的地图来绘制作为高清图像）

for (int i = 0; i <this.Count; i++)
{
    for (int j = 0; j < this[i].Count; j++)
    {
        if (faultyData[i][j])
        {
            faultyPoints = faultyPoints.Union<int[]>(allPairsLocal[i, j], myComparer).ToList();
        }
    }
}
for (int i = 0; i <this.Count; i++)
{
    for (int j = 0; j < this[i].Count; j++)
    {
        allPairsLocal[i, j] = allPairsLocal[i, j].Except(faultyPoints, myComparer).ToList();
    }
}

在我的第二个实现中，我尝试使用较小的 faultyPoints 列表。因此，我所做的是，对于每个报告错误的传感器，使用它的列表从所有其他传感器（以及它自己的）中删除地图点。这使列表的维度更小，但代价是更多的 annidated 循环。

for (int i = 0; i <this.Count; i++)
{
    for (int j = 0; j < this[i].Count; j++)
    {
        if (faultyData[i][j])
        {
            faultyPoints = allPairsLocal[i, j]. ToList();
            for (int x = 0; x < this.Count; x++)
            {
                for (int y = 0; y < this[x].Count; y++)
                {
                    allPairsLocal[x, y] = allPairsLocal[x, y].Except(faultyPoints, myComparer).ToList();
                }
            }
        }
    }
}

这两种实现都非常慢，我想这至少部分是因为数据集的大小。两者都比对整个数据集执行数据分析步骤花费更长的时间。有没有办法进行类似的操作但实现速度更快？有些步骤可能是平行的，但这并不会真正改变实质。是否有具有 O(1) 方法的数据结构来实现我在这里使用 Union 和 except 所做的事情？

再次感谢您阅读我的整篇文章。我很感激任何反馈，即使它不是一个完整的答案，我也很乐意澄清我能做什么。

4

2 回答 2

2

如果我理解正确的话，一旦你填充了pointsLocal数组，每个传感器都有以下内容(i,j)：

  • this[i][j]= 来自传感器的数据(i,j)
  • pointsLocal[i,j]= 传感器的地图点列表(i,j)
  • faultyData[i][j]= 如果来自传感器(i,j)的数据不正确，则为 true，否则为 false

考虑“反转”您的数据，以便给定一个地图点(x,y)，您可以有效地

  • 查明该点是否有故障（即地图点的任何传感器报告数据有故障）
  • 获取报告与地图点相关的数据的传感器列表

为此，我们可以创建一个使用您已经编写的比较器的字典。每个键是一(x,y)对（即一个int[2]），代表一个地图点；返回的值（如果有）是对该点有贡献的已知传感器列表。返回值null表示地图点被有故障的传感器“感染”，应该被忽略。如果字典中根本不存在给定的对，则意味着没有传感器对该点有贡献。

var mapPoints = new Dictionary<int[], List<int[]>)(PointsEqualityComparer);

for (int i = 0; i <this.Count; i++)
{
    for (int j = 0; j < this[i].Count; j++)
    {
        foreach (var point in pointsLocal[i,j]) 
        {
            if (faultyData[i][j])
            {
                // infected point
                mapPoints[point] = null;  
            }
            else
            {
                // Add the current sensor's indices (i,j) to the list of 
                // known sensors for the current map point

                List<int[]> sensors = null;
                if (!mapPoints.TryGetValue(point, out sensors)) 
                {
                    sensors = new List<int[]>();
                    mapPoints[point] = sensors;
                }

                // null here means that we previously determined that the
                // current map point is infected 
                if (sensors != null) 
                {
                    // Add sensor to list for this map point
                    sensors.Add(new int[] { i, j });
                }
            }
        }
    } 
}

现在您可以枚举所有地图点，将每个点分类为好或坏：

var faultyPoints = new List<int[]>();  // not sure you really need this? 
var goodPoints = new List<int[]>();
foreach (var point in mapPoints.Keys)
{
    var sensors = mapPoints[point];
    if (sensors == null)
         faultyPoints.Add(point);
    else
         goodPoints.Add(point);
}

最后，您可以为每个好的地图点枚举传感器，以进行分析：

foreach (var point in goodPoints) 
{
    var sensors = mapPoints[point]; 
    // for current point, aggregate data for each sensor listed in "sensors"
}

请注意，我没有更改allPairsLocal，因为分析步骤似乎没有必要。但是，如果您真的需要从中删除错误的地图点，您也可以有效地做到这一点：

for (int i = 0; i <this.Count; i++)
{
    for (int j = 0; j < this[i].Count; j++)
    {
        var points = allPairsLocal[i][j];
        var cleanedUp = new List<int[]>();
        foreach (var point in points) 
        {
            // Important: do NOT use 'faultyPoints' here. It will kill performance
            if (mapPoints[point] != null)
            {
               cleanedUp.Add(point); 
            }
        }
        allPairsLocal[i][j] = cleanedUp;   
    }
}

所有这一切的性能改进来自于使用字典来查找单个地图点，无论何时您需要知道它是否有故障或其贡献的传感器是什么。如果您的哈希函数良好，则查找本质上是一个恒定时间操作（摊销）。

您可以在此处进行许多优化。例如，您真的需要知道传感器索引来为每个地图点进行聚合吗？还是您只需要数据值？如果是后者，那么您的字典将是Dictionary<List<double>>. 最后，通过使用 Linq（而不是循环）进行许多枚举，可以使代码更加紧凑。

于 2013-09-18T07:58:06.943 回答
1

你是对的。这是因为 Union 和 except 操作的复杂性。
您有 N×M 传感器表（您在上面将其命名为Lists of map-points）。每个传感器都会影响一组点（您将其命名为allPairsLocal[i, j]）。并且每个点数组都是全局预定点数组 ( points on a X-by-Y map) 的子集。
如果我是对的，那么：

  1. X-by-Y 地图上的点 - 这是一个全局点数组。更重要的是，由于您可以比较点，因此您可以对它们进行排序并保持该数组排序（我的意思是实际上可能没有排序，但具有良好的读取操作复杂性）。用于Dictionary<int[], int>关键点的坐标，值的顺序索引（插入所有点后设置）。

  2. 现在我们有了一组传感器和点（让我们Dictionary<int[], int>从第 1 步开始命名为点）。我们需要构建 2 个映射 - 一个sensors2points（命名为 s2p）和points2sensors（命名为 p2s）。你有allPairsLocal作为sensors2points 并且看起来像List<int[]>[][]，即每个传感器的点坐标列表。但是我们需要将索引列表保留为每个传感器的点坐标，即转换int[]为它的顺序索引points：

    // straight and inverted mappings
var s2p = new List<int>[N*M];
var p2s = new List<List<int>>(point.Count);
//and initialize p2s inner lists
for (int i = 0; i < p2s.Count; i++)
    p2s[i] = new List<int>();

for (int i = 0; i < N * M; i++)
{
    s2p[i] = new List<int>(allPairsLocal[i/M][i%M].Count);

    //convert list of points coordinates to list of it's indices
    // and construct inverted mapping
    foreach(int[] p in allPairsLocal[i/M][i%M])
    {
        // points[p] - index of point p in Dictionary if you remember
        s2p[i].Add(points[p]);
        p2s[points[p]].Add(i);
    }            
}

我认为很明显，步骤 1 和 2 只需要在初始化时执行一次。然后选择您需要的有趣点：

//I don't know which set you need as a result - valid points or sensors so I do both

// false - correct, true - error. Initialized with false
BitArray sensorsMask = new BitArray(sensors.Count);
BitArray pointsMask = new BitArray(points.Count);

for (int i = 0; i < N * M; i++)
{
    if (faultyData[i / M][i % M])
        sensorsMask[i] = true; // means error in sensor

    foreach(int p in s2p[i])
        pointsMask[p] = true;
}

// so you can get only valid sensors
var validSensors = new List<int>();
for (int i = 0; i < N * M; i++)
    if (!sensorsMask[i])
        validSensors.Add(i);

// or only valid points
var validPoints = new List<int[]>();
foreach (var pair in points)
    if (!pointsMask[pair.Value])
        validPoints.Add(points.Key);

这可能不是很有效的方式（很难说出你到底想得到什么），但它比使用集合操作要好。我的意思是玩 mask-array vs sets。希望它会有所帮助。

于 2013-09-18T19:53:58.887 回答