diff --git a/CurryDB.cabal b/CurryDB.cabal new file mode 100644 --- /dev/null +++ b/CurryDB.cabal @@ -0,0 +1,115 @@ +name: CurryDB +version: 0.1.0.0 +synopsis: CurryDB: In-memory Key/Value Database +description: CurryDB: Simple, Persistent, Polymorphic, Transactional, In-memory Key/Value Database +license: BSD3 +license-file: LICENSE +author: Hideyuki Tanaka +maintainer: Hideyuki Tanaka +copyright: (c) 2012, Hideyuki Tanaka +category: Database +build-type: Simple +cabal-version: >=1.8 + +source-repository head + type: git + location: https://github.com/tanakh/CurryDB.git + +library + exposed-modules: Database.Curry + Database.Curry.Binary + Database.Curry.Commands + Database.Curry.Storage + Database.Curry.Types + Database.Memcached + Database.Memcached.Commands + Database.Memcached.Server + Database.Redis + Database.Redis.Builder + Database.Redis.Commands + Database.Redis.Parser + Database.Redis.Server + Database.Redis.Types + + build-depends: base >=4.5 && < 5 + , bytestring >=0.9 + , mtl >=2.1 + , transformers >=0.3 + , transformers-base >=0.4 + , unordered-containers >=0.2.2 + , containers >=0.4 + , data-lens >= 2.10 + , data-lens-fd >= 2.0 + , data-lens-template >= 2.1 + , conduit >=0.5 + , time + , stm >=2.4 + , monad-control >=0.3 + , text >= 0.10 + , attoparsec >=0.10 + , attoparsec-conduit + , blaze-builder >=0.3 + , blaze-textual + , network >=2.4 + , network-conduit >=0.6.1.1 + , data-default >=0.5 + , monad-logger >=0.2 + , fast-logger >=0.3 + , template-haskell + , system-filepath >=0.4.7 + , system-fileio >=0.3.10 + , async >=2.0 + , binary >=0.6.1 + , lifted-base >=0.2 + + ghc-prof-options: -auto-all + +executable curry-memcached + hs-source-dirs: memcached + main-is: main.hs + build-depends: base + , network-conduit + , CurryDB + +executable curry-redis + hs-source-dirs: redis + main-is: main.hs + ghc-options: -O3 + build-depends: base + , network-conduit + , system-filepath + , optparse-applicative + , ekg >=0.3.1 + , CurryDB + +test-suite doctests + type: exitcode-stdio-1.0 + hs-source-dirs: tests + main-is: doctests.hs + build-depends: base + , filepath + , directory + , doctest >=0.9.1 + +test-suite hspec + type: exitcode-stdio-1.0 + hs-source-dirs: tests + main-is: hspec.hs + build-depends: base + , hspec >=1.3 + , mtl + , stm + , conduit + , CurryDB + +benchmark curry-bench + type: exitcode-stdio-1.0 + hs-source-dirs: bench + main-is: bench.hs + ghc-options: -rtsopts -O2 + ghc-prof-options: -auto-all + build-depends: base + , bytestring + , mtl >=2.1 + , mersenne-random-pure64 + , CurryDB diff --git a/Database/Curry.hs b/Database/Curry.hs new file mode 100644 --- /dev/null +++ b/Database/Curry.hs @@ -0,0 +1,49 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.Curry ( + -- run DBM Monad + runDBMT, + + module Database.Curry.Commands, + module Database.Curry.Types, + ) where + +import Control.Applicative +import Control.Concurrent.Async +import Control.Concurrent.STM +import qualified Control.Exception.Lifted as EL +import Control.Monad.State.Strict +import Control.Monad.Trans.Control +import Control.Monad.Trans.Identity +import Data.Binary +import Data.Default +import qualified Data.HashMap.Strict as HMS +import System.IO +import System.Log.FastLogger + +import Database.Curry.Binary () +import Database.Curry.Commands +import Database.Curry.Storage +import Database.Curry.Types + +initDBMState :: Config -> STM () -> IO (DBMState v) +initDBMState conf upd = + DBMState + <$> newTVarIO HMS.empty + <*> pure upd + <*> mkLogger True stdout + <*> pure conf + +-- | Run 'DBMT' monad. +runDBMT :: (MonadIO m, MonadBaseControl IO m, Binary v) + => Config -> DBMT v m a -> m a +runDBMT conf m = do + (upd, reset, saveReq) <- liftIO $ createNotifyer $ configSaveStrategy conf + st <- liftIO $ initDBMState conf upd + (`evalStateT` st) $ runIdentityT $ unDBMT $ do + loadFromFile + control $ \run -> do + _ <- async $ run $ saveThread saveReq reset + run (m `EL.finally` saveToFile) diff --git a/Database/Curry/Binary.hs b/Database/Curry/Binary.hs new file mode 100644 --- /dev/null +++ b/Database/Curry/Binary.hs @@ -0,0 +1,12 @@ +{-# LANGUAGE FlexibleInstances #-} + +module Database.Curry.Binary where + +import Control.Applicative +import Data.Binary +import qualified Data.ByteString as S +import qualified Data.HashMap.Strict as HMS + +instance Binary v => Binary (HMS.HashMap S.ByteString v) where + put = put . HMS.toList + get = HMS.fromList <$> get diff --git a/Database/Curry/Commands.hs b/Database/Curry/Commands.hs new file mode 100644 --- /dev/null +++ b/Database/Curry/Commands.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE BangPatterns #-} + +module Database.Curry.Commands ( + -- Commands + insert, + insertWith, + delete, + Database.Curry.Commands.lookup, + lookupDefault, + keys, + + -- Transaction + transaction, + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Monad.State.Strict +import Control.Monad.Trans.Identity +import qualified Data.ByteString as S +import Data.Conduit +import Data.Default +import qualified Data.HashMap.Strict as HMS +import Data.Lens +import Data.Maybe + +import Database.Curry.Types + +insert :: S.ByteString -> v -> DBMS v () +insert !key !val = do + table <- access dbmTable + liftSTM $ modifyTVar' table $ HMS.insert key val + update +{-# INLINE insert #-} + +insertWith :: (v -> v -> v) -> S.ByteString -> v -> DBMS v () +insertWith !f !key !val = do + htvar <- access dbmTable + liftSTM $ modifyTVar' htvar $ HMS.insertWith f key val + update +{-# INLINE insertWith #-} + +delete :: S.ByteString -> DBMS v () +delete !key = do + htvar <- access dbmTable + liftSTM $ modifyTVar' htvar $ HMS.delete key + update +{-# INLINE delete #-} + +lookup :: S.ByteString -> DBMS v (Maybe v) +lookup !key = do + htvar <- access dbmTable + liftSTM $ HMS.lookup key <$> readTVar htvar +{-# INLINE lookup #-} + +lookupDefault :: Default v => S.ByteString -> DBMS v v +lookupDefault !key = do + htvar <- access dbmTable + liftSTM $ fromMaybe def . HMS.lookup key <$> readTVar htvar +{-# INLINE lookupDefault #-} + +keys :: Monad m => DBMS v (Source (DBMT v m) S.ByteString) +keys = do + htvar <- access dbmTable + ht <- liftSTM $ readTVar htvar + return $ mapM_ yield $ HMS.keys ht +{-# INLINE keys #-} + +update ::DBMS v () +update = liftSTM =<< access dbmUpdate +{-# INLINE update #-} + +transaction :: MonadIO m => DBMS v a -> DBMT v m a +transaction = + lift . mapStateT (liftIO . atomically) . runIdentityT . unDBMT +{-# INLINE transaction #-} diff --git a/Database/Curry/Storage.hs b/Database/Curry/Storage.hs new file mode 100644 --- /dev/null +++ b/Database/Curry/Storage.hs @@ -0,0 +1,113 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.Curry.Storage ( + saveThread, + createNotifyer, + + saveToFile, + loadFromFile, + ) where + +import Control.Applicative +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.Async +import Control.Concurrent.STM +import qualified Control.Exception as E +import Control.Monad.Logger +import Control.Monad.State.Strict +import Data.Binary +import qualified Data.ByteString.Lazy as L +import Data.Lens +import Data.Monoid +import qualified Data.Text as T +import Data.Time +import qualified Filesystem as FS +import qualified Filesystem.Path.CurrentOS as FP +import System.IO + +import Database.Curry.Binary () +import Database.Curry.Types + +saveThread :: (Functor m, MonadIO m, Binary v) + => TVar Bool -> STM () -> DBMT v m () +saveThread saveReq reset = forever $ do + liftIO $ atomically $ do + req <- readTVar saveReq + when (not req) retry + writeTVar saveReq False + reset + saveToFile + +createNotifyer :: [SaveStrategy] -> IO (STM (), STM (), TVar Bool) +createNotifyer strats = do + timer <- createTimer + saveReq <- newTVarIO False + + let notify (SaveByFrequency {..})= do + upd <- newTVarIO 0 + forkIO $ forever $ do + start <- readTVarIO timer + atomically $ do + cur <- readTVar timer + when (cur `diffUTCTime` start < fromIntegral freqSecond) retry + num <- readTVar upd + writeTVar upd 0 + when (num >= freqUpdates) $ writeTVar saveReq True + return upd + + upds <- mapM notify strats + let upd = mapM_ (\tv -> modifyTVar' tv (+1)) upds + reset = mapM_ (\tv -> writeTVar tv 0) upds + + return (upd, reset, saveReq) + +createTimer :: IO (TVar UTCTime) +createTimer = do + curV <- newTVarIO =<< getCurrentTime + _ <- async $ forever $ do + threadDelay $ 10 ^ (6 :: Int) + time <- getCurrentTime + atomically $ writeTVar curV time + return curV + +saveToFile :: (MonadIO m, Binary v) => DBMT v m () +saveToFile = do + Config {..} <- access dbmConfig + case configPath of + Nothing -> return () + Just fname -> do + $logInfo "save to file..." + tbl <- liftIO . readTVarIO =<< access dbmTable + err <- liftIO $ E.try $ atomicWriteFile fname tbl + case err of + Right _ -> + return () + Left ioerr -> + $logError $ "save error: " <> (T.pack $ show $ (ioerr :: IOError)) + +loadFromFile :: (MonadIO m, Binary v) => DBMT v m () +loadFromFile = do + Config {..} <- access dbmConfig + case configPath of + Nothing -> return () + Just path -> do + $logInfo "load from file..." + etbl <- liftIO $ E.try $ do + v <- decode . L.fromChunks . (\x -> [x]) <$> FS.readFile path + E.evaluate v -- force and raise exception when data is corrupted. + case etbl of + Right tbl -> do + tv <- access dbmTable + liftIO $ atomically $ writeTVar tv tbl + Left err -> do + $logInfo $ "fail to load " <> ": " <> (T.pack $ show (err :: E.SomeException)) + +atomicWriteFile :: Binary b => FP.FilePath -> b -> IO () +atomicWriteFile path b = do + let tmpPath = path FP.<.> "tmp" + FS.withFile tmpPath WriteMode $ \h -> + L.hPut h $ encode b + -- rename operation is atomic + FS.rename tmpPath path diff --git a/Database/Curry/Types.hs b/Database/Curry/Types.hs new file mode 100644 --- /dev/null +++ b/Database/Curry/Types.hs @@ -0,0 +1,119 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +module Database.Curry.Types ( + -- DBM monad + DBMT, unDBMT, + DBMS, + liftSTM, + + -- State and lenses + DBMState(..), + dbmTable, + dbmUpdate, + dbmLogger, + dbmConfig, + + -- Configuration + Config(..), def, + SaveStrategy(..), + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.Base +import Control.Monad.Logger +import Control.Monad.State.Strict +import Control.Monad.Trans.Control +import Control.Monad.Trans.Identity +import qualified Data.ByteString as S +import Data.Conduit +import Data.Default +import qualified Data.HashMap.Strict as HMS +import Data.Lens.Template +import qualified Filesystem.Path.CurrentOS as FP +import Language.Haskell.TH.Syntax (Loc (..)) +import System.Log.FastLogger + +import Database.Curry.Binary () + +type DBMT v m = DBMT_ (StateT (DBMState v) m) +type DBMS v = DBMT v STM + +newtype DBMT_ m a = + DBMT_ { unDBMT :: IdentityT m a } + deriving + ( Functor, Applicative, Monad + , Alternative + , MonadIO, MonadTrans, MonadBase b + , MonadThrow, MonadResource + ) + +deriving instance MonadState (DBMState v) m => MonadState (DBMState v) (DBMT_ m) + +instance MonadTransControl DBMT_ where + newtype StT DBMT_ a = + StDBMT { unStDBM :: a } + liftWith f = + DBMT_ $ lift $ f $ liftM StDBMT . runIdentityT . unDBMT + restoreT = + DBMT_ . lift . liftM unStDBM + +instance MonadBaseControl b m => MonadBaseControl b (DBMT_ m) where + newtype StM (DBMT_ m) a = StMT { unStMT :: ComposeSt DBMT_ m a } + liftBaseWith = defaultLiftBaseWith StMT + restoreM = defaultRestoreM unStMT + +instance MonadIO m => MonadLogger (DBMT v m) where + monadLoggerLog loc level msg = do + logger <- gets _dbmLogger + date <- liftIO $ loggerDate logger + let (row, col) = loc_start loc + liftIO $ loggerPutStr logger + [ toLogStr date, LB " " + , LB "[", LS (show level), LB "] " + , toLogStr (loc_module loc), LB ":", LS (show row), LB ":", LS (show col), LB ": " + , toLogStr msg + , LB "\n" + ] + +data DBMState v + = DBMState + { _dbmTable :: TVar (HMS.HashMap S.ByteString v) + , _dbmUpdate :: STM () + , _dbmLogger :: Logger + , _dbmConfig :: Config + } + +data Config + = Config + { configPath :: Maybe FP.FilePath + , configSaveStrategy :: [SaveStrategy] + , configVerbosity :: LogLevel + } + +data SaveStrategy + = SaveByFrequency + { freqSecond :: Int + , freqUpdates :: Int + } + +makeLens ''DBMState + +instance Default Config where + def = Config + { configPath = Nothing + , configSaveStrategy = [] + , configVerbosity = LevelInfo + } + +liftSTM :: STM a -> DBMS v a +liftSTM = lift . lift +{-# INLINE liftSTM #-} diff --git a/Database/Memcached.hs b/Database/Memcached.hs new file mode 100644 --- /dev/null +++ b/Database/Memcached.hs @@ -0,0 +1,5 @@ +module Database.Memcached ( + module Database.Memcached.Server + ) where + +import Database.Memcached.Server diff --git a/Database/Memcached/Commands.hs b/Database/Memcached/Commands.hs new file mode 100644 --- /dev/null +++ b/Database/Memcached/Commands.hs @@ -0,0 +1,195 @@ +{-# LANGUAGE OverloadedStrings, TupleSections #-} + +module Database.Memcached.Commands ( + MemcachedT, + + Command(..), + Response(..), + Error(..), + + parseCommand, + fromResponse, + + execCommand, + ) where + +import Blaze.ByteString.Builder +import Blaze.ByteString.Builder.Char8 +import Control.Applicative +import Control.Monad +import Data.Attoparsec.ByteString.Char8 as A +import qualified Data.ByteString.Char8 as S +import qualified Data.Char as C +import Data.Monoid +import qualified Data.Text as T +import Data.Word +import Prelude hiding (lookup) +import Data.Maybe + +import Database.Curry as Curry + +-- protocol: + +type MemcachedT m = DBMT S.ByteString m + +data Command + -- Storage + = Set S.ByteString Word32 Int S.ByteString + | Add S.ByteString Word32 Int S.ByteString + | Replace S.ByteString Word32 Int S.ByteString + | Append S.ByteString Word32 Int S.ByteString + | Prepend S.ByteString Word32 Int S.ByteString + | Cas S.ByteString Word32 Int Word64 S.ByteString + + -- Retrieval + | Get [S.ByteString] + | Gets [S.ByteString] + + -- Deletion + | Delete S.ByteString + + -- Other + | Incr S.ByteString Word64 + | Decr S.ByteString Word64 + | Touch S.ByteString Int + -- | Slabs + | Stats + | FlushAll + | Version + | Verbosity Int + | Quit + deriving (Show) + +data Response + = Stored + | NotStored + | Exists + | NotFound + | Values [(S.ByteString, S.ByteString)] + | Deleted + | Value S.ByteString + +data Error + = Error + | ClientError T.Text + | ServerError T.Text + +-- memcached text protocol parser + +parseCommand :: A.Parser Command +parseCommand = + Set <$> "set " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " <*> bytes + <|> Add <$> "add " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " <*> bytes + <|> Replace <$> "replace " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " <*> bytes + <|> Append <$> "append " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " <*> bytes + <|> Prepend <$> "prepend " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " <*> bytes + + <|> do cas <- Cas <$> "cas " .*> key <*. " " <*> decimal <*. " " <*> decimal <*. " " + len <- decimal <*. " " + cas <$> decimal <*. crlf <*> A.take len <*. crlf + + <|> Get <$> "get" .*> some (" " .*> key) <*. crlf + <|> Gets <$> "gets" .*> some (" " .*> key) <*. crlf + + <|> Delete <$> "delete " .*> key <*. crlf + + <|> Incr <$> "incr " .*> key <*> " " .*> decimal <*. crlf + <|> Decr <$> "decr " .*> key <*> " " .*> decimal <*. crlf + <|> Touch <$> "touch " .*> key <*> " " .*> decimal <*. crlf + + <|> pure Stats <*. "stats" <*. crlf + <|> pure FlushAll <*. "flush_all" <*. crlf + <|> pure Version <*. "version" <*. crlf + <|> pure Verbosity <*. "verbosity" <*> decimal <*. crlf + <|> pure Quit <*. "quit" <*. crlf + + where + key = A.takeWhile1 $ \c -> not (C.isSpace c || C.isControl c) + bytes = decimal <*. crlf >>= \len -> A.take len <*. crlf + +fromResponse :: Response -> Builder +fromResponse resp = case resp of + Stored -> + fromByteString "STORED\r\n" + NotStored -> + fromByteString "NOT_STORED\r\n" + Exists -> + fromByteString "EXISTS\r\n" + NotFound -> + fromByteString "NOT_FOUND\r\n" + + Values kvs -> mconcat + [ fromByteString "VALUE " <> fromByteString key <> fromByteString " 0 " <> + fromString (show $ S.length val) <> fromByteString "\r\n" <> + fromByteString val <> fromByteString "\r\n" + | (key, val) <- kvs + ] <> fromByteString "END\r\n" + + Deleted -> + fromByteString "DELETED\r\n" + + Value bs -> + fromByteString bs + +crlf :: S.ByteString +crlf = "\r\n" + +----- + +execCommand :: Command -> MemcachedT IO Response +execCommand req = transaction $ case req of + Set key _flags _exptime val -> do + Curry.insert key val + return Stored + + Add key _flags _exptime val -> do + Curry.lookup key >>= \mb -> case mb of + Nothing -> do + Curry.insert key val + return Stored + Just _ -> + return NotStored + + Replace key _flags _exptime val -> do + Curry.lookup key >>= \mb -> case mb of + Nothing -> + return NotStored + Just _ -> do + Curry.insert key val + return Stored + + Append key _flags _exptime val -> do + Curry.insertWith (\new old -> old <> new) key val + return Stored + + Prepend key _flags _exptime val -> do + Curry.insertWith (\new old -> new <> old) key val + return Stored + + Get ks -> + Values . catMaybes . zipWith (\k v -> (k, ) <$> v) ks <$> mapM Curry.lookup ks + + Delete key -> do + Curry.lookup key >>= \mb -> case mb of + Nothing -> + return NotFound + Just _ -> do + Curry.delete key + return Deleted + + Incr key val -> incr key (+ val) + Decr key val -> incr key (subtract val) + + where + incr key f = do + Curry.lookup key >>= \mb -> case mb of + Nothing -> + return NotFound + Just bs -> + case S.readInt bs of + Just (cur, "") -> do + let next = S.pack $ show (f $ fromIntegral cur) + Curry.insert key next + return $ Value next + _ -> + return NotFound diff --git a/Database/Memcached/Server.hs b/Database/Memcached/Server.hs new file mode 100644 --- /dev/null +++ b/Database/Memcached/Server.hs @@ -0,0 +1,28 @@ +module Database.Memcached.Server ( + runServer, + + ServerSettings, + serverSettings, + ) where + +import Blaze.ByteString.Builder +import Control.Monad.Trans +import Data.Conduit +import Data.Conduit.Attoparsec +import Data.Conduit.Network +import Network (withSocketsDo) + +import Database.Curry +import Database.Memcached.Commands + +runServer :: ServerSettings (MemcachedT IO) -> IO () +runServer ss = withSocketsDo $ runDBMT def $ runTCPServer ss server + +server :: Application (MemcachedT IO) +server ss = + appSource ss $$ conduitParser parseCommand =$ awaitForever p =$ appSink ss + where + p (_range, req) = do + liftIO $ putStrLn $ "server: " ++ show req + resp <- lift $ execCommand req + yield . toByteString . fromResponse $ resp diff --git a/Database/Redis.hs b/Database/Redis.hs new file mode 100644 --- /dev/null +++ b/Database/Redis.hs @@ -0,0 +1,5 @@ +module Database.Redis ( + module Database.Redis.Server + ) where + +import Database.Redis.Server diff --git a/Database/Redis/Builder.hs b/Database/Redis/Builder.hs new file mode 100644 --- /dev/null +++ b/Database/Redis/Builder.hs @@ -0,0 +1,41 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Database.Redis.Builder ( + fromReply, + ) where + +import Blaze.ByteString.Builder +import Blaze.Text +import qualified Data.ByteString as S +import Data.Monoid + +import Database.Redis.Types + +fromReply :: Reply -> Builder +fromReply rep = case rep of + StatusReply stat -> + fromByteString "+" <> fromByteString stat <> crlf + ErrorReply err -> + fromByteString "-" <> fromByteString err <> crlf + IntReply n -> + fromByteString ":" <> integral n <> crlf + BulkReply mb -> + fromBulkReply mb + MultiBulkReply Nothing -> + fromByteString "*-1" <> crlf + MultiBulkReply (Just bss) -> + fromByteString "*" <> integral (length bss) <> crlf <> + mconcat (map fromBulkReply bss) +{-# INLINE fromReply #-} + +fromBulkReply :: Maybe S.ByteString -> Builder +fromBulkReply Nothing = + fromByteString "$-1" <> crlf +fromBulkReply (Just bs) = + fromByteString "$" <> integral (S.length bs) <> crlf <> + fromByteString bs <> crlf +{-# INLINE fromBulkReply #-} + +crlf :: Builder +crlf = fromByteString "\r\n" +{-# INLINE crlf #-} diff --git a/Database/Redis/Commands.hs b/Database/Redis/Commands.hs new file mode 100644 --- /dev/null +++ b/Database/Redis/Commands.hs @@ -0,0 +1,166 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ViewPatterns #-} + +module Database.Redis.Commands ( + process, + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Monad.Trans (MonadIO) +import qualified Data.ByteString.Char8 as S +import Data.Foldable +import qualified Data.HashSet as HS +import Data.Maybe +import qualified Data.Sequence as Seq + +import Database.Curry as Curry +import Database.Redis.Types + +type RedisCommand = RedisT STM Reply + +ping :: RedisCommand +ping = return $ StatusReply "PONG" + +set :: S.ByteString -> S.ByteString -> RedisCommand +set key val = do + Curry.insert key (VString val) + return replyOK + +mset :: [S.ByteString] -> RedisCommand +mset ss = go ss >> return replyOK where + go (key: val: rest) = do + Curry.insert key (VString val) + go rest + go _ = return () + +get :: S.ByteString -> RedisCommand +get key = do + x <- Curry.lookup key + return $ case x of + Just (VString val) -> BulkReply $ Just val + Nothing -> BulkReply Nothing + _ -> typeErr + +incr, decr :: S.ByteString -> RedisCommand +incr = modInt succ +decr = modInt pred + +modInt :: (Int -> Int) -> S.ByteString -> RedisCommand +modInt f key = do + x <- Curry.lookup key + case x of + Just (VString (toInt -> Just (f -> val))) -> do + Curry.insert key $ toVString val + return $ IntReply val + Nothing -> do + Curry.insert key $ toVString $ f 0 + return $ IntReply $ f 0 + Just (VString _) -> do + return notIntErr + _ -> + return typeErr + +lpush :: S.ByteString -> [S.ByteString] -> RedisCommand +lpush key vals = do + x <- Curry.lookup key + case x of + Just (VList ls) -> do + Curry.insert key $ VList $ foldl' (\ys y -> y Seq.<| ys) ls vals + return $ IntReply $ Seq.length ls + length vals + Nothing -> do + Curry.insert key $ VList $ foldl' (\ys y -> y Seq.<| ys) Seq.empty vals + return $ IntReply $ length vals + _ -> + return typeErr + +lpop :: S.ByteString -> RedisCommand +lpop key = do + x <- fromMaybe (VList Seq.empty) <$> Curry.lookup key + case x of + VList ls -> do + case Seq.viewl ls of + val Seq.:< rest -> do + Curry.insert key $ VList rest + return $ BulkReply $ Just val + _ -> + return $ BulkReply Nothing + _ -> + return typeErr + +lrange :: S.ByteString -> S.ByteString -> S.ByteString -> RedisCommand +lrange key sstart sstop = do + x <- fromMaybe (VList Seq.empty) <$> Curry.lookup key + case (x, toInt sstart, toInt sstop) of + (VList ss, Just start, Just stop) -> do + return + $ MultiBulkReply $ Just $ map Just $ toList + $ Seq.take (stop - start) $ Seq.drop start ss + _ -> + return typeErr + +sadd :: S.ByteString -> [S.ByteString] -> RedisCommand +sadd key vals = do + x <- fromMaybe (VSet HS.empty) <$> Curry.lookup key + case x of + VSet ss -> do + let nss = foldl' (flip HS.insert) ss vals + Curry.insert key $ VSet nss + return $ IntReply $ HS.size nss - HS.size ss + _ -> + return typeErr + +spop :: S.ByteString -> RedisCommand +spop key = do + x <- fromMaybe (VSet HS.empty) <$> Curry.lookup key + case x of + VSet ss -> do + case HS.toList ss of + (arb:_) -> do + Curry.insert key $ VSet $ HS.delete arb ss + return $ BulkReply $ Just arb + _ -> do + return $ BulkReply Nothing + _ -> + return typeErr + +----- + +toInt :: S.ByteString -> Maybe Int +toInt bs = + case S.readInt bs of + Just (n, "") -> Just n + _ -> Nothing + +toVString :: Int -> Value +toVString = VString . S.pack . show + +replyOK, typeErr, notIntErr :: Reply +replyOK = StatusReply "OK" +typeErr = ErrorReply "ERR Operation against a key holding the wrong kind of value" +notIntErr = ErrorReply "ERR value is not an integer or out of range" + +----- + +process :: (Functor m, Applicative m, MonadIO m) + => (a, Request) -> RedisT m Reply +process (_pos, req)= transaction $ case req of + Request ["PING"] -> ping + + Request ["SET", key, val] -> set key val + Request ("MSET": args) -> mset args + Request ["GET", key] -> get key + Request ["INCR", key] -> incr key + Request ["DECR", key] -> decr key + + Request ("LPUSH": key: vals) -> lpush key vals + Request ["LPOP", key] -> lpop key + Request ["LRANGE", key, start, stop] -> lrange key start stop + + Request ("SADD": key: vals) -> sadd key vals + Request ["SPOP", key] -> spop key + + _ -> do + return $ ErrorReply "Bad Request" +{-# INLINE process #-} diff --git a/Database/Redis/Parser.hs b/Database/Redis/Parser.hs new file mode 100644 --- /dev/null +++ b/Database/Redis/Parser.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Database.Redis.Parser ( + parseRequest, + ) where + +import Control.Applicative +import Control.Monad +import Data.Attoparsec.Char8 as A + +import Database.Redis.Types + +parseRequest :: Parser Request +parseRequest = unified <|> inline where + unified = do + n <- char '*' *> decimal <* crlf + Request <$> replicateM n arg + + arg = do + n <- char '$' *> decimal <* crlf + A.take n <* crlf + + inline = do + cmd <- A.takeWhile1 A.isAlpha_ascii + Request <$> ((cmd:) <$> many (char ' ' *> A.takeTill isSpace)) <* crlf + + crlf = string "\r\n" + +{-# INLINE parseRequest #-} diff --git a/Database/Redis/Server.hs b/Database/Redis/Server.hs new file mode 100644 --- /dev/null +++ b/Database/Redis/Server.hs @@ -0,0 +1,37 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} + +module Database.Redis.Server ( + runServer, + ServerSettings, serverSettings, + def, + ) where + +import Blaze.ByteString.Builder +import Control.Monad.Logger +import Data.Conduit +import Data.Conduit.Attoparsec (conduitParser) +import Data.Conduit.Internal (sinkToPipe, sourceToPipe) +import qualified Data.Conduit.List as CL +import Data.Conduit.Network +import Data.Monoid +import qualified Data.Text as T +import Network (withSocketsDo) + +import Database.Curry +import Database.Redis.Builder +import Database.Redis.Commands +import Database.Redis.Parser +import Database.Redis.Types + +runServer :: Config -> ServerSettings (RedisT IO) -> IO () +runServer conf ss = withSocketsDo $ runDBMT conf $ do + $logInfo $ "listen on port " <> (T.pack $ show $ serverPort ss) <> "." + runTCPServer ss $ \ad -> do + runPipe + $ sourceToPipe (appSource ad) + >+> injectLeftovers (conduitParser parseRequest) + -- >+> CL.mapM (\req -> $logInfo (T.pack $ show $ snd req) >> return req) + >+> CL.mapM (fmap (toByteString . fromReply) . process) + >+> sinkToPipe (appSink ad) diff --git a/Database/Redis/Types.hs b/Database/Redis/Types.hs new file mode 100644 --- /dev/null +++ b/Database/Redis/Types.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE DeriveDataTypeable #-} + +module Database.Redis.Types ( + RedisT, + Value(..),Score, SortedSet, + Request(..), Reply(..), + ) where + +import Control.Applicative +import Data.Binary +import qualified Data.ByteString as S +import qualified Data.HashMap.Strict as HMS +import qualified Data.HashSet as HS +import Data.Int +import qualified Data.Sequence as Seq +import qualified Data.Set as Set + +import Database.Curry + +type RedisT m = DBMT Value m + +data Value + = VString {-# UNPACK #-} !S.ByteString + | VList !(Seq.Seq S.ByteString) + | VSet !(HS.HashSet S.ByteString) + | VHash !(HMS.HashMap S.ByteString S.ByteString) + | VSortedSet {-# UNPACK #-} !SortedSet + +type Score = Int32 + +type SortedSet = (Set.Set (Score, S.ByteString), HMS.HashMap S.ByteString Score) + +data Request + = Request [S.ByteString] + deriving (Show) + +data Reply + = StatusReply {-# UNPACK #-} !S.ByteString + | ErrorReply {-# UNPACK #-} !S.ByteString + | IntReply {-# UNPACK #-} !Int + | BulkReply !(Maybe S.ByteString) + | MultiBulkReply !(Maybe [Maybe S.ByteString]) + deriving (Show) + +instance Binary Value where + put (VString bs) = put (0 :: Word8) >> put bs + put (VList ls) = put (1 :: Word8) >> put ls + put (VSet ss) = put (2 :: Word8) >> put (HS.toList ss) + put (VHash ha) = put (3 :: Word8) >> put ha + put (VSortedSet ss) = put (4 :: Word8) >> put ss + + get = get >>= \tag -> case (tag :: Word8) of + 0 -> VString <$> get + 1 -> VList <$> get + 2 -> VSet . HS.fromList <$> get + 3 -> VHash <$> get + 4 -> VSortedSet <$> get + _ -> fail "data corrupted" diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2012, Hideyuki Tanaka + +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 Hideyuki Tanaka 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/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/bench/bench.hs b/bench/bench.hs new file mode 100644 --- /dev/null +++ b/bench/bench.hs @@ -0,0 +1,39 @@ + +main :: IO () +main = return () + +{- +import Control.Applicative +import Control.Monad +import Control.Monad.Trans +import qualified Data.ByteString.Char8 as S +import Prelude hiding (lookup) +import Data.Char +import System.Random.Mersenne.Pure64 +import System.IO.Unsafe +import Data.IORef + +ior :: IORef PureMT +ior = unsafePerformIO $ newIORef =<< newPureMT +{-# NOINLINE ior #-} + +randomString :: Int -> PureMT -> (String, PureMT) +randomString n m = go 0 "" m where + go i ss mt + | i == n = (ss, mt) + | otherwise = + let (x, nt) = randomInt mt in + go (i+1) (f x :ss) nt + f x = chr (ord 'a' + x `mod` 26) + +main :: IO () +main = runDBMT $ do + replicateM_ 1000000 $ do + (key, val) <- liftIO $ do + mt1 <- readIORef ior + let (key, mt2) = randomString 8 mt1 + let (val, mt3) = randomString 8 mt2 + writeIORef ior mt3 + return (key, val) + insert (S.pack key) (S.pack val) +-} diff --git a/memcached/main.hs b/memcached/main.hs new file mode 100644 --- /dev/null +++ b/memcached/main.hs @@ -0,0 +1,6 @@ +{-# LANGUAGE OverloadedStrings #-} + +import Database.Memcached.Server + +main :: IO () +main = runServer (serverSettings 3334 "*") diff --git a/redis/main.hs b/redis/main.hs new file mode 100644 --- /dev/null +++ b/redis/main.hs @@ -0,0 +1,44 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +import Database.Curry +import Database.Redis +import Options.Applicative +-- import System.Remote.Monitoring (forkServer) +import Filesystem.Path.CurrentOS (decodeString) + +data Option = Option + { optPort :: Int + , optPath :: Maybe FilePath + } + +opts :: Parser Option +opts = Option + <$> option + ( long "port" & short 'p' & metavar "PORT" + & value 8854 + & help "Port Number" + ) + <*> option + ( long "file" & short 'f' & metavar "FILE" + & reader (Just . Just) + & value Nothing + & help "dump file" + ) + +main :: IO () +main = execParser optsInfo >>= \Option{..} -> do + -- forkServer "localhost" 8000 -- monitoring + (`runServer` serverSettings optPort "*") $ def + { configPath = fmap decodeString optPath + , configSaveStrategy = + [ SaveByFrequency 900 1 + , SaveByFrequency 300 10 + , SaveByFrequency 60 10000 + ] + } + where + optsInfo = info (helper <*> opts) + ( fullDesc + & header "curry-redis - a redis clone over CurryDB" + ) diff --git a/tests/doctests.hs b/tests/doctests.hs new file mode 100644 --- /dev/null +++ b/tests/doctests.hs @@ -0,0 +1,26 @@ +import Control.Applicative +import Control.Monad +import Data.List +import System.Directory +import System.FilePath +import Test.DocTest + +main :: IO () +main = getSources >>= \sources -> doctest $ + "-i." + : "-idist/build/autogen" + : "-optP-include" + : "-optPdist/build/autogen/cabal_macros.h" + : sources + +getSources :: IO [FilePath] +getSources = filter (isSuffixOf ".hs") <$> go "Database" + where + go dir = do + (dirs, files) <- getFilesAndDirectories dir + (files ++) . concat <$> mapM go dirs + +getFilesAndDirectories :: FilePath -> IO ([FilePath], [FilePath]) +getFilesAndDirectories dir = do + c <- map (dir ) . filter (`notElem` ["..", "."]) <$> getDirectoryContents dir + (,) <$> filterM doesDirectoryExist c <*> filterM doesFileExist c diff --git a/tests/hspec.hs b/tests/hspec.hs new file mode 100644 --- /dev/null +++ b/tests/hspec.hs @@ -0,0 +1,110 @@ +{-# LANGUAGE OverloadedStrings #-} + +import Control.Applicative +import Control.Monad +import Control.Monad.Trans +import Data.Conduit +import qualified Data.Conduit.List as CL +import Data.List (sort) + +import Test.Hspec +-- import Test.QuickCheck +-- import Test.HUnit + +import Database.Curry + +import Prelude hiding (lookup) + +main :: IO () +main = hspec $ do + describe "core apis" $ do + it "insert correctly" $ do + runDBMT def $ do + v1 <- transaction $ lookup "foo" + liftIO $ v1 `shouldBe` Nothing + transaction $ insert "foo" (123 :: Int) + v2 <- transaction $ lookup "foo" + liftIO $ v2 `shouldBe` Just 123 + v3 <- transaction $ lookup "bar" + liftIO $ v3 `shouldBe` Nothing + return () :: IO () + + it "insertWith correctly" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + transaction $ insertWith (+) "foo" 456 + v1 <- transaction $ lookup "foo" + liftIO $ v1 `shouldBe` Just (123 + 456) + return () :: IO () + + it "delete correctly" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + v1 <- transaction $ lookup "foo" + liftIO $ v1 `shouldBe` Just 123 + transaction $ delete "foo" + v2 <- transaction $ lookup "foo" + liftIO $ v2 `shouldBe` Nothing + return () :: IO () + + it "lookupDefault correctly" $ do + runDBMT def $ do + v1 <- transaction $ lookupDefault "foo" + liftIO $ v1 `shouldBe` 0 + transaction $ insert "foo" (123 :: Int) + v2 <- transaction $ lookupDefault "foo" + liftIO $ v2 `shouldBe` 123 + return () :: IO () + + it "keys correctly" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + transaction $ insert "bar" (456 :: Int) + transaction $ insert "baz" (789 :: Int) + ks <- transaction $ keys + ls <- ks $$ CL.consume + liftIO $ sort ls `shouldBe` sort ["foo", "bar", "baz"] + return () :: IO () + + it "keys correctly even after updates" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + transaction $ insert "bar" (456 :: Int) + transaction $ insert "baz" (789 :: Int) + ks <- transaction $ keys + transaction $ insert "hoge" (123 :: Int) + transaction $ insert "moge" (456 :: Int) + ls <- ks $$ CL.consume + liftIO $ sort ls `shouldBe` sort ["foo", "bar", "baz"] + return () :: IO () + + it "transaction correctly" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + transaction $ insert "bar" (456 :: Int) + transaction $ do + insertWith (+) "foo" 456 + insertWith subtract "bar" 123 + v1 <- transaction $ lookup "foo" + liftIO $ v1 `shouldBe` Just (123 + 456) + v2 <- transaction $ lookup "bar" + liftIO $ v2 `shouldBe` Just (456 - 123) + return () :: IO () + + it "transaction rollbacks correctly" $ do + runDBMT def $ do + transaction $ insert "foo" (123 :: Int) + transaction $ insert "bar" (456 :: Int) + transaction $ do + insertWith (+) "foo" 456 + insertWith subtract "bar" 123 + insert "baz" 789 + lift mzero + <|> return () + v1 <- transaction $ lookup "foo" + liftIO $ v1 `shouldBe` Just 123 + v2 <- transaction $ lookup "bar" + liftIO $ v2 `shouldBe` Just 456 + v3 <- transaction $ lookup "baz" + liftIO $ v3 `shouldBe` Nothing + return () :: IO ()