Вложенные Итерации

Question

Я работаю с конкретной базой данных, в которой после успешного запроса вы можете получить доступ к группе фрагментов полученных данных с помощью определенной команды:

getResultData :: IO (ResponseCode, ByteString)

Теперь getResultData вернет код ответа и некоторые данные, где коды ответа выглядят так:

response = GET_DATA_FAILED | OPERATION_SUCCEEDED | NO_MORE_DATA

ByteString - это один, несколько или все фрагменты:

http://desmond.imageshack.us/Himg189/scaled.php?server=189&filename=chunksjpeg.png&res=medium

На этом история не заканчивается. Существует поток групп:

http://desmond.imageshack.us/Himg695/scaled.php?server=695&filename=chunkgroupsjpeg.png&res=medium

После получения ответа NO_MORE_DATA от getResultData, вызов getNextItem будет повторять поток, позволяя мне снова начать вызовы getResultData. Как только getNextItem возвращает STREAM_FINISHED, это все, что она написала; У меня есть мои данные.

Теперь я хочу изменить это явление с помощью Date.Iteratee или Data.Enumerator. Поскольку мое существующее решение Data.Iteratee работает, оно покажется очень наивным, и я чувствую, что мне следует моделировать его с помощью вложенных итераций, а не одного большого итеративного объекта, как в настоящее время реализуется мое решение.

Я искал код Data.Iteratee 0.8.6.2, и я немного запутался, когда дело доходит до вложенных вещей.

Являются ли вложенные итерации правильным курсом действий? Если да, то как бы это смоделировать с помощью вложенных итераций?

С уважением

Answer 1

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

Куски и группы

Первая проблема - правильно настроить источник данных. По сути, логические разделы, которые вы описали, дают вам поток, эквивалентный [[ByteString]], Если вы создадите перечислитель для непосредственного создания этого, каждый элемент в потоке будет полной группой фрагментов, которых, вероятно, вы хотите избежать (по соображениям памяти). Вы можете сгладить все в один [ByteString], но тогда вам нужно будет заново ввести границы, что было бы довольно расточительно, так как БД делает это за вас.

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

enumGroup :: Enumerator ByteString IO a
enumGroup = enumFromCallback cb ()
 where
  cb () = do
    (code, data) <- getResultData
    case code of
        OPERATION_SUCCEEDED -> return $ Right ((True, ()), data)
        NO_MORE_DATA        -> return $ Right ((False, ()), data)
        GET_DATA_FAILED     -> return $ Left MyException

Поскольку куски имеют фиксированный размер, вы можете легко разделить это на Data.Iteratee.group,

enumGroupChunked :: Iteratee [ByteString] IO a -> IO (Iteratee ByteString IO a)
enumGroupChunked = enumGroup . joinI . group groupSize

Сравните тип этого с Enumerator

type Enumerator s m a = Iteratee s m a -> m (Iteratee s m a)

Так enumGroupChunked в основном причудливый перечислитель, который меняет тип потока. Это означает, что он принимает потребителя [ByteString] iteratee и возвращает его, который использует простые строки байтов. Часто тип возвращаемого значения перечислителя не имеет значения; это просто итератор, с которым вы оцениваете run (или же tryRun) чтобы получить на выходе, чтобы вы могли сделать то же самое здесь:

evalGroupChunked :: Iteratee [ByteString] IO a -> IO a
evalGroupChunked i = enumGroupChunked i >>= run

Если у вас есть более сложная обработка для каждой группы, проще всего сделать это в enumGroupChunked функция.

Поток групп

Теперь это не так, что делать с потоком групп? Ответ зависит от того, как вы хотите их потреблять. Если вы хотите по существу обрабатывать каждую группу в потоке независимо, я бы сделал нечто подобное:

foldStream :: Iteratee [ByteString] IO a -> (b -> a -> b) -> b -> IO b
foldStream iter f acc0 = do
  val <- evalGroupChunked iter
  res <- getNextItem
  case res of 
        OPERATION_SUCCEEDED -> foldStream iter f $! f acc0 val
        NO_MORE_DATA        -> return $ f acc0 val
        GET_DATA_FAILED     -> error "had a problem"

Однако предположим, что вы хотите выполнить некоторую потоковую обработку всего набора данных, а не только отдельных групп. То есть у вас есть

bigProc :: Iteratee [ByteString] IO a

что вы хотите запустить весь набор данных. Здесь полезен итератор возврата перечислителя. Некоторый предыдущий код теперь будет немного отличаться:

enumGroupChunked' :: Iteratee [ByteString] IO a
  -> IO (Iteratee ByteString IO (Iteratee [ByteString] IO a))
enumGroupChunked' = enumGroup . group groupSize

procStream :: Iteratee [ByteString] IO a -> a
procStream iter = do
  i' <- enumGroupChunked' iter >>= run
  res <- getNextItem
  case res of 
        OPERATION_SUCCEEDED -> procStream i'
        NO_MORE_DATA        -> run i'
        GET_DATA_FAILED     -> error "had a problem"

Это использование вложенных итераторов (т.е. Iteratee s1 m (Iteratee s2 m a)) немного необычен, но особенно полезен, когда вы хотите последовательно обрабатывать данные из нескольких перечислителей. Ключ должен признать, что run Использование внешнего итератора даст вам итератора, который готов получать больше данных. Это модель, которая хорошо работает в этом случае, потому что вы можете перечислять каждую группу независимо, но обрабатывать их как один поток.

Одно предупреждение: внутренний итератор будет в том состоянии, в котором он был оставлен. Предположим, что последний фрагмент группы может быть меньше полного фрагмента, например

   Group A               Group B               Group C
   1024, 1024, 512       1024, 1024, 1024      1024, 1024, 1024

Что произойдет в этом случае, потому что group объединяет данные в блоки размером 1024, объединяет последний фрагмент группы A с первыми 512 байтами группы B. Это не проблема с foldStream пример, потому что этот код завершает внутренний итератор (с joinI). Это означает, что группы действительно независимы, поэтому вы должны относиться к ним как к таковым. Если вы хотите объединить группы, как в procStream Вы должны думать обо всем потоке. Если это ваш случай, то вам нужно использовать что-то более сложное, чем просто group,

Data.Iteratee против Data.Enumerator

Не вдаваясь в дискуссию о достоинствах любого из пакетов, не говоря уже об IterIO (я, по общему признанию, предвзят), я хотел бы указать на то, что я считаю наиболее существенным различием между этими двумя понятиями: абстракция потока.

В Data.Iteratee, потребитель Iteratee ByteString m a работает с условной байтовой строкой некоторой длины, с доступом к одному фрагменту ByteString в одно время

В Data.Enumerator, потребитель Iteratee ByteString m a оперирует условным [ByteString] с доступом к одному или нескольким элементам (строкам байтов) одновременно.

Это означает, что большинство операций Data.Iteratee ориентированы на элемент, т.е. Iteratee ByteString они будут работать на одном Word8 в то время как операции Data.Enumerator сфокусированы на чанках и работают на ByteString,

Вы можете думать о Data.Iteratee.Iteratee [s] m a === Data.Enumerator.Iteratee s m a,