diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Li Meng Jun (c) 2017 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Li Meng Jun nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# periodic-common + +Periodic task system common. diff --git a/periodic-common.cabal b/periodic-common.cabal new file mode 100644 --- /dev/null +++ b/periodic-common.cabal @@ -0,0 +1,46 @@ +name: periodic-common +version: 1.1.7.0 +synopsis: Periodic task system common. +description: Periodic task system common library. +homepage: https://github.com/Lupino/haskell-periodic/tree/master/periodic-common#readme +license: BSD3 +license-file: LICENSE +author: Li Meng Jun +maintainer: lmjubuntu@gmail.com +copyright: MIT +category: System,Web +build-type: Simple +extra-source-files: README.md +cabal-version: >=1.10 + +library + hs-source-dirs: src + exposed-modules: Periodic.Types + , Periodic.Types.ServerCommand + , Periodic.Types.ClientCommand + , Periodic.Types.WorkerCommand + , Periodic.Types.Internal + , Periodic.Types.ClientType + , Periodic.Types.Packet + , Periodic.Types.Error + , Periodic.Types.Job + , Periodic.Node + , Periodic.IOList + , Periodic.CRC32 + + build-depends: base >= 4.7 && < 5 + , bytestring + , byteable + , entropy + , hashable + , text + , binary + , vector + , metro + , unliftio + , hslogger + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/Lupino/haskell-periodic diff --git a/src/Periodic/CRC32.hs b/src/Periodic/CRC32.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/CRC32.hs @@ -0,0 +1,100 @@ +{-# LANGUAGE TypeFamilies #-} +module Periodic.CRC32 + ( CRC32(..) + , digest + , updateDigest32 + ) where + + +------------------------------------------------------------------------------- +import Data.Bits +import Data.ByteString as BS +import Data.Vector.Unboxed as V +import Data.Word + + +newtype CRC32 = CRC32 { crc32 :: Word32 } deriving (Show, Eq, Ord) + + +------------------------------------------------------------------------------- +updateDigest32 :: CRC32 -> ByteString -> CRC32 +updateDigest32 crc = xorFinal . BS.foldl go crc + where + xorFinal (CRC32 x) = CRC32 (x `xor` 0xffffffff) + go (CRC32 crc') b8 = + let idx = fromIntegral (crc' `xor` b32) .&. 0xff + b32 = fromIntegral b8 + in CRC32 (((crc' `shiftR` 8) .&. 0x00ffffff) `xor` (table ! idx)) + + +------------------------------------------------------------------------------- +table :: V.Vector Word32 +table = V.fromList + [ 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba + , 0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3 + , 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988 + , 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91 + , 0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de + , 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7 + , 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec + , 0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5 + , 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172 + , 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b + , 0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940 + , 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59 + , 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116 + , 0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f + , 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924 + , 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d + , 0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a + , 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433 + , 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818 + , 0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01 + , 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e + , 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457 + , 0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c + , 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65 + , 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2 + , 0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb + , 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0 + , 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9 + , 0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086 + , 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f + , 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4 + , 0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad + , 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a + , 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683 + , 0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8 + , 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1 + , 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe + , 0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7 + , 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc + , 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5 + , 0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252 + , 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b + , 0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60 + , 0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79 + , 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236 + , 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f + , 0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04 + , 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d + , 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a + , 0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713 + , 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38 + , 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21 + , 0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e + , 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777 + , 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c + , 0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45 + , 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2 + , 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db + , 0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0 + , 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9 + , 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6 + , 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf + , 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94 + , 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d + ] + +digest :: ByteString -> CRC32 +digest = updateDigest32 (CRC32 0xffffffff) diff --git a/src/Periodic/IOList.hs b/src/Periodic/IOList.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/IOList.hs @@ -0,0 +1,55 @@ +module Periodic.IOList + ( IOList + , newIOList + , insert + , append + , elem + , elemSTM + , delete + , deleteSTM + , toList + , toListSTM + , clearSTM + , fromList + ) where + +import qualified Data.List as L +import Prelude hiding (elem) +import UnliftIO (MonadIO, STM, TVar, atomically, modifyTVar', + newTVarIO, readTVar, readTVarIO, writeTVar) + + +newtype IOList a = IOList (TVar [a]) + +newIOList :: MonadIO m => m (IOList a) +newIOList = IOList <$> newTVarIO [] + +fromList :: MonadIO m => [a] -> m (IOList a) +fromList l = IOList <$> newTVarIO l + +insert :: MonadIO m => IOList a -> a -> m () +insert (IOList h) a = atomically . modifyTVar' h $ \v -> a:v + +append :: MonadIO m => IOList a -> a -> m () +append (IOList h) a = atomically . modifyTVar' h $ \v -> v ++ [a] + +elem :: (Eq a, MonadIO m) => IOList a -> a -> m Bool +elem (IOList h) a = L.elem a <$> readTVarIO h + +elemSTM :: (Eq a) => IOList a -> a -> STM Bool +elemSTM (IOList h) a = L.elem a <$> readTVar h + +delete :: (Eq a, MonadIO m) => IOList a -> a -> m () +delete (IOList h) a = atomically . modifyTVar' h $ L.delete a + +deleteSTM :: (Eq a) => IOList a -> a -> STM () +deleteSTM (IOList h) a = modifyTVar' h $ L.delete a + +toList :: MonadIO m => IOList a -> m [a] +toList (IOList h) = readTVarIO h + +toListSTM :: IOList a -> STM [a] +toListSTM (IOList h) = readTVar h + +clearSTM :: IOList a -> STM () +clearSTM (IOList h) = writeTVar h [] diff --git a/src/Periodic/Node.hs b/src/Periodic/Node.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Node.hs @@ -0,0 +1,43 @@ +module Periodic.Node + ( Nid (..) + , NodeEnv + , NodeT + , SessionT + , SessionEnv + , SessionEnv1 + , NodeEnvList + , runNodeT + , sessionGen + , defaultSessionHandler + ) where + +import Metro.IOHashMap (IOHashMap) +import qualified Metro.Node as M (NodeEnv1, NodeT, runNodeT1) +import qualified Metro.Session as M (SessionEnv, SessionEnv1, SessionT, + getSessionId) +import Periodic.Types (Msgid (..), Nid (..), Packet, msgidLength) +import System.Entropy (getEntropy) +import System.Log.Logger (errorM) +import UnliftIO (MonadIO, liftIO) + +type NodeEnv u rpkt = M.NodeEnv1 u Nid Msgid (Packet rpkt) + +type NodeT u rpkt = M.NodeT u Nid Msgid (Packet rpkt) + +type SessionT u rpkt = M.SessionT u Nid Msgid (Packet rpkt) + +type SessionEnv u rpkt = M.SessionEnv u Nid Msgid (Packet rpkt) +type SessionEnv1 u rpkt = M.SessionEnv1 u Nid Msgid (Packet rpkt) + +sessionGen :: IO Msgid +sessionGen = Msgid <$> getEntropy msgidLength + +type NodeEnvList u rpkt tp = IOHashMap Nid (NodeEnv u rpkt tp) + +runNodeT :: Monad m => NodeEnv u rpkt tp -> NodeT u rpkt tp m a -> m a +runNodeT = M.runNodeT1 + +defaultSessionHandler :: MonadIO m => SessionT u rpkt tp m () +defaultSessionHandler = do + pid <- M.getSessionId + liftIO $ errorM "Periodic.Node" $ "Session [" ++ show pid ++ "] not found." diff --git a/src/Periodic/Types.hs b/src/Periodic/Types.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types.hs @@ -0,0 +1,9 @@ +module Periodic.Types + ( module X + ) where + +import Periodic.Types.ClientType as X +import Periodic.Types.Error as X +import Periodic.Types.Internal as X +import Periodic.Types.Job as X +import Periodic.Types.Packet as X diff --git a/src/Periodic/Types/ClientCommand.hs b/src/Periodic/Types/ClientCommand.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/ClientCommand.hs @@ -0,0 +1,87 @@ +{-# LANGUAGE OverloadedStrings #-} +module Periodic.Types.ClientCommand + ( ClientCommand (..) + ) where + +import Data.Binary +import Data.Binary.Get (getWord32be) +import Data.Binary.Put (putWord32be) +import Periodic.Types.Internal +import Periodic.Types.Job (FuncName, Job, JobName) + +data ClientCommand = SubmitJob Job + | Status + | Ping + | DropFunc FuncName + | RemoveJob FuncName JobName + | ConfigGet ConfigKey + | ConfigSet ConfigKey Int + | Dump + | Load [Job] + | Shutdown + | RunJob Job + deriving (Show) + +instance Binary ClientCommand where + get = do + tp <- getWord8 + case tp of + 13 -> SubmitJob <$> get + 14 -> pure Status + 9 -> pure Ping + 15 -> DropFunc <$> get + 17 -> do + fn <- get + RemoveJob fn <$> get + 20 -> pure Shutdown + 22 -> ConfigGet <$> get + 23 -> do + key <- get + val <- getWord32be + pure . ConfigSet key $ fromIntegral val + 18 -> pure Dump + 19 -> Load <$> get + 25 -> RunJob <$> get + _ -> error $ "Error ClientCommand" ++ show tp + + put (SubmitJob job) = do + putWord8 13 + put job + put Status = putWord8 14 + put Ping = putWord8 9 + put (DropFunc func) = do + putWord8 15 + put func + put (RemoveJob fn jn) = do + putWord8 17 + put fn + put jn + put Shutdown = putWord8 20 + put (ConfigGet key) = do + putWord8 22 + put key + put (ConfigSet k v) = do + putWord8 23 + put k + putWord32be $ fromIntegral v + put Dump = putWord8 18 + put (Load jobs) = do + putWord8 19 + put jobs + put (RunJob job) = do + putWord8 25 + put job + +instance Validatable ClientCommand where + validate (SubmitJob job) = validate job + validate (DropFunc func) = validate func + validate (RemoveJob fn jn) = do + validate fn + validate jn + validate (ConfigGet key) = validate key + validate (ConfigSet k v) = do + validate k + validateNum "ConfigValue" 0 0xFFFFFFFF v + validate (Load jobs) = validate jobs + validate (RunJob job) = validate job + validate _ = Right () diff --git a/src/Periodic/Types/ClientType.hs b/src/Periodic/Types/ClientType.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/ClientType.hs @@ -0,0 +1,21 @@ +{-# LANGUAGE OverloadedStrings #-} +module Periodic.Types.ClientType + ( ClientType (..) + ) where + +import Data.Binary + +data ClientType = TypeClient + | TypeWorker + deriving (Eq, Show) + +instance Binary ClientType where + get = do + tp <- getWord8 + case tp of + 1 -> pure TypeClient + 2 -> pure TypeWorker + _ -> error $ "Error ClientType " ++ show tp + + put TypeClient = putWord8 1 + put TypeWorker = putWord8 2 diff --git a/src/Periodic/Types/Error.hs b/src/Periodic/Types/Error.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/Error.hs @@ -0,0 +1,17 @@ +module Periodic.Types.Error + ( Error (..) + ) where + +import Control.Exception (Exception) + +data Error = MagicNotMatch + | PacketDecodeError String + | TransportClosed + | TransportTimeout + | DataTooLarge + | InValidError String + | CRCNotMatch + | EmptyError + deriving (Show, Eq, Ord) + +instance Exception Error diff --git a/src/Periodic/Types/Internal.hs b/src/Periodic/Types/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/Internal.hs @@ -0,0 +1,123 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE UndecidableInstances #-} + +module Periodic.Types.Internal + ( FromBS (..) + , ConfigKey (..) + , LockName (..) + , Validatable (..) + , validateLength + , validateNum + , Nid (..) + , Msgid (..) + , msgidLength + ) where + +import Data.Binary (Binary (..)) +import Data.Binary.Get (getByteString, getWord8) +import Data.Binary.Put (putByteString, putWord8) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B (length, pack, unpack) +import qualified Data.ByteString.Lazy as LB (ByteString, fromStrict) +import Data.Hashable +import Data.Int (Int32) +import Data.String (IsString (..)) +import Data.Text (Text) +import qualified Data.Text as T (unpack) +import Data.Text.Encoding (decodeUtf8With) +import Data.Text.Encoding.Error (ignore) +import qualified Data.Text.Lazy as LT (Text, fromStrict) +import GHC.Generics (Generic) + +class FromBS a where + fromBS :: ByteString -> a + +instance FromBS Text where + fromBS = decodeUtf8With ignore + +instance FromBS [Char] where + fromBS = T.unpack . fromBS + +instance FromBS LT.Text where + fromBS = LT.fromStrict . fromBS + +instance FromBS LB.ByteString where + fromBS = LB.fromStrict + +instance FromBS ByteString where + fromBS = id + +newtype ConfigKey = ConfigKey String + deriving (Show) + +instance Binary ConfigKey where + get = do + size <- getWord8 + dat <- getByteString $ fromIntegral size + return $ ConfigKey $ B.unpack dat + put (ConfigKey dat) = do + putWord8 . fromIntegral $ length dat + putByteString $ B.pack dat + +newtype LockName = LockName ByteString + deriving (Generic, Eq, Ord, Show) + +instance Hashable LockName + +instance Binary LockName where + get = do + size <- getWord8 + dat <- getByteString $ fromIntegral size + return $ LockName dat + put (LockName dat) = do + putWord8 . fromIntegral $ B.length dat + putByteString dat + +instance IsString LockName where + fromString = LockName . fromString + +class Validatable a where + validate :: a -> Either String () + +instance (Validatable a) => Validatable [a] where + validate [] = Right () + validate (x:xs) = do + validate x + validate xs + +instance Validatable ByteString where + validate bs = validateLength "Data" 0 maxBound $ B.length bs + +validateLength :: String -> Int32 -> Int32 -> Int -> Either String () +validateLength n min' max' l' + | l < min' = Left $ n ++ " is to short" + | l > max' = Left $ n ++ " is to long" + | otherwise = Right () + where l = fromIntegral l' + +validateNum :: (Ord a) => String -> a -> a -> a -> Either String () +validateNum n min' max' l + | l < min' = Left $ n ++ " is to small" + | l > max' = Left $ n ++ " is to big" + | otherwise = Right () + +instance Validatable LockName where + validate (LockName bs) = validateLength "LockName" 1 255 $ B.length bs + +instance Validatable ConfigKey where + validate (ConfigKey k) = validateLength "ConfigKey" 1 255 $ length k + +newtype Nid = Nid ByteString + deriving (Generic, Eq, Ord, Show) + +instance Hashable Nid + +newtype Msgid = Msgid ByteString + deriving (Generic, Eq, Ord, Show) + +instance Hashable Msgid + + +msgidLength :: Int +msgidLength = 4 diff --git a/src/Periodic/Types/Job.hs b/src/Periodic/Types/Job.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/Job.hs @@ -0,0 +1,259 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE RecordWildCards #-} + +module Periodic.Types.Job + ( FuncName (..) + , JobName (..) + , Workload (..) + , JobHandle + , Job + , initJob + , setWorkload + , setSchedAt + , setCount + , setTimeout + , getFuncName + , getName + , getWorkload + , getSchedAt + , getCount + , getTimeout + , getHandle + , unHandle + , jobHandle + ) where + +import Data.Byteable (Byteable (..)) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B (empty, length) +import Data.ByteString.Lazy (toStrict) +import Data.Hashable +import Data.Int (Int64) +import GHC.Generics (Generic) + +import Data.String (IsString (..)) +import Periodic.Types.Internal + +import Data.Binary +import Data.Binary.Get +import Data.Binary.Put + +newtype FuncName = FuncName {unFN :: ByteString} + deriving (Generic, Eq, Ord, Show) + +instance Hashable FuncName + +instance IsString FuncName where + fromString = FuncName . fromString + +instance FromBS FuncName where + fromBS = FuncName . fromBS + +instance Binary FuncName where + get = do + size <- getWord8 + dat <- getByteString $ fromIntegral size + return $ FuncName dat + put (FuncName dat) = do + putWord8 . fromIntegral $ B.length dat + putByteString dat + +instance Validatable FuncName where + validate (FuncName n) = validateLength "FuncName" 1 255 $ B.length n + +newtype JobName = JobName {unJN :: ByteString} + deriving (Generic, Eq, Ord, Show) + +instance Hashable JobName + +instance IsString JobName where + fromString = JobName . fromString + +instance FromBS JobName where + fromBS = JobName . fromBS + +instance Binary JobName where + get = do + size <- getWord8 + dat <- getByteString $ fromIntegral size + return $ JobName dat + put (JobName dat) = do + putWord8 . fromIntegral $ B.length dat + putByteString dat + +instance Validatable JobName where + validate (JobName n) = validateLength "JobName" 1 255 $ B.length n + +data JobHandle = JobHandle FuncName JobName + deriving (Generic, Eq, Ord, Show) + +instance Hashable JobHandle + +instance Binary JobHandle where + get = do + fn <- get + JobHandle fn <$> get + put (JobHandle fn jn) = do + put fn + put jn + +instance Validatable JobHandle where + validate (JobHandle fn jn) = do + validate fn + validate jn + +newtype Workload = Workload {unWL :: ByteString} + deriving (Generic, Eq, Ord, Show) + +instance Hashable Workload + +instance IsString Workload where + fromString = Workload . fromString + +instance FromBS Workload where + fromBS = Workload . fromBS + +instance Binary Workload where + get = do + size <- getWord32be + dat <- getByteString $ fromIntegral size + return $ Workload dat + put (Workload dat) = do + putWord32be . fromIntegral $ B.length dat + putByteString dat + +instance Validatable Workload where + validate (Workload n) = validateLength "Workload" 0 maxBound $ B.length n + +data Job = Job + { jFuncName :: FuncName + , jName :: JobName + , jWorkload :: Workload + , jSchedAt :: Int64 + , jCount :: Int + , jTimeout :: Int + } + deriving (Show) + +instance Byteable Job where + toBytes = toStrict . encode + +data JVer = V0 + | V1 + | V2 + | V3 + +toVer :: Int -> JVer +toVer 0 = V0 +toVer 1 = V1 +toVer 2 = V2 +toVer 3 = V3 +toVer _ = V0 + +fromVer :: JVer -> Int +fromVer V0 = 0 +fromVer V1 = 1 +fromVer V2 = 2 +fromVer V3 = 3 + +calcVer :: Job -> JVer +calcVer Job{jCount = count, jTimeout = to} + | count > 0 && to > 0 = V3 + | to > 0 = V2 + | count > 0 = V1 + | otherwise = V0 + + +instance Binary Job where + get = do + jFuncName <- get + jName <- get + jWorkload <- get + jSchedAt <- getInt64be + ver <- toVer . fromIntegral <$> getWord8 + (jCount, jTimeout) <- + case ver of + V0 -> pure (0, 0) + V1 -> do + v <- fromIntegral <$> getInt32be + pure (v, 0) + V2 -> do + v <- fromIntegral <$> getInt32be + pure (0, v) + V3 -> do + v0 <- fromIntegral <$> getInt32be + v1 <- fromIntegral <$> getInt32be + pure (v0, v1) + return Job {..} + put j@Job {..} = do + put jFuncName + put jName + put jWorkload + putInt64be jSchedAt + let ver = calcVer j + + putWord8 $ fromIntegral $ fromVer ver + + case ver of + V0 -> pure () + V1 -> putInt32be $ fromIntegral jCount + V2 -> putInt32be $ fromIntegral jTimeout + V3 -> do + putInt32be $ fromIntegral jCount + putInt32be $ fromIntegral jTimeout + +instance Validatable Job where + validate (Job fn jn w _ c t) = do + validate fn + validate jn + validate w + validateNum "JobCount" 0 maxBound c + validateNum "JobTimeout" 0 maxBound t + +initJob :: FuncName -> JobName -> Job +initJob jFuncName jName = Job + { jWorkload = Workload B.empty + , jSchedAt = 0 + , jCount = 0 + , jTimeout = 0 + , .. + } + +setSchedAt :: Int64 -> Job -> Job +setSchedAt schedAt job = job {jSchedAt = schedAt} + +setWorkload :: Workload -> Job -> Job +setWorkload w job = job {jWorkload = w} + +setCount :: Int -> Job -> Job +setCount c job = job {jCount = c} + +setTimeout :: Int -> Job -> Job +setTimeout t job = job {jTimeout = t} + +getFuncName :: Job -> FuncName +getFuncName = jFuncName + +getName :: Job -> JobName +getName = jName + +getSchedAt :: Job -> Int64 +getSchedAt = jSchedAt + +getWorkload :: Job -> Workload +getWorkload = jWorkload + +getCount :: Job -> Int +getCount = jCount + +getTimeout :: Job -> Int +getTimeout = jTimeout + +getHandle :: Job -> JobHandle +getHandle job = jobHandle (getFuncName job) (getName job) + +unHandle :: JobHandle -> (FuncName, JobName) +unHandle (JobHandle fn jn) = (fn, jn) + +jobHandle :: FuncName -> JobName -> JobHandle +jobHandle = JobHandle diff --git a/src/Periodic/Types/Packet.hs b/src/Periodic/Types/Packet.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/Packet.hs @@ -0,0 +1,167 @@ +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} + +module Periodic.Types.Packet + ( Magic (..) + , Packet + , getPacketData + , getPacketMagic + , packetREQ + , packetRES + , getResult + + , RegPacket + , regPacketREQ + , regPacketRES + , getClientType + ) where + +import Data.Binary (Binary (..), decode, decodeOrFail, + encode) +import Data.Binary.Get (getByteString, getWord32be) +import Data.Binary.Put (putByteString, putWord32be) +import Data.ByteString (ByteString) +import qualified Data.ByteString as B (drop, empty, length) +import Data.ByteString.Lazy (fromStrict, toStrict) +import Metro.Class (GetPacketId (..), RecvPacket (..), + SendPacket (..), SetPacketId (..)) +import Periodic.CRC32 (CRC32 (..), digest) +import Periodic.Types.Error (Error (..)) +import Periodic.Types.Internal (Msgid (..)) +import UnliftIO (throwIO) + +data Magic = REQ + | RES + deriving (Show, Eq) + +instance Binary Magic where + get = do + bs <- getByteString 4 + case bs of + "\x00REQ" -> pure REQ + "\x00RES" -> pure RES + _ -> fail $ "No such magic " ++ show bs + + put REQ = putByteString "\x00REQ" + put RES = putByteString "\x00RES" + +magicLength :: Int +magicLength = 4 + +discoverMagic :: Monad m => ByteString -> (Int -> m ByteString) -> m (Magic, ByteString) +discoverMagic "\0REQ" _ = return (REQ, "\0REQ") +discoverMagic "\0RES" _ = return (REQ, "\0RES") +discoverMagic prev recv = do + bs <- (prev <>) <$> recv 1 + if B.length bs > magicLength then discoverMagic (B.drop (B.length bs - magicLength) bs) recv + else discoverMagic bs recv + +newtype PacketLength = PacketLength Int + deriving (Show, Eq) + +instance Binary PacketLength where + get = PacketLength . fromIntegral <$> getWord32be + put (PacketLength l) = putWord32be $ fromIntegral l + +instance Binary CRC32 where + get = CRC32 <$> getWord32be + put (CRC32 l) = putWord32be l + +putBS bs = do + put $ PacketLength $ B.length bs + put $ digest bs + putByteString bs + +getHead = do + magic <- get + PacketLength _ <- get + crc <- get + return (magic, crc) + +data Packet a = Packet + { packetMagic :: Magic + , packetCRC :: CRC32 + , packetId :: Msgid + , packetData :: a + } + deriving (Show, Eq) + +instance Binary a => Binary (Packet a) where + get = do + (magic, crc) <- getHead + pid <- getByteString 4 + Packet magic crc (Msgid pid) <$> get + put (Packet magic _ (Msgid pid) body) = do + put magic + putBS $ pid <> toStrict (encode body) + +commonRecvPacket f recv = do + (_, magicbs) <- discoverMagic B.empty recv + hbs <- recv 4 + crcbs <- recv 4 + case decode (fromStrict hbs) of + PacketLength len -> do + bs <- recv len + case decodeOrFail (fromStrict $ magicbs <> hbs <> crcbs <> bs) of + Left (_, _, e1) -> throwIO $ PacketDecodeError $ "Packet: " <> e1 + Right (_, _, pkt) -> + if digest bs == f pkt then return pkt + else throwIO CRCNotMatch + +instance Binary a => RecvPacket (Packet a) where + recvPacket = commonRecvPacket packetCRC + +instance Binary a => SendPacket (Packet a) where + + +instance GetPacketId Msgid (Packet a) where + getPacketId = packetId + +instance SetPacketId Msgid (Packet a) where + setPacketId k pkt = pkt { packetId = k } + + +getPacketData :: Packet a -> a +getPacketData = packetData + +getPacketMagic :: Packet a -> Magic +getPacketMagic = packetMagic + +packetREQ :: a -> Packet a +packetREQ = Packet REQ (CRC32 0) (Msgid "0000") + +packetRES :: a -> Packet a +packetRES = Packet RES (CRC32 0) (Msgid "0000") + +getResult :: a -> (b -> a) -> Maybe (Packet b) -> a +getResult defv _ Nothing = defv +getResult _ f (Just rpkt) = f (getPacketData rpkt) + +data RegPacket a = RegPacket + { regMagic :: Magic + , regCRC :: CRC32 + , regType :: a + } + deriving (Show, Eq) + +instance Binary a => Binary (RegPacket a) where + get = do + (magic, crc) <- getHead + RegPacket magic crc <$> get + put (RegPacket magic _ body) = do + put magic + putBS $ toStrict (encode body) + +instance Binary a => RecvPacket (RegPacket a) where + recvPacket = commonRecvPacket regCRC + +instance Binary a => SendPacket (RegPacket a) where + +regPacketREQ :: a -> RegPacket a +regPacketREQ = RegPacket REQ (CRC32 0) + +regPacketRES :: a -> RegPacket a +regPacketRES = RegPacket RES (CRC32 0) + +getClientType :: RegPacket a -> a +getClientType = regType diff --git a/src/Periodic/Types/ServerCommand.hs b/src/Periodic/Types/ServerCommand.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/ServerCommand.hs @@ -0,0 +1,84 @@ +{-# LANGUAGE OverloadedStrings #-} +module Periodic.Types.ServerCommand + ( ServerCommand (..) + , isSuccess + , isPong + ) where + +import Periodic.Types.Internal +import Periodic.Types.Job (Job) + +import Data.Binary +import Data.Binary.Get (getRemainingLazyByteString, + getWord32be) +import Data.Binary.Put (putByteString, putWord32be) +import Data.ByteString (ByteString) +import Data.ByteString.Lazy (toStrict) + +data ServerCommand = Noop + | JobAssign Job + | NoJob + | Pong + | Unknown + | Success + | Config Int + | Acquired Bool + | NoWorker + | Data ByteString + deriving (Show) + +instance Binary ServerCommand where + get = do + tp <- getWord8 + case tp of + 0 -> pure Noop + 5 -> JobAssign <$> get + 6 -> pure NoJob + 10 -> pure Pong + 12 -> pure Unknown + 16 -> pure Success + 24 -> do + val <- getWord32be + pure . Config $ fromIntegral val + 26 -> do + v <- getWord8 + pure $ Acquired $ v == 1 + 29 -> pure NoWorker + 30 -> Data . toStrict <$> getRemainingLazyByteString + _ -> error $ "Error ServerCommand" ++ show tp + + put Noop = putWord8 0 + put (JobAssign job) = do + putWord8 5 + put job + put NoJob = putWord8 6 + put Pong = putWord8 10 + put Unknown = putWord8 12 + put Success = putWord8 16 + put (Config v) = do + putWord8 24 + putWord32be $ fromIntegral v + put (Acquired True) = do + putWord8 26 + putWord8 1 + put (Acquired False) = do + putWord8 26 + putWord8 0 + put NoWorker = putWord8 29 + put (Data bs) = do + putWord8 30 + putByteString bs + +instance Validatable ServerCommand where + validate (JobAssign job) = validate job + validate (Config v) = validateNum "ConfigValue" 0 0xFFFFFFFF v + validate _ = Right () + + +isSuccess :: ServerCommand -> Bool +isSuccess Success = True +isSuccess _ = False + +isPong :: ServerCommand -> Bool +isPong Pong = True +isPong _ = False diff --git a/src/Periodic/Types/WorkerCommand.hs b/src/Periodic/Types/WorkerCommand.hs new file mode 100644 --- /dev/null +++ b/src/Periodic/Types/WorkerCommand.hs @@ -0,0 +1,110 @@ +{-# LANGUAGE OverloadedStrings #-} +module Periodic.Types.WorkerCommand + ( WorkerCommand (..) + ) where + +import Data.Int (Int64) +import Periodic.Types.Internal +import Periodic.Types.Job (FuncName, JobHandle) + +import Data.Binary +import Data.Binary.Get +import Data.Binary.Put +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as B (length) +import Data.ByteString.Lazy (toStrict) + +data WorkerCommand = GrabJob + | SchedLater JobHandle Int64 Int + | WorkDone JobHandle ByteString + | WorkFail JobHandle + | Sleep + | Ping + | CanDo FuncName + | CantDo FuncName + | Broadcast FuncName + | Acquire LockName Int JobHandle + | Release LockName JobHandle + deriving (Show) + +instance Binary WorkerCommand where + get = do + tp <- getWord8 + case tp of + 1 -> pure GrabJob + 2 -> do + jh <- get + later <- getInt64be + step <- fromIntegral <$> getInt16be + pure (SchedLater jh later step) + 3 -> do + jh <- get + WorkDone jh . toStrict <$> getRemainingLazyByteString + 4 -> WorkFail <$> get + 11 -> pure Sleep + 9 -> pure Ping + 7 -> CanDo <$> get + 8 -> CantDo <$> get + 21 -> Broadcast <$> get + 27 -> do + n <- get + c <- fromIntegral <$> getInt16be + Acquire n c <$> get + 28 -> do + n <- get + Release n <$> get + _ -> error $ "Error WorkerCommand " ++ show tp + + put GrabJob = putWord8 1 + put (SchedLater jh later step) = do + putWord8 2 + put jh + putInt64be later + putInt16be $ fromIntegral step + put (WorkDone jh w) = do + putWord8 3 + put jh + putByteString w + put (WorkFail jh) = do + putWord8 4 + put jh + put Sleep = putWord8 11 + put Ping = putWord8 9 + put (CanDo fn) = do + putWord8 7 + put fn + put (CantDo fn) = do + putWord8 8 + put fn + put (Broadcast fn) = do + putWord8 21 + put fn + put (Acquire n c jh) = do + putWord8 27 + put n + putInt16be $ fromIntegral c + put jh + put (Release n jh) = do + putWord8 28 + put n + put jh + +instance Validatable WorkerCommand where + validate (SchedLater jh _ step) = do + validate jh + validateNum "Step" 0 0xFFFF step + validate (WorkDone jh w) = do + validate jh + validateLength "WorkData" 0 maxBound $ B.length w + validate (WorkFail jh) = validate jh + validate (CanDo fn) = validate fn + validate (CantDo fn) = validate fn + validate (Broadcast fn) = validate fn + validate (Acquire n c jh) = do + validate n + validateNum "LockCount" 1 0xFFFF c + validate jh + validate (Release n jh) = do + validate n + validate jh + validate _ = Right ()