96d0dd6 · curryer · git.iko.soy

git.iko.soy/curryer

setup criterion benchmarking

commit96d0dd6d61b128bab593d6b4389f3f596c30770a authorAgentM <agentm@themactionfaction.com> dateSun 18 Oct 2020, 13:41 (6 years ago) parentedcdd54

6 files changed+107−65

compare with master

unifiedsplit

README.markdown+1−1 bench.sh+1 bench/Basic.hs+29−3 curryer.cabal+4−2 src/Network/RPC/Curryer/Client.hs+32−18 src/Network/RPC/Curryer/Server.hs+40−41

README.markdown+1−1

@@ -9,4 +9,4 @@ Curryer is inspired by the now unmaintained [distributed-process](https://hackag

99* blocking and non-blocking remote function calls

1010* asynchronous server-to-client callbacks (for server-initiated notifications)

1111* timeouts

12−

12+* utilizes versionable shared ADT instead of shared function `StaticPtr`s

bench.shnew file+1

@@ -0,0 +1 @@

1+cabal run perf -- +RTS -p && ~/.cabal/bin/ghc-prof-flamegraph < perf.prof > perf.html

bench/Basic.hs+29−3

@@ -9,6 +9,7 @@ import Codec.Winery

99import Control.Concurrent.Async

1010import Control.Concurrent

1111import Control.Monad

12+import qualified Data.ByteString as BS

1213

1314main :: IO ()

1415main = do

@@ -20,17 +21,33 @@ main = do

2021 clientConn <- connect (const (pure ())) localHostAddr port

2122

2223 _ <- syncWaitRequest clientConn 100

23− defaultMain [bgroup "wait sync" [bench "100 ms" (nfIO $ syncWaitRequest clientConn 100)]]

24+ let syncWait ms = nfIO (syncWaitRequest clientConn ms)

25+ syncByteString bs = nfIO (syncByteStringRequest clientConn bs)

26+ bs0 = BS.empty

27+ bs100 = BS.pack (replicate 100 3)

28+ bs1000 = BS.pack (replicate 1000 3)

29+ bs1M = BS.pack (replicate 1000000 3)

30+ defaultMain [bgroup "wait sync" [bench "0 ms" (syncWait 0),

31+ bench "10 ms" (syncWait 10),

32+ bench "100 ms" (syncWait 100)],

33+ bgroup "bytestring roundtrip" [bench "0 bytes" (syncByteString bs0),

34+ bench "100 bytes" (syncByteString bs100),

35+ bench "1000 bytes" (syncByteString bs1000),

36+ bench "1000000 bytes" (syncByteString bs1M)

37+ ]]

38+

39+ --TODO: bench for > PIPE_BUF bytes

2440

25− putStrLn "benchmark complete"

2641 close clientConn

2742 cancel server

2843

29−data BenchRequest = WaitMillisecondsReq Int

44+data BenchRequest = WaitMillisecondsReq Int --ask the server to respond in X milliseconds

45+ | WaitByteStringReq BS.ByteString --ask the server to round trip arbitrary bytes

3046 deriving (Generic, Show)

3147 deriving Serialise via WineryVariant BenchRequest

3248

3349data BenchResponse = WaitMillisecondsResp

50+ | WaitByteStringResp BS.ByteString

3451 deriving (Generic, Show, Eq)

3552 deriving Serialise via WineryVariant BenchResponse

3653

@@ -38,6 +55,7 @@ benchmarkServerMessageHandler :: BenchRequest -> IO (HandlerResponse BenchRespon

3855benchmarkServerMessageHandler (WaitMillisecondsReq ms) = do

3956 threadDelay (1000 * ms)

4057 pure (HandlerResponse WaitMillisecondsResp)

58+benchmarkServerMessageHandler (WaitByteStringReq bs) = pure (HandlerResponse (WaitByteStringResp bs))

4159

4260--perform 10000 small synchronous requests

4361syncWaitRequest :: Connection BenchResponse -> Int -> IO Int

@@ -46,3 +64,11 @@ syncWaitRequest conn ms = do

4664 when (ret /= Right WaitMillisecondsResp) (error "ret failed")

4765 pure ms

4866

67+syncByteStringRequest :: Connection BenchResponse -> BS.ByteString -> IO BS.ByteString

68+syncByteStringRequest conn bs = do

69+ ret <- call conn (WaitByteStringReq bs)

70+ case ret of

71+ Right (WaitByteStringResp bs') -> pure bs'

72+ Right _ -> error "invalid resp"

73+ Left err -> error (show err)

74+

curryer.cabal+4−2

@@ -59,9 +59,11 @@ Test-Suite test

5959Benchmark perf

6060 Default-Language: Haskell2010

6161 Default-Extensions: OverloadedStrings

62− Build-Depends: base, criterion, curryer, network, winery, async

62+ Build-Depends: base, criterion, curryer, network, winery, async, bytestring

63+

6364 Main-Is: Basic.hs

6465 Type: exitcode-stdio-1.0

65− GHC-Options: -Wall -threaded -rtsopts

66+ GHC-Options: -Wall -threaded -rtsopts -fprof-auto

6667 HS-Source-Dirs: ./bench

68+

6769

src/Network/RPC/Curryer/Client.hs+32−18

@@ -1,3 +1,4 @@

1+{-# LANGUAGE TypeApplications, RankNTypes, ScopedTypeVariables #-}

12module Network.RPC.Curryer.Client where

23import Network.RPC.Curryer.Server

34import Network.Socket as Socket

@@ -10,34 +11,47 @@ import Control.Concurrent.MVar

1011import GHC.Conc

1112import Data.Time.Clock

1213import System.Timeout

14+import Data.Proxy

1315

1416type SyncMap a = STMMap.Map UUID (MVar (Either ConnectionError a), UTCTime)

1517-- the request map holds

16−data Connection a = Connection Socket (Async ()) (SyncMap a)

18+data Connection a = Connection { _conn_sock :: Socket,

19+ _conn_asyncThread :: Async (),

20+ _conn_syncmap :: SyncMap a

21+ }

1722

18−connect :: (Serialise msg) =>

19− AsyncMessageHandler msg ->

23+connect :: forall clientmsg. (Serialise clientmsg) =>

24+ AsyncMessageHandler clientmsg ->

2025 HostAddr ->

2126 PortNumber ->

22− IO (Connection msg)

27+ IO (Connection clientmsg)

2328connect notificationCallback hostAddr portNum = do

2429 sock <- TCP.connect hostAddr portNum

2530 syncmap <- STMMap.newIO

26− asyncThread <- async (clientAsync sock syncmap notificationCallback)

27− pure (Connection sock asyncThread syncmap)

31+ let decoder :: Decoder (Message clientmsg)

32+ decoder = case getDecoder (schema (Proxy @(Message clientmsg))) of

33+ Left err -> error (show err)

34+ Right dec -> dec

35+ asyncThread <- async (clientAsync sock syncmap decoder notificationCallback)

36+ pure (Connection {

37+ _conn_sock = sock,

38+ _conn_asyncThread = asyncThread,

39+ _conn_syncmap = syncmap

40+ })

2841

2942close :: Connection a -> IO ()

30−close (Connection sock asyncThread _) = do

31− Socket.close sock

32− cancel asyncThread

43+close conn = do

44+ Socket.close (_conn_sock conn)

45+ cancel (_conn_asyncThread conn)

3346

3447-- async thread for handling client-side incoming messages- dispatch to proper waiting thread or handler asynchronous notifications

35−clientAsync :: Serialise b =>

48+clientAsync :: forall b. Serialise b =>

3649 Socket ->

3750 SyncMap b ->

51+ Decoder (Message b) ->

3852 AsyncMessageHandler b ->

3953 IO ()

40−clientAsync sock syncmap asyncHandler = do

54+clientAsync sock syncmap decoder asyncHandler = do

4155 -- ping proper thread to continue

4256 let findRequest requestId =

4357 atomically $ do

@@ -46,7 +60,7 @@ clientAsync sock syncmap asyncHandler = do

4660 pure val'

4761

4862 let responseHandler responseMsg = do

49− putStrLn "client-side message handler"

63+ --putStrLn "client-side message handler"

5064 case responseMsg of

5165 Response requestId val -> do

5266 varval <- findRequest requestId

@@ -62,21 +76,21 @@ clientAsync sock syncmap asyncHandler = do

6276 case varval of

6377 Nothing -> error "dumped unrequested timeout response"

6478 Just (mVar,_) -> putMVar mVar (Left TimeoutError)

65−

66− drainSocketMessages sock responseHandler

79+ drainSocketMessages sock decoder responseHandler

6780

6881call :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError response)

6982call = callTimeout Nothing

7083

7184-- | Send a request to the remote server and returns a response.

7285callTimeout :: (Serialise request, Serialise response) => Maybe Int -> Connection response -> request -> IO (Either ConnectionError response)

73−callTimeout mTimeout (Connection sock _ mVarMap) msg = do

86+callTimeout mTimeout conn msg = do

87+ let mVarMap = _conn_syncmap conn

7488 requestID <- UUID <$> UUIDBase.nextRandom

7589 -- setup mvar to wait for response

7690 responseMVar <- newEmptyMVar

7791 now <- getCurrentTime

7892 atomically $ STMMap.insert (responseMVar, now) requestID mVarMap

79− sendMessage (ResponseExpectedRequest requestID mTimeout msg) sock

93+ sendMessage (ResponseExpectedRequest requestID mTimeout msg) (_conn_sock conn)

8094 let timeoutMicroseconds =

8195 case mTimeout of

8296 Just timeout' -> timeout' + 100 --add 100 ms to account for unknown network latency

@@ -90,8 +104,8 @@ callTimeout mTimeout (Connection sock _ mVarMap) msg = do

90104 Just res -> pure res

91105

92106asyncCall :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError ())

93−asyncCall (Connection sock _ _) msg = do

107+asyncCall conn msg = do

94108 requestID <- UUID <$> UUIDBase.nextRandom

95− sendMessage (AsyncRequest requestID msg) sock

109+ sendMessage (AsyncRequest requestID msg) (_conn_sock conn)

96110 pure (Right ())

97111

src/Network/RPC/Curryer/Server.hs+40−41

@@ -1,4 +1,4 @@

1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications #-}

1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP #-}

22{- HLINT ignore "Use lambda-case" -}

33module Network.RPC.Curryer.Server where

44import Streamly

@@ -24,34 +24,31 @@ import qualified Streamly.Internal.Data.Stream.IsStream as S

2424import qualified Data.Binary as B

2525import qualified Data.UUID as UUIDBase

2626import qualified Data.UUID.V4 as UUIDBase

27−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA

27+

2828import qualified Network.RPC.Curryer.StreamlyAdditions as SA

2929import Control.Monad

3030import Data.Hashable

3131import System.Timeout

3232import qualified Network.ByteOrder as BO

33+import Data.Proxy

3334

3435-- for toArrayS conversion

35−import qualified Data.ByteString.Internal as BSI

36+{-import qualified Data.ByteString.Internal as BSI

37+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA

3638import Foreign.ForeignPtr (plusForeignPtr)

3739import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

3840import GHC.Ptr (plusPtr)

39−

40−--import Debug.Trace

41−

42−data ClientHelloMessage = ClientHelloMessage Int Int

43− deriving Generic

44− deriving Serialise via WineryVariant ClientHelloMessage

45−

46−data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],

47− serverInfo :: String

48− }

49− deriving Generic

50− deriving Serialise via WineryRecord ServerHelloMessage

51−

52−--request-response token

53−--client-side timeout

54−--server-to-client and client-to-server async request (no response requested)

41+-}

42+#if CURRYER_SHOW_BYTES == 1

43+import Debug.Trace

44+#endif

45+

46+traceBytes :: Applicative f => String -> BS.ByteString -> f ()

47+#if CURRYER_SHOW_BYTES == 1

48+traceBytes msg bs = traceShowM (msg, BS.length bs, bs)

49+#else

50+traceBytes _ _ = pure ()

51+#endif

5552

5653type Timeout = Int

5754

@@ -104,30 +101,35 @@ localHostAddr = (127,0,0,1)

104101-- Each message is length-prefixed by a 32-bit unsigned length.

105102messageBoundaryP :: Parser IO Word8 BS.ByteString

106103messageBoundaryP = do

107− let x = FL.toList

108− w4x8 <- P.take 4 x

104+ let s = FL.toList

105+ w4x8 <- P.take 4 s

109106 --traceShowM ("w4x8"::String, w4x8)

110107 let c = fromIntegral (BO.word32 (BS.pack w4x8))

111108 --traceShowM ("c"::String, c)

112− vals <- P.take c x

109+ vals <- P.take c s

113110 let bytes = BS.pack vals

114111 --traceShowM ("parsedBytes"::String, c, BS.length bytes, bytes)

115112 pure bytes

116113

117114type NewConnectionHandler msg = IO (Maybe msg)

118115

119−type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)

116+type NewMessageHandler req resp = req -> IO (HandlerResponse resp)

120117

121−serve :: (Show msg, Serialise msg, Serialise resp) =>

118+serve :: forall req resp. (Show req, Serialise req, Serialise resp) =>

122119 NewConnectionHandler resp ->

123− NewMessageHandler msg resp ->

120+ NewMessageHandler req resp ->

124121 HostAddr ->

125122 PortNumber ->

126123 Maybe (MVar SockAddr) ->

127124 IO Bool

128125serve connhandler userMsgHandler hostaddr port mSockLock = do

129− let handleSock sock = do

130− putStrLn "handleSock"

126+ let

127+ decoder :: NewMessageHandler req resp -> Decoder (Message req)

128+ decoder _ =

129+ case getDecoder (schema (Proxy @(Message req))) of

130+ Left exc -> error (show exc)

131+ Right dec -> dec

132+ handleSock sock = do

131133 -- allow the server to send an async welcome message to the new client, if necessary

132134 mResp <- connhandler

133135 case mResp of

@@ -135,8 +137,8 @@ serve connhandler userMsgHandler hostaddr port mSockLock = do

135137 Just resp -> do

136138 requestID <- UUID <$> UUIDBase.nextRandom

137139 sendMessage (AsyncRequest requestID resp) sock

138− drainSocketMessages sock (serverMessageHandler sock userMsgHandler)

139− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & serially . S.mapM (handleWithM handleSock) & S.drain

140+ drainSocketMessages sock (decoder userMsgHandler) (serverMessageHandler sock userMsgHandler)

141+ serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain

140142 pure True

141143

142144serverMessageHandler :: forall req resp. (Show req,

@@ -147,7 +149,7 @@ serverMessageHandler :: forall req resp. (Show req,

147149 -> Message req

148150 -> IO ()

149151serverMessageHandler sock requestMessageHandler msg = do

150− putStrLn $ "GOT MSG " ++ show msg

152+ --putStrLn $ "GOT MSG " ++ show msg

151153 let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))

152154 runTimeout mTimeout m = case mTimeout of

153155 Nothing -> Just <$> m

@@ -155,7 +157,6 @@ serverMessageHandler sock requestMessageHandler msg = do

155157 case msg of

156158 Response{} -> putStrLn "client sent response"

157159 ResponseExpectedRequest requestID mTimeout val -> do

158− putStrLn "server received ResponseExpectedRequest"

159160 resp <- runTimeout mTimeout (requestMessageHandler val)

160161 case resp of

161162 Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sock

@@ -164,7 +165,6 @@ serverMessageHandler sock requestMessageHandler msg = do

164165 Nothing ->

165166 sendMessage (TimedOutResponse @(Message resp) requestID) sock

166167 AsyncRequest _ val -> do

167− putStrLn "AsyncReq"

168168 void $ requestMessageHandler val

169169 --no response necessary

170170 ExceptionResponse{} -> putStrLn "client sent exception response"

@@ -177,29 +177,26 @@ type AsyncMessageHandler a = a -> IO ()

177177

178178

179179

180−drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()

181−drainSocketMessages sock msgHandler = do

180+drainSocketMessages :: Serialise msg => Socket -> Decoder (Message msg) -> MessageHandler msg -> IO ()

181+drainSocketMessages sock decoder msgHandler = do

182182 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)

183183 handler bs = do

184− case deserialise bs of

185− Left err ->

186− print err

187− Right val ->

188− msgHandler val --add response function

184+ let decoded = evalDecoder decoder bs

185+ msgHandler decoded

189186 S.drain $ serially $ S.parseMany messageBoundaryP sockStream & S.mapM handler

190187

191188--send length-tagged bytestring, perhaps should be in network byte order?

192189sendMessage :: Serialise a => Message a -> Socket -> IO ()

193190sendMessage msg socket' = do

194191 let

195− msgbytes = serialise msg

192+ msgbytes = serialiseOnly msg

196193 fullbytes = lenbytes <> msgbytes

197194 len = BS.length msgbytes

198195 lenbytes = BO.bytestring32 (fromIntegral len)

199196

200197 byteCount <- Socket.send socket' fullbytes

201198 when (byteCount /= BS.length fullbytes) (error "bytes sent mismatch")

202− --traceShowM ("sent bytes:" :: String, byteCount, fullbytes)

199+ traceBytes "sent bytes" fullbytes

203200

204201

205202

@@ -207,6 +204,7 @@ sendMessage msg socket' = do

207204-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the

208205-- 'ByteString' and wraps it with 'Array' constructors and hence the operation

209206-- is performed in constant time.

207+{-

210208{-# INLINE toArrayS #-}

211209toArrayS :: BS.ByteString -> SA.Array Word8

212210--slow path

@@ -220,3 +218,4 @@ toArraySlow :: BS.ByteString -> SA.Array Word8

220218toArraySlow bs = arr

221219 where

222220 arr = SA.fromList (BS.unpack bs)

221+-}