具有早期中止的 Haskell 并行搜索

Question

我想搜索一个列表，测试属性 X 的每个元素，然后在找到具有属性 X 的元素时返回。

这个列表非常大，可以从并行性中受益，但相对于计算时间而言， Spark 的成本相当高。 parListChunk会很棒，但是它必须搜索整个列表。

有什么方法可以写出 parListChunk 之类的东西吗？但提前中止？

这是天真的搜索代码:

hasPropertyX :: Object -> Bool

anyObjectHasPropertyX :: [Object] -> Bool
anyObjectHasPropertyX [] = False
anyObjectHasPropertyX l
| hasPropertyX (head l) == True = True
| otherwise = anyObjectHasPropertyX (tail l)

这是我第一次尝试并行:

anyObjectHasPropertyXPar [] = False
anyObjectHasPropertyXPar [a] = hasPropertyX a
anyObjectHasPropertyXPar (a:b:rest) = runEval $ do c1 <- rpar (force (hasPropertyX a))
                                                   c2 <- rpar (force (hasPropertyX b))
                                                   rseq c1
                                                   rseq c2
                                                   if (c1 == True) || (c2 == True) then return True else return (anyObjectHasPropertyXPar rest)

这确实比简单的代码运行得稍快(即使使用 -N1 ，很奇怪)，但速度并不快(它通过扩展并行计算的数量有所帮助)。我相信它并没有太大的好处，因为它必须为列表中的每个元素触发一个线程。

有没有类似于 parListChunk 的方法那只会触发n个线程并允许提前中止？

编辑:我在考虑这个问题时遇到了问题，因为我似乎需要监视所有线程的返回值。如果我省略 rseq的并且有类似的东西

if (c1 == True) || (c2 == True) then ...

运行时环境是否足够智能以监视两个线程并在其中一个返回时继续？

Answer 1

我认为您使用 Control.Parallel.Strategies 不会有太大的运气。 .该模块的一个关键特性是它表达了“确定性并行性”，因此程序的结果不受并行评估的影响。您描述的问题基本上是不确定的，因为线程正在竞相寻找第一个匹配项。

更新:我现在看到你只返回 True如果找到该元素，则所需的行为在技术上是确定的。所以，也许有办法欺骗 Strategies模块投入工作。不过，下面的实现似乎满足要求。

这是一个并行查找的实现 parFind使用 Control.Concurrent 在 IO monad 中运行原语，似乎做你想做的事。两个MVars使用:runningV记录有多少线程仍在运行，以允许最后一个线程检测搜索失败；和 resultV用于返回 Just结果或Nothing当最后一个线程检测到搜索失败时。请注意，与这个玩具示例不同，除非测试(您的 hasPropertyX 以上)比列表遍历工作量大得多，否则它的性能不太可能比单线程实现更好。

import Control.Monad
import Control.Concurrent
import Data.List
import System.Environment

-- Thin a list to every `n`th element starting with index `i`
thin :: Int -> Int -> [a] -> [a]
thin i n = unfoldr step . drop i
  where step [] = Nothing
        step (y:ys) = Just (y, drop (n-1) ys)

-- Use `n` parallel threads to find first element of `xs` satisfying `f`
parFind :: Int -> (a -> Bool) -> [a] -> IO (Maybe a)
parFind n f xs = do
  resultV <- newEmptyMVar
  runningV <- newMVar n
  comparisonsV <- newMVar 0
  threads <- forM [0..n-1] $ \i -> forkIO $ do
    case find f (thin i n xs) of
      Just x -> void (tryPutMVar resultV (Just x))
      Nothing -> do m <- takeMVar runningV
                    if m == 1
                      then void (tryPutMVar resultV Nothing)
                      else putMVar runningV (m-1)
  result <- readMVar resultV
  mapM_ killThread threads
  return result

myList :: [Int]
myList = [1..1000000000]

-- Use `n` threads to find first element equal to `y` in `myList`
run :: Int -> Int -> IO ()
run n y = do x <- parFind n (== y) myList
             print x

-- e.g.,  stack ghc -- -O2 -threaded SearchList.hs
--        time ./SearchList +RTS -N4 -RTS 4 12345  # find 12345 using 4 threads -> 0.018s
--        time ./SearchList +RTS -N4 -RTS 4 -1     # full search w/o match -> 6.7s
main :: IO ()
main = do [n,y] <- getArgs
          run (read n) (read y)

另外，请注意，此版本在交错的子列表上运行线程，而不是将主列表分成连续的 block 。我这样做是因为(1)更容易证明“早期”元素很快被发现； (2) 我的巨大列表意味着如果需要将整个列表保存在内存中，内存使用量可能会激增。

事实上，这个例子有点像一个性能定时炸弹——它的内存使用是不确定的，如果一个线程远远落后，它可能会爆炸，因此整个列表的大部分需要保存在内存中。

在一个真实世界的例子中，整个列表可能保存在内存中并且属性测试很昂贵，您可能会发现将列表分成 block 更快。