continuum (empty) → 0.1.0.0
raw patch · 7 files changed
+476/−0 lines, 7 filesdep +basedep +bytestringdep +cerealsetup-changed
Dependencies added: base, bytestring, cereal, containers, data-default, foldl, leveldb-haskell-fork, mtl, nanomsg-haskell, parallel-io, resourcet, stm, suspend, time, timers, transformers, transformers-base
Files
- LICENSE +0/−0
- Setup.hs +2/−0
- continuum.cabal +72/−0
- src/Continuum/Client/Base.hs +34/−0
- src/Continuum/Common/Serialization.hs +215/−0
- src/Continuum/Common/Types.hs +147/−0
- src/Continuum/Server/Base.hs +6/−0
+ LICENSE view
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ continuum.cabal view
@@ -0,0 +1,72 @@+name: continuum+stability: experimental+description: Continuum Database Client+version: 0.1.0.0+license: MIT+license-file: LICENSE+author: Alex Petrov+maintainer: alexp@coffeenco.de+build-type: Simple+cabal-version: >=1.10++Flag Binaries+ description: Build examples+ default: False+ manual: True++Flag Profiling+ description: Profiling mode+ default: False+ manual: True++source-repository head+ type: git+ location: https://github.com/ifesdjeen/continuum.git++library+ default-language: Haskell2010+ exposed-modules: Continuum.Client.Base+ , Continuum.Common.Types+ other-modules: Continuum.Common.Serialization+ if flag(Profiling)+ ghc-options: -fprof-auto -rtsopts -rtsopts -O2 -Wall -threaded+ else+ ghc-options: -Wall -O2 -threaded -threaded -rtsopts+ hs-source-dirs: src+ build-depends: base >= 4 && < 5+ , containers >0.5+ , bytestring >= 0.10.4.0+ , cereal >= 0.4.0.1+ , nanomsg-haskell >= 0.2.2+ , time >=1.4.2+ , mtl >2.1++executable continuum-server+ Default-Language: Haskell2010+ hs-source-dirs: src+ other-extensions: CPP, MagicHash+ extra-libraries: hyperleveldb+ build-depends: base >= 4 && < 5+ , foldl >= 1.0.5+ , mtl >2.1+ , containers >0.5+ , time >=1.4.2+ , timers >=0.2.0.0+ , stm >=2.4.3+ , suspend >=0.1.0.1+ , leveldb-haskell-fork >= 0.3.4.1+ , resourcet >= 1.1.2.2+ , bytestring >= 0.10.4.0+ , parallel-io >= 0.3.3+ , cereal >= 0.4.0.1+ , data-default+ , transformers >= 0.4.1.0+ , transformers-base >= 0.4.1 && < 0.5+ , nanomsg-haskell >= 0.2.2++ if flag(Profiling)+ ghc-options: -fprof-auto -rtsopts -rtsopts -O2 -Wall -threaded+ else+ ghc-options: -Wall -rtsopts -O2 -threaded++ main-is: Continuum/Server/Base.hs
+ src/Continuum/Client/Base.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}++module Continuum.Client.Base+ (module Continuum.Client.Base+ , module Continuum.Common.Types+ )+ where++import qualified Nanomsg as N+import Data.Serialize (encode, decode)+import qualified Continuum.Common.Serialization as S+import Continuum.Common.Types++data ContinuumClient = ContinuumClient { clientSocket :: N.Socket N.Bus }++connect :: String+ -> String+ -> IO ContinuumClient+connect host port = do+ socket <- N.socket N.Bus+ _ <- N.connect socket ("tcp://" ++ host ++ ":" ++ port)+ return ContinuumClient { clientSocket = socket }+++executeQuery :: ContinuumClient+ -> Query+ -> IO (DbErrorMonad DbResult)+executeQuery client query = do+ let sock = (clientSocket client)+ _ <- N.send sock (encode query)+ bs <- N.recv sock+ return $ (S.decodeDbResult bs)
+ src/Continuum/Common/Serialization.hs view
@@ -0,0 +1,215 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Continuum.Common.Serialization where++import Foreign+import Continuum.Common.Types++import Data.Serialize as S+import qualified Data.ByteString as B+import qualified Data.Map as Map++import Control.Applicative ( (<$>) )+import Data.List ( elemIndex )+import Data.Maybe ( isJust, fromJust, catMaybes )+import Control.Monad.Except ( forM_, throwError )++-- import Debug.Trace++data Success = Success++-- validate :: DbSchema -> DbRecord -> Either String Success+-- validate = error "Not Implemented"++-- |+-- | ENCODING+-- |++encodeRecord :: DbSchema -> DbRecord -> Integer -> (B.ByteString, B.ByteString)+encodeRecord schema (DbRecord timestamp vals) sid = (encodeKey, encodeValue)+ where encodeKey = B.concat [(packWord64 timestamp), (packWord64 sid)]+ encodedParts = fmap fastEncodeValue $ catMaybes $ (\x -> Map.lookup x vals) <$> (fields schema)+ lengths = B.length <$> encodedParts+ encodeValue = runPut $ do+ -- Change to B.Pack+ forM_ lengths (putWord8 . fromIntegral)+ forM_ encodedParts putByteString+ -- encode . catMaybes $ fmap (\x -> Map.lookup x vals) (fields schema)++encodeBeginTimestamp :: Integer -> B.ByteString+encodeBeginTimestamp timestamp =+ B.concat [(packWord64 timestamp), (packWord64 0)]++encodeEndTimestamp :: Integer -> B.ByteString+encodeEndTimestamp timestamp =+ B.concat [(packWord64 timestamp), (packWord64 999999)]++encodeSchema :: DbSchema -> B.ByteString+encodeSchema = encode++decodeSchema :: (B.ByteString, B.ByteString)+ -> DbErrorMonad (B.ByteString, DbSchema)+decodeSchema (dbName, encodedSchema) =+ case (decode encodedSchema) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right schema) -> return (dbName, schema)++decodeQuery :: B.ByteString+ -> DbErrorMonad Query+decodeQuery encodedQuery =+ case (decode encodedQuery) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right query) -> return query++decodeDbResult :: B.ByteString+ -> DbErrorMonad DbResult+decodeDbResult encodedDbResult =+ case (decode encodedDbResult) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right query) -> return query+++-- |+-- | DECODING+-- |++decodeRecord :: Decoding+ -> DbSchema+ -> (B.ByteString, B.ByteString)+ -> DbErrorMonad DbResult++decodeRecord (Field field) schema !(k, bs) = do+ decodedK <- decodeKey k+ decodedVal <- if isJust idx+ then decodeFieldByIndex schema indices (fromJust idx) bs+ else throwError FieldNotFoundError+ return $! FieldRes (decodedK, decodedVal)+ where idx = elemIndex field (fields schema)+ indices = decodeIndexes schema bs++decodeRecord Record schema !(k, bs) = do+ timestamp <- decodeKey k+ decodedValue <- decodeValues schema bs+ return $! RecordRes $ DbRecord timestamp (Map.fromList $ zip (fields schema) decodedValue)++decodeRecord (Fields flds) schema (k, bs) = do+ decodedK <- decodeKey k+ decodedVal <- if isJust idxs+ then mapM (\idx -> decodeFieldByIndex schema (decodeIndexes schema bs) idx bs) (fromJust idxs)+ else throwError FieldNotFoundError+ return $! FieldsRes (decodedK, decodedVal)+ where+ {-# INLINE idxs #-}+ idxs = mapM (`elemIndex` (fields schema)) flds+{-# INLINE decodeRecord #-}++-- |+-- | DECODING Utility Functions+-- |++decodeKey :: B.ByteString -> DbErrorMonad Integer+decodeKey x = unpackWord64 (B.take 8 x)+{-# INLINE decodeKey #-}++decodeIndexes :: DbSchema -> B.ByteString -> [Int]+decodeIndexes schema bs =+ map fromIntegral $ B.unpack $ B.take (length (fields schema)) bs+{-# INLINE decodeIndexes #-}++slide :: [a] -> [(a, a)]+slide (f:s:xs) = (f,s) : slide (s:xs)+slide _ = []++decodeValues :: DbSchema -> B.ByteString -> DbErrorMonad [DbValue]+decodeValues schema bs = mapM (\(t, s) -> fastDecodeValue t s) (zip (schemaTypes schema) bytestrings)+ where+ indices = decodeIndexes schema bs+ bytestrings = snd (foldl step (B.drop (length indices) bs, []) indices)+ step (remaining', acc) n = (B.drop n remaining', acc ++ [B.take n remaining'])++-- | Decodes field by index+decodeFieldByIndex :: DbSchema+ -> [Int]+ -> Int+ -> B.ByteString+ -> DbErrorMonad DbValue+decodeFieldByIndex schema indices idx bs = fastDecodeValue ((schemaTypes schema) !! idx) bytestring+ where+ {-# INLINE bytestring #-}+ bytestring = B.take (indices !! idx) $ B.drop startFrom bs+ {-# INLINE startFrom #-}+ startFrom = (length indices) + (sum $ take idx indices)+{-# INLINE decodeFieldByIndex #-}+++-- |+-- | "FAST" CUSTOM SERIALIZATION+-- |++fastEncodeValue :: DbValue -> B.ByteString+fastEncodeValue (DbInt value) = packWord64 value+fastEncodeValue (DbString value) = value++fastDecodeValue :: DbType -> B.ByteString -> DbErrorMonad DbValue+fastDecodeValue DbtInt bs = DbInt <$> unpackWord64 bs+fastDecodeValue DbtString bs = return $ DbString bs++packWord64 :: Integer -> B.ByteString+packWord64 i =+ let w = (fromIntegral i :: Word64)+ in B.pack [ (fromIntegral (w `shiftR` 56) :: Word8)+ , (fromIntegral (w `shiftR` 48) :: Word8)+ , (fromIntegral (w `shiftR` 40) :: Word8)+ , (fromIntegral (w `shiftR` 32) :: Word8)+ , (fromIntegral (w `shiftR` 24) :: Word8)+ , (fromIntegral (w `shiftR` 16) :: Word8)+ , (fromIntegral (w `shiftR` 8) :: Word8)+ , (fromIntegral (w) :: Word8)]+{-# INLINE packWord64 #-}++unpackWord64 :: B.ByteString -> Either DbError Integer+unpackWord64 s = return $+ (fromIntegral (s `B.index` 0) `shiftL` 56) .|.+ (fromIntegral (s `B.index` 1) `shiftL` 48) .|.+ (fromIntegral (s `B.index` 2) `shiftL` 40) .|.+ (fromIntegral (s `B.index` 3) `shiftL` 32) .|.+ (fromIntegral (s `B.index` 4) `shiftL` 24) .|.+ (fromIntegral (s `B.index` 5) `shiftL` 16) .|.+ (fromIntegral (s `B.index` 6) `shiftL` 8) .|.+ (fromIntegral (s `B.index` 7) )+{-# INLINE unpackWord64 #-}++-- Group Queries++byField :: B.ByteString -> DbRecord -> DbValue+byField f (DbRecord _ m) = fromJust $ Map.lookup f m+byField _ _ = DbInt 1 -- WTF++byFieldMaybe :: B.ByteString -> DbRecord -> Maybe DbValue+byFieldMaybe f (DbRecord _ m) = Map.lookup f m+byFieldMaybe _ _ = Just $ DbInt 1 -- WTF++byTime :: Integer -> DbRecord -> Integer+byTime interval (DbRecord t _) = interval * (t `quot` interval)+++unpackString :: DbValue -> B.ByteString+unpackString (DbString i) = i+unpackString _ = error "Can't unpack Int"++unpackInt :: DbValue -> Integer+unpackInt (DbInt i) = i+unpackInt _ = error "Can't unpack Int"++unpackFloat :: DbValue -> Float+unpackFloat (DbFloat i) = i+unpackFloat _ = error "Can't unpack Float"++unpackDouble :: DbValue -> Double+unpackDouble (DbDouble i) = i+unpackDouble _ = error "Can't unpack Double"
+ src/Continuum/Common/Types.hs view
@@ -0,0 +1,147 @@+{-# LANGUAGE DeriveGeneric #-}++module Continuum.Common.Types where++import Data.ByteString ( ByteString )+import GHC.Generics ( Generic )+import qualified Data.Serialize as S+import qualified Data.Map as Map++data DbType =+ DbtInt+ | DbtString+ deriving(Show, Generic)++instance S.Serialize DbType++data DbError =+ IndexesDecodeError String+ | FieldDecodeError String ByteString+ | ValuesDecodeError String+ | ValueDecodeError String+ | KeyDecodeError String+ | DecodeFieldByIndexError String [Int]+ | FieldNotFoundError+ | NoSuchDatabaseError+ | NoAggregatorAvailable+ | SchemaDecodingError String+ | OtherError+ deriving (Show, Eq, Ord, Generic)++instance S.Serialize DbError++type DbErrorMonad = Either DbError++data DbValue =+ EmptyValue+ | DbInt Integer+ | DbFloat Float+ | DbDouble Double+ | DbString ByteString+ | DbTimestamp Integer+ | DbSequenceId Integer+ -- DbList [DbValue]+ -- DbMap [(DbValue, DbValue)]+ deriving (Show, Eq, Ord, Generic)++instance S.Serialize DbValue++data DbRecord =+ DbRecord Integer (Map.Map ByteString DbValue)+ deriving(Generic, Show, Eq)++instance S.Serialize DbRecord++-- |+-- | DB RESULT+-- |++data DbResult =+ EmptyRes+ | ErrorRes DbError+ | RecordRes DbRecord+ | FieldRes (Integer, DbValue)+ | FieldsRes (Integer, [DbValue])+ | CountStep Integer+ | CountRes Integer+ | GroupRes (Map.Map DbValue DbResult)+ deriving(Generic, Show, Eq)++instance S.Serialize DbResult++-- |+-- | RANGE+-- |++data KeyRange =+ OpenEnd ByteString+ | TsOpenEnd Integer+ | SingleKey ByteString+ | TsSingleKey Integer+ | KeyRange ByteString ByteString+ | TsKeyRange Integer Integer+ | EntireKeyspace+ deriving(Show)+-- |+-- | AGGREGATES+-- |++data Decoding =+ Field ByteString+ | Fields [ByteString]+ | Record+ deriving(Generic, Show)++instance S.Serialize Decoding++-- |+-- |+-- |++data DbSchema = DbSchema+ { fieldMappings :: Map.Map ByteString Int+ , fields :: [ByteString]+ , indexMappings :: Map.Map Int ByteString+ , schemaMappings :: Map.Map ByteString DbType+ , schemaTypes :: [DbType]+ } deriving (Generic, Show)++instance S.Serialize DbSchema++-- |+-- | QUERIES+-- |++data Query =+ Count+ | Distinct+ | Min+ | Max+ | Group Query+ | Insert ByteString ByteString+ | CreateDb ByteString DbSchema+ | RunQuery ByteString Decoding Query+ deriving (Generic, Show)++instance S.Serialize Query++-- |+-- | External Protocol Specification+-- |++-- TODO: Split client and server requests++data Node = Node String String+ deriving(Generic, Show, Eq, Ord)++instance S.Serialize Node++data Request =+ ImUp Node+ | Introduction Node+ | NodeList [Node]+ | Heartbeat Node+ | Query Query+ deriving(Generic, Show)++instance S.Serialize Request
+ src/Continuum/Server/Base.hs view
@@ -0,0 +1,6 @@+module Main where++import Continuum.Cluster++main :: IO ()+main = startNode