Я реализую программу на Haskell, которая сравнивает каждую строку файла с каждой другой строкой в файле.Который может быть реализован однопоточным следующим образом
distance :: Int -> Int -> Int
distance a b = (a-b)*(a-b)
sumOfDistancesOnSmallFile :: FilePath -> IO Int
sumOfDistancesOnSmallFile path = do
fileContents <- readFile path
return $ allDistances $ map read $ lines $ fileContents
where
allDistances (x:xs) = (allDistances xs) + ( sum $ map (distance x) xs)
allDistances _ = 0
Это будет выполнено за O (n ^ 2) времени и должно хранить полный список целых чисел в памяти все время.В моей настоящей программе строка содержит больше чисел, из которых я создаю немного более сложный тип данных, чем Int.Это дало мне нехватку памяти для данных, которые я должен обработать.
Таким образом, необходимо внести два улучшения в вышеупомянутое однопоточное решение.Во-первых, ускорить фактическое время работы.Во-вторых, найдите способ не хранить весь список в памяти все время.Я знаю, что это требует анализа всего файла n раз.Таким образом, будет выполнено O (n ^ 2) сравнений и O (n ^ 2) строк проанализировано.Это нормально для меня, так как я предпочел бы медленную успешную программу, чем неудачную.Когда входной файл достаточно мал, я всегда могу найти более простую версию.
Чтобы использовать несколько ядер процессора, я взял реализацию Mapreduce из Real World Haskell (глава 24, доступна здесь ).
Я изменил функцию чанкинга из книги, чтобы вместо деления всего файла на чанки возвращать столько чанков, сколько строк, причем каждый чанк представляет один элемент
tails . lines . readFile
, потому что ячтобы программа была также масштабируемой по размеру файла, я изначально использовал lazy IO .Это, однако, терпит неудачу с «Слишком много открытых файлов», о котором я спросил в предыдущем вопросе (дескрипторы файлов были удалены GC слишком поздно).Полная ленивая версия ввода-вывода размещена там.
Как объясняет принятый ответ, строгий ввод-вывод может решить проблему.Это действительно решает проблему «Слишком много открытых файлов» для файлов размером 2 КБ, но завершается с «нехваткой памяти» для файла размером 50 КБ.
Обратите внимание, что первая однопоточная реализация (безmapreduce) способен обрабатывать файл размером 50 тыс.
Альтернативное решение, которое мне также больше всего нравится, - это использовать iteratee IO .Я ожидал, что это решит как дескриптор файла, так и исчерпание ресурсов памяти.Однако моя реализация все еще не работает с ошибкой «Слишком много открытых файлов» в файле со строкой 2 КБ.
Версия IO iteratee имеет ту же функцию mapReduce , что и в книге, но имеет измененный chunkedFileEnum , чтобы позволить ему работать с Enumerator .
Таким образом, мой вопрос таков;что не так со следующей базовой реализацией ИО?Где Laziness?.
import Control.Monad.IO.Class (liftIO)
import Control.Monad.Trans (MonadIO, liftIO)
import System.IO
import qualified Data.Enumerator.List as EL
import qualified Data.Enumerator.Text as ET
import Data.Enumerator hiding (map, filter, head, sequence)
import Data.Text(Text)
import Data.Text.Read
import Data.Maybe
import qualified Data.ByteString.Char8 as Str
import Control.Exception (bracket,finally)
import Control.Monad(forM,liftM)
import Control.Parallel.Strategies
import Control.Parallel
import Control.DeepSeq (NFData)
import Data.Int (Int64)
--Goal: in a file with n values, calculate the sum of all n*(n-1)/2 squared distances
--My operation for one value pair
distance :: Int -> Int -> Int
distance a b = (a-b)*(a-b)
combineDistances :: [Int] -> Int
combineDistances = sum
--Test file generation
createTestFile :: Int -> FilePath -> IO ()
createTestFile n path = writeFile path $ unlines $ map show $ take n $ infiniteList 0 1
where infiniteList :: Int->Int-> [Int]
infiniteList i j = (i + j) : infiniteList j (i+j)
--Applying my operation simply on a file
--(Actually does NOT throw an Out of memory on a file generated by createTestFile 50000)
--But i want to use multiple cores..
sumOfDistancesOnSmallFile :: FilePath -> IO Int
sumOfDistancesOnSmallFile path = do
fileContents <- readFile path
return $ allDistances $ map read $ lines $ fileContents
where
allDistances (x:xs) = (allDistances xs) + ( sum $ map (distance x) xs)
allDistances _ = 0
--Setting up an enumerator of read values from a text stream
readerEnumerator :: Monad m =>Integral a => Reader a -> Step a m b -> Iteratee Text m b
readerEnumerator reader = joinI . (EL.concatMapM transformer)
where transformer input = case reader input of
Right (val, remainder) -> return [val]
Left err -> return [0]
readEnumerator :: Monad m =>Integral a => Step a m b -> Iteratee Text m b
readEnumerator = readerEnumerator (signed decimal)
--The iteratee version of my operation
distancesFirstToTailIt :: Monad m=> Iteratee Int m Int
distancesFirstToTailIt = do
maybeNum <- EL.head
maybe (return 0) distancesOneToManyIt maybeNum
distancesOneToManyIt :: Monad m=> Int -> Iteratee Int m Int
distancesOneToManyIt base = do
maybeNum <- EL.head
maybe (return 0) combineNextDistance maybeNum
where combineNextDistance nextNum = do
rest <- distancesOneToManyIt base
return $ combineDistances [(distance base nextNum),rest]
--The mapreduce algorithm
mapReduce :: Strategy b -- evaluation strategy for mapping
-> (a -> b) -- map function
-> Strategy c -- evaluation strategy for reduction
-> ([b] -> c) -- reduce function
-> [a] -- list to map over
-> c
mapReduce mapStrat mapFunc reduceStrat reduceFunc input =
mapResult `pseq` reduceResult
where mapResult = parMap mapStrat mapFunc input
reduceResult = reduceFunc mapResult `using` reduceStrat
--Applying the iteratee operation using mapreduce
sumOfDistancesOnFileWithIt :: FilePath -> IO Int
sumOfDistancesOnFileWithIt path = chunkedFileEnum chunkByLinesTails (distancesUsingMapReduceIt) path
distancesUsingMapReduceIt :: [Enumerator Text IO Int] -> IO Int
distancesUsingMapReduceIt = mapReduce rpar (runEnumeratorAsMapFunc)
rpar (sumValuesAsReduceFunc)
where runEnumeratorAsMapFunc :: Enumerator Text IO Int -> IO Int
runEnumeratorAsMapFunc = (\source->run_ (source $$ readEnumerator $$ distancesFirstToTailIt))
sumValuesAsReduceFunc :: [IO Int] -> IO Int
sumValuesAsReduceFunc = liftM sum . sequence
--Working with (file)chunk enumerators:
data ChunkSpec = CS{
chunkOffset :: !Int
, chunkLength :: !Int
} deriving (Eq,Show)
chunkedFileEnum :: (NFData (a)) => MonadIO m =>
(FilePath-> IO [ChunkSpec])
-> ([Enumerator Text m b]->IO a)
-> FilePath
-> IO a
chunkedFileEnum chunkCreator funcOnChunks path = do
(chunks, handles)<- chunkedEnum chunkCreator path
r <- funcOnChunks chunks
(rdeepseq r `seq` (return r)) `finally` mapM_ hClose handles
chunkedEnum :: MonadIO m=>
(FilePath -> IO [ChunkSpec])
-> FilePath
-> IO ([Enumerator Text m b], [Handle])
chunkedEnum chunkCreator path = do
chunks <- chunkCreator path
liftM unzip . forM chunks $ \spec -> do
h <- openFile path ReadMode
hSeek h AbsoluteSeek (fromIntegral (chunkOffset spec))
let chunk = ET.enumHandle h --Note:chunklength not taken into account, so just to EOF
return (chunk,h)
-- returns set of chunks representing tails . lines . readFile
chunkByLinesTails :: FilePath -> IO[ChunkSpec]
chunkByLinesTails path = do
bracket (openFile path ReadMode) hClose $ \h-> do
totalSize <- fromIntegral `liftM` hFileSize h
let chunkSize = 1
findChunks offset = do
let newOffset = offset + chunkSize
hSeek h AbsoluteSeek (fromIntegral newOffset)
let findNewline lineSeekOffset = do
eof <- hIsEOF h
if eof
then return [CS offset (totalSize - offset)]
else do
bytes <- Str.hGet h 256
case Str.elemIndex '\n' bytes of
Just n -> do
nextChunks <- findChunks (lineSeekOffset + n + 1)
return (CS offset (totalSize-offset):nextChunks)
Nothing -> findNewline (lineSeekOffset + Str.length bytes)
findNewline newOffset
findChunks 0
Кстати, я работаю на HaskellPlatform 2011.2.0 на Mac OS X 10.6.7 (снежный барс)
со следующими пакетами:
bytestring 0.9.1.10
параллельно 3.1.0.1
счетчик 0.4.8, с руководством здесь