content: {-# LANGUAGE CPP #-}
  -- |
  -- Module      : Streamly.Internal.Data.Array
  -- Copyright   : (c) 2019 Composewell Technologies
  -- License     : BSD3
  -- Maintainer  : streamly@composewell.com
  -- Stability   : experimental
  -- Portability : GHC
  --
  module Streamly.Internal.Data.Array
      (
      -- * Setup
      -- $setup
  
      -- * Design Notes
      -- $design
  
      -- * The Array Type
        module Streamly.Internal.Data.Array.Type
  
      -- * Construction
      -- Monadic Folds
      , createOfLast
  
      -- * Random Access
      -- , getIndicesFrom    -- read from a given position to the end of file
      -- , getIndicesUpto    -- read from beginning up to the given position
      -- , getIndicesFromTo
      -- , getIndicesFromRev  -- read from a given position to the beginning of file
      -- , getIndicesUptoRev  -- read from end to the given position in file
      , indexReader
      , indexReaderFromThenTo
  
      -- * Search
      , binarySearch
      , findIndicesOf
      -- getIndicesOf
      , indexFinder -- see splitOn
      -- , findIndexOf
      -- , find
  
      -- * Casting
      , cast
      , asBytes
      , unsafeCast
      , asCStringUnsafe -- XXX asCString
      , asCWString
  
      -- * Subarrays
      -- , sliceOffLen
      , indexerFromLen
      , splitterFromLen
  
      -- * Streaming Operations
      , streamTransform
  
      -- * Folding
      , streamFold
      , foldM
      , foldRev
  
      -- * Stream of Arrays
      , concatSepBy
      , concatEndBy
      , concatEndBySeq
  
      , compactMax
      , compactMax'
      , compactSepByByte_
      , compactEndByByte_
      , compactEndByLn_
  
      -- * Parsing Stream of Arrays
      , foldBreakChunks -- Uses Stream, bad perf on break
      , foldChunks
      , foldBreak
      -- , parseBreakChunksK -- XXX uses Parser. parseBreak is better?
      , toParserK
      , parseBreak
      , parseBreakPos
      , parse
      , parsePos
  
      -- * Serialization
      , encodeAs
      , serialize
      , serialize'
      , deserialize
  
      -- * Deprecated
      , slicerFromLen
      , sliceIndexerFromLen
      , castUnsafe
      , getSliceUnsafe
      , pinnedSerialize
      , genSlicesFromLen
      , getSlicesFromLen
      , getIndices
      , writeLastN
      , interpose
      , interposeSuffix
      , intercalateSuffix
      , compactLE
      , pinnedCompactLE
      , compactOnByte
      , compactOnByteSuffix
      , splitOn
      , fold
      , foldBreakChunksK
      )
  where
  
  #include "assert.hs"
  #include "deprecation.h"
  #include "inline.hs"
  #include "ArrayMacros.h"
  
  import Control.Monad.IO.Class (MonadIO(..))
  -- import Data.Bifunctor (first)
  -- import Data.Either (fromRight)
  import Data.Functor.Identity (Identity(..))
  import Data.Proxy (Proxy(..))
  import Data.Word (Word8)
  import Foreign.C.String (CString, CWString)
  import GHC.Types (SPEC(..))
  import Streamly.Internal.Data.Unbox (Unbox(..))
  import Prelude hiding (length, null, last, map, (!!), read, concat)
  
  import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..), MutByteArray)
  import Streamly.Internal.Data.Serialize.Type (Serialize)
  import Streamly.Internal.Data.Fold.Type (Fold(..))
  import Streamly.Internal.Data.Parser (ParseError(..), ParseErrorPos(..))
  import Streamly.Internal.Data.ParserK.Type
      (ParserK, ParseResult(..), Input(..), Step(..))
  import Streamly.Internal.Data.Stream (Stream(..))
  import Streamly.Internal.Data.StreamK.Type (StreamK)
  import Streamly.Internal.Data.SVar.Type (adaptState, defState)
  import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
  import Streamly.Internal.Data.Unfold.Type (Unfold(..))
  import Streamly.Internal.System.IO (unsafeInlineIO)
  
  import qualified Streamly.Internal.Data.Fold.Type as Fold
  import qualified Streamly.Internal.Data.Serialize.Type as Serialize
  import qualified Streamly.Internal.Data.MutByteArray.Type as MBA
  import qualified Streamly.Internal.Data.MutArray as MA
  import qualified Streamly.Internal.Data.RingArray as RB
  import qualified Streamly.Internal.Data.ParserDrivers as Drivers
  import qualified Streamly.Internal.Data.Parser.Type as ParserD
  import qualified Streamly.Internal.Data.ParserK.Type as ParserK
  import qualified Streamly.Internal.Data.Stream as D
  import qualified Streamly.Internal.Data.Stream as Stream
  import qualified Streamly.Internal.Data.StreamK.Type as StreamK
  import qualified Streamly.Internal.Data.Unfold as Unfold
  
  import Streamly.Internal.Data.Array.Type
  
  #include "DocTestDataArray.hs"
  
  -- $design
  --
  -- To summarize:
  --
  --  * Arrays are finite and fixed in size
  --  * provide /O(1)/ access to elements
  --  * store only data and not functions
  --  * provide efficient IO interfacing
  --
  -- 'Foldable' instance is not provided because the implementation would be much
  -- less efficient compared to folding via streams.  'Semigroup' and 'Monoid'
  -- instances should be used with care; concatenating arrays using binary
  -- operations can be highly inefficient.  Instead, use
  -- 'Streamly.Internal.Data.Stream.Chunked.toArray' to concatenate N
  -- arrays at once.
  --
  -- Each array is one pointer visible to the GC.  Too many small arrays (e.g.
  -- single byte) are only as good as holding those elements in a Haskell list.
  -- However, small arrays can be compacted into large ones to reduce the
  -- overhead. To hold 32GB memory in 32k sized buffers we need 1 million arrays
  -- if we use one array for each chunk. This is still significant to add
  -- pressure to GC.
  
  -------------------------------------------------------------------------------
  -- Folds with Array as the container
  -------------------------------------------------------------------------------
  
  -- NOTE: We could possible write this in terms of "MutArray.createOfLast" but
  -- this causes regression. This is probably because mapping inside "Fold.ifThen"
  -- is more efficient than mapping over "Fold.ifTen".
  --
  -- | @createOfLast n@ folds a maximum of @n@ elements from the end of the input
  -- stream to an 'Array'.
  --
  {-# INLINE createOfLast #-}
  createOfLast ::
         (Unbox a, MonadIO m) => Int -> Fold m a (Array a)
  createOfLast n = Fold.ifThen (pure (n <= 0)) (Fold.fromPure empty) lst
  
      where
  
      lst =
          let f = fmap unsafeFreeze . RB.toMutArray
           in Fold.rmapM f $ RB.createOfLast n
  
  {-# DEPRECATED writeLastN "Please use createOfLast instead." #-}
  {-# INLINE writeLastN #-}
  writeLastN :: (Unbox a, MonadIO m) => Int -> Fold m a (Array a)
  writeLastN = createOfLast
  
  -------------------------------------------------------------------------------
  -- Random Access
  -------------------------------------------------------------------------------
  
  -------------------------------------------------------------------------------
  -- Searching
  -------------------------------------------------------------------------------
  
  -- | Given a sorted array, perform a binary search to find the given element.
  -- Returns the index of the element if found.
  --
  -- /Unimplemented/
  {-# INLINE binarySearch #-}
  binarySearch :: a -> Array a -> Maybe Int
  binarySearch = undefined
  
  -- find/findIndex etc can potentially be implemented more efficiently on arrays
  -- compared to streams by using SIMD instructions.
  -- We can also return a bit array instead.
  
  -- Can use SIMD.
  
  -- | Perform a linear search to find all the indices where a given element is
  -- present in an array.
  --
  -- /Unimplemented/
  indexFinder :: (a -> Bool) -> Unfold Identity (Array a) Int
  indexFinder = undefined
  
  -- |
  -- /Unimplemented/
  findIndicesOf :: (a -> Bool) -> Array a -> Stream Identity Int
  findIndicesOf p = Stream.unfold (indexFinder p)
  
  {-
  findIndexOf :: (a -> Bool) -> Array a -> Maybe Int
  findIndexOf p = Unfold.fold Fold.one . Stream.unfold (indexFinder p)
  
  find :: (a -> Bool) -> Array a -> Bool
  find = Unfold.fold Fold.null . Stream.unfold (indexFinder p)
  -}
  
  -------------------------------------------------------------------------------
  -- Folds
  -------------------------------------------------------------------------------
  
  -- XXX We can potentially use SIMD instructions on arrays to fold faster.
  
  -------------------------------------------------------------------------------
  -- Slice
  -------------------------------------------------------------------------------
  
  getSliceUnsafe ::
         forall a. Unbox a
      => Int -- ^ starting index
      -> Int -- ^ length of the slice
      -> Array a
      -> Array a
  RENAME(getSliceUnsafe,unsafeSliceOffLen)
  
  splitOn :: (Monad m, Unbox a) =>
      (a -> Bool) -> Array a -> Stream m (Array a)
  RENAME(splitOn,splitEndBy_)
  
  {-# INLINE indexerFromLen #-}
  indexerFromLen, sliceIndexerFromLen :: forall m a. (Monad m, Unbox a)
      => Int -- ^ from index
      -> Int -- ^ length of the slice
      -> Unfold m (Array a) (Int, Int)
  indexerFromLen from len =
      Unfold.lmap unsafeThaw (MA.indexerFromLen from len)
  RENAME(sliceIndexerFromLen,indexerFromLen)
  
  {-# DEPRECATED genSlicesFromLen "Please use indexerFromLen instead." #-}
  {-# INLINE genSlicesFromLen #-}
  genSlicesFromLen :: forall m a. (Monad m, Unbox a)
      => Int -- ^ from index
      -> Int -- ^ length of the slice
      -> Unfold m (Array a) (Int, Int)
  genSlicesFromLen = indexerFromLen
  
  -- | Generate a stream of slices of specified length from an array, starting
  -- from the supplied array index. The last slice may be shorter than the
  -- requested length.
  --
  -- /Pre-release//
  {-# INLINE splitterFromLen #-}
  splitterFromLen, slicerFromLen :: forall m a. (Monad m, Unbox a)
      => Int -- ^ from index
      -> Int -- ^ length of the slice
      -> Unfold m (Array a) (Array a)
  splitterFromLen from len =
      fmap unsafeFreeze
          $ Unfold.lmap unsafeThaw (MA.splitterFromLen from len)
  RENAME(slicerFromLen,splitterFromLen)
  
  {-# DEPRECATED getSlicesFromLen "Please use splitterFromLen instead." #-}
  {-# INLINE getSlicesFromLen #-}
  getSlicesFromLen :: forall m a. (Monad m, Unbox a)
      => Int -- ^ from index
      -> Int -- ^ length of the slice
      -> Unfold m (Array a) (Array a)
  getSlicesFromLen = splitterFromLen
  
  -------------------------------------------------------------------------------
  -- Random reads and writes
  -------------------------------------------------------------------------------
  
  -- | Given a stream of array indices, read the elements on those indices from
  -- the supplied Array. An exception is thrown if an index is out of bounds.
  --
  -- This is the most general operation. We can implement other operations in
  -- terms of this:
  --
  -- @
  -- read =
  --      let u = lmap (\arr -> (0, length arr - 1)) Unfold.enumerateFromTo
  --       in Unfold.lmap f (indexReader arr)
  --
  -- readRev =
  --      let i = length arr - 1
  --       in Unfold.lmap f (indexReaderFromThenTo i (i - 1) 0)
  -- @
  --
  -- /Pre-release/
  {-# INLINE indexReader #-}
  indexReader :: (Monad m, Unbox a) => Stream m Int -> Unfold m (Array a) a
  indexReader m =
      let unf = MA.indexReaderWith (return . unsafeInlineIO) m
       in Unfold.lmap unsafeThaw unf
  
  -- XXX DO NOT REMOVE, change the signature to use Stream instead of unfold
  {-# DEPRECATED getIndices "Please use getIndices instead." #-}
  {-# INLINE getIndices #-}
  getIndices :: (Monad m, Unbox a) => Stream m Int -> Unfold m (Array a) a
  getIndices = indexReader
  
  -- | Unfolds @(from, then, to, array)@ generating a finite stream whose first
  -- element is the array value from the index @from@ and the successive elements
  -- are from the indices in increments of @then@ up to @to@. Index enumeration
  -- can occur downwards or upwards depending on whether @then@ comes before or
  -- after @from@.
  --
  -- @
  -- getIndicesFromThenTo =
  --     let f (from, next, to, arr) =
  --             (Stream.enumerateFromThenTo from next to, arr)
  --      in Unfold.lmap f getIndices
  -- @
  --
  -- /Unimplemented/
  {-# INLINE indexReaderFromThenTo #-}
  indexReaderFromThenTo :: Unfold m (Int, Int, Int, Array a) a
  indexReaderFromThenTo = undefined
  
  -------------------------------------------------------------------------------
  -- Transform via stream operations
  -------------------------------------------------------------------------------
  
  -- for non-length changing operations we can use the original length for
  -- allocation. If we can predict the length then we can use the prediction for
  -- new allocation. Otherwise we can use a hint and adjust dynamically.
  
  {-
  -- | Transform an array into another array using a pipe transformation
  -- operation.
  --
  {-# INLINE runPipe #-}
  runPipe :: (MonadIO m, Unbox a, Unbox b)
      => Pipe m a b -> Array a -> m (Array b)
  runPipe f arr = P.runPipe (toArrayMinChunk (length arr)) $ f (read arr)
  -}
  
  -- XXX For transformations that cannot change the number of elements e.g. "map"
  -- we can use a predetermined array length.
  --
  -- | Transform an array into another array using a stream transformation
  -- operation.
  --
  -- /Pre-release/
  {-# INLINE streamTransform #-}
  streamTransform :: forall m a b. (MonadIO m, Unbox a, Unbox b)
      => (Stream m a -> Stream m b) -> Array a -> m (Array b)
  streamTransform f arr =
      Stream.fold (createWith (length arr)) $ f (read arr)
  
  -------------------------------------------------------------------------------
  -- Casts
  -------------------------------------------------------------------------------
  
  -- | Cast an array having elements of type @a@ into an array having elements of
  -- type @b@. The array size must be a multiple of the size of type @b@
  -- otherwise accessing the last element of the array may result into a crash or
  -- a random value.
  --
  -- /Pre-release/
  --
  unsafeCast, castUnsafe ::
  #ifdef DEVBUILD
      Unbox b =>
  #endif
      Array a -> Array b
  unsafeCast (Array contents start end) =
      Array contents start end
  RENAME(castUnsafe,unsafeCast)
  
  -- | Cast an @Array a@ into an @Array Word8@.
  --
  --
  asBytes :: Array a -> Array Word8
  asBytes = unsafeCast
  
  -- | Cast an array having elements of type @a@ into an array having elements of
  -- type @b@. The length of the array should be a multiple of the size of the
  -- target element otherwise 'Nothing' is returned.
  --
  --
  cast :: forall a b. (Unbox b) => Array a -> Maybe (Array b)
  cast arr =
      let len = byteLength arr
          r = len `mod` SIZE_OF(b)
       in if r /= 0
          then Nothing
          else Just $ unsafeCast arr
  
  -- | Convert an array of any element type into a null terminated CString Ptr.
  -- The array is copied to pinned memory.
  --
  -- /Unsafe/
  --
  -- /O(n) Time: (creates a copy of the array)/
  --
  -- /Pre-release/
  --
  asCStringUnsafe :: Array a -> (CString -> IO b) -> IO b
  asCStringUnsafe arr = MA.asCString (unsafeThaw arr)
  
  -- | Convert an array of any element type into a null terminated CWString Ptr.
  -- The array is copied to pinned memory.
  --
  -- /Unsafe/
  --
  -- /O(n) Time: (creates a copy of the array)/
  --
  -- /Pre-release/
  --
  asCWString :: Array a -> (CWString -> IO b) -> IO b
  asCWString arr = MA.asCWString (unsafeThaw arr)
  
  -------------------------------------------------------------------------------
  -- Folds
  -------------------------------------------------------------------------------
  
  -- XXX Use runIdentity for pure fold
  -- XXX Rename fold to foldM, we can then use "fold" for pure folds.
  -- XXX We do not need an INLINE on fold?
  
  -- | Fold an array using a 'Fold'.
  --
  -- /Pre-release/
  {-# INLINE foldM #-}
  fold, foldM :: (Monad m, Unbox a) => Fold m a b -> Array a -> m b
  foldM f arr = Stream.fold f (read arr)
  RENAME(fold,foldM)
  
  foldRev :: Unbox a => Fold.Fold Identity a b -> Array a -> b
  foldRev f arr = runIdentity $ Stream.fold f (readRev arr)
  
  -- | Fold an array using a stream fold operation.
  --
  -- /Pre-release/
  {-# INLINE streamFold #-}
  streamFold :: (Monad m, Unbox a) => (Stream m a -> m b) -> Array a -> m b
  streamFold f arr = f (read arr)
  
  --------------------------------------------------------------------------------
  -- Serialization
  --------------------------------------------------------------------------------
  
  {-# INLINE encodeAs #-}
  encodeAs :: forall a. Serialize a => PinnedState -> a -> Array Word8
  encodeAs ps a =
      unsafeInlineIO $ do
          let len = Serialize.addSizeTo 0 a
          mbarr <- MBA.newAs ps len
          off <- Serialize.serializeAt 0 mbarr a
          assertM(len == off)
          pure $ Array mbarr 0 off
  
  -- |
  -- Properties:
  -- 1. Identity: @deserialize . serialize == id@
  -- 2. Encoded equivalence: @serialize a == serialize a@
  {-# INLINE serialize #-}
  serialize :: Serialize a => a -> Array Word8
  serialize = encodeAs Unpinned
  
  -- | Serialize a Haskell type to a pinned byte array. The array is allocated
  -- using pinned memory so that it can be used directly in OS APIs for writing
  -- to file or sending over the network.
  --
  -- Properties:
  --
  -- 1. Identity: @deserialize . serialize' == id@
  -- 2. Encoded equivalence: @serialize' a == serialize' a@
  {-# INLINE serialize' #-}
  pinnedSerialize, serialize' :: Serialize a => a -> Array Word8
  serialize' = encodeAs Pinned
  RENAME_PRIME(pinnedSerialize,serialize)
  
  -- XXX We can deserialize it like MutArray, returning the remaining slice.
  
  -- | Decode a Haskell type from a byte array containing its serialized
  -- representation.
  {-# INLINE deserialize #-}
  deserialize :: Serialize a => Array Word8 -> (a, Array Word8)
  deserialize arr =
      let (a, b) = unsafeInlineIO $ MA.deserialize (unsafeThaw arr)
       in (a, unsafeFreeze b)
  
  -------------------------------------------------------------------------------
  -- Streams of Arrays
  -------------------------------------------------------------------------------
  
  -- TODO: efficiently compare two streams of arrays. Two streams can have chunks
  -- of different sizes, we can handle that in the stream comparison abstraction.
  -- This could be useful e.g. to fast compare whether two files differ.
  
  -- | Insert the given element between arrays and flatten.
  --
  -- >>> concatSepBy x = Stream.unfoldEachSepBy x Array.reader
  --
  {-# INLINE concatSepBy #-}
  concatSepBy, interpose :: (Monad m, Unbox a) =>
      a -> Stream m (Array a) -> Stream m a
  concatSepBy x = D.unfoldEachSepBy x reader
  
  RENAME(interpose,concatSepBy)
  
  data FlattenState s =
        OuterLoop s
      | InnerLoop s !MutByteArray !Int !Int
  
  -- | Insert the given element after each array and flatten. This is similar to
  -- unlines.
  --
  -- >>> concatEndBy x = Stream.unfoldEachEndBy x Array.reader
  --
  {-# INLINE_NORMAL concatEndBy #-}
  concatEndBy, interposeSuffix :: forall m a. (Monad m, Unbox a)
      => a -> Stream m (Array a) -> Stream m a
  -- concatEndBy x = D.unfoldEachEndBy x reader
  concatEndBy sep (D.Stream step state) = D.Stream step' (OuterLoop state)
  
      where
  
      {-# INLINE_LATE step' #-}
      step' gst (OuterLoop st) = do
          r <- step (adaptState gst) st
          return $ case r of
              D.Yield Array{..} s ->
                  D.Skip (InnerLoop s arrContents arrStart arrEnd)
              D.Skip s -> D.Skip (OuterLoop s)
              D.Stop -> D.Stop
  
      step' _ (InnerLoop st _ p end) | p == end =
          return $ D.Yield sep $ OuterLoop st
  
      step' _ (InnerLoop st contents p end) = do
          let !x = unsafeInlineIO $ peekAt p contents
          return $ D.Yield x (InnerLoop st contents (INDEX_NEXT(p,a)) end)
  
  RENAME(interposeSuffix,concatEndBy)
  
  -- | Insert the given array after each array and flatten.
  --
  -- >>> concatEndBySeq x = Stream.unfoldEachEndBySeq x Array.reader
  --
  {-# INLINE concatEndBySeq #-}
  concatEndBySeq, intercalateSuffix :: (Monad m, Unbox a)
      => Array a -> Stream m (Array a) -> Stream m a
  concatEndBySeq x = D.unfoldEachEndBySeq x reader
  
  RENAME(intercalateSuffix,concatEndBySeq)
  
  -- | @compactMax n@ coalesces adjacent arrays in the input stream
  -- only if the combined size would be less than or equal to n.
  --
  -- Generates unpinned arrays irrespective of the pinning status of input
  -- arrays.
  {-# INLINE_NORMAL compactMax #-}
  compactMax, compactLE :: (MonadIO m, Unbox a)
      => Int -> Stream m (Array a) -> Stream m (Array a)
  compactMax n stream =
      D.map unsafeFreeze $ MA.compactMax n $ D.map unsafeThaw stream
  
  RENAME(compactLE,compactMax)
  
  -- | Like 'compactMax' but generates pinned arrays.
  {-# INLINE_NORMAL compactMax' #-}
  compactMax', pinnedCompactLE :: (MonadIO m, Unbox a)
      => Int -> Stream m (Array a) -> Stream m (Array a)
  compactMax' n stream =
      D.map unsafeFreeze $ MA.compactMax' n $ D.map unsafeThaw stream
  
  {-# DEPRECATED pinnedCompactLE "Please use compactMax' instead." #-}
  {-# INLINE pinnedCompactLE #-}
  pinnedCompactLE = compactMax'
  
  -- | Split a stream of byte arrays on a given separator byte, dropping the
  -- separator and coalescing all the arrays between two separators into a single
  -- array.
  --
  {-# INLINE compactSepByByte_ #-}
  compactSepByByte_, compactOnByte
      :: (MonadIO m)
      => Word8
      -> Stream m (Array Word8)
      -> Stream m (Array Word8)
  compactSepByByte_ byte =
      fmap unsafeFreeze . MA.compactSepByByte_ byte . fmap unsafeThaw
  
  RENAME(compactOnByte,compactSepByByte_)
  
  -- | Like 'compactSepByByte_', but considers the separator in suffix position
  -- instead of infix position.
  {-# INLINE compactEndByByte_ #-}
  compactEndByByte_, compactOnByteSuffix
      :: (MonadIO m)
      => Word8
      -> Stream m (Array Word8)
      -> Stream m (Array Word8)
  compactEndByByte_ byte =
      fmap unsafeFreeze . MA.compactEndByByte_ byte . fmap unsafeThaw
  -- compactEndByByte_ byte = chunksEndBy_ (== byte) . concat
  
  RENAME(compactOnByteSuffix,compactEndByByte_)
  
  -- XXX On windows we should compact on "\r\n". We can just compact on '\n' and
  -- drop the last byte in each array if it is '\r'.
  
  -- | Compact byte arrays on newline character, dropping the newline char.
  {-# INLINE compactEndByLn_ #-}
  compactEndByLn_ :: MonadIO m
      => Stream m (Array Word8)
      -> Stream m (Array Word8)
  compactEndByLn_ = compactEndByByte_ 10
  
  -------------------------------------------------------------------------------
  -- Folding Streams of Arrays
  -------------------------------------------------------------------------------
  
  -- XXX This should not be used for breaking a stream as the D.cons used in
  -- reconstructing the stream could be very bad for performance. This can only
  -- be useful in folding without breaking.
  {-# INLINE_NORMAL foldBreakChunks #-}
  foldBreakChunks :: forall m a b. (MonadIO m, Unbox a) =>
      Fold m a b -> Stream m (Array a) -> m (b, Stream m (Array a))
  foldBreakChunks (Fold fstep initial _ final) stream@(Stream step state) = do
      res <- initial
      case res of
          Fold.Partial fs -> go SPEC state fs
          Fold.Done fb -> return $! (fb, stream)
  
      where
  
      {-# INLINE go #-}
      go !_ st !fs = do
          r <- step defState st
          case r of
              Stream.Yield (Array contents start end) s ->
                  let fp = Tuple' end contents
                   in goArray SPEC s fp start fs
              Stream.Skip s -> go SPEC s fs
              Stream.Stop -> do
                  b <- final fs
                  return (b, D.nil)
  
      goArray !_ s (Tuple' end _) !cur !fs
          | cur == end = do
              go SPEC s fs
      goArray !_ st fp@(Tuple' end contents) !cur !fs = do
          x <- liftIO $ peekAt cur contents
          res <- fstep fs x
          let next = INDEX_NEXT(cur,a)
          case res of
              Fold.Done b -> do
                  let arr = Array contents next end
                  return $! (b, D.cons arr (D.Stream step st))
              Fold.Partial fs1 -> goArray SPEC st fp next fs1
  
  -- This may be more robust wrt fusion compared to unfoldMany?
  
  -- | Fold a stream of arrays using a 'Fold'. This is equivalent to the
  -- following:
  --
  -- >>> foldChunks f = Stream.fold f . Stream.unfoldEach Array.reader
  --
  foldChunks :: (MonadIO m, Unbox a) => Fold m a b -> Stream m (Array a) -> m b
  foldChunks f s = fmap fst (foldBreakChunks f s)
  -- foldStream f = Stream.fold f . Stream.unfoldEach reader
  
  -- | Fold a stream of arrays using a 'Fold' and return the remaining stream.
  --
  -- The following alternative to this function allows composing the fold using
  -- the parser Monad:
  --
  -- @
  -- foldBreakStreamK f s =
  --       fmap (first (fromRight undefined))
  --     $ StreamK.parseBreakChunks (ParserK.adaptC (Parser.fromFold f)) s
  -- @
  --
  -- We can compare perf and remove this one or define it in terms of that.
  --
  foldBreak, foldBreakChunksK :: forall m a b. (MonadIO m, Unbox a) =>
      Fold m a b -> StreamK m (Array a) -> m (b, StreamK m (Array a))
  {-
  foldBreakChunksK f s =
        fmap (first (fromRight undefined))
      $ StreamK.parseBreakChunks (ParserK.adaptC (Parser.fromFold f)) s
  -}
  foldBreak (Fold fstep initial _ final) stream = do
      res <- initial
      case res of
          Fold.Partial fs -> go fs stream
          Fold.Done fb -> return (fb, stream)
  
      where
  
      {-# INLINE go #-}
      go !fs st = do
          let stop = (, StreamK.nil) <$> final fs
              single a = yieldk a StreamK.nil
              yieldk (Array contents start end) r =
                  let fp = Tuple' end contents
                   in goArray fs r fp start
           in StreamK.foldStream defState yieldk single stop st
  
      goArray !fs st (Tuple' end _) !cur
          | cur == end = do
              go fs st
      goArray !fs st fp@(Tuple' end contents) !cur = do
          x <- liftIO $ peekAt cur contents
          res <- fstep fs x
          let next = INDEX_NEXT(cur,a)
          case res of
              Fold.Done b -> do
                  let arr = Array contents next end
                  return $! (b, StreamK.cons arr st)
              Fold.Partial fs1 -> goArray fs1 st fp next
  
  RENAME(foldBreakChunksK,foldBreak)
  
  {-
  -- This can be generalized to any type provided it can be unfolded to a stream
  -- and it can be combined using a semigroup operation.
  --
  {-# INLINE_NORMAL parseBreakD #-}
  parseBreakD ::
         forall m a b. (MonadIO m, MonadThrow m, Unbox a)
      => PRD.Parser a m b
      -> D.Stream m (Array.Array a)
      -> m (b, D.Stream m (Array.Array a))
  parseBreakD
      (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do
  
      res <- initial
      case res of
          PRD.IPartial s -> go SPEC state (List []) s
          PRD.IDone b -> return (b, stream)
          PRD.IError err -> throwM $ ParseError err
  
      where
  
      -- "backBuf" contains last few items in the stream that we may have to
      -- backtrack to.
      --
      -- XXX currently we are using a dumb list based approach for backtracking
      -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
      -- That will allow us more efficient random back and forth movement.
      go !_ st backBuf !pst = do
          r <- step defState st
          case r of
              D.Yield (Array contents start end) s ->
                  gobuf SPEC s backBuf
                      (Tuple' end contents) start pst
              D.Skip s -> go SPEC s backBuf pst
              D.Stop -> do
                  b <- extract pst
                  return (b, D.nil)
  
      -- Use strictness on "cur" to keep it unboxed
      gobuf !_ s backBuf (Tuple' end _) !cur !pst
          | cur == end = do
              go SPEC s backBuf pst
      gobuf !_ s backBuf fp@(Tuple' end contents) !cur !pst = do
          x <- liftIO $ peekByteIndex contents cur
          pRes <- pstep pst x
          let next = INDEX_NEXT(cur,a)
          case pRes of
              PR.Partial 0 pst1 ->
                   gobuf SPEC s (List []) fp next pst1
              PR.Partial n pst1 -> do
                  assert (n <= Prelude.length (x:getList backBuf)) (return ())
                  let src0 = Prelude.take n (x:getList backBuf)
                      arr0 = A.fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  let !(Array cont1 start end1) = src
                      fp1 = Tuple' end1 cont1
                  gobuf SPEC s (List []) fp1 start pst1
              PR.Continue 0 pst1 ->
                  gobuf SPEC s (List (x:getList backBuf)) fp next pst1
              PR.Continue n pst1 -> do
                  assert (n <= Prelude.length (x:getList backBuf)) (return ())
                  let (src0, buf1) = splitAt n (x:getList backBuf)
                      arr0 = A.fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  let !(Array cont1 start end1) = src
                      fp1 = Tuple' end1 cont1
                  gobuf SPEC s (List buf1) fp1 start pst1
              PR.Done 0 b -> do
                  let arr = Array contents next end
                  return (b, D.cons arr (D.Stream step s))
              PR.Done n b -> do
                  assert (n <= Prelude.length (x:getList backBuf)) (return ())
                  let src0 = Prelude.take n (x:getList backBuf)
                      -- XXX create the array in reverse instead
                      arr0 = A.fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      -- XXX Use StreamK to avoid adding arbitrary layers of
                      -- constructors every time.
                      str = D.cons arr0 (D.cons arr1 (D.Stream step s))
                  return (b, str)
              PR.SError err -> throwM $ ParseError err
  
  -- | Parse an array stream using the supplied 'Parser'.  Returns the parse
  -- result and the unconsumed stream. Throws 'ParseError' if the parse fails.
  --
  -- 'parseBreak' is an alternative to this function which allows composing the
  -- parser using the parser Monad.
  --
  -- We can compare perf and remove this one or define it in terms of that.
  --
  -- /Internal/
  --
  {-# INLINE_NORMAL parseBreakChunksK #-}
  parseBreakChunksK ::
         forall m a b. (MonadIO m, Unbox a)
      => Parser a m b
      -> StreamK m (Array a)
      -> m (Either ParseError b, StreamK m (Array a))
  parseBreakChunksK (Parser pstep initial extract) stream = do
      res <- initial
      case res of
          IPartial s -> go s stream []
          IDone b -> return (Right b, stream)
          IError err -> return (Left (ParseError err), stream)
  
      where
  
      -- "backBuf" contains last few items in the stream that we may have to
      -- backtrack to.
      --
      -- XXX currently we are using a dumb list based approach for backtracking
      -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.
      -- That will allow us more efficient random back and forth movement.
      go !pst st backBuf = do
          let stop = goStop pst backBuf -- (, K.nil) <$> extract pst
              single a = yieldk a StreamK.nil
              yieldk arr r = goArray pst backBuf r arr
           in StreamK.foldStream defState yieldk single stop st
  
      -- Use strictness on "cur" to keep it unboxed
      goArray !pst backBuf st (Array _ cur end) | cur == end = go pst st backBuf
      goArray !pst backBuf st (Array contents cur end) = do
          x <- liftIO $ peekAt cur contents
          pRes <- pstep pst x
          let next = INDEX_NEXT(cur,a)
          case pRes of
              Parser.SPartial 1 s ->
                   goArray s [] st (Array contents next end)
              Parser.SPartial m s -> do
                  assertM(m <= 1)
                  let n = 1 - m
                  assert (n <= Prelude.length (x:backBuf)) (return ())
                  let src0 = Prelude.take n (x:backBuf)
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  goArray s [] st src
              Parser.SContinue 1 s ->
                  goArray s (x:backBuf) st (Array contents next end)
              Parser.SContinue m s -> do
                  assertM(m <= 1)
                  let n = 1 - m
                  assert (n <= Prelude.length (x:backBuf)) (return ())
                  let (src0, buf1) = Prelude.splitAt n (x:backBuf)
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  goArray s buf1 st src
              Parser.SDone 1 b -> do
                  let arr = Array contents next end
                  return (Right b, StreamK.cons arr st)
              Parser.SDone m b -> do
                  assertM(m <= 1)
                  let n = 1 - m
                  assert (n <= Prelude.length (x:backBuf)) (return ())
                  let src0 = Prelude.take n (x:backBuf)
                      -- XXX Use fromListRevN once implemented
                      -- arr0 = A.fromListRevN n src0
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      str = StreamK.cons arr0 (StreamK.cons arr1 st)
                  return (Right b, str)
              Parser.SError err -> do
                  let n = Prelude.length backBuf
                      arr0 = fromListN n (Prelude.reverse backBuf)
                      arr1 = Array contents cur end
                      str = StreamK.cons arr0 (StreamK.cons arr1 st)
                  return (Left (ParseError err), str)
  
      -- This is a simplified goArray
      goExtract !pst backBuf (Array _ cur end)
          | cur == end = goStop pst backBuf
      goExtract !pst backBuf (Array contents cur end) = do
          x <- liftIO $ peekAt cur contents
          pRes <- pstep pst x
          let next = INDEX_NEXT(cur,a)
          case pRes of
              Parser.SPartial 0 s ->
                   goExtract s [] (Array contents next end)
              Parser.SPartial m s -> do
                  assertM(m <= 0)
                  let n = negate m
                  assert (n <= Prelude.length (x:backBuf)) (return ())
                  let src0 = Prelude.take n (x:backBuf)
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  goExtract s [] src
              Parser.SContinue 0 s ->
                  goExtract s backBuf (Array contents next end)
              Parser.SContinue m s -> do
                  assertM(m <= 0)
                  let n = negate m
                  assert (n <= Prelude.length (x:backBuf)) (return ())
                  let (src0, buf1) = Prelude.splitAt n (x:backBuf)
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      src = arr0 <> arr1
                  goExtract s buf1 src
              Parser.SDone 0 b -> do
                  let arr = Array contents next end
                  return (Right b, StreamK.fromPure arr)
              Parser.SDone m b -> do
                  assertM(m <= 0)
                  let n = negate m
                  assert (n <= Prelude.length backBuf) (return ())
                  let src0 = Prelude.take n (x:backBuf)
                      -- XXX Use fromListRevN once implemented
                      -- arr0 = A.fromListRevN n src0
                      arr0 = fromListN n (Prelude.reverse src0)
                      arr1 = Array contents next end
                      str = StreamK.cons arr0 (StreamK.fromPure arr1)
                  return (Right b, str)
              Parser.SError err -> do
                  let n = Prelude.length backBuf
                      arr0 = fromListN n (Prelude.reverse backBuf)
                      arr1 = Array contents cur end
                      str = StreamK.cons arr0 (StreamK.fromPure arr1)
                  return (Left (ParseError err), str)
  
      -- This is a simplified goExtract
      {-# INLINE goStop #-}
      goStop !pst backBuf = do
          pRes <- extract pst
          case pRes of
              Parser.SPartial _ _ -> error "Bug: parseBreak: Partial in extract"
              Parser.SContinue 0 s ->
                  goStop s backBuf
              Parser.SContinue m s -> do
                  assertM(m <= 0)
                  let n = negate m
                  assert (n <= Prelude.length backBuf) (return ())
                  let (src0, buf1) = Prelude.splitAt n backBuf
                      arr = fromListN n (Prelude.reverse src0)
                  goExtract s buf1 arr
              Parser.SDone 0 b ->
                  return (Right b, StreamK.nil)
              Parser.SDone m b -> do
                  assertM(m <= 0)
                  let n = negate m
                  assert (n <= Prelude.length backBuf) (return ())
                  let src0 = Prelude.take n backBuf
                      -- XXX Use fromListRevN once implemented
                      -- arr0 = A.fromListRevN n src0
                      arr0 = fromListN n (Prelude.reverse src0)
                  return (Right b, StreamK.fromPure arr0)
              Parser.SError err -> do
                  let n = Prelude.length backBuf
                      arr0 = fromListN n (Prelude.reverse backBuf)
                  return (Left (ParseError err), StreamK.fromPure arr0)
  -}
  
  -- | Run a 'ParserK' over a 'StreamK' of Arrays and return the parse result and
  -- the remaining Stream.
  {-# INLINE parseBreak #-}
  parseBreak
      :: (Monad m, Unbox a)
      => ParserK (Array a) m b
      -> StreamK m (Array a)
      -> m (Either ParseError b, StreamK m (Array a))
  parseBreak = Drivers.parseBreakChunks
  
  -- | Like 'parseBreak' but includes stream position information in the error
  -- messages.
  --
  {-# INLINE parseBreakPos #-}
  parseBreakPos
      :: (Monad m, Unbox a)
      => ParserK (Array a) m b
      -> StreamK m (Array a)
      -> m (Either ParseErrorPos b, StreamK m (Array a))
  parseBreakPos = Drivers.parseBreakChunksPos
  
  {-# INLINE parse #-}
  parse :: (Monad m, Unbox a) =>
      ParserK (Array a) m b -> StreamK m (Array a) -> m (Either ParseError b)
  parse f = fmap fst . parseBreak f
  
  -- | Like 'parse' but includes stream position information in the error
  -- messages.
  --
  {-# INLINE parsePos #-}
  parsePos :: (Monad m, Unbox a) =>
      ParserK (Array a) m b -> StreamK m (Array a) -> m (Either ParseErrorPos b)
  parsePos f = fmap fst . parseBreakPos f
  
  -------------------------------------------------------------------------------
  -- Convert ParserD to ParserK
  -------------------------------------------------------------------------------
  
  {-# INLINE adaptCWith #-}
  adaptCWith
      :: forall m a s b r. (Monad m, Unbox a)
      => (s -> a -> m (ParserD.Step s b))
      -> m (ParserD.Initial s b)
      -> (s -> m (ParserD.Final s b))
      -> (ParseResult b -> Int -> Input (Array a) -> m (Step (Array a) m r))
      -> Int
      -> Int
      -> Input (Array a)
      -> m (Step (Array a) m r)
  adaptCWith pstep initial extract cont !offset0 !usedCount !input = do
      res <- initial
      case res of
          ParserD.IPartial pst -> do
              case input of
                  Chunk arr -> parseContChunk usedCount offset0 pst arr
                  None -> parseContNothing usedCount pst
          ParserD.IDone b -> cont (Success offset0 b) usedCount input
          ParserD.IError err -> cont (Failure offset0 err) usedCount input
  
      where
  
      -- XXX We can maintain an absolute position instead of relative that will
      -- help in reporting of error location in the stream.
      {-# NOINLINE parseContChunk #-}
      parseContChunk !count !offset !state arr@(Array contents start end) = do
           if offset >= 0
           then go SPEC (start + offset * SIZE_OF(a)) state
           else return $ Continue offset (parseCont count state)
  
          where
  
          {-# INLINE onDone #-}
          onDone n b =
              assert (n <= length arr)
                  (cont (Success n b) (count + n - offset) (Chunk arr))
  
          {-# INLINE callParseCont #-}
          callParseCont constr n pst1 =
              assert (n < 0 || n >= length arr)
                  (return $ constr n (parseCont (count + n - offset) pst1))
  
          {-# INLINE onPartial #-}
          onPartial = callParseCont Partial
  
          {-# INLINE onContinue #-}
          onContinue = callParseCont Continue
  
          {-# INLINE onError #-}
          onError n err =
              cont (Failure n err) (count + n - offset) (Chunk arr)
  
          {-# INLINE onBack #-}
          onBack offset1 elemSize constr pst = do
              let pos = offset1 - start
               in if pos >= 0
                  then go SPEC offset1 pst
                  else constr (pos `div` elemSize) pst
  
          -- Note: div may be expensive but the alternative is to maintain an element
          -- offset in addition to a byte offset or just the element offset and use
          -- multiplication to get the byte offset every time, both these options
          -- turned out to be more expensive than using div.
          go !_ !cur !pst | cur >= end =
              onContinue ((end - start) `div` SIZE_OF(a))  pst
          go !_ !cur !pst = do
              let !x = unsafeInlineIO $ peekAt cur contents
              pRes <- pstep pst x
              let elemSize = SIZE_OF(a)
                  next = INDEX_NEXT(cur,a)
                  move n = cur + n * elemSize
                  curOff = (cur - start) `div` elemSize
                  nextOff = (next - start) `div` elemSize
              case pRes of
                  ParserD.SDone 1 b ->
                      onDone nextOff b
                  ParserD.SDone 0 b ->
                      onDone curOff b
                  ParserD.SDone n b ->
                      onDone ((move n - start) `div` elemSize) b
                  ParserD.SPartial 1 pst1 ->
                      go SPEC next pst1
                  ParserD.SPartial 0 pst1 ->
                      go SPEC cur pst1
                  ParserD.SPartial n pst1 ->
                      onBack (move n) elemSize onPartial pst1
                  ParserD.SContinue 1 pst1 ->
                      go SPEC next pst1
                  ParserD.SContinue 0 pst1 ->
                      go SPEC cur pst1
                  ParserD.SContinue n pst1 ->
                      onBack (move n) elemSize onContinue pst1
                  ParserD.SError err ->
                      onError curOff err
  
      {-# NOINLINE parseContNothing #-}
      parseContNothing !count !pst = do
          r <- extract pst
          case r of
              ParserD.FDone n b ->
                  assert (n <= 0) (cont (Success n b) (count + n) None)
              ParserD.FContinue n pst1 ->
                  assert (n <= 0)
                      (return $ Continue n (parseCont (count + n) pst1))
              ParserD.FError err ->
                  -- XXX It is called only when there is no input arr. So using 0
                  -- as the position is correct?
                  cont (Failure 0 err) count None
  
      -- XXX Maybe we can use two separate continuations instead of using
      -- Just/Nothing cases here. That may help in avoiding the parseContJust
      -- function call.
      {-# INLINE parseCont #-}
      parseCont !cnt !pst (Chunk arr) = parseContChunk cnt 0 pst arr
      parseCont !cnt !pst None = parseContNothing cnt pst
  
  -- | Convert a 'Parser' to 'ParserK' working on an Array stream.
  --
  -- /Pre-release/
  --
  {-# INLINE_LATE toParserK #-}
  toParserK :: (Monad m, Unbox a) => ParserD.Parser a m b -> ParserK (Array a) m b
  toParserK (ParserD.Parser step initial extract) =
      ParserK.MkParser $ adaptCWith step initial extract
  