Work-in-progress

Author: LaurentRDC

packages/network-transport-quic/network-transport-quic.cabal

@@ -0,0 +1,52 @@
+cabal-version: 3.0
+Name:          network-transport-quic
+Version:       0.1.0
+build-Type:    Simple
+License:       BSD-3-Clause
+License-file:  LICENSE
+Copyright:     Well-Typed LLP, Tweag I/O Limited
+Author:        Duncan Coutts, Nicolas Wu, Edsko de Vries
+maintainer:    The Distributed Haskell team
+Stability:     experimental
+Homepage:      http://haskell-distributed.github.com
+Bug-Reports:   https://github.com/haskell-distributed/distributed-process/issues
+Synopsis:      Networking layer for Cloud Haskell based on QUIC
+Description:   Networking layer for Cloud Haskell based on QUIC
+tested-with:   GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1
+Category:      Network
+extra-doc-files: ChangeLog
+
+source-repository head
+  Type:     git
+  Location: https://github.com/haskell-distributed/distributed-process
+  SubDir:   packages/network-transport-quic
+
+common warnings
+    ghc-options: -Wall
+                 -Wcompat
+                 -Widentities
+                 -Wincomplete-uni-patterns
+                 -Wincomplete-record-updates
+                 -Wredundant-constraints
+                 -fhide-source-paths
+                 -Wpartial-fields
+                 -Wunused-packages
+
+library
+  import:             warnings
+  build-depends:      async >= 2.2 && < 2.3,
+                      base >= 4.14 && < 5,
+                      binary,
+                      bytestring >= 0.10 && < 0.13,
+                      containers,
+                      network >= 3.1 && < 3.3,
+                      network-transport >= 0.5 && < 0.6,
+                      quic ^>=0.2,
+                      stm >=2.4 && <2.6
+  exposed-modules:    Network.Transport.QUIC
+  other-modules:      Network.Transport.QUIC.EndpointState
+  default-language:   Haskell2010
+  default-extensions: ImportQualifiedPost
+  -- The -threaded option is /required/ to use the quic library
+  ghc-options:        -threaded
+  hs-source-dirs:     src

packages/network-transport-quic/src/Network/Transport/QUIC.hs

@@ -0,0 +1,80 @@
+module Network.Transport.QUIC (
+    createTransport,
+    QUICAddr (..),
+) where
+
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TQueue (
+    TQueue,
+    newTQueueIO,
+    readTQueue,
+    writeTQueue,
+ )
+import Data.ByteString qualified as BS
+import Data.ByteString.Char8 qualified as BS8
+import Network.QUIC qualified as QUIC
+import Network.QUIC.Server (defaultServerConfig)
+import Network.QUIC.Server qualified as QUIC.Server
+import Network.Transport (ConnectionId, EndPoint (..), EndPointAddress (EndPointAddress), Event (..), NewEndPointErrorCode, Transport (..), TransportError (..))
+
+import Network.Socket (HostName, ServiceName)
+
+{- | Create a new Transport.
+
+Only a single transport should be created per Haskell process
+(threads can, and should, create their own endpoints though).
+-}
+createTransport :: QUICAddr -> IO Transport
+createTransport quicAddr = do
+    pure $ Transport (newEndpoint quicAddr) closeQUICTransport
+
+data QUICAddr = QUICAddr
+    { quicBindHost :: !HostName
+    , quicBindPort :: !ServiceName
+    }
+
+newEndpoint :: QUICAddr -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
+newEndpoint quicAddr = do
+    eventQueue <- newTQueueIO
+
+    QUIC.Server.run
+        defaultServerConfig
+        ( \conn -> do
+            -- TODO: create a bidirectional stream
+            -- which can be re-used for sending
+            quicStream <- QUIC.acceptStream conn
+            -- TODO: how to ensure positivity of ConnectionId? QUIC StreamID should be a 62 bit integer,
+            -- so there's room to make it a positive 64 bit integer (ConnectionId ~ Word64)
+            let connId = fromIntegral (QUIC.streamId quicStream)
+            receiveLoop connId quicStream eventQueue
+        )
+
+    pure . Right $
+        EndPoint
+            (atomically (readTQueue eventQueue))
+            (encodeQUICAddr quicAddr)
+            _
+            _
+            _
+            _
+  where
+    receiveLoop ::
+        ConnectionId ->
+        QUIC.Stream ->
+        TQueue Event ->
+        IO ()
+    receiveLoop connId stream eventQueue = do
+        incoming <- QUIC.recvStream stream 1024 -- TODO: variable length?
+        -- TODO: check some state whether we should stop all connections
+        if BS.null incoming
+            then do
+                atomically (writeTQueue eventQueue (ConnectionClosed connId))
+            else do
+                atomically (writeTQueue eventQueue (Received connId [incoming]))
+                receiveLoop connId stream eventQueue
+
+    encodeQUICAddr :: QUICAddr -> EndPointAddress
+    encodeQUICAddr (QUICAddr host port) = EndPointAddress (BS8.pack $ host <> ":" <> port)
+
+closeQUICTransport :: IO ()
+closeQUICTransport = pure ()

packages/network-transport-quic/src/Network/Transport/QUIC/EndpointState.hs

@@ -0,0 +1,29 @@
+module Network.Transport.QUIC.EndpointState (
+    EndpointState,
+    newConnection,
+) where
+
+import Data.Map.Strict (Map)
+import Network.QUIC (Stream, StreamId)
+import Network.QUIC qualified as QUIC
+import Network.Transport (ConnectionId)
+
+import Data.Map.Strict qualified as Map
+
+newtype EndpointState = EndpointState
+    { streamIds :: Map StreamId ConnectionId
+    , streams   :: Map 
+    }
+
+newConnection :: Stream -> EndpointState -> EndpointState
+newConnection stream (EndpointState sids) =
+    EndpointState
+        ( Map.insert
+            (QUIC.streamId stream)
+            -- TODO: how to ensure positivity? QUIC StreamID should be a 62 bit integer,
+            -- so there's room to make it a positive 64 bit integer (ConnectionId ~ Word64)
+            (fromIntegral $ QUIC.streamId stream)
+            sids
+        )
+
+lookupConnectionId :: EndpointState -> ConnectionId