Torch.Data.StreamedPipeline

Defining a Datastream

We will show how to retrieve the IMDB dataset as an example datastream. The dataset used here can be found at https://ai.stanford.edu/~amaas/data/sentiment/

import Pipes
import qualified Pipes.Safe as Safe
import qualified Pipes.Prelude as P
import System.Directory

newtype Imdb = Imdb { dataDir :: String }

data Sentiment = Positive | Negative

instance (MonadBaseControl IO m, MonadSafe m) => Datastream m Sentiment Imdb (Text, Sentiment) where
  streamSamples Imdb{..} sent = Select $ do
    rawFilePaths <- zip (repeat sent) <$> (liftIO $ listDirectory (dataDir </> sentToPath sent))
    let filePaths = fmap (second $ mappend (dataDir </> sentToPath sent)) rawFilePaths
    for (each filePaths) $ \(rev, fp) -> Safe.withFile fp ReadMode $ \fh ->
      P.zip (PT.fromHandleLn fh) (yield rev)
        where sentToPath Pos = "pos" ++ pure pathSeparator
              sentToPath Neg = "neg" ++ pure pathSeparator

This streams in movie reviews from each file in either the positive review directory or the negative review directory, depending on the seed value used.

This highlights a use of seed values that is more interesting than just specifying the thread count, but also has some problems. When running this datastream with either streamFrom or 'streamFrom'', you need to supply both Positive and Negative values as seeds to retrieve the entire IMDB dataset, and in this case positive and negative reviews will be streamed in concurrently. The problem with designing a datastream in this fashion is you limit the amount of concurrency (2 threads in this case) without duplicating data. Ultimately though seeds should be quite flexible and allow you to design the concurrency how you see fit. Be careful not to use duplicate seed values unless you want duplicate data.

Datastream

class Monad m => Datastream m seed dataset sample | dataset -> sample where

The base datastream class. A dataset returns a stream of samples based on a seed value.

Methods

streamSamples :: dataset -> seed -> ListT m sample

Instances

(MonadBaseControl IO m, MonadSafe m, FromRecord batch) => Datastream m () (CsvDatastream batch) (Vector batch)

Defined in Torch.Data.CsvDatastream

Methods

streamSamples :: CsvDatastream batch -> () -> ListT m (Vector batch)

(MonadBaseControl IO m, MonadSafe m, FromNamedRecord batch) => Datastream m () (CsvDatastreamNamed batch) (Vector batch)

Defined in Torch.Data.CsvDatastream

Methods

streamSamples :: CsvDatastreamNamed batch -> () -> ListT m (Vector batch)

Monad m => Datastream m Int (MNIST m) (Tensor, Tensor)

Defined in Torch.Vision

Methods

streamSamples :: MNIST m -> Int -> ListT m (Tensor, Tensor)

Datastream m seed dataset batch => Datastream m seed (CollatedDataset m dataset batch collatedBatch) collatedBatch

Defined in Torch.Data.Dataset

Methods

streamSamples :: CollatedDataset m dataset batch collatedBatch -> seed -> ListT m collatedBatch

newtype DatastreamOptions

Datastream options used when looding datastreams. Currently only buffer size is configurable, since thread count is controlled by the number of seeds (see streamFrom functions).

Constructors

DatastreamOptions

Fields

• bufferSize :: Int

  Max number of samples stored in each buffer at a given time.

datastreamOpts :: DatastreamOptions

Default dataloader options, you should override the fields in this record.

Dataloading

streamFrom :: forall sample m dataset seed b. (Datastream m seed dataset sample, MonadBaseControl IO m, MonadBase IO m) => DatastreamOptions -> dataset -> ListT m seed -> ContT b m (ListT m sample)

Return a stream of samples from the given dataset as a continuation. A stream of samples is generated for every seed in the given stream of seeds, and all of these streams are merged into the output stream in a non-deterministic order (if you need determinism see 'streamFrom''). Every stream created for each seed value is made in its own thread.

streamFrom' :: forall sample m f dataset seed b. (Show sample, Datastream m seed dataset sample, MonadBaseControl IO m, MonadBase IO m, MonadIO m, Foldable f) => DatastreamOptions -> dataset -> f seed -> ContT b m (ListT m sample)

This function is the same as streamFrom except the seeds are specified as a Foldable, and the stream returned has a deterministic ordering. The results from each given seed are interspersed in the order defined by the Foldable of seeds.

Reexports

class (Applicative b, Applicative m, Monad b, Monad m) => MonadBase (b :: Type -> Type) (m :: Type -> Type) | m -> b where

Methods

liftBase :: b α -> m α

Lift a computation from the base monad

Instances

MonadBase Identity Identity

Defined in Control.Monad.Base

Methods

liftBase :: Identity α -> Identity α

MonadBase STM STM

Defined in Control.Monad.Base

Methods

liftBase :: STM α -> STM α

MonadBase IO IO

Defined in Control.Monad.Base

Methods

liftBase :: IO α -> IO α

MonadBase Maybe Maybe

Defined in Control.Monad.Base

Methods

liftBase :: Maybe α -> Maybe α

MonadBase List List

Defined in Control.Monad.Base

Methods

liftBase :: [α] -> [α]

MonadBase b m => MonadBase b (SafeT m)

Defined in Pipes.Safe

Methods

liftBase :: b α -> SafeT m α

MonadBase b m => MonadBase b (MaybeT m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> MaybeT m α

(Functor f, MonadBase b m) => MonadBase b (FreeT f m)

Defined in Control.Monad.Trans.Free

Methods

liftBase :: b α -> FreeT f m α

(Monoid w, MonadBase b m) => MonadBase b (AccumT w m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> AccumT w m α

MonadBase b m => MonadBase b (ExceptT e m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> ExceptT e m α

MonadBase b m => MonadBase b (IdentityT m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> IdentityT m α

MonadBase b m => MonadBase b (ReaderT r m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> ReaderT r m α

MonadBase b m => MonadBase b (SelectT r m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> SelectT r m α

MonadBase b m => MonadBase b (StateT s m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> StateT s m α

MonadBase b m => MonadBase b (StateT s m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> StateT s m α

(Monoid w, MonadBase b m) => MonadBase b (WriterT w m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> WriterT w m α

(Monoid w, MonadBase b m) => MonadBase b (WriterT w m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> WriterT w m α

(Monoid w, MonadBase b m) => MonadBase b (WriterT w m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> WriterT w m α

MonadBase b m => MonadBase b (ContT r m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> ContT r m α

(Monoid w, MonadBase b m) => MonadBase b (RWST r w s m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> RWST r w s m α

(Monoid w, MonadBase b m) => MonadBase b (RWST r w s m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> RWST r w s m α

(Monoid w, MonadBase b m) => MonadBase b (RWST r w s m)

Defined in Control.Monad.Base

Methods

liftBase :: b α -> RWST r w s m α

MonadBase IO m => MonadBase IO (Proxy a' a b' b m)

Defined in Torch.Data.Internal

Methods

liftBase :: IO α -> Proxy a' a b' b m α

MonadBase (ST s) (ST s)

Defined in Control.Monad.Base

Methods

liftBase :: ST s α -> ST s α

MonadBase (Either e) (Either e)

Defined in Control.Monad.Base

Methods

liftBase :: Either e α -> Either e α

MonadBase (ST s) (ST s)

Defined in Control.Monad.Base

Methods

liftBase :: ST s α -> ST s α

MonadBase ((->) r) ((->) r)

Defined in Control.Monad.Base

Methods

liftBase :: (r -> α) -> r -> α

class MonadBase b m => MonadBaseControl (b :: Type -> Type) (m :: Type -> Type) | m -> b where

Writing instances

The usual way to write a MonadBaseControl instance for a transformer stack over a base monad B is to write an instance MonadBaseControl B B for the base monad, and MonadTransControl T instances for every transformer T. Instances for MonadBaseControl are then simply implemented using ComposeSt, defaultLiftBaseWith, defaultRestoreM.

Associated Types

type StM (m :: Type -> Type) a

Monadic state that m adds to the base monad b.

For all base (non-transformed) monads, StM m a ~ a:

StM IO         a ~ a
StM Maybe      a ~ a
StM (Either e) a ~ a
StM []         a ~ a
StM ((->) r)   a ~ a
StM Identity   a ~ a
StM STM        a ~ a
StM (ST s)     a ~ a

If m is a transformed monad, m ~ t b, StM is the monadic state of the transformer t (given by its StT from MonadTransControl). For a transformer stack, StM is defined recursively:

StM (IdentityT  m) a ~ ComposeSt IdentityT m a ~ StM m a
StM (MaybeT     m) a ~ ComposeSt MaybeT    m a ~ StM m (Maybe a)
StM (ErrorT e   m) a ~ ComposeSt ErrorT    m a ~ Error e => StM m (Either e a)
StM (ExceptT e  m) a ~ ComposeSt ExceptT   m a ~ StM m (Either e a)
StM (ListT      m) a ~ ComposeSt ListT     m a ~ StM m [a]
StM (ReaderT r  m) a ~ ComposeSt ReaderT   m a ~ StM m a
StM (StateT s   m) a ~ ComposeSt StateT    m a ~ StM m (a, s)
StM (WriterT w  m) a ~ ComposeSt WriterT   m a ~ Monoid w => StM m (a, w)
StM (RWST r w s m) a ~ ComposeSt RWST      m a ~ Monoid w => StM m (a, s, w)

Methods

liftBaseWith :: (RunInBase m b -> b a) -> m a

liftBaseWith is similar to liftIO and liftBase in that it lifts a base computation to the constructed monad.

Instances should satisfy similar laws as the MonadIO and MonadBase laws:

liftBaseWith (\_ -> return a) = return a

liftBaseWith (\_ -> m >>= f)  =  liftBaseWith (\_ -> m) >>= (\a -> liftBaseWith (\_ -> f a))

As Li-yao Xia explains, parametricity guarantees that

f $ liftBaseWith q = liftBaseWith $ runInBase -> f $ q runInBase

The difference with liftBase is that before lifting the base computation liftBaseWith captures the state of m. It then provides the base computation with a RunInBase function that allows running m computations in the base monad on the captured state:

withFileLifted :: MonadBaseControl IO m => FilePath -> IOMode -> (Handle -> m a) -> m a
withFileLifted file mode action = liftBaseWith (\runInBase -> withFile file mode (runInBase . action)) >>= restoreM
                             -- = control $ \runInBase -> withFile file mode (runInBase . action)
                             -- = liftBaseOp (withFile file mode) action

liftBaseWith is usually not implemented directly, but using defaultLiftBaseWith.

restoreM :: StM m a -> m a

Construct a m computation from the monadic state of m that is returned from a RunInBase function.

Instances should satisfy:

liftBaseWith (\runInBase -> runInBase m) >>= restoreM = m

restoreM is usually not implemented directly, but using defaultRestoreM.

Instances

MonadBaseControl Identity Identity

Defined in Control.Monad.Trans.Control

Associated Types

type StM Identity a

Methods

liftBaseWith :: (RunInBase Identity Identity -> Identity a) -> Identity a

restoreM :: StM Identity a -> Identity a

MonadBaseControl STM STM

Defined in Control.Monad.Trans.Control

Associated Types

type StM STM a

Methods

liftBaseWith :: (RunInBase STM STM -> STM a) -> STM a

restoreM :: StM STM a -> STM a

MonadBaseControl IO IO

Defined in Control.Monad.Trans.Control

Associated Types

type StM IO a

Methods

liftBaseWith :: (RunInBase IO IO -> IO a) -> IO a

restoreM :: StM IO a -> IO a

MonadBaseControl Maybe Maybe

Defined in Control.Monad.Trans.Control

Associated Types

type StM Maybe a

Methods

liftBaseWith :: (RunInBase Maybe Maybe -> Maybe a) -> Maybe a

restoreM :: StM Maybe a -> Maybe a

MonadBaseControl List List

Defined in Control.Monad.Trans.Control

Associated Types

type StM List a

Methods

liftBaseWith :: (RunInBase List List -> [a]) -> [a]

restoreM :: StM List a -> [a]

MonadBaseControl b m => MonadBaseControl b (SafeT m)

Defined in Pipes.Safe

Associated Types

type StM (SafeT m) a

Methods

liftBaseWith :: (RunInBase (SafeT m) b -> b a) -> SafeT m a

restoreM :: StM (SafeT m) a -> SafeT m a

MonadBaseControl b m => MonadBaseControl b (MaybeT m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (MaybeT m) a

Methods

liftBaseWith :: (RunInBase (MaybeT m) b -> b a) -> MaybeT m a

restoreM :: StM (MaybeT m) a -> MaybeT m a

MonadBaseControl b m => MonadBaseControl b (ExceptT e m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (ExceptT e m) a

Methods

liftBaseWith :: (RunInBase (ExceptT e m) b -> b a) -> ExceptT e m a

restoreM :: StM (ExceptT e m) a -> ExceptT e m a

MonadBaseControl b m => MonadBaseControl b (IdentityT m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (IdentityT m) a

Methods

liftBaseWith :: (RunInBase (IdentityT m) b -> b a) -> IdentityT m a

restoreM :: StM (IdentityT m) a -> IdentityT m a

MonadBaseControl b m => MonadBaseControl b (ReaderT r m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (ReaderT r m) a

Methods

liftBaseWith :: (RunInBase (ReaderT r m) b -> b a) -> ReaderT r m a

restoreM :: StM (ReaderT r m) a -> ReaderT r m a

MonadBaseControl b m => MonadBaseControl b (StateT s m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (StateT s m) a

Methods

liftBaseWith :: (RunInBase (StateT s m) b -> b a) -> StateT s m a

restoreM :: StM (StateT s m) a -> StateT s m a

MonadBaseControl b m => MonadBaseControl b (StateT s m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (StateT s m) a

Methods

liftBaseWith :: (RunInBase (StateT s m) b -> b a) -> StateT s m a

restoreM :: StM (StateT s m) a -> StateT s m a

(Monoid w, MonadBaseControl b m) => MonadBaseControl b (WriterT w m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (WriterT w m) a

Methods

liftBaseWith :: (RunInBase (WriterT w m) b -> b a) -> WriterT w m a

restoreM :: StM (WriterT w m) a -> WriterT w m a

(Monoid w, MonadBaseControl b m) => MonadBaseControl b (WriterT w m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (WriterT w m) a

Methods

liftBaseWith :: (RunInBase (WriterT w m) b -> b a) -> WriterT w m a

restoreM :: StM (WriterT w m) a -> WriterT w m a

(Monoid w, MonadBaseControl b m) => MonadBaseControl b (RWST r w s m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (RWST r w s m) a

Methods

liftBaseWith :: (RunInBase (RWST r w s m) b -> b a) -> RWST r w s m a

restoreM :: StM (RWST r w s m) a -> RWST r w s m a

(Monoid w, MonadBaseControl b m) => MonadBaseControl b (RWST r w s m)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (RWST r w s m) a

Methods

liftBaseWith :: (RunInBase (RWST r w s m) b -> b a) -> RWST r w s m a

restoreM :: StM (RWST r w s m) a -> RWST r w s m a

MonadBaseControl (ST s) (ST s)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (ST s) a

Methods

liftBaseWith :: (RunInBase (ST s) (ST s) -> ST s a) -> ST s a

restoreM :: StM (ST s) a -> ST s a

MonadBaseControl (Either e) (Either e)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (Either e) a

Methods

liftBaseWith :: (RunInBase (Either e) (Either e) -> Either e a) -> Either e a

restoreM :: StM (Either e) a -> Either e a

MonadBaseControl (ST s) (ST s)

Defined in Control.Monad.Trans.Control

Associated Types

type StM (ST s) a

Methods

liftBaseWith :: (RunInBase (ST s) (ST s) -> ST s a) -> ST s a

restoreM :: StM (ST s) a -> ST s a

MonadBaseControl ((->) r) ((->) r)

Defined in Control.Monad.Trans.Control

Associated Types

type StM ((->) r) a

Methods

liftBaseWith :: (RunInBase ((->) r) ((->) r) -> r -> a) -> r -> a

restoreM :: StM ((->) r) a -> r -> a