diff --git a/Network/HTTP2.hs b/Network/HTTP2.hs --- a/Network/HTTP2.hs +++ b/Network/HTTP2.hs @@ -6,6 +6,8 @@ Frame(..) , FrameHeader(..) , FramePayload(..) + , HeaderBlockFragment + , Padding , isPaddingDefined -- * Encoding , encodeFrame @@ -23,13 +25,12 @@ , FrameType , fromFrameTypeId , toFrameTypeId - -- * Types - , HeaderBlockFragment - , Padding - , Weight + -- * Priority , Priority(..) + , Weight , defaultPriority , highestPriority + , controlPriority -- * Stream identifier , StreamId , isControl @@ -54,8 +55,8 @@ , setPriority -- * SettingsList , SettingsList - , SettingsValue , SettingsKeyId(..) + , SettingsValue , fromSettingsKeyId , toSettingsKeyId , checkSettingsList @@ -68,8 +69,6 @@ , defaultInitialWindowSize , maxWindowSize , isWindowOverflow - -- * Misc - , recommendedConcurrency -- * Error code , ErrorCode , ErrorCodeId(..) @@ -78,11 +77,12 @@ -- * Error , HTTP2Error(..) , errorCodeId - -- * Magic + -- * Predefined values , connectionPreface , connectionPrefaceLength , frameHeaderLength , maxPayloadLength + , recommendedConcurrency ) where import Data.ByteString (ByteString) diff --git a/Network/HTTP2/Decode.hs b/Network/HTTP2/Decode.hs --- a/Network/HTTP2/Decode.hs +++ b/Network/HTTP2/Decode.hs @@ -131,6 +131,7 @@ ---------------------------------------------------------------- +-- | The type for frame payload decoder. type FramePayloadDecoder = FrameHeader -> ByteString -> Either HTTP2Error FramePayload @@ -149,6 +150,8 @@ ] -- | Decoding an HTTP/2 frame payload. +-- This function is considered to return a frame payload decoder +-- according to a frame type. decodeFramePayload :: FrameTypeId -> FramePayloadDecoder decodeFramePayload (FrameUnknown typ) = checkFrameSize $ decodeUnknownFrame typ decodeFramePayload ftyp = checkFrameSize decoder @@ -157,9 +160,11 @@ ---------------------------------------------------------------- +-- | Frame payload decoder for DATA frame. decodeDataFrame :: FramePayloadDecoder decodeDataFrame header bs = decodeWithPadding header bs DataFrame +-- | Frame payload decoder for HEADERS frame. decodeHeadersFrame :: FramePayloadDecoder decodeHeadersFrame header bs = decodeWithPadding header bs $ \bs' -> if hasPriority then @@ -171,12 +176,15 @@ where hasPriority = testPriority $ flags header +-- | Frame payload decoder for PRIORITY frame. decodePriorityFrame :: FramePayloadDecoder decodePriorityFrame _ bs = Right $ PriorityFrame $ priority bs +-- | Frame payload decoder for RST_STREAM frame. decoderstStreamFrame :: FramePayloadDecoder decoderstStreamFrame _ bs = Right $ RSTStreamFrame $ toErrorCodeId (word32 bs) +-- | Frame payload decoder for SETTINGS frame. decodeSettingsFrame :: FramePayloadDecoder decodeSettingsFrame FrameHeader{..} (PS fptr off _) = Right $ SettingsFrame alist where @@ -196,15 +204,18 @@ let v = fromIntegral w32 settings n' (p +. 6) (builder. ((k,v):)) +-- | Frame payload decoder for PUSH_PROMISE frame. decodePushPromiseFrame :: FramePayloadDecoder decodePushPromiseFrame header bs = decodeWithPadding header bs $ \bs' -> let (bs0,bs1) = BS.splitAt 4 bs' sid = streamIdentifier (word32 bs0) in PushPromiseFrame sid bs1 +-- | Frame payload decoder for PING frame. decodePingFrame :: FramePayloadDecoder decodePingFrame _ bs = Right $ PingFrame bs +-- | Frame payload decoder for GOAWAY frame. decodeGoAwayFrame :: FramePayloadDecoder decodeGoAwayFrame _ bs = Right $ GoAwayFrame sid ecid bs2 where @@ -213,6 +224,7 @@ sid = streamIdentifier (word32 bs0) ecid = toErrorCodeId (word32 bs1) +-- | Frame payload decoder for WINDOW_UPDATE frame. decodeWindowUpdateFrame :: FramePayloadDecoder decodeWindowUpdateFrame _ bs | wsi == 0 = Left $ ConnectionError ProtocolError "window update must not be 0" @@ -220,6 +232,7 @@ where !wsi = fromIntegral (word32 bs `clearBit` 31) +-- | Frame payload decoder for CONTINUATION frame. decodeContinuationFrame :: FramePayloadDecoder decodeContinuationFrame _ bs = Right $ ContinuationFrame bs diff --git a/Network/HTTP2/Priority.hs b/Network/HTTP2/Priority.hs --- a/Network/HTTP2/Priority.hs +++ b/Network/HTTP2/Priority.hs @@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-} +{-# LANGUAGE BangPatterns #-} -- | This is partial implementation of the priority of HTTP/2. -- @@ -9,18 +10,20 @@ -- -- This queue is fair for weight. Consider two weights: 201 and 101. -- Repeating enqueue/dequeue probably produces --- 201, 201, 101, 201, 201, 101, ... based on randomness. +-- 201, 201, 101, 201, 201, 101, ... -- -- Only one entry per stream should be enqueued. --- If multiple entries for a stream are inserted, the ordering --- is not preserved because of the randomness. module Network.HTTP2.Priority ( + -- * PriorityTree PriorityTree , newPriorityTree + -- * PriorityTree functions , prepare , enqueue , dequeue + , delete + , clear ) where #if __GLASGOW_HASKELL__ < 709 @@ -28,93 +31,120 @@ #endif import Control.Concurrent.STM import Control.Monad (when, unless) -import qualified Data.IntMap.Strict as Map import Data.IntMap.Strict (IntMap) -import Network.HTTP2.RandomSkewHeap (Heap) -import qualified Network.HTTP2.RandomSkewHeap as Heap +import qualified Data.IntMap.Strict as Map +import Network.HTTP2.Priority.Queue (TPriorityQueue) +import qualified Network.HTTP2.Priority.Queue as Q import Network.HTTP2.Types ---------------------------------------------------------------- -type Struct a = (PriorityQueue a, Priority) -- | Abstract data type for priority trees. -data PriorityTree a = PriorityTree (TVar (IntMap (Struct a))) - (PriorityQueue a) --- INVARIANT: Empty PriorityQueue is never enqueued in --- another PriorityQueue. -type PriorityQueue a = TPQueue (Element a) +data PriorityTree a = PriorityTree (TVar (Glue a)) + (TNestedPriorityQueue a) + (TQueue (StreamId, a)) + +type Glue a = IntMap (TNestedPriorityQueue a, Priority) + +-- INVARIANT: Empty TNestedPriorityQueue is never enqueued in +-- another TNestedPriorityQueue. +type TNestedPriorityQueue a = TPriorityQueue (Element a) + data Element a = Child a - | Parent (PriorityQueue a) + | Parent (TNestedPriorityQueue a) +---------------------------------------------------------------- + -- | Creating a new priority tree. newPriorityTree :: IO (PriorityTree a) -newPriorityTree = PriorityTree <$> newTVarIO Map.empty <*> atomically newTPQueue - -newPriorityQueue :: STM (PriorityQueue a) -newPriorityQueue = TPQueue <$> newTVar Heap.empty +newPriorityTree = PriorityTree <$> newTVarIO Map.empty + <*> atomically Q.new + <*> newTQueueIO ---------------------------------------------------------------- -- | Bringing up the structure of the priority tree. -- This must be used for Priority frame. prepare :: PriorityTree a -> StreamId -> Priority -> IO () -prepare (PriorityTree var _) sid p = atomically $ do - q <- newPriorityQueue +prepare (PriorityTree var _ _) sid p = atomically $ do + q <- Q.new modifyTVar' var $ Map.insert sid (q, p) --- | Enqueuing an element to the priority tree. +-- | Enqueuing an entry to the priority tree. -- This must be used for Header frame. -enqueue :: PriorityTree a -> a -> Priority -> IO () -enqueue (PriorityTree var q0) a p0 = atomically $ do +-- If 'controlPriority' is specified, +-- it is treated as a control frame and top-queued. +enqueue :: PriorityTree a -> StreamId -> Priority -> a -> IO () +enqueue (PriorityTree _ _ cq) sid p0 x + | p0 == controlPriority = atomically $ writeTQueue cq (sid,x) +enqueue (PriorityTree var q0 _) sid p0 x = atomically $ do m <- readTVar var - loop m (Child a) p0 + let !el = Child x + loop m el p0 where loop m el p - | pid == 0 = writeTPQueue q0 el p + | pid == 0 = Q.enqueue q0 sid w el | otherwise = case Map.lookup pid m of - Nothing -> writeTPQueue q0 el defaultPriority -- error case: checkme + Nothing -> Q.enqueue q0 sid w el -- error case: checkme Just (q', p') -> do - notQueued <- isTPQueueEmpty q' - writeTPQueue q' el p - when notQueued $ loop m (Parent q') p' + notQueued <- Q.isEmpty q' + Q.enqueue q' sid w el + when notQueued $ do + let !el' = Parent q' + loop m el' p' where pid = streamDependency p + w = weight p --- | Dequeuing an element from the priority tree. -dequeue :: PriorityTree a -> IO (a, Priority) -dequeue (PriorityTree _ q0) = atomically (loop q0) +-- | Dequeuing an entry from the priority tree. +dequeue :: PriorityTree a -> IO (StreamId, a) +dequeue (PriorityTree _ q0 cq) = atomically $ do + mx <- tryReadTQueue cq + case mx of + Just x -> return x + Nothing -> loop q0 where loop q = do - (el, w) <- readTPQueue q + (sid,w,el) <- Q.dequeue q case el of - Child a -> return (a, w) - p@(Parent q') -> do - r <- loop q' - empty <- isTPQueueEmpty q' - unless empty $ writeTPQueue q p w - return r - ----------------------------------------------------------------- --- --- The following code is originally written by Fumiaki Kinoshita --- - -newtype TPQueue a = TPQueue (TVar (Heap (a,Priority))) - -newTPQueue :: STM (TPQueue a) -newTPQueue = TPQueue <$> newTVar Heap.empty - -readTPQueue :: TPQueue a -> STM (a, Priority) -readTPQueue (TPQueue th) = do - h <- readTVar th - case Heap.uncons h of - Nothing -> retry - Just (ap, _, h') -> do - writeTVar th h' - return ap + Child x -> return $! (sid, x) + Parent q' -> do + entr <- loop q' + empty <- Q.isEmpty q' + unless empty $ Q.enqueue q sid w el + return entr -writeTPQueue :: TPQueue a -> a -> Priority -> STM () -writeTPQueue (TPQueue th) a p = modifyTVar' th $ Heap.insert (a,p) (weight p) +-- | Deleting the entry corresponding to 'StreamId'. +-- 'delete' and 'enqueue' are used to change the priority of +-- a live stream. +delete :: PriorityTree a -> StreamId -> Priority -> IO (Maybe a) +delete (PriorityTree var q0 _) sid p + | pid == 0 = atomically $ del q0 + | otherwise = atomically $ do + m <- readTVar var + case Map.lookup pid m of + Nothing -> return Nothing + Just (q,_) -> del q + where + pid = streamDependency p + del q = do + mel <- Q.delete sid q + case mel of + Nothing -> return Nothing + Just el -> case el of + Child x -> return $ Just x + Parent _ -> return Nothing -- fixme: this is error -isTPQueueEmpty :: TPQueue a -> STM Bool -isTPQueueEmpty (TPQueue th) = Heap.isEmpty <$> readTVar th +-- | Clearing the internal state for 'StreamId' from 'PriorityTree'. +-- When a stream is closed, this function MUST be called +-- to prevent memory leak. +clear :: PriorityTree a -> StreamId -> Priority -> IO () +clear (PriorityTree var q0 _) sid p + | pid == 0 = atomically $ Q.clear sid q0 + | otherwise = atomically $ do + m <- readTVar var + case Map.lookup pid m of + Nothing -> return () + Just (q,_) -> Q.clear sid q + where + pid = streamDependency p diff --git a/Network/HTTP2/Priority/PSQ.hs b/Network/HTTP2/Priority/PSQ.hs new file mode 100644 --- /dev/null +++ b/Network/HTTP2/Priority/PSQ.hs @@ -0,0 +1,99 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.HTTP2.Priority.PSQ ( + PriorityQueue + , empty + , isEmpty + , enqueue + , dequeue + , delete + , clear + ) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative ((<$>)) +#endif +import Data.Array (Array, listArray, (!)) +import Data.IntMap.Strict (IntMap) +import qualified Data.IntMap.Strict as I +import Data.IntPSQ (IntPSQ) +import qualified Data.IntPSQ as P +import Data.Word (Word64) + +---------------------------------------------------------------- + +type Key = Int +type Weight = Int +type Deficit = Word64 +type Heap a = IntPSQ Deficit (Weight, a) + +-- FIXME: The base (Word64) would be overflowed. +-- In that case, the heap must be re-constructed. +data PriorityQueue a = PriorityQueue { + baseDeficit :: {-# UNPACK #-} !Deficit + , deficitMap :: IntMap Deficit + , queue :: Heap a + } + +---------------------------------------------------------------- + +deficitSteps :: Int +deficitSteps = 65536 + +deficitList :: [Deficit] +deficitList = map calc idxs + where + idxs = [1..256] :: [Double] + calc w = round (fromIntegral deficitSteps / w) + +deficitTable :: Array Int Deficit +deficitTable = listArray (1,256) deficitList + +weightToDeficit :: Weight -> Deficit +weightToDeficit w = deficitTable ! w + +---------------------------------------------------------------- + +empty :: PriorityQueue a +empty = PriorityQueue 0 I.empty P.empty + +isEmpty :: PriorityQueue a -> Bool +isEmpty PriorityQueue{..} = P.null queue + +enqueue :: Key -> Weight -> a -> PriorityQueue a -> PriorityQueue a +enqueue k w x PriorityQueue{..} = + PriorityQueue baseDeficit deficitMap' queue' + where + !d = weightToDeficit w + !forNew = baseDeficit + d + f _ _ old = old + d + (!mold, !deficitMap') = I.insertLookupWithKey f k forNew deficitMap + !deficit' = case mold of + Nothing -> forNew + Just old -> old + d + !queue' = P.insert k deficit' (w,x) queue + +dequeue :: PriorityQueue a -> Maybe (Key, Weight, a, PriorityQueue a) +dequeue PriorityQueue{..} = case P.minView queue of + Nothing -> Nothing + Just (k, deficit, (w,x), queue') + | P.null queue' -> Just (k, w, x, empty) + | otherwise -> Just (k, w, x, PriorityQueue deficit deficitMap queue') + +delete :: Key -> PriorityQueue a -> (Maybe a, PriorityQueue a) +delete k PriorityQueue{..} = case P.findMin queue' of + Nothing -> (mx, empty) + Just (_,deficit,_) -> let !deficitMap' = I.delete k deficitMap + in (mx, PriorityQueue deficit deficitMap' queue') + where + (!mx,!queue') = P.alter f k queue + f Nothing = (Nothing, Nothing) + f (Just (_, (_,x))) = (Just x, Nothing) + +clear :: Key -> PriorityQueue a -> PriorityQueue a +clear k PriorityQueue{..} = PriorityQueue baseDeficit deficitMap' queue' + where + !deficitMap' = I.delete k deficitMap + !queue' = P.delete k queue -- just in case diff --git a/Network/HTTP2/Priority/Queue.hs b/Network/HTTP2/Priority/Queue.hs new file mode 100644 --- /dev/null +++ b/Network/HTTP2/Priority/Queue.hs @@ -0,0 +1,53 @@ +{-# LANGUAGE CPP #-} + +module Network.HTTP2.Priority.Queue ( + TPriorityQueue + , new + , isEmpty + , enqueue + , dequeue + , delete + , clear + ) where + +#if __GLASGOW_HASKELL__ < 709 +import Control.Applicative ((<$>)) +#endif +import Control.Concurrent.STM +import Network.HTTP2.Priority.PSQ (PriorityQueue) +import qualified Network.HTTP2.Priority.PSQ as Q + +---------------------------------------------------------------- + +type Key = Int +type Weight = Int + +newtype TPriorityQueue a = TPriorityQueue (TVar (PriorityQueue a)) + +new :: STM (TPriorityQueue a) +new = TPriorityQueue <$> newTVar Q.empty + +isEmpty :: TPriorityQueue a -> STM Bool +isEmpty (TPriorityQueue th) = Q.isEmpty <$> readTVar th + +enqueue :: TPriorityQueue a -> Key -> Weight -> a -> STM () +enqueue (TPriorityQueue th) k w x = modifyTVar' th $ Q.enqueue k w x + +dequeue :: TPriorityQueue a -> STM (Key, Weight, a) +dequeue (TPriorityQueue th) = do + h <- readTVar th + case Q.dequeue h of + Nothing -> retry + Just (k, w, x, h') -> do + writeTVar th h' + return (k, w, x) + +delete :: Key -> TPriorityQueue a -> STM (Maybe a) +delete k (TPriorityQueue th) = do + q <- readTVar th + let (mx, q') = Q.delete k q + writeTVar th q' + return mx + +clear :: Key -> TPriorityQueue a -> STM () +clear k (TPriorityQueue th) = modifyTVar' th $ \q -> Q.clear k q diff --git a/Network/HTTP2/RandomSkewHeap.hs b/Network/HTTP2/RandomSkewHeap.hs deleted file mode 100644 --- a/Network/HTTP2/RandomSkewHeap.hs +++ /dev/null @@ -1,88 +0,0 @@ -{-# LANGUAGE BangPatterns #-} - --- This data structure is based on skew heap. --- --- If we take weight as priority, a typical heap (priority queue) --- is not fair enough. Consider two weight 201 for A and 101 for B. --- A typical heap would generate A(201), A(200), A(199), A(198), ...., --- and finaly A(101), B(101), A(100), B(100). --- What we want is A, A, B, A, A, B... --- --- So, we introduce randomness to Skew Heap. --- --- In the random binary tree, --- an element is selected as the root with probability of --- 1 / (n + 1) where n is the size of the original tree. --- In the random skew heap, an element is selected as the root --- with the probability of weight / total_weight. --- --- Since this data structure uses random numbers, APIs should be --- essentially impure. But since this is used with STM, --- APIs are made to be pure with unsafePerformIO. - -module Network.HTTP2.RandomSkewHeap ( - Heap - , empty - , isEmpty - , singleton - , insert - , uncons - ) where - -import Network.HTTP2.Types (Weight) -import System.IO.Unsafe (unsafePerformIO) -import System.Random.MWC (createSystemRandom, uniformR, GenIO) - -data Heap a = Leaf | Node Weight -- total - a Weight !(Heap a) !(Heap a) deriving Show - -empty :: Heap a -empty = Leaf - -isEmpty :: Heap a -> Bool -isEmpty Leaf = True -isEmpty _ = False - -singleton :: a -> Weight -> Heap a -singleton a w = Node w a w Leaf Leaf - -insert :: a -> Weight -> Heap a -> Heap a -insert a w t = merge (singleton a w) t - --- if l is a singleton, w1 == tw1. -merge :: Heap t -> Heap t -> Heap t -merge t Leaf = t -merge Leaf t = t -merge l@(Node tw1 x1 w1 ll lr) r@(Node tw2 x2 w2 rl rr) - | g <= tw1 = Node tw x1 w1 lr $ merge ll r - | otherwise = Node tw x2 w2 rr $ merge rl l - where - tw = tw1 + tw2 - g = unsafePerformIO $ uniformR (1,tw) gen -{-# NOINLINE merge #-} - -uncons :: Heap a -> Maybe (a, Weight, Heap a) -uncons Leaf = Nothing -uncons (Node _ a w l r) = Just (a, w, t) - where - !t = merge l r - -gen :: GenIO -gen = unsafePerformIO createSystemRandom -{-# NOINLINE gen #-} - -{- -main :: IO () -main = do - let q = insert "c" 1 $ insert "b" 101 $ insert "a" 201 empty - loop 1000 q - where - loop :: Int -> Heap String -> IO () - loop 0 _ = return () - loop n q = do - case uncons q of - Nothing -> error "Nothing" - Just (x, w, q') -> do - putStrLn x - loop (n-1) (insert x w q') --} diff --git a/Network/HTTP2/Types.hs b/Network/HTTP2/Types.hs --- a/Network/HTTP2/Types.hs +++ b/Network/HTTP2/Types.hs @@ -71,6 +71,7 @@ , Priority(..) , defaultPriority , highestPriority + , controlPriority , Padding ) where @@ -92,8 +93,10 @@ ---------------------------------------------------------------- +-- | The type for raw error code. type ErrorCode = Word32 +-- | The type for error code. See . data ErrorCodeId = NoError | ProtocolError | InternalError @@ -112,7 +115,7 @@ | UnknownErrorCode ErrorCode deriving (Show, Read, Eq, Ord) --- | +-- | Converting 'ErrorCodeId' to 'ErrorCode'. -- -- >>> fromErrorCodeId NoError -- 0 @@ -135,7 +138,7 @@ fromErrorCodeId HTTP11Required = 0xd fromErrorCodeId (UnknownErrorCode w) = w --- | +-- | Converting 'ErrorCode' to 'ErrorCodeId'. -- -- >>> toErrorCodeId 0 -- NoError @@ -176,6 +179,7 @@ ---------------------------------------------------------------- +-- | The type for SETTINGS key. data SettingsKeyId = SettingsHeaderTableSize | SettingsEnablePush | SettingsMaxConcurrentStreams @@ -184,9 +188,10 @@ | SettingsMaxHeaderBlockSize deriving (Show, Read, Eq, Ord, Enum, Bounded) +-- | The type for raw SETTINGS value. type SettingsValue = Int -- Word32 --- | +-- | Converting 'SettingsKeyId' to raw value. -- -- >>> fromSettingsKeyId SettingsHeaderTableSize -- 1 @@ -201,7 +206,7 @@ maxSettingsKeyId :: Word16 maxSettingsKeyId = fromIntegral $ fromEnum (maxBound :: SettingsKeyId) --- | +-- | Converting raw value to 'SettingsKeyId'. -- -- >>> toSettingsKeyId 0 -- Nothing @@ -220,7 +225,7 @@ ---------------------------------------------------------------- --- | Settings containing raw values. +-- | Association list of SETTINGS. type SettingsList = [(SettingsKeyId,SettingsValue)] -- | Checking 'SettingsList' and reporting an error if any. @@ -282,6 +287,7 @@ update (SettingsMaxFrameSize,x) def = def { maxFrameSize = x } update (SettingsMaxHeaderBlockSize,x) def = def { maxHeaderBlockSize = Just x } +-- | The type for window size. type WindowSize = Int -- | The default initial window size. @@ -319,8 +325,10 @@ ---------------------------------------------------------------- +-- | The type for weight in priority. Its values are from 1 to 256. type Weight = Int +-- | Type for stream priority data Priority = Priority { exclusive :: Bool , streamDependency :: StreamId @@ -341,8 +349,13 @@ highestPriority :: Priority highestPriority = Priority False 0 256 +-- | Priority for control frames to be top-queued. +controlPriority :: Priority +controlPriority = Priority False 0 (-1) + ---------------------------------------------------------------- +-- | The type for raw frame type. type FrameType = Word8 minFrameType :: FrameType @@ -351,7 +364,7 @@ maxFrameType :: FrameType maxFrameType = 9 --- Valid frame types +-- | The type for frame type. data FrameTypeId = FrameData | FrameHeaders | FramePriority @@ -365,7 +378,7 @@ | FrameUnknown FrameType deriving (Show, Eq, Ord) --- | +-- | Converting 'FrameTypeId' to 'FrameType'. -- -- >>> fromFrameTypeId FrameData -- 0 @@ -386,7 +399,7 @@ fromFrameTypeId FrameContinuation = 9 fromFrameTypeId (FrameUnknown x) = x --- | +-- | Converting 'FrameType' to 'FrameTypeId'. -- -- >>> toFrameTypeId 0 -- FrameData @@ -419,69 +432,79 @@ ---------------------------------------------------------------- -- Flags +-- | The type for flags. type FrameFlags = Word8 --- | +-- | The initial value of flags. No flags are set. +-- -- >>> defaultFlags -- 0 defaultFlags :: FrameFlags defaultFlags = 0 --- | +-- | Checking if the END_STREAM flag is set. -- >>> testEndStream 0x1 -- True testEndStream :: FrameFlags -> Bool testEndStream x = x `testBit` 0 --- | +-- | Checking if the ACK flag is set. -- >>> testAck 0x1 -- True testAck :: FrameFlags -> Bool testAck x = x `testBit` 0 -- fixme: is the spec intentional? --- | +-- | Checking if the END_HEADERS flag is set. +-- -- >>> testEndHeader 0x4 -- True testEndHeader :: FrameFlags -> Bool testEndHeader x = x `testBit` 2 --- | +-- | Checking if the PADDED flag is set. +-- -- >>> testPadded 0x8 -- True testPadded :: FrameFlags -> Bool testPadded x = x `testBit` 3 --- | +-- | Checking if the PRIORITY flag is set. +-- -- >>> testPriority 0x20 -- True testPriority :: FrameFlags -> Bool testPriority x = x `testBit` 5 --- | +-- | Setting the END_STREAM flag. +-- -- >>> setEndStream 0 -- 1 setEndStream :: FrameFlags -> FrameFlags setEndStream x = x `setBit` 0 --- | +-- | Setting the ACK flag. +-- -- >>> setAck 0 -- 1 setAck :: FrameFlags -> FrameFlags setAck x = x `setBit` 0 -- fixme: is the spec intentional? --- | +-- | Setting the END_HEADERS flag. +-- -- >>> setEndHeader 0 -- 4 setEndHeader :: FrameFlags -> FrameFlags setEndHeader x = x `setBit` 2 --- | +-- | Setting the PADDED flag. +-- -- >>> setPadded 0 -- 8 setPadded :: FrameFlags -> FrameFlags setPadded x = x `setBit` 3 --- | +-- | Setting the PRIORITY flag. +-- -- >>> setPriority 0 -- 32 setPriority :: FrameFlags -> FrameFlags @@ -489,9 +512,11 @@ ---------------------------------------------------------------- +-- | The type for stream identifier type StreamId = Int --- | +-- | Checking if the stream identifier for control. +-- -- >>> isControl 0 -- True -- >>> isControl 1 @@ -500,7 +525,8 @@ isControl 0 = True isControl _ = False --- | +-- | Checking if the stream identifier for request. +-- -- >>> isRequest 0 -- False -- >>> isRequest 1 @@ -508,7 +534,8 @@ isRequest :: StreamId -> Bool isRequest = odd --- | +-- | Checking if the stream identifier for response. +-- -- >>> isResponse 0 -- False -- >>> isResponse 2 @@ -517,18 +544,24 @@ isResponse 0 = False isResponse n = even n -testExclusive :: Int -> Bool +-- | Checking if the exclusive flag is set. +testExclusive :: StreamId -> Bool testExclusive n = n `testBit` 31 -setExclusive :: Int -> Int +-- | Setting the exclusive flag. +setExclusive :: StreamId -> StreamId setExclusive n = n `setBit` 31 -clearExclusive :: Int -> Int +-- | Clearing the exclusive flag. +clearExclusive :: StreamId -> StreamId clearExclusive n = n `clearBit` 31 ---------------------------------------------------------------- +-- | The type for fragments of a header encoded with HPACK. type HeaderBlockFragment = ByteString + +-- | The type for padding in payloads. type Padding = ByteString ---------------------------------------------------------------- @@ -583,6 +616,11 @@ ---------------------------------------------------------------- -- | Checking if padding is defined in this frame type. +-- +-- >>> isPaddingDefined $ DataFrame "" +-- True +-- >>> isPaddingDefined $ PingFrame "" +-- False isPaddingDefined :: FramePayload -> Bool isPaddingDefined (DataFrame _) = True isPaddingDefined (HeadersFrame _ _) = True diff --git a/bench/Main.hs b/bench/Main.hs new file mode 100644 --- /dev/null +++ b/bench/Main.hs @@ -0,0 +1,225 @@ +{-# LANGUAGE BangPatterns #-} + +module Main where + +import Control.Concurrent.STM +import Criterion.Main +import Data.List (foldl') +import System.Random + +import qualified ArrayOfQueue as A +import qualified ArrayOfQueueIO as AIO +import qualified BinaryHeap as B +import qualified BinaryHeapIO as BIO +import qualified Heap as O +import qualified Network.HTTP2.Priority.PSQ as P +import qualified RandomSkewHeap as R + +type Key = Int +type Weight = Int + +numOfStreams :: Int +numOfStreams = 100 + +numOfTrials :: Int +numOfTrials = 10000 + +main :: IO () +main = do + gen <- getStdGen + let ks = [1,3..] + ws = take numOfStreams $ randomRs (1,256) gen + xs = zip ks ws + defaultMain [ + bgroup "enqueue & dequeue" [ + bench "Random Skew Heap" $ whnf enqdeqR xs + , bench "Okasaki Heap" $ whnf enqdeqO xs + , bench "Priority Search Queue" $ whnf enqdeqP xs + , bench "Binary Heap STM" $ nfIO (enqdeqB xs) + , bench "Binary Heap IO" $ nfIO (enqdeqBIO xs) + , bench "Array of Queue STM" $ nfIO (enqdeqA xs) + , bench "Array of Queue IO" $ nfIO (enqdeqAIO xs) + ] + , bgroup "delete" [ + bench "Random Skew Heap" $ whnf deleteR xs + , bench "Okasaki Heap" $ whnf deleteO xs + , bench "Priority Search Queue" $ whnf deleteP xs + , bench "Binary Heap STM" $ nfIO (deleteB xs) + , bench "Binary Heap IO" $ nfIO (deleteBIO xs) + , bench "Array of Queue IO" $ nfIO (deleteAIO xs) + ] + ] + +---------------------------------------------------------------- + +enqdeqR :: [(Key,Weight)] -> () +enqdeqR xs = loop pq numOfTrials + where + !pq = createR xs R.empty + loop _ 0 = () + loop q !n = case R.dequeue q of + Nothing -> error "enqdeqR" + Just (k,w,x,q') -> let !q'' = R.enqueue k w x q' + in loop q'' (n - 1) + +deleteR :: [(Key,Weight)] -> R.PriorityQueue Int +deleteR xs = foldl' (\p k -> snd (R.delete k p)) pq ks + where + !pq = createR xs R.empty + (ks,_) = unzip xs + +createR :: [(Key,Weight)] -> R.PriorityQueue Int -> R.PriorityQueue Int +createR [] !q = q +createR ((k,w):xs) !q = createR xs q' + where + !q' = R.enqueue k w k q + +---------------------------------------------------------------- + +enqdeqO :: [(Key,Weight)] -> O.PriorityQueue Int +enqdeqO xs = loop pq numOfTrials + where + !pq = createO xs O.empty + loop !q 0 = q + loop !q !n = case O.dequeue q of + Nothing -> error "enqdeqO" + Just (k,w,x,q') -> loop (O.enqueue k w x q') (n - 1) + +deleteO :: [(Key,Weight)] -> O.PriorityQueue Int +deleteO xs = foldl' (\p k -> snd (O.delete k p)) pq ks + where + !pq = createO xs O.empty + (ks,_) = unzip xs + +createO :: [(Key,Weight)] -> O.PriorityQueue Int -> O.PriorityQueue Int +createO [] !q = q +createO ((k,w):xs) !q = createO xs q' + where + !q' = O.enqueue k w k q + +---------------------------------------------------------------- + +enqdeqP :: [(Key,Weight)] -> P.PriorityQueue Int +enqdeqP xs = loop pq numOfTrials + where + !pq = createP xs P.empty + loop !q 0 = q + loop !q !n = case P.dequeue q of + Nothing -> error "enqdeqP" + Just (k,w,x,q') -> loop (P.enqueue k w x q') (n - 1) + +deleteP :: [(Key,Weight)] -> P.PriorityQueue Int +deleteP xs = foldl' (\p k -> snd (P.delete k p)) pq ks + where + !pq = createP xs P.empty + (ks,_) = unzip xs + +createP :: [(Key,Weight)] -> P.PriorityQueue Int -> P.PriorityQueue Int +createP [] !q = q +createP ((k,w):xs) !q = createP xs (P.enqueue k w k q) + +---------------------------------------------------------------- + +enqdeqB :: [(Key,Weight)] -> IO () +enqdeqB xs = do + q <- atomically (B.new numOfStreams) + createB xs q + loop q numOfTrials + where + loop _ 0 = return () + loop q !n = do + Just (k,w,x) <- atomically $ B.dequeue q + atomically $ B.enqueue k w x q + loop q (n - 1) + +deleteB :: [(Key,Weight)] -> IO () +deleteB xs = do + q <- atomically $ B.new numOfStreams + createB xs q + mapM_ (\k -> atomically $ B.delete k q) keys + where + (keys,_) = unzip xs + +createB :: [(Key,Weight)] -> B.PriorityQueue Int -> IO () +createB [] _ = return () +createB ((k,w):xs) !q = do + atomically $ B.enqueue k w k q + createB xs q + +---------------------------------------------------------------- + +enqdeqBIO :: [(Key,Weight)] -> IO () +enqdeqBIO xs = do + q <- BIO.new numOfStreams + createBIO xs q + loop q numOfTrials + where + loop _ 0 = return () + loop q !n = do + Just (k,w,x) <- BIO.dequeue q + BIO.enqueue k w x q + loop q (n - 1) + +deleteBIO :: [(Key,Weight)] -> IO () +deleteBIO xs = do + q <- BIO.new numOfStreams + createBIO xs q + mapM_ (\k -> BIO.delete k q) keys + where + (keys,_) = unzip xs + +createBIO :: [(Key,Weight)] -> BIO.PriorityQueue Int -> IO () +createBIO [] _ = return () +createBIO ((k,w):xs) !q = do + BIO.enqueue k w k q + createBIO xs q + +---------------------------------------------------------------- + +enqdeqA :: [(Key,Weight)] -> IO () +enqdeqA xs = do + q <- atomically A.new + createA xs q + loop q numOfTrials + where + loop _ 0 = return () + loop q !n = do + Just (k,w,x) <- atomically $ A.dequeue q + atomically $ A.enqueue k w x q + loop q (n - 1) + +createA :: [(Key,Weight)] -> A.PriorityQueue Int -> IO () +createA [] _ = return () +createA ((k,w):xs) !q = do + atomically $ A.enqueue k w k q + createA xs q + +---------------------------------------------------------------- + +enqdeqAIO :: [(Key,Weight)] -> IO () +enqdeqAIO xs = do + q <- AIO.new + createAIO xs q + loop q numOfTrials + where + loop _ 0 = return () + loop q !n = do + Just (k,w,x) <- AIO.dequeue q + AIO.enqueue k w x q + loop q (n - 1) + +deleteAIO :: [(Key,Weight)] -> IO () +deleteAIO xs = do + q <- AIO.new + createAIO xs q + mapM_ (\k -> AIO.delete k q) keys + where + (keys,_) = unzip xs + +createAIO :: [(Key,Weight)] -> AIO.PriorityQueue Int -> IO () +createAIO [] _ = return () +createAIO ((k,w):xs) !q = do + AIO.enqueue k w k q + createAIO xs q + +---------------------------------------------------------------- diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -1,5 +1,5 @@ Name: http2 -Version: 1.1.0 +Version: 1.2.0 Author: Kazu Yamamoto Maintainer: Kazu Yamamoto License: BSD3 @@ -55,14 +55,15 @@ Network.HPACK.Types Network.HTTP2.Decode Network.HTTP2.Encode - Network.HTTP2.RandomSkewHeap + Network.HTTP2.Priority.PSQ + Network.HTTP2.Priority.Queue Network.HTTP2.Types Build-Depends: base >= 4.6 && < 5 , array , bytestring >= 0.10 , bytestring-builder , containers >= 0.5 - , mwc-random + , psqueues , stm , unordered-containers @@ -98,6 +99,7 @@ , hex , hspec >= 1.3 , mwc-random + , psqueues , stm , unordered-containers , word8 @@ -236,3 +238,20 @@ , http2 , text , unordered-containers + +Benchmark criterion + Type: exitcode-stdio-1.0 + Default-Language: Haskell2010 + Hs-Source-Dirs: bench, . + Ghc-Options: -Wall + Main-Is: Main.hs + Build-Depends: base + , array + , containers + , criterion + , hashtables + , heaps + , mwc-random + , psqueues + , random + , stm diff --git a/test/HTTP2/PrioritySpec.hs b/test/HTTP2/PrioritySpec.hs --- a/test/HTTP2/PrioritySpec.hs +++ b/test/HTTP2/PrioritySpec.hs @@ -1,47 +1,78 @@ +{-# LANGUAGE BangPatterns #-} + module HTTP2.PrioritySpec where +import Data.List (group, sort) import Test.Hspec -import Control.Monad (void) import Network.HTTP2.Priority +import qualified Network.HTTP2.Priority.PSQ as P import Network.HTTP2.Types spec :: Spec spec = do - describe "enqueue & dequeue" $ do - it "enqueue and dequeue frames from Firefox properly" $ do - pt <- newPriorityTree :: IO (PriorityTree Int) - prepare pt 3 $ Priority False 0 201 - prepare pt 5 $ Priority False 0 101 - prepare pt 7 $ Priority False 0 1 - prepare pt 9 $ Priority False 7 1 - prepare pt 11 $ Priority False 3 1 - enqueue pt 13 $ Priority False 11 32 - (sid13,_) <- dequeue pt - sid13 `shouldBe` 13 - enqueue pt 15 $ Priority False 3 32 - enqueue pt 17 $ Priority False 3 32 - enqueue pt 19 $ Priority False 3 32 - enqueue pt 21 $ Priority False 3 32 - enqueue pt 23 $ Priority False 3 32 - enqueue pt 25 $ Priority False 3 32 - enqueue pt 27 $ Priority False 11 22 - enqueue pt 29 $ Priority False 11 22 - enqueue pt 31 $ Priority False 11 22 - enqueue pt 33 $ Priority False 5 32 - enqueue pt 35 $ Priority False 5 32 - enqueue pt 37 $ Priority False 5 32 - -- Currently, just checking no errors. - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt - void $ dequeue pt + describe "priority tree" $ do + it "enqueue and dequeue frames from Firefox properly" $ firefox + describe "base priority queue" $ do + it "queues entries based on weight" $ do + let q = P.enqueue 5 1 5 $ + P.enqueue 3 101 3 $ + P.enqueue 1 201 1 P.empty + xs = enqdeq q 1000 + map length (group (sort xs)) `shouldBe` [664,333,3] +firefox :: IO () +firefox = do + pt <- newPriorityTree :: IO (PriorityTree Int) + prepare pt 3 (pri 0 201) + prepare pt 5 (pri 0 101) + prepare pt 7 (pri 0 1) + prepare pt 9 (pri 7 1) + prepare pt 11 (pri 3 1) + enQ pt 13 (pri 11 32) + dequeue pt `shouldReturn` (13,13) + enQ pt 15 (pri 3 32) + enQ pt 17 (pri 3 32) + enQ pt 19 (pri 3 32) + enQ pt 21 (pri 3 32) + enQ pt 23 (pri 3 32) + enQ pt 25 (pri 3 32) + enQ pt 27 (pri 11 22) + enQ pt 29 (pri 11 22) + enQ pt 31 (pri 11 22) + enQ pt 33 (pri 5 32) + enQ pt 35 (pri 5 32) + enQ pt 37 (pri 5 32) + dequeue pt `shouldReturn` (15,15) + clear pt 15 (pri 3 32) + clear pt 15 (pri 3 32) + dequeue pt `shouldReturn` (33,33) + dequeue pt `shouldReturn` (17,17) + clear pt 17 (pri 3 32) + clear pt 17 (pri 3 32) + delete pt 17 (pri 3 32) `shouldReturn` Nothing + delete pt 31 (pri 11 22) `shouldReturn` Just 31 + dequeue pt `shouldReturn` (19,19) + dequeue pt `shouldReturn` (35,35) + dequeue pt `shouldReturn` (21,21) + dequeue pt `shouldReturn` (23,23) + dequeue pt `shouldReturn` (37,37) + dequeue pt `shouldReturn` (25,25) + dequeue pt `shouldReturn` (27,27) + dequeue pt `shouldReturn` (29,29) + enQ pt 39 (pri 3 32) + dequeue pt `shouldReturn` (39,39) + +enQ :: PriorityTree Int -> StreamId -> Priority -> IO () +enQ pt sid p = enqueue pt sid p sid + +pri :: StreamId -> Weight -> Priority +pri dep w = Priority False dep w + +enqdeq :: P.PriorityQueue Int -> Int -> [Int] +enqdeq pq num = loop pq num [] + where + loop _ 0 xs = xs + loop !q !n xs = case P.dequeue q of + Nothing -> error "enqdeq" + Just (k,w,x,q') -> loop (P.enqueue k w x q') (n - 1) (x:xs)