Haskell 中可变非整数类型的可变列表

Question

我正在尝试从二进制中解析一个巨大的复杂值 3d 数据数组。稍后这应该变成 l 矩阵 (n x m)。因为我要处理这些矩阵，所以我仅限于矩阵库——hmatrix 似乎很有前途。数据布局不是我要求的格式，所以我必须在 (i,j,k) -> (k,i,j) 位置跳转，其中 i 和 j 是 n 和 m 的元素和 l.

的 k 元素

我认为阅读本文的唯一方法是使用可变变量，否则我将得到数 TB 的垃圾。我的想法是使用盒装互数组或互矩阵向量（来自 Numeric.LinearAlgebra.Devel 的 STMatrix），所以我最终得到类似的东西：

data MVector s (STMatrix s t)

但我不确定如何正确使用它们：我可以使用 modify:

修改 MVector 的一个元素

modify :: PrimMonad m => MVector (PrimState m) a -> (a -> a) -> Int -> m ()

或者使用modifyM（奇怪：栈中vector-0.12.3.0没有modifyM...）

modifyM :: PrimMonad m => MVector (PrimState m) a -> (a -> m a) -> Int -> m ()

所以我可以使用函数调用 (a -> a) 到 runST 例程来修改 SMatrix。我不确定是否应该在 IO 中放置 ST (?)

尽管如此 - 我认为，这应该有效，但只是有用，当我想修改整个矩阵时，调用这个 (a->a)-routine n x m x l- 次会有点开销（也许它会被优化掉......）。所以我最终会在编组数组时通过指针修改内容 (i,j,k) -> (k,i,j) 并逐个读取 Matrix 的所有内容 - 但这感觉不对，我想避免这种肮脏的技巧。

您有什么想法可以让这件事变得更...干净一点吗？泰

编辑： 感谢 K. A. Buhr。他的解决方案到目前为止有效。现在，我只运行了解一些性能影响。如果我比较解决方案：

{-# LANGUAGE BangPatterns #-}
module Main where
import Data.List
import Numeric.LinearAlgebra
import qualified Data.Vector as V
import qualified Data.Vector.Storable as VS
import qualified Data.Vector.Storable.Mutable as VSM

-- Create an l-length list of n x m hmatrix Matrices
toMatrices :: Int -> Int -> Int -> [C] -> [Matrix C]
toMatrices l n m dats = map (reshape m) $ VS.createT $ do
  mats <- V.replicateM l $ VSM.unsafeNew (m*n)
  sequence_ $ zipWith (\(i,j,k) x ->
      VSM.unsafeWrite (mats V.! k) (loc i j) x) idxs (dats ++ repeat 0)
  return $ V.toList mats

  where idxs = (,,) <$> [0..n-1] <*> [0..m-1] <*> [0..l-1]
        loc i j = i*m + j

test1 = toMatrices 1000 1000 100 (fromIntegral <$> [1..])

main = do
  let !a = test1
  print "done"

使用最简单的 C 代码：

#include <stdlib.h>
#include <stdio.h>
void main() 
{

    const int n = 1000;
    const int m = 1000;
    const int l = 100;

    double *src = malloc(n*m*l * sizeof(double));
    for (int i = 0; i < n*m*l; i++) {
        src[i] = (double)i;
    }

    double *dest = malloc(n*m*l * sizeof(double));
    for (int i = 0; i < n; i++) {
        for (int j = 0; j < m; j++) {
            for (int k = 0; k < l; k++) {
                dest[k*n*m+i*m+j] = src[i*m*l+j*l+k];
            }
        }
    }
    printf("done: %f\n", dest[n*m*l - 1]); // Need to access the array, otherwise it'll get lost by -O2
    free(src);
    free(dest);
}

两者都使用 -O2 进行编译，给出以下性能猜测：

real    0m5,611s
user    0m14,845s
sys 0m2,759s

对比

real    0m0,441s
user    0m0,200s
sys 0m0,240s

这是每核性能大约 2 个数量级。从分析中我了解到

      VSM.unsafeWrite (mats V.! k) (loc i j) x

是昂贵的功能。由于我将在类似分钟的间隔内使用此过程，因此我希望将解析时间保持在与磁盘访问时间一样短的时间内。我看看能不能加快速度

PS：这是为了一些测试，如果我可以将通常的 DSP 从 C-like 移动到 Haskell

编辑 2 : 好的，这是我尝试求和后得到的结果：

{-# LANGUAGE BangPatterns #-}

module Main where

import Data.List
import qualified Data.Vector as V
import qualified Data.Vector.Storable as VS
import qualified Data.Vector.Storable.Mutable as VSM
import Numeric.LinearAlgebra

-- Create an l-length list of n x m hmatrix Matrices
toMatrices :: Int -> Int -> Int -> VS.Vector C -> V.Vector (Matrix C)
toMatrices l n m dats =
  V.map (reshape m) newMat
   where
    newMat = VS.createT $
      V.generateM l $ \k -> do
      curMat <- VSM.unsafeNew (m * n)
      VS.mapM_
        (\i ->
           VS.mapM_
             (\j -> VSM.unsafeWrite curMat (loc i j) (dats VS.! (oldLoc i j k)))
            idjs)
        idis
      return curMat
    loc i j = i * m + j
    oldLoc i j k = i * m * l + j * l + k
    !idis = VS.generate n (\a->a)
    !idjs = VS.generate m (\a->a)

test1 = toMatrices 100 1000 1000 arr
  where
    arr = VS.generate (1000 * 1000 * 100) fromIntegral :: VS.Vector C

main = do
  let !a = test1
  print "done"

它给出了一些关于：

real    0m1,816s
user    0m1,636s
sys 0m1,120s

，所以比 C 代码慢 ~4 倍。我想我可以忍受这个。我想，我正在用这段代码破坏向量的所有流功能。如果有任何建议以类似的速度让他们回来，我将不胜感激！

Answer 1

据我了解，您有一组“庞大”的数据，顺序为 i-大，j-中，k-小，您想要加载它变成由 k 索引的矩阵，其元素具有 i 索引的行和 j 索引的列，对吗？所以，你想要一个像这样的函数：

import Numeric.LinearAlgebra

-- load into "l" matrices of size "n x m"
toMatrices :: Int -> Int -> Int -> [C] -> [Matrix C]
toMatrices l n m dats = ...

请注意，您在上面编写了 n x m 矩阵，将 i 与 n 相关联，将 j 与 m 相关联。翻转 n 和 m 的角色会更常见，但我坚持使用你的符号，所以请注意这一点。

如果整个数据列表 [C] 可以轻松地放入内存中，您可以通过编写如下内容来实现不变：

import Data.List
import Data.List.Split
import Numeric.LinearAlgebra

toMatrices :: Int -> Int -> Int -> [C] -> [Matrix C]
toMatrices l n m = map (reshape m . fromList) . transpose . chunksOf l

这会将输入数据分成 l 大小的块，将它们转置为 l 列表，并将每个列表转换为矩阵。如果有某种方法可以并行强制所有 Matrix C 值，则可以通过一次遍历数据来完成，而无需保留整个列表。不幸的是，单个 Matrix C 值只能一个一个地强制执行，并且需要保留整个列表，直到可以强制执行所有这些值为止。

因此，如果“巨大的”[C] 列表对于内存来说太大了，您可能是正确的，您需要将数据加载到（部分）可变结构中。代码编写起来有些困难，但它的最终形式还算不错。我相信以下方法会起作用：

import Data.List
import Numeric.LinearAlgebra
import qualified Data.Vector as V
import qualified Data.Vector.Storable as VS
import qualified Data.Vector.Storable.Mutable as VSM

-- Create an l-length list of n x m hmatrix Matrices
toMatrices :: Int -> Int -> Int -> [C] -> [Matrix C]
toMatrices l n m dats = map (reshape m) $ VS.createT $ do
  mats <- V.replicateM l $ VSM.unsafeNew (m*n)
  sequence_ $ zipWith (\(i,j,k) x ->
      VSM.unsafeWrite (mats V.! k) (loc i j) x) idxs (dats ++ repeat 0)
  return $ V.toList mats

  where idxs = (,,) <$> [0..n-1] <*> [0..m-1] <*> [0..l-1]
        loc i j = i*m + j

test1 = toMatrices 4 3 2 (fromIntegral <$> [1..24])
test2 = toMatrices 1000 1000 100 (fromIntegral <$> [1..])

main = do
  print $ test1
  print $ norm_Inf . foldl1' (+) $ test2

用-O2编译，最大驻留约为1.6Gigs，这与在内存中保存100个100万个16字节复数值的矩阵所需的预期内存相匹配，因此看起来是正确的。

无论如何，这个版本的 toMatrices 由于使用了三种不同的向量变体而变得有些复杂。 hmatrix中有Vector，与vector中的不可变可存储VS.Vector相同； vector 还有两种类型：不可变的盒装 V.Vector 和可变的可存储 VSM.Vector.

do-block 创建了一个 V.Vector 的 VSM.Vectors 并用跨 index/value 对执行的一系列 monadic 动作填充它们。您可以通过修改 idxs 的定义以匹配数据流的顺序以任何顺序加载数据。 do-block returns 列表中的最后 VSM.Vectors，辅助函数 VS.createT 将它们全部冻结为 VS.Vectors（即 Vector 来自 hmatrix)，并且 reshape 被映射到向量中以将它们变成 m 列矩阵。

请注意，在您的实际应用程序中，您必须注意从文件中读取的数据项列表不是由 toMatrices 以外的代码保留的，无论是原始文本形式或解析的数字形式。这不应该很难做到正确，但您可能希望在将计算机锁定在真实数据集上之前测试中等规模的测试输入。

Haskell 中可变非整数类型的可变列表

Mutable list of mutabale non-integral types in Haskell

haskell

mutable

matrix

hmatrix