Алгоритм с эффективным использованием памяти для задачи `take n (sort xs)` (sorted prefix)

Итак, мне действительно удалось решить проблему. Идея состоит в том, чтобы отбросить причудливые структуры данных и работать вручную ;-) По сути, мы разбиваем список ввода на части, сортируем их и сворачиваем [[Int]] список, выбирая n наименьшие элементы на каждом шаге. Уловка заключается в правильном слиянии аккумулятора с отсортированным блоком. Мы должны использовать seq, иначе лень вас укусит, и результат все равно потребует много памяти для вычислений. Кроме того, я смешиваю слияние с take n, просто для большей оптимизации. Вот вся программа вместе с предыдущими попытками:

{-# LANGUAGE ScopedTypeVariables, PackageImports #-}     
module Main where

import qualified Data.List as List
import qualified Data.List.Split as Split
import qualified "heaps" Data.Heap as Heap -- qualified import from "heaps" package

import System.Random.MWC
import qualified Data.Vector.Unboxed as Vec

import System.Environment

limitSortL n xs = take n (List.sort xs)
limitSortH n xs = List.unfoldr Heap.uncons (List.foldl' (\ acc x -> Heap.take n (Heap.insert x acc) ) Heap.empty xs)
takeSortMerge n inp = List.foldl' 
                        (\acc lst -> (merge n acc (List.sort lst))) 
                        [] (Split.splitEvery n inp)
    where
     merge 0 _ _ = []
     merge _ [] xs = xs
     merge _ ys [] = ys
     merge f (x:xs) (y:ys) | x < y = let tail = merge (f-1) xs (y:ys) in tail `seq` (x:tail) 
                           | otherwise = let tail = merge (f-1) (x:xs) ys in tail `seq` (y:tail)


main = do
  st <- create

  let n1 = 10^7
      n2 = 20

  rxs :: [Int] <- Vec.toList `fmap` uniformVector st (n1)

  args <- getArgs

  case args of
    ["LIST"] ->  print (limitSortL n2 rxs)
    ["HEAP"] ->  print (limitSortH n2 rxs)
    ["MERGE"] -> print (takeSortMerge n2 rxs)
    _ -> putStrLn "Nothing..."

  return ()

Производительность во время выполнения, потребление памяти, время сборки мусора:

LIST       3.96s   1168 MB    75 %
HEAP       35.29s    78 MB    3.6 %
MERGE      1.00s     78 MB    3.0 %
just rxs   0.21s     78 MB    0.0 %  -- just evaluating the random vector

Существует множество алгоритмов выбора, специализирующихся именно на этом. Алгоритм, основанный на разделах, является «классическим», но, как и Quicksort, он не совсем подходит для списков Haskell. Википедия не показывает ничего, что связано с функциональным программированием, хотя я подозреваю, что описанный «выбор турнира» такой же или не сильно отличается от вашего текущего решения для сортировки слиянием.

Если вас беспокоит потребление памяти, вы можете использовать очередь с приоритетом - она ​​использует O (K) памяти и время O (N * logK) в целом:

queue := first k elements
for each element in the rest:
    add the element to the queue
    remove the largest element from the queue
convert the queue to a sorted list

«Быстрая сортировка и k-е наименьшие элементы», всегда заинтересованный Генрих Апфельмус: http://apfelmus.nfshost.com/articles/quicksearch.html

Простите, если я не могу расшифровать

 Vec.toList `fmap` uniformVector st (10^7)

но как долго будет этот список? Ясно ли, что какой бы ленивой ни была сортировка слиянием, она должна будет реализовать, по крайней мере, весь список?

Обновлять:

Я слышал, что сортировка слиянием, реализованная в Data.List.sort, является ленивой и не создает больше элементов, чем необходимо.

Это ничего не говорит о потреблении пространства для слияния, прежде чем оно сможет начать доставку первых элементов списка. В любом случае ему придется пройти (и тем самым реализовать) весь список, выделить для объединения подсписки и т. Д. Вот ссылка от http://www.inf.fh-flensburg.de/lang/algorithmmen/sortieren/merge/mergen.htm

Недостатком сортировки слиянием является то, что для временного массива b требуется дополнительное пространство размером Θ (n).

Есть разные возможности для реализации слияния функций. Самый действенный из них - вариант б. Он требует вдвое меньше дополнительного места, он быстрее, чем другие варианты, и он стабилен.

Возможно, вы неправильно оцениваете проблему. Это могло быть скорее из-за слишком лени, чем из-за недостатка.

Может быть, вам стоит попробовать более строгую структуру данных или изменяемый массив в монаде ST.

Для подхода с изменяемым массивом вы можете ограничить количество перемещений на вставку до n/2 вместо n-1, записав индекс h, который «указывает на» заголовок очереди, хранящейся в массиве, и разрешив очереди «оборачиваться» внутри массива.

Эффективность памяти редко бывает сильной стороной haskell. Тем не менее, не так сложно создать алгоритм сортировки, который был бы более ленивым, чем сортировка слиянием. Например, вот простая быстрая сортировка:

qsort [] = []
qsort (x:xs) = qcombine (qsort a) b (qsort c) where
    (a,b,c) = qpart x (x:xs) ([],[],[])
qpart _ [] ac = ac
qpart n (x:xs) (a,b,c)
    | x > n = qpart n xs (a,b,x:c)
    | x < n = qpart n xs (x:a,b,c)
    | otherwise = qpart n xs (a,x:b,c)
qcombine (a:as) b c = a:qcombine as b c
qcombine [] (b:bs) c = b:qcombine [] bs c
qcombine [] [] c = c

Я использовал явную рекурсию, чтобы было очевидно, что происходит. Каждая часть здесь действительно ленивая, что означает, что qcombine никогда не будет вызывать qsort c, если это не нужно. Это должно снизить использование памяти, если вам нужны только первые несколько элементов.

Вы можете создать лучший алгоритм сортировки для этой конкретной задачи, который использует части стиля быстрой сортировки, чтобы получить первые n элементов списка в несортированном порядке. Затем просто вызовите высокоэффективный алгоритм сортировки, если они вам нужны по порядку.

пример такого подхода:

qselect 0 _ = []
qselect n [] = error ("cant produce " ++ show n ++ " from empty list")
qselect n (x:xs)
    | al > n = qselect n a
    | al + bl > n = a ++ take (al - n) b
    | otherwise = a ++ b ++ (qselect (n - al - bl) c) where
        (a,al,b,bl,c,cl) = qpartl x (x:xs) ([],0,[],0,[],0)

qpartl _ [] ac = ac
qpartl n (x:xs) (a,al,b,bl,c,cl)
    | x > n = qpartl n xs (a,al,b,bl,x:c,cl+1)
    | x < n = qpartl n xs (x:a,al+1,b,bl,c,cl+1)
    | otherwise = qpartl n xs (a,al,x:b,bl+1,c,cl)

Опять же, этот код не самый чистый, но я хочу прояснить, что он делает.

Для случая, когда вы хотите взять очень маленькое число, оптимальна сортировка выбора. Например, если вам нужен только самый высокий элемент в списке, вы можете перебирать его один раз, давая большую тэту размера списка.

С другой стороны, если вам нужен почти весь список, но вы не заботитесь о его порядке, вы можете многократно «удалить» самые нижние элементы в списке.

Оба этих подхода, а также быстрая сортировка выше, равны O (n ^ 2), но вы хотите иметь стратегию, которая часто работает в больших O (k * n) и, как правило, не использует тонну пространства.

Другой вариант - использовать алгоритм сортировки на месте для управления использованием памяти. Я не знаю ни одного ленивого вида на месте, но если они существуют, это было бы идеально.