docs/User/Project/Upgrading-0.8-to-0.9.md · streamly

git.iko.soy/streamly

streamly/docs/User/Project/Upgrading-0.8-to-0.9.md

384 lines 13 kB history raw

1# Upgrading to streamly 0.9.0

3Also see the detailed changelog describing all the changes in the release:

5* [0.9.0 API Changelog](/docs/User/Project/ApiChangelogs/0.8.3-0.9.0.txt) or

6* [0.9.0 API Changelog](https://hackage.haskell.org/package/streamly-0.9.0/docs/docs/User/ProjectRelated/ApiChangelogs/0.8.3-0.9.0.txt)

8`Streamly.Prelude` module has been deprecated, equivalent

9functionality is covered by the `Streamly.Data.Stream`,

10`Streamly.Data.Stream.Prelude`, and `Streamly.Data.Fold` modules. The

11new modules use a monomorphic `Stream` type instead of the polymorphic

12`IsStream t` type.

14`Streamly.Data.Stream` module and the `Stream` type are designed

15for writing high-performance fused pipelines not involving explicit

16recursion. For writing code that may require recursive function calls,

17`Streamly.Data.Stream.StreamK` module and the `StreamK` type have been

18introduced which provide a CPS based stream implementation. `Stream` and

19`StreamK` types can be easily interconverted. These changes have been made to

20make performance robust and not rely on GHC rewrite rules which could be

21fragile. For example, GHC 9.0.x had broken the rewrite rule which was

22not fixed until GHC-9.2.2. This split also gives more power and

23transparency of the performance behavior to the programmer.

25Instead of using separate stream types for concurrent code, now

26you have to use explicit concurrent combinators with appropriate

27concurrency parameters. These combinators are available in the

28`Streamly.Data.Stream.Prelude` module. This change has been made to allow

29programmers to control concurrent behavior in a more robust way and

30reduce pitfalls.

32The old code can be adapted to use the new modules with some

33changes. More details about this is supplied in the following sections.

34Please open an issue if you cannot find a way to adapt the code to the new

35release.

37Assume the following imports in the code snippets below:

39```haskell docspec

40>>> import qualified Streamly.Data.Stream as Stream

41>>> import qualified Streamly.Data.StreamK as StreamK

42>>> import qualified Streamly.Data.Stream.Prelude as Stream

43>>> import qualified Streamly.Data.Fold as Fold

44>>> import qualified Streamly.Data.Parser as Parser

45```

47## Dependencies in cabal file

49If you allow streamly < 0.9 in version bounds in your package then you

50may not be able to build if any of the following is true:

52* there is a dependency on `streamly-core` (module name conflicts)

53* there is a dependency on `streamly-bytestring` >= 0.2 (type conflict)

55You can define and use a build flag in your package to allow older

56streamly versions:

58```

59flag old-streamly

60 description: Allow streamly versions lower than 0.9

61 manual: True

62 default: False

63```

65```

66 if flag(old-streamly)

67 build-depends:

68 streamly >=0.8.3 && < 0.9

69 else

70 build-depends:

71 streamly-core >= 0.1.0 && < 0.2,

72 streamly >=0.9.0 && < 0.10

73```

75## The `Stream` and `StreamK` types

77The following types are removed:

79* `IsStream`

80* `ZipSerialM`

81* `ZipAsyncM`

82* `WSerialT`

83* `WAsyncT`

84* `SerialT`

85* `ParallelT`

86* `AsyncT`

87* `AheadT`

89In the new release, the `Stream` type is the primary stream type that

90you will use most of the time. You can think of it as a replacement for

91the `SerialT` type. However, it does not provide an Applicative or Monad

92instance.

94The `CrossStream` type in `Streamly.Internal.Data.Stream` is a

95wrapper over `Stream` type supplying the Monad instance. However, see

96the "Stream Fusion" section in the `Streamly.Data.Stream` module for

97limitations of the `Stream` type. The `StreamK` type and `CrossStreamK`

98(in Streamly.Internal.Data.Stream.StreamK) could be used to overcome the

99limitations of `Stream` type.

100

101If required, you can use the template-haskell functions in

102`Streamly.Data.Stream.MkType` to create stream type wrappers (like

103ZipSerialM or WSerialT) with custom Applicative or Monadic properties.

104But in general, try to avoid specific types and use explicit functions

105from the stream module.

106

107Here are the TH macro based recipes to define types equivalent to all the older

108types:

109

110```

111{-# LANGUAGE StandaloneDeriving #-}

112{-# LANGUAGE TemplateHaskell #-}

113{-# LANGUAGE TypeFamilies #-}

114{-# LANGUAGE UndecidableInstances #-}

115

116import Streamly.Data.Stream.MkType

117import qualified Streamly.Data.Stream.Prelude as Stream

118import qualified Streamly.Internal.Data.StreamK as StreamK

119import Language.Haskell.TH

120```

121

122`SerialT`: Use `CrossStreamK` from `Streamly.Internal.Data.StreamK` equivalent

123Monad and other instances.

124

125For `WSerialT`, create a newtype wrapper using the following monad bind

126operation:

127

128```

129bind = StreamK.bindWith StreamK.interleave

130```

131

132`WSerialT` requires `StreamK`, therefore you can not generate the code directly

133using TH macros. However, you can print the code generated by the TH macro.

134Then you can tweak that code, replacing `Stream` type with `StreamK`.

135

136```

137expr <- runQ (mkCrossType "WSerialT" "bind" True)

138putStrLn $ pprint expr

139```

140

141For `AsyncT`:

142

143```

144bind = flip (Stream.parConcatMap id)

145$(mkCrossType "AsyncT" "bind" True)

146```

147

148For `WAsyncT`:

149

150```

151bind = flip (Stream.parConcatMap (Stream.interleaved True))

152$(mkCrossType "WAsyncT" "bind" True)

153```

154

155For `AheadT`:

156

157```

158bind = flip (Stream.parConcatMap (Stream.ordered True))

159$(mkCrossType "AheadT" "bind" True)

160```

161

162For `ParallelT`:

163

164```

165bind = flip (Stream.parConcatMap (Stream.eager True))

166$(mkCrossType "ParallelT" "bind" True)

167```

168

169For `ZipSerialM`:

170

171```

172apply = Stream.zipWith ($)

173$(mkZipType "ZipSerialM" "apply" False)

174```

175

176For `ZipAsync`:

177

178```

179apply = Stream.parApply id

180$(mkZipType "ZipAsync" "apply" False)

181```

182

183`adapt` is not needed anymore.

184

185```haskell docspec

186>>> (.:) = StreamK.cons

187>>> cons = StreamK.cons

188>>> wSerial = StreamK.interleave

189>>> serial = StreamK.append

190>>> fromIndicesM f = Stream.mapM f $ Stream.enumerateFrom 0

191>>> fromIndices f = fmap f $ Stream.enumerateFrom 0

192>>> fromListM = Stream.sequence . Stream.fromList

193>>> fromFoldable = StreamK.toStream . StreamK.fromFoldable

194>>> fromFoldableM = Stream.sequence . fromFoldable

195```

196

197## Stream folding functions

198

199Explicit stream fold functions have been omitted from the new stream

200module. You can use the following equivalent definitions:

201

202```haskell docspec

203>>> the = Stream.fold Fold.the

204>>> sum = Stream.fold Fold.sum

205>>> product = Stream.fold Fold.product

206>>> or = Stream.fold Fold.or

207>>> null = Stream.fold Fold.null

208>>> elemIndex a = Stream.fold (Fold.elemIndex a)

209>>> elem a = Stream.fold (Fold.elem a)

210>>> notElem a = Stream.fold (Fold.notElem a)

211>>> minimumBy ordering = Stream.fold (Fold.minimumBy ordering)

212>>> minimum = Stream.fold Fold.minimum

213>>> maximumBy ordering = Stream.fold (Fold.maximumBy ordering)

214>>> maximum = Stream.fold Fold.maximum

215>>> mapM_ f = Stream.fold (Fold.drainMapM f)

216>>> lookup a = Stream.fold (Fold.lookup a)

217>>> length = Stream.fold Fold.length

218>>> last = Stream.fold Fold.latest

219>>> head = Stream.fold Fold.one

220>>> foldlM' f a = Stream.fold (Fold.foldlM' f a)

221>>> foldl1 f = Stream.fold (Fold.foldl1' f)

222>>> foldl' f a = Stream.fold (Fold.foldl' f a)

223>>> findIndex eq = Stream.fold (Fold.findIndex eq)

224>>> find eq = Stream.fold (Fold.find eq)

225>>> findM eq = Stream.fold (Fold.findM eq)

226>>> drainWhile p = Stream.fold Fold.drain . Stream.takeWhile p

227>>> drainN i = Stream.fold Fold.drain . Stream.take i

228>>> drain = Stream.fold Fold.drain

229>>> any p = Stream.fold (Fold.any p)

230>>> and = Stream.fold Fold.and

231>>> all p = Stream.fold (Fold.all p)

232>>> (!!) i = Stream.fold (Fold.index i)

233>>> tail = Streamly.Internal.Data.StreamK.tail

234>>> init = Streamly.Internal.Data.StreamK.init

235```

236

237Mapping functions:

238

239```haskell docspec

240>>> map = fmap

241```

242

243Similarly for scanning use `Stream.scan` or `Stream.postscan`

244with an appropriate fold.

245

246```haskell docspec

247>>> scanl' f z = Stream.scan (Fold.foldl' f z)

248>>> scanlM' f z = Stream.scan (Fold.foldlM' f z)

249>>> postscanl' f z = Stream.postscan (Fold.foldl' f z)

250>>> postscanlM' f z = Stream.postscan (Fold.foldlM' f z)

251>>> scanl1' f = Stream.catMaybes . Stream.scan (Fold.foldl1' f)

252>>> scanl1M' f = Stream.catMaybes . Stream.scan (Fold.foldlM1' f)

253>>> concatMapWith = StreamK.concatMapWith

254>>> concatFoldableWith f = Prelude.foldr f StreamK.nil

255>>> concatMapFoldableWith f g = Prelude.foldr (f . g) StreamK.nil

256>>> concatForFoldableWith f xs g = Prelude.foldr (f . g) StreamK.nil xs

257```

258

259Filters:

260

261```haskell docspec

262>>> deleteBy cmp x = Stream.scanMaybe (Fold.deleteBy cmp x)

263>>> findIndices p = Stream.scanMaybe (Fold.findIndices p)

264>>> elemIndices a = findIndices (== a)

265```

266

267Custom implementations of most of these folds, scans and filters are also

268available in the `Streamly.Internal.Data.Stream` module.

269

270## Stream splitting and grouping functions

271

272Stream splitting and grouping functions like `splitOn`, `wordsBy`, and

273`groupsBy` have been omitted from the new stream module as these can

274now be implemented using `foldMany` and an appropriate fold from the

275`Streamly.Data.Fold` module, or using `parseMany` and an appropriate

276parser from the `Streamly.Data.Parser` module.

277

278```haskell docspec

279>>> uniq = Stream.scanMaybe (Fold.uniqBy (==))

280>>> splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)

281>>> splitOn = Streamly.Internal.Data.Stream.splitOn

282>>> splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)

283>>> indexedR = Streamly.Internal.Data.Stream.indexedR

284>>> groupsBy eq fld = Stream.parseMany (Parser.groupBy eq fld)

285>>> groups = groupsBy (==)

286>>> groupsByRolling = Streamly.Internal.Data.Stream.groupsRollingBy

287>>> wordsBy p f = Stream.parseMany (Parser.wordBy p f)

288>>> chunksOf n f = Stream.foldMany (Fold.take n f)

289```

290

291Direct implementation of these are also available in

292`Streamly.Internal.Data.Stream`.

293

294## Concurrency

295

296Earlier, concurrent and non-concurrent code used the same

297combinators. The code was made concurrent by using different concurrent

298stream types such as `AsyncT`, `ParallelT` etc. Now you use the same

299stream type everywhere, you have to choose a concurrent combinator

300for concurrent behavior. For example, use `mapM` for serial behavior

301and `parMapM` for concurrent behavior. Concurrent combinators can

302be imported from `Streamly.Data.Stream.Prelude` module. Concurrent

303combinators are prefixed with `par`.

304

305Parallel combinators take a concurrency config argument to specify the

306concurrency control parameters. The following combinators have the same

307meaning as before except that they are used to set the config parameters

308instead of being applied on the stream:

309

310* `rate`

311* `maxRate`

312* `constRate`

313* `avgRate`

314* `minRate`

315* `maxThreads`

316* `maxBuffer`

317

318A stream is evaluated asynchronously using `parEval`:

319

320```haskell docspec

321>>> :set -XFlexibleContexts

322>>> mkAsync = Stream.parEval id

323```

324

325Earlier `consM` was used to create an implicitly concurrent stream of

326actions. In the new release, an equivalent effect is achieved by using

327a serial `consM` to create a stream of actions and then explicitly using

328`Stream.parEval` on it to evaluate it concurrently.

329

330```haskell docspec

331>>> consM = StreamK.consM

332>>> (|:) = consM

333```

334

335Note that you will have to use `StreamK.toStream` to covert it to `Stream`

336before using `parEval` on it.

337

338Existing generation combinators that can be implemented using new primitives:

339

340```haskell docspec

341>>> repeatM = Stream.parRepeatM

342>>> replicateM = Stream.parReplicateM

343>>> unfoldrM step = Stream.parEval id . Stream.unfoldrM step

344>>> iterateM step = Stream.parEval id . Stream.iterateM step

345>>> fromIndicesM f = Stream.parEval id . fromIndicesM f

346>>> fromListM = Stream.parSequence id . Stream.fromList

347>>> fromFoldableM = Stream.parSequence id . StreamK.toStream . StreamK.fromFoldable

348```

349

350Existing transformation combinators that can be implemented using `parEval`:

351

352```haskell docspec

353>>> (|$.) f = f . Stream.parEval id

354>>> (|&.) = flip (|$.)

355>>> (|$) f = f . Stream.parEval id

356>>> (|&) = flip (|$)

357>>> sequence = Stream.parSequence

358>>> mapM = Stream.parMapM

359```

360

361`parList` is used to evaluate multiple streams concurrently and combine the

362outputs. Existing combinators that can be implemented using `parList`:

363

364```haskell docspec

365>>> async x y = Stream.parList id [x, y]

366>>> wAsync x y = Stream.parList (Stream.interleaved True) [x, y]

367>>> parallel x y = Stream.parList (Stream.eager True) [x, y]

368>>> ahead x y = Stream.parList (Stream.ordered True) [x, y]

369```

370

371Concurrent zipping and merging combinators:

372

373```haskell docspec

374>>> zipAsyncWithM = Stream.parZipWith id

375>>> zipAsyncWith = Stream.parZipWith id

376>>> mergeAsyncByM = Stream.parMergeByM id

377>>> mergeAsyncBy = Stream.parMergeBy id

378```

379

380The equivalent of `concatMapWith` using a concurrent combining operation in the

381new release is `parConcatMap`. The config argument in `parConcatMap` can

382specify an equivalent of the combining operation. Similarly, concurrent

383`concatFoldableWith`, `concatMapFoldableWith`, `concatForFoldableWith` can also

384be expressed using `parConcatMap`.