4f70b0d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Rename Socket BufferOf APIs

4f70b0dRanjeet Kumar Ranjan5 years ago

docs/Changelog.md+6

compare with master

unifiedsplit

@@ -1,7 +1,13 @@

1# Changelog

3

4+## Unreleased

6+* Following APIs have been renamed in Streamly.Network.Socket module:

7+ readWithBufferOf -> readWith

8+ readChunksWithBufferOf -> readChunksWith

9+ writeWithBufferOf -> writeWith

10+ writeChunksWithBufferOf -> writeChunksWith

5## 0.8.2 (Mar 2022)

11## 0.8.2 (Mar 2022)

7* Fix performance issues for GHC-9. These changes coupled with GHC changes

13* Fix performance issues for GHC-9. These changes coupled with GHC changes

src/Streamly/Internal/Network/Socket.hs+78−39

@@ -24,7 +24,7 @@ module Streamly.Internal.Network.Socket

25 -- * Read from connection

26 , read

27− , readWithBufferOf

27+ , readWith

28 -- , readUtf8

29 -- , readLines

30 -- , readFrames

@@ -35,9 +35,9 @@ module Streamly.Internal.Network.Socket

35 -- , readChunksUpto

36 , readChunk

37 , readChunks

38− , readChunksWithBufferOf

38+ , readChunksWith

40− , toChunksWithBufferOf

40+ , toChunksWith

41 , toChunks

42 , toBytes

@@ -46,19 +46,25 @@ module Streamly.Internal.Network.Socket

46 -- , writeUtf8

47 -- , writeUtf8ByLines

48 -- , writeByFrames

49− , writeWithBufferOf

49+ , writeWith

50− , writeMaybesWithBufferOf

50+ , writeMaybesWith

52 , putChunks

53− , putBytesWithBufferOf

53+ , putBytesWith

54 , putBytes

56 -- -- * Array Write

57 , writeChunk

58 , writeChunks

59− , writeChunksWithBufferOf

59+ , writeChunksWith

61 -- reading/writing datagrams

62+

63+ -- * Deprecated

64+ , readWithBufferOf

65+ , readChunksWithBufferOf

66+ , writeWithBufferOf

67+ , writeChunksWithBufferOf

62 )

68 )

63where

69where

@@ -337,15 +343,15 @@ _readChunksUptoWith f size h = go

337 then stp

343 then stp

338 else yld arr go

344 else yld arr go

339

345

340−-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.

346+-- | @toChunksWith size h@ reads a stream of arrays from file handle @h@.

341-- The maximum size of a single array is limited to @size@.

347-- The maximum size of a single array is limited to @size@.

342-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

348-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'

343-- on the 'Handle'.

349-- on the 'Handle'.

344−{-# INLINE_NORMAL toChunksWithBufferOf #-}

350+{-# INLINE_NORMAL toChunksWith #-}

345−toChunksWithBufferOf :: (IsStream t, MonadIO m)

351+toChunksWith :: (IsStream t, MonadIO m)

346 => Int -> Socket -> t m (Array Word8)

352 => Int -> Socket -> t m (Array Word8)

347−-- toChunksWithBufferOf = _readChunksUptoWith readChunk

353+-- toChunksWith = _readChunksUptoWith readChunk

348−toChunksWithBufferOf size h = fromStreamD (D.Stream step ())

354+toChunksWith size h = fromStreamD (D.Stream step ())

349 where

355 where

350 {-# INLINE_LATE step #-}

356 {-# INLINE_LATE step #-}

351 step _ _ = do

357 step _ _ = do

@@ -363,17 +369,17 @@ toChunksWithBufferOf size h = fromStreamD (D.Stream step ())

363-- @since 0.7.0

369-- @since 0.7.0

364{-# INLINE toChunks #-}

370{-# INLINE toChunks #-}

365toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

371toChunks :: (IsStream t, MonadIO m) => Socket -> t m (Array Word8)

366−toChunks = toChunksWithBufferOf defaultChunkSize

372+toChunks = toChunksWith defaultChunkSize

367

373

368-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

374-- | Unfold the tuple @(bufsize, socket)@ into a stream of 'Word8' arrays.

369-- Read requests to the socket are performed using a buffer of size @bufsize@.

375-- Read requests to the socket are performed using a buffer of size @bufsize@.

370-- The size of an array in the resulting stream is always less than or equal to

376-- The size of an array in the resulting stream is always less than or equal to

371-- @bufsize@.

377-- @bufsize@.

372--

378--

373−-- @since 0.7.0

379+-- @since 0.9.0

374−{-# INLINE_NORMAL readChunksWithBufferOf #-}

380+{-# INLINE_NORMAL readChunksWith #-}

375−readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)

381+readChunksWith :: MonadIO m => Unfold m (Int, Socket) (Array Word8)

376−readChunksWithBufferOf = Unfold step return

382+readChunksWith = Unfold step return

377 where

383 where

378 {-# INLINE_LATE step #-}

384 {-# INLINE_LATE step #-}

379 step (size, h) = do

385 step (size, h) = do

@@ -383,6 +389,14 @@ readChunksWithBufferOf = Unfold step return

383 0 -> D.Stop

389 0 -> D.Stop

384 _ -> D.Yield arr (size, h)

390 _ -> D.Yield arr (size, h)

385

391

392+-- | Same as 'readChunksWith'

393+--

394+-- @since 0.7.0

395+{-# DEPRECATED readChunksWithBufferOf "Please use 'readChunksWith' instead" #-}

396+{-# INLINE_NORMAL readChunksWithBufferOf #-}

397+readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Socket) (Array Word8)

398+readChunksWithBufferOf = readChunksWith

399+

386-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket

400-- | Unfolds a socket into a stream of 'Word8' arrays. Requests to the socket

387-- are performed using a buffer of size

401-- are performed using a buffer of size

388-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The

402-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The

@@ -392,7 +406,7 @@ readChunksWithBufferOf = Unfold step return

392-- @since 0.7.0

406-- @since 0.7.0

393{-# INLINE readChunks #-}

407{-# INLINE readChunks #-}

394readChunks :: MonadIO m => Unfold m Socket (Array Word8)

408readChunks :: MonadIO m => Unfold m Socket (Array Word8)

395−readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf

409+readChunks = UF.supplyFirst defaultChunkSize readChunksWith

396

410

397-------------------------------------------------------------------------------

411-------------------------------------------------------------------------------

398-- Read File to Stream

412-- Read File to Stream

@@ -403,13 +417,13 @@ readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf

403-- also control the read throughput in mbps or IOPS.

417-- also control the read throughput in mbps or IOPS.

404

418

405{-

419{-

406−-- | @readWithBufferOf bufsize handle@ reads a byte stream from a file

420+-- | @readWith bufsize handle@ reads a byte stream from a file

407-- handle, reads are performed in chunks of up to @bufsize@. The stream ends

421-- handle, reads are performed in chunks of up to @bufsize@. The stream ends

408-- as soon as EOF is encountered.

422-- as soon as EOF is encountered.

409--

423--

410−{-# INLINE readWithBufferOf #-}

424+{-# INLINE readWith #-}

411−readWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

425+readWith :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

412−readWithBufferOf chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

426+readWith chunkSize h = A.flattenArrays $ readChunksUpto chunkSize h

413-}

427-}

414

428

415-- TODO

429-- TODO

@@ -427,10 +441,18 @@ toBytes = AS.concat . toChunks

427-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests

441-- | Unfolds the tuple @(bufsize, socket)@ into a byte stream, read requests

428-- to the socket are performed using buffers of @bufsize@.

442-- to the socket are performed using buffers of @bufsize@.

429--

443--

444+-- @since 0.9.0

445+{-# INLINE readWith #-}

446+readWith :: MonadIO m => Unfold m (Int, Socket) Word8

447+readWith = UF.many A.read readChunksWith

448+

449+-- | Same as 'readWith'

450+--

430-- @since 0.7.0

451-- @since 0.7.0

452+{-# DEPRECATED readWithBufferOf "Please use 'readWith' instead" #-}

431{-# INLINE readWithBufferOf #-}

453{-# INLINE readWithBufferOf #-}

432readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8

454readWithBufferOf :: MonadIO m => Unfold m (Int, Socket) Word8

433−readWithBufferOf = UF.many A.read readChunksWithBufferOf

455+readWithBufferOf = readWith

434

456

435-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are

457-- | Unfolds a 'Socket' into a byte stream. IO requests to the socket are

436-- performed in sizes of

458-- performed in sizes of

@@ -439,7 +461,7 @@ readWithBufferOf = UF.many A.read readChunksWithBufferOf

439-- @since 0.7.0

461-- @since 0.7.0

440{-# INLINE read #-}

462{-# INLINE read #-}

441read :: MonadIO m => Unfold m Socket Word8

463read :: MonadIO m => Unfold m Socket Word8

442−read = UF.supplyFirst defaultChunkSize readWithBufferOf

464+read = UF.supplyFirst defaultChunkSize readWith

443

465

444-------------------------------------------------------------------------------

466-------------------------------------------------------------------------------

445-- Writing

467-- Writing

@@ -461,17 +483,26 @@ putChunks h = S.mapM_ (liftIO . writeChunk h)

461writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()

483writeChunks :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()

462writeChunks h = FL.drainBy (liftIO . writeChunk h)

484writeChunks h = FL.drainBy (liftIO . writeChunk h)

463

485

464−-- | @writeChunksWithBufferOf bufsize socket@ writes a stream of arrays to

486+-- | @writeChunksWith bufsize socket@ writes a stream of arrays to

465-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.

487-- @socket@ after coalescing the adjacent arrays in chunks of @bufsize@.

466-- Multiple arrays are coalesed as long as the total size remains below the

488-- Multiple arrays are coalesed as long as the total size remains below the

467-- specified size. It never splits an array, if a single array is bigger than

489-- specified size. It never splits an array, if a single array is bigger than

468-- the specified size it emitted as it is.

490-- the specified size it emitted as it is.

469--

491--

470−-- @since 0.8.0

492+-- @since 0.9.0

493+{-# INLINE writeChunksWith #-}

494+writeChunksWith :: (MonadIO m, Storable a)

495+ => Int -> Socket -> Fold m (Array a) ()

496+writeChunksWith n h = lpackArraysChunksOf n (writeChunks h)

497+

498+-- | Same as 'writeChunksWith'

499+--

500+-- @since 0.7.0

501+{-# DEPRECATED writeChunksWithBufferOf "Please use 'writeChunksWith' instead" #-}

471{-# INLINE writeChunksWithBufferOf #-}

502{-# INLINE writeChunksWithBufferOf #-}

472writeChunksWithBufferOf :: (MonadIO m, Storable a)

503writeChunksWithBufferOf :: (MonadIO m, Storable a)

473 => Int -> Socket -> Fold m (Array a) ()

504 => Int -> Socket -> Fold m (Array a) ()

474−writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)

505+writeChunksWithBufferOf = writeChunksWith

475

506

476-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

507-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

477--

508--

@@ -485,33 +516,41 @@ writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)

485-- be written to the IO device as soon as we collect the specified number of

516-- be written to the IO device as soon as we collect the specified number of

486-- input elements.

517-- input elements.

487--

518--

488−-- @since 0.7.0

519+-- @since 0.9.0

489−{-# INLINE putBytesWithBufferOf #-}

520+{-# INLINE putBytesWith #-}

490−putBytesWithBufferOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

521+putBytesWith :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

491−putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m

522+putBytesWith n h m = putChunks h $ AS.arraysOf n m

492

523

493-- | Write a byte stream to a socket. Accumulates the input in chunks of

524-- | Write a byte stream to a socket. Accumulates the input in chunks of

494-- specified number of bytes before writing.

525-- specified number of bytes before writing.

495--

526--

527+-- @since 0.9.0

528+{-# INLINE writeWith #-}

529+writeWith :: MonadIO m => Int -> Socket -> Fold m Word8 ()

530+writeWith n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)

531+

532+-- | Same as 'writeWith'

533+--

496-- @since 0.7.0

534-- @since 0.7.0

535+{-# DEPRECATED writeWithBufferOf "Please use 'writeWith' instead" #-}

497{-# INLINE writeWithBufferOf #-}

536{-# INLINE writeWithBufferOf #-}

498writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

537writeWithBufferOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

499−writeWithBufferOf n h = FL.chunksOf n (A.writeNUnsafe n) (writeChunks h)

538+writeWithBufferOf = writeWith

500

539

501-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an

540-- | Write a stream of 'Maybe' values. Keep buffering the 'Just' values in an

502-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered

541-- array. Write the array to the 'Handle' as soon as a 'Nothing' is encountered

503-- or the buffer size exceeds the specified limit.

542-- or the buffer size exceeds the specified limit.

504--

543--

505-- /Pre-release/

544-- /Pre-release/

506−{-# INLINE writeMaybesWithBufferOf #-}

545+{-# INLINE writeMaybesWith #-}

507−writeMaybesWithBufferOf :: (MonadIO m )

546+writeMaybesWith :: (MonadIO m )

508 => Int -> Socket -> Fold m (Maybe Word8) ()

547 => Int -> Socket -> Fold m (Maybe Word8) ()

509−writeMaybesWithBufferOf n h =

548+writeMaybesWith n h =

510 let writeNJusts = FL.lmap fromJust $ A.writeN n

549 let writeNJusts = FL.lmap fromJust $ A.writeN n

511 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

550 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

512 in FL.many writeOnNothing (writeChunks h)

551 in FL.many writeOnNothing (writeChunks h)

513

552

514−-- > write = 'writeWithBufferOf' defaultChunkSize

553+-- > write = 'writeWith' defaultChunkSize

515--

554--

516-- | Write a byte stream to a file handle. Combines the bytes in chunks of size

555-- | Write a byte stream to a file handle. Combines the bytes in chunks of size

517-- up to 'defaultChunkSize' before writing. Note that the write behavior

556-- up to 'defaultChunkSize' before writing. Note that the write behavior

@@ -520,19 +559,19 @@ writeMaybesWithBufferOf n h =

520-- @since 0.7.0

559-- @since 0.7.0

521{-# INLINE putBytes #-}

560{-# INLINE putBytes #-}

522putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

561putBytes :: MonadIO m => Socket -> SerialT m Word8 -> m ()

523−putBytes = putBytesWithBufferOf defaultChunkSize

562+putBytes = putBytesWith defaultChunkSize

524

563

525-- | Write a byte stream to a socket. Accumulates the input in chunks of

564-- | Write a byte stream to a socket. Accumulates the input in chunks of

526-- up to 'defaultChunkSize' bytes before writing.

565-- up to 'defaultChunkSize' bytes before writing.

527--

566--

528-- @

567-- @

529−-- write = 'writeWithBufferOf' 'defaultChunkSize'

568+-- write = 'writeWith' 'defaultChunkSize'

530-- @

569-- @

531--

570--

532-- @since 0.7.0

571-- @since 0.7.0

533{-# INLINE write #-}

572{-# INLINE write #-}

534write :: MonadIO m => Socket -> Fold m Word8 ()

573write :: MonadIO m => Socket -> Fold m Word8 ()

535−write = writeWithBufferOf defaultChunkSize

574+write = writeWith defaultChunkSize

536

575

537{-

576{-

538{-# INLINE write #-}

577{-# INLINE write #-}

src/Streamly/Network/Socket.hs+10−4

@@ -101,20 +101,26 @@ module Streamly.Network.Socket

101

102 -- * Read

103 , read

104− , readWithBufferOf

104+ , readWith

105 , readChunks

106− , readChunksWithBufferOf

106+ , readChunksWith

107 , readChunk

108

109 -- * Write

110 , write

111− , writeWithBufferOf

111+ , writeWith

112 , writeChunks

113− , writeChunksWithBufferOf

113+ , writeChunksWith

114 , writeChunk

115

116 -- * Exceptions

117 , forSocketM

118+

119+ -- * Deprecated

120+ , readWithBufferOf

121+ , readChunksWithBufferOf

122+ , writeWithBufferOf

123+ , writeChunksWithBufferOf

118 )

124 )

119where

125where

120

126

test/Streamly/Test/Network/Socket.hs+9−9

@@ -42,7 +42,7 @@ testDataSource = concat $ replicate 1000 testData

42------------------------------------------------------------------------------

43handlerChunksWithBuffer :: Socket -> IO ()

44handlerChunksWithBuffer sk =

45− Stream.unfold Socket.readChunksWithBufferOf (100, sk)

45+ Stream.unfold Socket.readChunksWith (100, sk)

46 & Stream.fold (Socket.writeChunks sk)

47 & discard

@@ -54,8 +54,8 @@ handlerChunks sk =

55handlerwithbuffer :: Socket -> IO ()

56handlerwithbuffer sk =

57− Stream.unfold Socket.readWithBufferOf (100, sk)

57+ Stream.unfold Socket.readWith (100, sk)

58− & Stream.fold (Socket.writeWithBufferOf 100 sk)

58+ & Stream.fold (Socket.writeWith 100 sk)

59 & discard

61handlerRW :: Socket -> IO ()

@@ -108,8 +108,8 @@ execute port size handler = do

108 & Stream.finally (killThread tid)

109 return lst

110

111−validateWithBufferOf :: Property

111+validateWith :: Property

112−validateWithBufferOf = monadicIO $ do

112+validateWith = monadicIO $ do

113 res <- run $ do

114 ls2 <- execute basePort 45000 handlerwithbuffer

115 Stream.eqBy (==) (Stream.fromList testDataSource) ls2

@@ -130,8 +130,8 @@ validateChunks = monadicIO $ do

130 Stream.eqBy (==) (Stream.fromList testDataSource) ls2

131 assert res

132

133−validateChunksWithBufferOf :: Property

133+validateChunksWith :: Property

134−validateChunksWithBufferOf = monadicIO $ do

134+validateChunksWith = monadicIO $ do

135 res <- run $ do

136 ls2 <- execute (basePort + 3) 45000 handlerChunksWithBuffer

137 Stream.eqBy (==) (Stream.fromList testDataSource) ls2

@@ -146,6 +146,6 @@ main = hspec $ do

146 describe moduleName $ do

147 describe "Read/Write" $ do

148 prop "read/write" validateRW

149− prop "readWithBufferOf/writeWithBufferOf" validateWithBufferOf

149+ prop "readWith/writeWith" validateWith

150 prop "readChunks/writeChunks" validateChunks

151− prop "readChunksWithBufferOf" validateChunksWithBufferOf

151+ prop "readChunksWith" validateChunksWith