最好的方式“循环2
我发现为Haskell修复数组库非常有趣，并且想要制作一个简单的程序，试图了解如何使用它。 我也使用列表做了一个简单的实现，事实证明它更快。 我的主要问题是如何改进下面的Repa代码，使其成为最高效的（并且希望也是非常可读的）。 我使用Haskell很新，在Repa上找不到任何容易理解的教程[ 编辑 Haskell Wiki中有一个，当我写这个时我忘记了它]，所以不要以为我知道任何东西。 :)例如，我不确定何时使用强制或deepSeqArray。
该程序用于以下列方式近似计算球体的体积：
下面显示了两个版本，一个使用列表，另一个使用repa。 我知道代码效率很低，特别是对于这个用例，但是这个想法是让它在以后变得更加复杂。
对于下面的值，并使用“ghc -Odph -fllvm -fforce-recomp -rtsopts -threaded”进行编译，列表版本需要1.4秒，而重新版本需要12秒+ RTS -N1和10秒+ RTS - N2，虽然没有火花转换（我有一个双核英特尔机器（Core 2 Duo E7400 @ 2.8 GHz）运行Windows 7 64，GHC 7.0.2和llvm 2.8）。 （在下面的主要部分注释掉正确的行来运行其中一个版本。）
感谢您的任何帮助！
import Data.Array.Repa as R
import qualified Data.Vector.Unboxed as V
import Prelude as P
-- Calculate the volume of a sphere by putting it in a bath of coordinates. Generate coordinates (x,y,z) in a cuboid. Then, for each coordinate, check if it is inside the sphere. Sum those coordinates and multiply by the coordinate grid step size to find an approximate volume.
particles = [(0,0,0)] -- used for the list alternative --[(0,0,0),(0,2,0)]
particles_repa = [0,0,0::Double] -- used for the repa alternative, can currently just be one coordinate
-- Radius of the particle
a = 4
-- Generate the coordinates. Could this be done more efficiently, and at the same time simple? In Matlab I would use ndgrid.
step = 0.1 --0.05
xrange = [-10,-10+step..10] :: [Double]
yrange = [-10,-10+step..10]
zrange = [-10,-10+step..10]
-- All coordinates as triples. These are used directly in the list version below.
coords = [(x,y,z) | x <- xrange, y <- yrange, z <- zrange]
---- List code ----
volumeIndividuals = fromIntegral (length particles) * 4*pi*a**3/3
volumeInside = step**3 * fromIntegral (numberInsideParticles particles coords)
numberInsideParticles particles coords = length $ filter (==True) $ P.map (insideParticles particles) coords
insideParticles particles coord = any (==True) $ P.map (insideParticle coord) particles
insideParticle (xc,yc,zc) (xp,yp,zp) = ((xc-xp)^2+(yc-yp)^2+(zc-zp)^2) < a**2
---- End list code ----
---- Repa code ----
-- Put the coordinates in a Nx3 array.
xcoords = P.map ((x,_,_) -> x) coords
ycoords = P.map ((_,y,_) -> y) coords
zcoords = P.map ((_,_,z) -> z) coords
-- Total number of coordinates
num_coords = (length xcoords) ::Int
xcoords_r = fromList (Z :. num_coords :. (1::Int)) xcoords
ycoords_r = fromList (Z :. num_coords :. (1::Int)) ycoords
zcoords_r = fromList (Z :. num_coords :. (1::Int)) zcoords
rcoords = xcoords_r R.++ ycoords_r R.++ zcoords_r
-- Put the particle coordinates in an array, then extend (replicate) this array so that its size becomes the same as that of rcoords
particle = fromList (Z :. (1::Int) :. (3::Int)) particles_repa
particle_slice = slice particle (Z :. (0::Int) :. All)
particle_extended = extend (Z :. num_coords :. All) particle_slice
-- Calculate the squared difference between the (x,y,z) coordinates of the particle and the coordinates of the cuboid.
squared_diff = deepSeqArrays [rcoords,particle_extended] ((force2 rcoords) -^ (force2 particle_extended)) **^ 2
(**^) arr pow = R.map (**pow) arr
xslice = slice squared_diff (Z :. All :. (0::Int))
yslice = slice squared_diff (Z :. All :. (1::Int))
zslice = slice squared_diff (Z :. All :. (2::Int))
-- Calculate the distance between each coordinate and the particle center
sum_squared_diff = [xslice,yslice,zslice] `deepSeqArrays` xslice +^ yslice +^ zslice
-- Do the rest using vector, since I didn't get the repa variant working.
ssd_vec = toVector sum_squared_diff
-- Determine the number of the coordinates that are within the particle (instead of taking the square root to get the distances above, I compare to the square of the radius here, to improve performance)
total_within = fromIntegral (V.length $ V.filter (<a**2) ssd_vec)
--total_within = foldAll (x acc -> if x < a**2 then acc+1 else acc) 0 sum_squared_diff
-- Finally, calculate an approximation of the volume of the sphere by taking the volume of the cubes with side step, multiplied with the number of coordinates within the sphere.
volumeInside_repa = step**3 * total_within
-- Helper function that shows the size of a 2-D array.
rsize = reverse . listOfShape . (extent :: Array DIM2 Double -> DIM2)
---- End repa code ----
-- Comment out the list or the repa version if you want to time the calculations separately.
main = do
putStrLn $ "Step = " P.++ show step
putStrLn $ "Volume of individual particles = " P.++ show volumeIndividuals
putStrLn $ "Volume of cubes inside particles (list) = " P.++ show volumeInside
putStrLn $ "Volume of cubes inside particles (repa) = " P.++ show volumeInside_repa
编辑 ：一些背景解释了为什么我写了上面的代码：
我主要在Matlab中编写代码，而且我的性能改进经验主要来自该领域。 在Matlab中，您通常希望使用直接在矩阵上运行的函数进行计算，以提高性能。 在Matlab R2010b中，我使用下面显示的矩阵版本执行上述问题0.9秒，使用嵌套循环执行15秒。 虽然我知道Haskell与Matlab非常不同，但我希望从使用列表到使用Haskell中的Repa数组可以提高代码的性能。 从列表转换 - > Repa arrays->载体在那里，因为我不够熟练，以更好地替换它们。 这就是为什么我要求输入。 :)因此，上面的时间数字是主观的，因为它可能比我的表现更能衡量我的表现，但这对我来说是一个有效的指标，因为决定我要使用什么取决于我是否可以制作它工作与否。
tl; dr：我明白我上面的Repa代码可能是愚蠢的或病态的，但这是我现在可以做的最好的。 我希望能够写出更好的Haskell代码，并且我希望你能帮助我在这个方向上（已经做过）。 :)
function archimedes_simple()
particles = [0 0 0]';
a = 4;
step = 0.1;
xrange = [-10:step:10];
yrange = [-10:step:10];
zrange = [-10:step:10];
[X,Y,Z] = ndgrid(xrange,yrange,zrange);
dists2 = bsxfun(@minus,X,particles(1)).^2+ ...
bsxfun(@minus,Y,particles(2)).^2+ ...
bsxfun(@minus,Z,particles(3)).^2;
inside = dists2 < a^2;
num_inside = sum(inside(:));
disp('');
disp(['Step = ' num2str(step)]);
disp(['Volume of individual particles = ' num2str(size(particles,2)*4*pi*a^3/3)]);
disp(['Volume of cubes inside particles = ' num2str(step^3*num_inside)]);
end
编辑2 ：更快，更简单的Repa代码版本
我现在读了一些关于Repa的内容，并且想了一下。 以下是一个新的修复版本。 在这种情况下，我使用Repa扩展函数从值列表中创建x，y和z坐标作为三维数组（类似于ndgrid在Matlab中的工作方式）。 然后我映射这些阵列来计算到球形粒子的距离。 最后，我折叠得到的三维距离数组，计算球体内有多少个坐标，然后乘以一个常数因子以得到近似的体积。 我的算法实现现在更类似于上面的Matlab版本，并且不再有任何向量转换。
新版本在我的计算机上运行约5秒钟，这是从上面的相当大的改进。 如果我在编译时使用“线程”，与“+ RTS -N2”结合使用，但是线程版本在计算机的两个内核中使用最大值，那么时间是相同的。 不过，我确实看到了几秒钟的“-N2”运行到3.1秒，但以后不能再生成它们。 也许它对于同时运行的其他进程非常敏感？ 基准测试时，我已经关闭了计算机上的大部分程序，但仍有一些程序正在运行，例如后台进程。
如果我们使用“-N2”并添加运行时开关以关闭并行GC（-qg），则时间一直减少到约4.1秒，并使用-qa“使用操作系统设置线程关联（实验）”，时间缩短到约3.5秒。 用“+ RTS -s”查看运行程序的输出，使用-qg执行的GC更少。
今天下午，我会看看我是否可以在8核电脑上运行代码，只是为了好玩。 :)
import Data.Array.Repa as R
import Prelude as P
import qualified Data.List as L
-- Calculate the volume of a spherical particle by putting it in a bath of coordinates. Generate coordinates (x,y,z) in a cuboid. Then, for each coordinate, check if it is inside the sphere. Sum those coordinates and multiply by the coordinate grid step size to find an approximate volume.
particles :: [(Double,Double,Double)]
particles = [(0,0,0)]
-- Radius of the spherical particle
a = 4
volume_individuals = fromIntegral (length particles) * 4*pi*a^3/3
-- Generate the coordinates.
step = 0.1
coords_list = [-10,-10+step..10] :: [Double]
num_coords = (length coords_list) :: Int
coords :: Array DIM1 Double
coords = fromList (Z :. (num_coords ::Int)) coords_list
coords_slice :: Array DIM1 Double
coords_slice = slice coords (Z :. All)
-- x, y and z are 3-D arrays, where the same index into each array can be used to find a single coordinate, e.g. (x(i,j,k),y(i,j,k),z(i,j,k)).
x,y,z :: Array DIM3 Double
x = extend (Z :. All :. num_coords :. num_coords) coords_slice
y = extend (Z :. num_coords :. All :. num_coords) coords_slice
z = extend (Z :. num_coords :. num_coords :. All) coords_slice
-- Calculate the squared distance from each coordinate to the center of the spherical particle.
dist2 :: (Double, Double, Double) -> Array DIM3 Double
dist2 particle = ((R.map (squared_diff xp) x) + (R.map (squared_diff yp) y) + (R.map ( squared_diff zp) z))
where
(xp,yp,zp) = particle
squared_diff xi xa = (xa-xi)^2
-- Count how many of the coordinates are within the spherical particle.
num_inside_particle :: (Double,Double,Double) -> Double
num_inside_particle particle = foldAll (acc x -> if x<a^2 then acc+1 else acc) 0 (force $ dist2 particle)
-- Calculate the approximate volume covered by the spherical particle.
volume_inside :: [Double]
volume_inside = P.map ((*step^3) . num_inside_particle) particles
main = do
putStrLn $ "Step = " P.++ show step
putStrLn $ "Volume of individual particles = " P.++ show volume_individuals
putStrLn $ "Volume of cubes inside each particle (repa) = " P.++ (P.concat . ( L.intersperse ", ") . P.map show) volume_inside
-- As an alternative, y and z could be generated from x, but this was slightly slower in my tests (~0.4 s).
--y = permute_dims_3D x
--z = permute_dims_3D y
-- Permute the dimensions in a 3-D array, (forward, cyclically)
permute_dims_3D a = backpermute (swap e) swap a
where
e = extent a
swap (Z :. i:. j :. k) = Z :. k :. i :. j
新代码的空间分析
与唐斯图尔特在下面创建的相同类型的配置文件，但用于新的Repa代码。
代码审查笔记
rsize不使用（看起来很贵！） (**)所有用途都可以用廉价(^)替代Int 。 any (==True)与or相同 时间分析
COST CENTRE MODULE %time %alloc
squared_diff Main 25.0 27.3
insideParticle Main 13.8 15.3
sum_squared_diff Main 9.8 5.6
rcoords Main 7.4 5.6
particle_extended Main 6.8 9.0
particle_slice Main 5.0 7.6
insideParticles Main 5.0 4.4
yslice Main 3.6 3.0
xslice Main 3.0 3.0
ssd_vec Main 2.8 2.1
**^ Main 2.6 1.4
显示你的函数squared_diff有点可疑：
squared_diff :: Array DIM2 Double
squared_diff = deepSeqArrays [rcoords,particle_extended]
((force2 rcoords) -^ (force2 particle_extended)) **^ 2
尽管我没有看到任何明显的修复。
空间分析
空间配置文件中没有什么太神奇的：你清楚地看到列表阶段，然后是矢量阶段。 列表阶段分配很多，这会被回收。
按类型分解堆，最初我们看到很多列表和元组被分配（按需），然后分配并保存大块数组：
再次，有点我们期望看到的......数组的东西并不比列表代码分配更多（实际上，总体来说少一点），但它只需要花费更长的时间来运行。
使用保持器分析检查空间泄漏：
这里有一些有趣的事情，但没有什么惊人的。 zcoords被保留用于列表程序执行的长度，然后一些数组（SYSTEM）被分配用于重新运行。
检查核心
所以在这一点上，我首先假设你确实在列表和数组中实现了相同的算法（即在数组情况下没有做额外的工作），并且没有明显的空间泄漏。 所以我的怀疑是严重优化的维修代码。 让我们看看核心（用ghc-core。
内联所有的CAF
我在所有顶层数组定义中添加了内联编译指示，希望删除一些CAfs，并让GHC更难以优化数组代码。 这确实使GHC很难编译模块（在工作时分配高达4.3G和10分钟）。 这对我来说是一个线索，即GHC之前无法很好地优化这个程序，因为当我增加阈值时，有新的东西可以做。
操作
我改变了代码来强制rcoords和particle_extended ，并且发现我们直接在他们中间失去了大部分时间：
COST CENTRE MODULE %time %alloc
rcoords Main 32.6 34.4
particle_extended Main 21.5 27.2
**^ Main 9.8 12.7
对此代码的最大改进显然是以更好的方式生成这两个不变的输入。
请注意，这基本上是一种懒惰的流式传输算法，并且在这种情况下，时间浪费是一次性分配至少两个24361803元素数组的沉没成本，然后可能至少分配一次或两次或放弃共享。 这个代码的最好的例子，我认为，用一个非常好的优化器和一个zillion重写规则，将大致匹配列表版本（它也可以非常容易地并行化）。
我认为贝恩是合适的。 会对这个基准感兴趣，但是我绝对怀疑这对于一个严格的数组库来说不是一个很好的用例，我怀疑matlab是隐藏了它的ngrid函数背后的一些聪明的优化（优化，我会授予，它可能是有用的端口来维修）。]
编辑：
这是一个快速和肮脏的方式来并行列表代码。 导入Control.Parallel.Strategies ，然后将numberInsideParticles写为：
numberInsideParticles particles coords = length $ filter id $
withStrategy (parListChunk 2000 rseq) $ P.map (insideParticles particles) coords
随着我们扩展内核（一个内核的12s到8个内核的3.7s），这显示了很好的加速，但火花创建的开销意味着即使是8个内核，我们也只能匹配单核非平行版本。 我尝试了一些替代策略，并得到了类似的结果。 再次，我不确定我们可能做得比单线程列表版本好多少。 由于每个粒子的计算都很便宜，我们主要强调分配而不是计算。 在我看来，这样的大赢家可能是矢量化计算，而且据我所知，这非常需要手工编码。
还要注意的是，并行版本花费大约70％的时间用于GC，而单核版本花费其时间的1％（即，尽可能分配的资源被有效地融合）。
我已经添加了一些关于如何优化Repa程序到Haskell wiki的建议：http://www.haskell.org/haskellwiki/Numeric_Haskell:_A_Repa_Tutorial#Optimising_Repa_programs
链接地址: http://www.djcxy.com/p/59969.html上一篇: Best way of "looping over a 2
下一篇: Publishing Amazon AWS credentials for account with 1 permission