c9a821b · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add type specific Num instances instead of deriving from base

c9a821bHarendra Kumar9 years ago

src/Asyncly/AsyncT.hs+1−45

compare with master

unifiedsplit

@@ -42,7 +42,7 @@ module Asyncly.AsyncT

42 )

43where

45−import Control.Applicative (Alternative (..), liftA2)

45+import Control.Applicative (Alternative (..))

46import Control.Concurrent (ThreadId, forkIO,

47 myThreadId, threadDelay)

48import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

@@ -521,50 +521,6 @@ pushOneToCtx ctype m = do

521 pushWorker ctx

522 return ctx

523

524−------------------------------------------------------------------------------

525−-- Num

526−------------------------------------------------------------------------------

527−

528−instance (Monad m, Num a) => Num (Stream m a) where

529− fromInteger n = pure (fromInteger n)

530−

531− negate = fmap negate

532− abs = fmap abs

533− signum = fmap signum

534−

535− (+) = liftA2 (+)

536− (*) = liftA2 (*)

537− (-) = liftA2 (-)

538−

539−instance (Monad m, Fractional a) => Fractional (Stream m a) where

540− fromRational n = pure (fromRational n)

541−

542− recip = fmap recip

543−

544− (/) = liftA2 (/)

545−

546−instance (Monad m, Floating a) => Floating (Stream m a) where

547− pi = pure pi

548−

549− exp = fmap exp

550− sqrt = fmap sqrt

551− log = fmap log

552− sin = fmap sin

553− tan = fmap tan

554− cos = fmap cos

555− asin = fmap asin

556− atan = fmap atan

557− acos = fmap acos

558− sinh = fmap sinh

559− tanh = fmap tanh

560− cosh = fmap cosh

561− asinh = fmap asinh

562− atanh = fmap atanh

563− acosh = fmap acosh

564−

565− (**) = liftA2 (**)

566− logBase = liftA2 logBase

567−

568-------------------------------------------------------------------------------

524-------------------------------------------------------------------------------

569-- AsyncT transformer

525-- AsyncT transformer

570-------------------------------------------------------------------------------

526-------------------------------------------------------------------------------

src/Asyncly/Prelude.hs+195−17

@@ -71,8 +71,8 @@ import Asyncly.AsyncT

71------------------------------------------------------------------------------

73newtype StreamT m a = StreamT {getStreamT :: Stream m a}

74− deriving (Functor, Applicative, Monad, Semigroup, Monoid, Num, Fractional,

74+ deriving (Functor, Applicative, Monad, Semigroup, Monoid,

75− Floating, MonadTrans, MonadIO, MonadThrow)

75+ MonadTrans, MonadIO, MonadThrow)

77deriving instance MonadAsync m => Alternative (StreamT m)

78deriving instance MonadAsync m => MonadPlus (StreamT m)

@@ -85,15 +85,63 @@ instance Streaming StreamT where

85 toStream = getStreamT

86 fromStream = StreamT

88+-- XXX The Functor/Applicative/Num instances for all the types are exactly the

89+-- same, how can we reduce this boilerplate (use TH)? We cannot derive them

90+-- from a single base type because they depend on the Monad instance which is

91+-- different for each type.

92+

93+------------------------------------------------------------------------------

94+-- Num

95+------------------------------------------------------------------------------

96+

97+instance (Monad m, Num a) => Num (StreamT m a) where

98+ fromInteger n = pure (fromInteger n)

99+

100+ negate = fmap negate

101+ abs = fmap abs

102+ signum = fmap signum

103+

104+ (+) = liftA2 (+)

105+ (*) = liftA2 (*)

106+ (-) = liftA2 (-)

107+

108+instance (Monad m, Fractional a) => Fractional (StreamT m a) where

109+ fromRational n = pure (fromRational n)

110+

111+ recip = fmap recip

112+

113+ (/) = liftA2 (/)

114+

115+instance (Monad m, Floating a) => Floating (StreamT m a) where

116+ pi = pure pi

117+

118+ exp = fmap exp

119+ sqrt = fmap sqrt

120+ log = fmap log

121+ sin = fmap sin

122+ tan = fmap tan

123+ cos = fmap cos

124+ asin = fmap asin

125+ atan = fmap atan

126+ acos = fmap acos

127+ sinh = fmap sinh

128+ tanh = fmap tanh

129+ cosh = fmap cosh

130+ asinh = fmap asinh

131+ atanh = fmap atanh

132+ acosh = fmap acosh

133+

134+ (**) = liftA2 (**)

135+ logBase = liftA2 logBase

136+

88------------------------------------------------------------------------------

137------------------------------------------------------------------------------

89-- InterleavedT interleaves the iterations in the Monad and Applicative

138-- InterleavedT interleaves the iterations in the Monad and Applicative

90-- compositions. It does not make much sense for infinite streams.

139-- compositions. It does not make much sense for infinite streams.

91------------------------------------------------------------------------------

140------------------------------------------------------------------------------

141

93−-- XXX the applicative, functor and num instances may not be correct

94newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

142newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

95− deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

143+ deriving (Functor, Applicative, Semigroup, Monoid,

96− Floating, MonadTrans, MonadIO, MonadThrow)

144+ MonadTrans, MonadIO, MonadThrow)

145

98deriving instance MonadAsync m => Alternative (InterleavedT m)

146deriving instance MonadAsync m => Alternative (InterleavedT m)

99deriving instance MonadAsync m => MonadPlus (InterleavedT m)

147deriving instance MonadAsync m => MonadPlus (InterleavedT m)

@@ -106,9 +154,9 @@ instance Streaming InterleavedT where

106 toStream = getInterleavedT

154 toStream = getInterleavedT

107 fromStream = InterleavedT

155 fromStream = InterleavedT

108

156

157+-- | Execute a monadic action for each element in the stream, in a fairly

158+-- interleaved manner i.e. iterations yield alternately.

109instance Monad m => Monad (InterleavedT m) where

159instance Monad m => Monad (InterleavedT m) where

110− -- | Execute a monadic action for each element in the stream, in a fairly

111− -- interleaved manner i.e. iterations yield alternately.

112 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

160 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

113 let run x = (runStream x) Nothing stp yld

161 let run x = (runStream x) Nothing stp yld

114 yield a Nothing = run $ getInterleavedT (f a)

162 yield a Nothing = run $ getInterleavedT (f a)

@@ -117,6 +165,50 @@ instance Monad m => Monad (InterleavedT m) where

117 getInterleavedT ((InterleavedT r) >>= f)

165 getInterleavedT ((InterleavedT r) >>= f)

118 in m Nothing stp yield

166 in m Nothing stp yield

119

167

168+------------------------------------------------------------------------------

169+-- Num

170+------------------------------------------------------------------------------

171+

172+instance (Monad m, Num a) => Num (InterleavedT m a) where

173+ fromInteger n = pure (fromInteger n)

174+

175+ negate = fmap negate

176+ abs = fmap abs

177+ signum = fmap signum

178+

179+ (+) = liftA2 (+)

180+ (*) = liftA2 (*)

181+ (-) = liftA2 (-)

182+

183+instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where

184+ fromRational n = pure (fromRational n)

185+

186+ recip = fmap recip

187+

188+ (/) = liftA2 (/)

189+

190+instance (Monad m, Floating a) => Floating (InterleavedT m a) where

191+ pi = pure pi

192+

193+ exp = fmap exp

194+ sqrt = fmap sqrt

195+ log = fmap log

196+ sin = fmap sin

197+ tan = fmap tan

198+ cos = fmap cos

199+ asin = fmap asin

200+ atan = fmap atan

201+ acos = fmap acos

202+ sinh = fmap sinh

203+ tanh = fmap tanh

204+ cosh = fmap cosh

205+ asinh = fmap asinh

206+ atanh = fmap atanh

207+ acosh = fmap acosh

208+

209+ (**) = liftA2 (**)

210+ logBase = liftA2 logBase

211+

120------------------------------------------------------------------------------

212------------------------------------------------------------------------------

121-- AsyncT runs the iterations asynchronously in the Monad and Applicative

213-- AsyncT runs the iterations asynchronously in the Monad and Applicative

122-- compositions. More iterations are started in parallel on demand. It can work

214-- compositions. More iterations are started in parallel on demand. It can work

@@ -124,8 +216,7 @@ instance Monad m => Monad (InterleavedT m) where

124------------------------------------------------------------------------------

216------------------------------------------------------------------------------

125

217

126newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

218newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

127− deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

219+ deriving (Functor, Applicative, Semigroup, Monoid, MonadTrans)

128− Floating, MonadTrans)

129

220

130deriving instance MonadAsync m => MonadIO (AsyncT m)

221deriving instance MonadAsync m => MonadIO (AsyncT m)

131deriving instance MonadAsync m => MonadThrow (AsyncT m)

222deriving instance MonadAsync m => MonadThrow (AsyncT m)

@@ -155,14 +246,58 @@ parbind k m f = go m

155 yield a (Just r) = run $ f a `k` (go r)

246 yield a (Just r) = run $ f a `k` (go r)

156 in g Nothing stp yield

247 in g Nothing stp yield

157

248

249+-- | Execute a monadic action for each element in the stream, running

250+-- iterations in parallel, but giving preference to iterations started

251+-- earlier.

158instance MonadAsync m => Monad (AsyncT m) where

252instance MonadAsync m => Monad (AsyncT m) where

159− -- | Execute a monadic action for each element in the stream, running

160− -- iterations in parallel, but giving preference to iterations started

161− -- earlier.

162 (AsyncT m) >>= f = AsyncT $ parbind par m g

253 (AsyncT m) >>= f = AsyncT $ parbind par m g

163 where g x = getAsyncT (f x)

254 where g x = getAsyncT (f x)

164 par = parallel (CtxType Conjunction LIFO)

255 par = parallel (CtxType Conjunction LIFO)

165

256

257+------------------------------------------------------------------------------

258+-- Num

259+------------------------------------------------------------------------------

260+

261+instance (Monad m, Num a) => Num (AsyncT m a) where

262+ fromInteger n = pure (fromInteger n)

263+

264+ negate = fmap negate

265+ abs = fmap abs

266+ signum = fmap signum

267+

268+ (+) = liftA2 (+)

269+ (*) = liftA2 (*)

270+ (-) = liftA2 (-)

271+

272+instance (Monad m, Fractional a) => Fractional (AsyncT m a) where

273+ fromRational n = pure (fromRational n)

274+

275+ recip = fmap recip

276+

277+ (/) = liftA2 (/)

278+

279+instance (Monad m, Floating a) => Floating (AsyncT m a) where

280+ pi = pure pi

281+

282+ exp = fmap exp

283+ sqrt = fmap sqrt

284+ log = fmap log

285+ sin = fmap sin

286+ tan = fmap tan

287+ cos = fmap cos

288+ asin = fmap asin

289+ atan = fmap atan

290+ acos = fmap acos

291+ sinh = fmap sinh

292+ tanh = fmap tanh

293+ cosh = fmap cosh

294+ asinh = fmap asinh

295+ atanh = fmap atanh

296+ acosh = fmap acosh

297+

298+ (**) = liftA2 (**)

299+ logBase = liftA2 logBase

300+

166------------------------------------------------------------------------------

301------------------------------------------------------------------------------

167-- ParallelT runs all iterations in parallel in the Monad and Applicative

302-- ParallelT runs all iterations in parallel in the Monad and Applicative

168-- compositions. Iterations are scheduled in a round-robin fashion. It cannot

303-- compositions. Iterations are scheduled in a round-robin fashion. It cannot

@@ -170,8 +305,7 @@ instance MonadAsync m => Monad (AsyncT m) where

170------------------------------------------------------------------------------

305------------------------------------------------------------------------------

171

306

172newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

307newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

173− deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

308+ deriving (Functor, Applicative, Semigroup, Monoid, MonadTrans)

174− Floating, MonadTrans)

175

309

176deriving instance MonadAsync m => MonadIO (ParallelT m)

310deriving instance MonadAsync m => MonadIO (ParallelT m)

177deriving instance MonadAsync m => MonadThrow (ParallelT m)

311deriving instance MonadAsync m => MonadThrow (ParallelT m)

@@ -186,14 +320,58 @@ instance Streaming ParallelT where

186 toStream = getParallelT

320 toStream = getParallelT

187 fromStream = ParallelT

321 fromStream = ParallelT

188

322

323+-- | Execute a monadic action for each element in the stream, running

324+-- iterations in a fairly parallel manner, i.e. all iterations are equally

325+-- likely to run.

189instance MonadAsync m => Monad (ParallelT m) where

326instance MonadAsync m => Monad (ParallelT m) where

190− -- | Execute a monadic action for each element in the stream, running

191− -- iterations in a fairly parallel manner, i.e. all iterations are equally

192− -- likely to run.

193 (ParallelT m) >>= f = ParallelT $ parbind par m g

327 (ParallelT m) >>= f = ParallelT $ parbind par m g

194 where g x = getParallelT (f x)

328 where g x = getParallelT (f x)

195 par = parallel (CtxType Conjunction FIFO)

329 par = parallel (CtxType Conjunction FIFO)

196

330

331+------------------------------------------------------------------------------

332+-- Num

333+------------------------------------------------------------------------------

334+

335+instance (Monad m, Num a) => Num (ParallelT m a) where

336+ fromInteger n = pure (fromInteger n)

337+

338+ negate = fmap negate

339+ abs = fmap abs

340+ signum = fmap signum

341+

342+ (+) = liftA2 (+)

343+ (*) = liftA2 (*)

344+ (-) = liftA2 (-)

345+

346+instance (Monad m, Fractional a) => Fractional (ParallelT m a) where

347+ fromRational n = pure (fromRational n)

348+

349+ recip = fmap recip

350+

351+ (/) = liftA2 (/)

352+

353+instance (Monad m, Floating a) => Floating (ParallelT m a) where

354+ pi = pure pi

355+

356+ exp = fmap exp

357+ sqrt = fmap sqrt

358+ log = fmap log

359+ sin = fmap sin

360+ tan = fmap tan

361+ cos = fmap cos

362+ asin = fmap asin

363+ atan = fmap atan

364+ acos = fmap acos

365+ sinh = fmap sinh

366+ tanh = fmap tanh

367+ cosh = fmap cosh

368+ asinh = fmap asinh

369+ atanh = fmap atanh

370+ acosh = fmap acosh

371+

372+ (**) = liftA2 (**)

373+ logBase = liftA2 logBase

374+

197------------------------------------------------------------------------------

375------------------------------------------------------------------------------

198-- Serially Zipping Streams

376-- Serially Zipping Streams

199------------------------------------------------------------------------------

377------------------------------------------------------------------------------

test/Main.hs+112

@@ -229,6 +229,22 @@ main = hspec $ do

229 describe "Bind and compose" $

230 bindAndComposeHierarchy toListParallel (k g)

231

232+ -- Nest two lists using different styles of product compositions

233+ it "Nests two streams using monadic serial composition" nestTwoSerial

234+ it "Nests two streams using monadic interleaved composition" nestTwoInterleaved

235+ it "Nests two streams using monadic async composition" nestTwoAsync

236+ it "Nests two streams using monadic parallel composition" nestTwoParallel

237+

238+ it "Nests two streams using applicative serial composition" nestTwoSerialApp

239+ it "Nests two streams using applicative interleaved composition" nestTwoInterleavedApp

240+ it "Nests two streams using applicative async composition" nestTwoAsyncApp

241+ it "Nests two streams using applicative parallel composition" nestTwoParallelApp

242+

243+ it "Nests two streams using Num serial composition" nestTwoSerialNum

244+ it "Nests two streams using Num interleaved composition" nestTwoInterleavedNum

245+ it "Nests two streams using Num async composition" nestTwoAsyncNum

246+ it "Nests two streams using Num parallel composition" nestTwoParallelNum

247+

232 ---------------------------------------------------------------------------

248 ---------------------------------------------------------------------------

233 -- TBD Bind and Bind combinations

249 -- TBD Bind and Bind combinations

234 ---------------------------------------------------------------------------

250 ---------------------------------------------------------------------------

@@ -242,6 +258,102 @@ main = hspec $ do

242 describe "Async zipping" $

258 describe "Async zipping" $

243 zipOps A.zipAsyncWith A.zipAsyncWithM zippingAsync

259 zipOps A.zipAsyncWith A.zipAsyncWithM zippingAsync

244

260

261+nestTwoSerial :: Expectation

262+nestTwoSerial =

263+ let s1 = foldMapWith (<>) return [1..4]

264+ s2 = foldMapWith (<>) return [5..8]

265+ in toListSerial (do

266+ x <- s1

267+ y <- s2

268+ return (x + y)

269+ ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

270+

271+nestTwoSerialApp :: Expectation

272+nestTwoSerialApp =

273+ let s1 = foldMapWith (<>) return [1..4]

274+ s2 = foldMapWith (<>) return [5..8]

275+ in toListSerial ((+) <$> s1 <*> s2)

276+ `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

277+

278+nestTwoSerialNum :: Expectation

279+nestTwoSerialNum =

280+ let s1 = foldMapWith (<>) return [1..4]

281+ s2 = foldMapWith (<>) return [5..8]

282+ in toListSerial (s1 + s2)

283+ `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

284+

285+nestTwoInterleaved :: Expectation

286+nestTwoInterleaved =

287+ let s1 = foldMapWith (<>) return [1..4]

288+ s2 = foldMapWith (<>) return [5..8]

289+ in toListInterleaved (do

290+ x <- s1

291+ y <- s2

292+ return (x + y)

293+ ) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

294+

295+nestTwoInterleavedApp :: Expectation

296+nestTwoInterleavedApp =

297+ let s1 = foldMapWith (<>) return [1..4]

298+ s2 = foldMapWith (<>) return [5..8]

299+ in toListInterleaved ((+) <$> s1 <*> s2)

300+ `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

301+

302+nestTwoInterleavedNum :: Expectation

303+nestTwoInterleavedNum =

304+ let s1 = foldMapWith (<>) return [1..4]

305+ s2 = foldMapWith (<>) return [5..8]

306+ in toListInterleaved (s1 + s2)

307+ `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

308+

309+nestTwoAsync :: Expectation

310+nestTwoAsync =

311+ let s1 = foldMapWith (<>) return [1..4]

312+ s2 = foldMapWith (<>) return [5..8]

313+ in toListAsync (do

314+ x <- s1

315+ y <- s2

316+ return (x + y)

317+ ) `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

318+

319+nestTwoAsyncApp :: Expectation

320+nestTwoAsyncApp =

321+ let s1 = foldMapWith (<>) return [1..4]

322+ s2 = foldMapWith (<>) return [5..8]

323+ in toListAsync ((+) <$> s1 <*> s2)

324+ `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

325+

326+nestTwoAsyncNum :: Expectation

327+nestTwoAsyncNum =

328+ let s1 = foldMapWith (<>) return [1..4]

329+ s2 = foldMapWith (<>) return [5..8]

330+ in toListAsync (s1 + s2)

331+ `shouldReturn` ([6,7,8,9,7,8,9,10,8,9,10,11,9,10,11,12] :: [Int])

332+

333+nestTwoParallel :: Expectation

334+nestTwoParallel =

335+ let s1 = foldMapWith (<>) return [1..4]

336+ s2 = foldMapWith (<>) return [5..8]

337+ in toListParallel (do

338+ x <- s1

339+ y <- s2

340+ return (x + y)

341+ ) `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

342+

343+nestTwoParallelApp :: Expectation

344+nestTwoParallelApp =

345+ let s1 = foldMapWith (<>) return [1..4]

346+ s2 = foldMapWith (<>) return [5..8]

347+ in toListParallel ((+) <$> s1 <*> s2)

348+ `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

349+

350+nestTwoParallelNum :: Expectation

351+nestTwoParallelNum =

352+ let s1 = foldMapWith (<>) return [1..4]

353+ s2 = foldMapWith (<>) return [5..8]

354+ in toListParallel (s1 + s2)

355+ `shouldReturn` ([6,7,7,8,8,8,9,9,9,9,10,10,10,11,11,12] :: [Int])

356+

245zipOps :: (Streaming t, Applicative (t IO))

357zipOps :: (Streaming t, Applicative (t IO))

246 => (forall a b c. (a -> b -> c)

358 => (forall a b c. (a -> b -> c)

247 -> StreamT IO a -> StreamT IO b -> StreamT IO c)

359 -> StreamT IO a -> StreamT IO b -> StreamT IO c)