packages feed

sync-mht 0.2.0.0 → 0.2.1.0

raw patch · 10 files changed

+853/−27 lines, 10 files

Files

src/main/hs/Main.hs view
@@ -1,3 +1,4 @@+-- main module {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE OverloadedStrings #-} import System.Environment
+ src/main/hs/Sync/MerkleTree/Analyse.hs view
@@ -0,0 +1,55 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE OverloadedStrings #-}+module Sync.MerkleTree.Analyse+    ( analyseDirectory+    ) where++import Control.Monad+import System.FilePath+import Foreign.C.Types+import Prelude hiding (lookup)+import Sync.MerkleTree.Types+import System.Directory++import System.Posix.Types+import System.Posix.Files+import qualified Data.Text as T++isRealFile :: String -> Bool+isRealFile x+    | x `elem` [".", ".."] = False+    | otherwise = True++analyseDirectory :: FilePath -> [FilePath] -> Path -> IO [Entry]+analyseDirectory fp ignore path+    | fp `elem` ignore = return []+    | otherwise =+        do files <- getDirectoryContents fp+           liftM concat $ mapM (analyse fp ignore path) $ filter isRealFile files++analyse :: FilePath -> [FilePath] -> Path -> String -> IO [Entry]+analyse fp ignore path name+    | (fp </> name) `elem` ignore = return []+    | otherwise =+        do status <- getFileStatus fp'+           analyse' status+    where+      path' = Path (SerText $ T.pack name) path+      fp' = fp </> name+      analyse' status+          | isRegularFile status =+              let CTime modtime = modificationTime status+                  COff filesize = fileSize status+              in return+                  [ FileEntry $ File+                    { f_name = path'+                    , f_size = FileSize filesize+                    , f_modtime = FileModTime modtime+                    } ]+          | isDirectory status =+              liftM ((DirectoryEntry path'):) $ analyseDirectory fp' ignore path'+          | otherwise = return [] -- No support for devices, sockets yet.+
+ src/main/hs/Sync/MerkleTree/Client.hs view
@@ -0,0 +1,126 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+module Sync.MerkleTree.Client where++import Control.Monad+import Control.Monad.IO.Class+import Data.Foldable(Foldable)+import Data.Monoid(Monoid, mappend, mempty, Sum(..))+import Data.Set(Set)+import Data.List+import Foreign.C.Types+import qualified Data.ByteString as BS+import qualified Data.Foldable as F+import qualified Data.Set as S+import qualified Data.Map as M+import qualified Data.Text as T+import Sync.MerkleTree.CommTypes+import Sync.MerkleTree.Trie+import Sync.MerkleTree.Types+import System.Directory+import System.IO+import System.Posix.Files++data Diff a = Diff (Set a) (Set a)+    deriving Show++instance Ord a => Monoid (Diff a) where+    mempty = Diff S.empty S.empty+    mappend (Diff x1 y1) (Diff x2 y2) = Diff (x1 `S.union` x2) (y1 `S.union` y2)++showText :: (Show a) => a -> T.Text+showText = T.pack . show++dataSize :: (Foldable f) => f Entry -> FileSize+dataSize s = getSum $ F.foldMap sizeOf s+    where+      sizeOf (FileEntry f) = Sum $ f_size f+      sizeOf (DirectoryEntry _) = Sum $ FileSize 0++dataSizeText :: (Foldable f) => f Entry -> T.Text+dataSizeText s = T.concat [showText $ unFileSize $ dataSize s, " bytes"]++class (Protocol m, MonadIO m) => (ClientMonad m) where+    split :: (Monoid a) => [m a] -> m a++logClient :: (Protocol m) => T.Text -> m ()+logClient t =+    do True <- logReq $ SerText t+       return ()++analyseEntries :: Diff Entry -> ([Entry],[Entry],[Entry])+analyseEntries (Diff obsoleteEntries newEntries) =+    (M.elems deleteMap, M.elems changeMap, M.elems newMap)+    where+      deleteMap = M.difference obsMap updMap+      changeMap = M.intersection updMap obsMap+      newMap = M.difference updMap obsMap+      obsMap = M.fromList $ S.toList $ S.map keyValue obsoleteEntries+      updMap = M.fromList $ S.toList $ S.map keyValue newEntries+      keyValue x = (name x, x)+      name (FileEntry f) = f_name f+      name (DirectoryEntry f) = f++abstractClient :: (ClientMonad m) => ClientServerOptions -> FilePath -> Trie Entry -> m ()+abstractClient cs fp trie =+    do logClient $ T.concat [ "Client finished directory traversal: ", showText $ t_hash trie ]+       Diff oent nent <- nodeReq (rootLocation, trie)+       let (delEntries, changedEntries, newEntries) = analyseEntries (Diff oent nent)+       logClient $ T.concat+           [ "Client has ", showText $ length delEntries, " superfluos files of size "+           , dataSizeText delEntries, ", ", showText $ length changedEntries, " changed files "+           , "of size ", dataSizeText changedEntries, " and ", showText $ length newEntries+           , " missing files of size ", dataSizeText newEntries, "."+           ]+       when (cs_delete cs) $+           forM_ (reverse $ sort delEntries) $ \e ->+               case e of+                 FileEntry f -> liftIO $ removeFile $ toFilePath fp $ f_name f+                 DirectoryEntry p -> liftIO $ removeDirectory $ toFilePath fp p+       mapM_ (split . map (synchronizeNewOrChangedEntry fp))+           $ splitEvery _CONCURRENT_FILETRANSFER_SIZE_ $ sort+           $ [ e | cs_add cs, e <- newEntries ] ++ [ e | cs_update cs, e <- changedEntries ]+       True <- terminateReq+       return ()++_CONCURRENT_FILETRANSFER_SIZE_ :: Int+_CONCURRENT_FILETRANSFER_SIZE_ = 96++splitEvery :: Int -> [a] -> [[a]]+splitEvery n l+    | null l = []+    | (h,t) <- splitAt n l = h:(splitEvery n t)++synchronizeNewOrChangedEntry :: (ClientMonad m) => FilePath -> Entry -> m ()+synchronizeNewOrChangedEntry fp entry =+    case entry of+      FileEntry f ->+          do firstResult <- queryFileReq (f_name f)+             h <- liftIO $ openFile (toFilePath fp $ f_name f) WriteMode+             let loop result =+                     case result of+                       Final -> return ()+                       ToBeContinued content contHandle ->+                           (liftIO $ BS.hPut h content) >> queryFileContReq contHandle >>= loop+             loop firstResult+             liftIO $ hClose h+             let modTime = (CTime $ unModTime $ f_modtime f)+             liftIO $ setFileTimes (toFilePath fp $ f_name f) modTime modTime+      DirectoryEntry p -> liftIO $ createDirectory $ toFilePath fp p++nodeReq :: (ClientMonad m) => (TrieLocation, Trie Entry) -> m (Diff Entry)+nodeReq (loc,trie) =+    do fp <- queryHashReq loc+       case () of+         () | fp == toFingerprint trie ->+                return mempty+            | Node arr <- t_node trie, NodeType == f_nodeType fp ->+                split $ map nodeReq (expand loc arr)+            | otherwise ->+                do s' <- querySetReq loc+                   let s = getAll trie+                   return $ Diff (s `S.difference` s') (s' `S.difference` s)+++
+ src/main/hs/Sync/MerkleTree/CommTypes.hs view
@@ -0,0 +1,63 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleInstances #-}+module Sync.MerkleTree.CommTypes where++import GHC.Generics+import Data.Set(Set)+import Data.Serialize+import Data.ByteString(ByteString)+import Data.Typeable++import Sync.MerkleTree.Types+import Sync.MerkleTree.Trie++class Monad m => Protocol m where+    queryHashReq :: TrieLocation -> m Fingerprint+    querySetReq :: TrieLocation -> m (Set Entry)+    logReq :: SerText -> m Bool+    queryFileReq :: Path -> m QueryFileResponse+    queryFileContReq :: ContHandle -> m QueryFileResponse+    terminateReq :: m Bool++data ContHandle = ContHandle Int+    deriving (Show, Generic, Typeable)++instance Serialize ContHandle++data ClientServerOptions+    = ClientServerOptions+      { cs_add :: Bool+      , cs_update :: Bool+      , cs_delete :: Bool+      , cs_ignore :: [FilePath]+      }+      deriving (Generic, Show, Typeable)++instance Serialize ClientServerOptions++data Request+    = QuerySet TrieLocation+    | QueryHash TrieLocation+    | Log SerText+    | QueryFile Path+    | QueryFileCont ContHandle+    | Terminate+    deriving (Generic, Show, Typeable)++instance Serialize Request++data QueryFileResponse+    = Final+    | ToBeContinued ByteString ContHandle+    deriving (Generic, Show, Typeable)++instance Serialize QueryFileResponse++data ChildSide+    = Service FilePath ClientServerOptions+    | Client FilePath ClientServerOptions+    deriving (Show, Generic, Typeable)++instance Serialize ChildSide
+ src/main/hs/Sync/MerkleTree/Server.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+module Sync.MerkleTree.Server where++import Control.Monad.State+import Sync.MerkleTree.CommTypes+import Sync.MerkleTree.Trie+import qualified Data.Text.IO as T+import Sync.MerkleTree.Types+import qualified Data.Map as M+import Data.Map(Map)+import qualified Data.ByteString as BS+import System.IO++data ServerState+    = ServerState+    { st_handles :: Map Int Handle+    , st_nextHandle :: Int+    , st_trie :: Trie Entry+    , st_path :: FilePath+    }++type ServerMonad = StateT ServerState IO++startServerState :: FilePath -> Trie Entry -> ServerState+startServerState fp trie =+    ServerState+    { st_handles = M.empty+    , st_nextHandle = 0+    , st_trie = trie+    , st_path = fp+    }++instance Protocol ServerMonad where+    querySetReq l = get >>= (\s -> return $ querySet (st_trie s) l)+    queryHashReq l = get >>= (\s -> return $ queryHash (st_trie s) l)+    logReq (SerText msg) = liftIO (T.hPutStrLn stderr msg) >> return True+    queryFileContReq (ContHandle n) =+        do s <- get+           let Just h = M.lookup n (st_handles s)+           withHandle h n+    queryFileReq f =+        do s <- get+           h <- liftIO $ openFile (toFilePath (st_path s) f) ReadMode+           let n = st_nextHandle s+           put $ s { st_handles = M.insert n h (st_handles s), st_nextHandle = n + 1 }+           withHandle h n+    terminateReq = return True++withHandle :: Handle -> Int -> ServerMonad QueryFileResponse+withHandle h n =+    do bs <- liftIO $ BS.hGet h (2^(17::Int))+       case () of+         () | BS.null bs ->+             do liftIO $ hClose h+                modify (\s -> s { st_handles = M.delete n (st_handles s) })+                return $ Final+            | otherwise -> return $ ToBeContinued bs $ ContHandle n
+ src/main/hs/Sync/MerkleTree/Sync.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE OverloadedStrings #-}+module Sync.MerkleTree.Sync+    ( child+    , local+    , parent+    , openStreams+    , Direction(..)+    ) where++import Control.Monad+import Control.Monad.State+import Data.Monoid+import System.FilePath+import Prelude hiding (lookup)+import Sync.MerkleTree.Trie+import Sync.MerkleTree.Types+import System.IO+import System.IO.Streams(InputStream, OutputStream)+import Data.ByteString(ByteString)+import qualified Data.Serialize as SE+import qualified Data.ByteString as BS+import qualified System.IO.Streams as ST++import Sync.MerkleTree.Analyse+import Sync.MerkleTree.CommTypes+import Sync.MerkleTree.Client+import Sync.MerkleTree.Server+import Sync.MerkleTree.Util.RequestMonad++data StreamPair+    = StreamPair+    { sp_in :: InputStream ByteString+    , sp_out :: OutputStream ByteString+    }++openStreams :: Handle -> Handle -> IO StreamPair+openStreams hIn hOut =+    do inStream <- ST.handleToInputStream hIn+       outStream <- ST.handleToOutputStream hOut+       return $ StreamPair { sp_in = inStream, sp_out = outStream }++instance Protocol RequestMonad where+    queryHashReq = request . QueryHash+    querySetReq = request . QuerySet+    queryFileReq = request . QueryFile+    queryFileContReq = request . QueryFileCont+    logReq = request . Log+    terminateReq = request Terminate++instance ClientMonad RequestMonad where+    split = splitRequests++instance ClientMonad ServerMonad where+    split xs = liftM mconcat $ sequence xs++data Direction+    = FromRemote+    | ToRemote++child :: IO ()+child =+    do streams <- openStreams stdin stdout+       side <- getFromInputStream (sp_in streams)+       case side of+         Service dir clientServerOpts -> server dir clientServerOpts streams+         Client dir clientServerOpts -> client dir clientServerOpts streams++parent :: StreamPair -> FilePath -> FilePath -> Direction -> ClientServerOptions -> IO ()+parent streams source destination direction clientServerOpts =+    case direction of+      FromRemote ->+        do respond (sp_out streams) $ Service source clientServerOpts+           client destination clientServerOpts streams+      ToRemote ->+        do respond (sp_out streams) $ Client destination clientServerOpts+           server source clientServerOpts streams++respond :: (Show a, SE.Serialize a) => OutputStream ByteString -> a -> IO ()+respond os = mapM_ (flip ST.write os . Just) . (:[BS.empty]) . SE.encode++local :: ClientServerOptions -> FilePath -> FilePath -> IO ()+local cs source destination =+    do sourceDir <- liftM (mkTrie 0) $ analyseDirectory source (cs_ignore cs) Root+       destinationDir <- liftM (mkTrie 0) $ analyseDirectory destination (cs_ignore cs) Root+       evalStateT (abstractClient cs destination destinationDir) (startServerState source sourceDir)++server :: FilePath -> ClientServerOptions -> StreamPair -> IO ()+server fp cs streams =+    do dir <- analyseDirectory fp (cs_ignore cs) Root+       evalStateT loop (startServerState fp $ mkTrie 0 dir)+    where+       serverRespond = liftIO . respond (sp_out streams)+       loop =+           do req <- liftIO $ getFromInputStream (sp_in streams)+              case req of+                QueryHash l -> queryHashReq l >>= serverRespond >> loop+                QuerySet l -> querySetReq l >>= serverRespond >> loop+                QueryFile f -> queryFileReq f >>= serverRespond >> loop+                QueryFileCont c -> queryFileContReq c >>= serverRespond >> loop+                Log t -> logReq t >>= serverRespond >> loop+                Terminate -> terminateReq >>= serverRespond >> return ()++client :: FilePath -> ClientServerOptions -> StreamPair -> IO ()+client fp cs streams =+    do dir <- analyseDirectory fp (cs_ignore cs) Root+       runRequestMonad (sp_in streams) (sp_out streams) $ abstractClient cs fp $ mkTrie 0 dir+
+ src/main/hs/Sync/MerkleTree/Trie.hs view
@@ -0,0 +1,154 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}+module Sync.MerkleTree.Trie where++import Prelude hiding (lookup)+import qualified Data.List as L+import Data.Typeable+import Data.Byteable+import Crypto.Hash+import qualified Data.Set as S+import Data.Set(Set)+import Data.Array.IArray+import Control.Arrow hiding (arr, loop)+import qualified Data.Text.Encoding as TE+import qualified Data.Text as T+import GHC.Generics+import qualified Data.ByteString.Base16 as B16+import qualified Data.ByteString as BS+import qualified Data.Serialize as SE++data Hash = Hash { unHash :: BS.ByteString }+    deriving (Eq, Ord, Generic)++instance Show Hash where+    showsPrec i x = showsPrec i (go x)+        where+          go :: Hash -> String+          go (Hash dig) = T.unpack $ TE.decodeUtf8 $ B16.encode dig++instance Read Hash where+    readsPrec i = map (first go) . readsPrec i+        where+          go :: String -> Hash+          go s = Hash . fst . B16.decode $ TE.encodeUtf8 $ T.pack s++instance SE.Serialize Hash++data Trie a+    = Trie+      { t_hash :: !Hash+      , t_node :: !(TrieNode a)+      }+      deriving (Eq, Read, Show)++data TrieNode a+    = Node !(Array Int (Trie a))+    | Leave !(Set a)+    deriving (Eq, Read, Show)++data NodeType = NodeType | LeaveType+    deriving (Eq, Read, Show, Generic)+instance SE.Serialize NodeType++data TrieLocation+    = TrieLocation+    { tl_level :: Int+    , tl_index :: Int+    }+    deriving (Read, Show, Generic)++instance SE.Serialize TrieLocation++degree :: Int+degree = 64++class HasDigest a where+    digest :: a -> Digest SHA256++data Fingerprint+    = Fingerprint+      { f_hash :: Hash+      , f_nodeType :: NodeType+      }+      deriving (Eq, Read, Show, Generic, Typeable)++instance SE.Serialize Fingerprint++toFingerprint :: Trie a -> Fingerprint+toFingerprint (Trie h node) = Fingerprint h nodeType+     where+       nodeType =+           case node of+             Node _ -> NodeType+             Leave _ -> LeaveType++mkTrie :: (Ord a, HasDigest a) => Int -> [a] -> Trie a+mkTrie i ls+    | length ls < degree = mkLeave ls+    | otherwise =+        mkNode $ fmap (mkTrie (i+1)) $ accumArray (flip (:)) [] (0,degree-1) $ map ((groupOf i) &&& id) ls++mkNode :: (Array Int (Trie a)) -> Trie a+mkNode arr =+    Trie+    { t_hash = combineHash $ map t_hash $ elems arr+    , t_node = Node arr+    }++hashSHA256 :: BS.ByteString -> Digest SHA256+hashSHA256 = hash++combineHash :: [Hash] -> Hash+combineHash = Hash . toBytes . hashSHA256 . BS.concat . map unHash++groupOf :: (HasDigest a) => Int -> a -> Int+groupOf i x = fromInteger $ toInteger $ (h0 `mod` (fromInteger $ toInteger degree))+     where+       Just (h0, _t) = BS.uncons $ toBytes $ h+       h :: Digest SHA256+       h = hash $ BS.concat [BS.pack [fromInteger $ toInteger i], toBytes $ digest x]++mkLeave :: (HasDigest a, Ord a) => [a] -> Trie a+mkLeave ls =+    Trie+    { t_hash = combineHash $ map (Hash . toBytes . digest) $ L.sort ls+    , t_node = Leave $ S.fromList ls+    }++lookup :: Trie a -> TrieLocation -> Trie a+lookup trie (TrieLocation { tl_level = l, tl_index = i })+    | l < 0 || i < 0 || i >= degree^l = error "illegal index pair"+    | l > 0, (g, i') <- i `quotRem` (degree ^ (l-1)), Node arr <- t_node trie =+        lookup (arr ! g) (TrieLocation { tl_level = (l - 1), tl_index =  i'})+    | l == 0 = trie+    | otherwise = error "index pair to deep"++queryHash :: Trie a -> TrieLocation -> Fingerprint+queryHash trie = toFingerprint . lookup trie++querySet :: (Ord a) => Trie a -> TrieLocation -> Set a+querySet trie = getAll . lookup trie++getAll :: (Ord a) => Trie a -> Set a+getAll (Trie _ node) =+    case node of+      Node arr -> S.unions $ map getAll $ elems arr+      Leave s -> s++rootLocation :: TrieLocation+rootLocation =+    TrieLocation+    { tl_level = 0+    , tl_index = 0+    }++expand :: TrieLocation -> (Array Int (Trie a)) -> [(TrieLocation, Trie a)]+expand loc arr = map go [0..(degree - 1)]+    where+      go i =+          ( TrieLocation+            { tl_level = tl_level loc + 1+            , tl_index = degree * tl_index loc + i }+          , arr ! i+          )
+ src/main/hs/Sync/MerkleTree/Types.hs view
@@ -0,0 +1,79 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Sync.MerkleTree.Types where++import System.FilePath+import Data.Int+import Crypto.Hash+import Data.Ord+import GHC.Generics+import Data.Typeable+import qualified Data.Text.Encoding as TE+import qualified Data.Text as T+import Sync.MerkleTree.Trie+import qualified Data.Serialize as SE++data File+    = File+      { f_name :: Path+      , f_size :: FileSize+      , f_modtime :: FileModTime+      }+      deriving (Show, Eq, Ord, Generic, Typeable)++instance SE.Serialize File++data Entry+    = FileEntry File+    | DirectoryEntry Path+    deriving (Show, Eq, Generic, Typeable)++instance SE.Serialize Entry++newtype FileSize = FileSize { unFileSize :: Int64 }+    deriving (Show, Eq, Ord, Generic, Num, Typeable)++instance SE.Serialize FileSize++data FileModTime = FileModTime { unModTime :: !Int64 }+    deriving (Show, Eq, Ord, Generic)++instance SE.Serialize FileModTime++data Path+    = Root+    | Path SerText Path+    deriving (Eq, Ord, Generic)++instance Show Path where+    show x = toFilePath "/" x++toFilePath :: FilePath -> Path -> FilePath+toFilePath fp p =+    case p of+      Root -> fp+      Path x y -> (toFilePath fp y) </> (T.unpack $ unSerText x)++instance SE.Serialize Path++instance Ord Entry where+    compare = comparing withLevel+        where+          level Root = 0 :: Int+          level (Path _ p) = 1 + level p+          withLevel entry =+              case entry of+                DirectoryEntry p -> (level p, Right p)+                FileEntry f -> (level $ f_name f, Left f)++instance HasDigest Entry where+    digest = hash . SE.encode++data SerText = SerText { unSerText :: !T.Text }+    deriving (Ord, Show, Eq)++instance SE.Serialize SerText where+    get = SE.get >>= either (fail . show) (return . SerText) . TE.decodeUtf8'+    put = SE.put . TE.encodeUtf8 . unSerText+
+ src/main/hs/Sync/MerkleTree/Util/RequestMonad.hs view
@@ -0,0 +1,182 @@+-- | RequestMonad+--+-- This monad is used to increase the speed of communication between two processes - if there is+-- latency. It works by using the non-deterministic part of the communication protocol to send+-- multiple requests to the output-channel, before processing the responses from the input-channel.+--+-- An example:+-- @+--     foo = split [bar, baz]+--     bar = do x <- request (GetSumOf 1 2)+--              liftM Sum request (GetSumOf x 3)+--     baz = liftM Sum request (GetSumOf 4 5)+-- @+--+-- In this case+-- @+--    runRequestMonad inputHandle outputHandle foo+-- @+-- will send both messages GetSumOf 1 2, GetSumOf 4 5, without having to wait for the repsonse to+-- the first request. The last request GetSumOf 3 3 will be send after the response for the first+-- message has arrived.+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+module Sync.MerkleTree.Util.RequestMonad+    ( RequestMonad+    , request+    , runRequestMonad+    , splitRequests+    , getFromInputStream+    ) where++import Control.Applicative(Applicative(..))+import Control.Concurrent(Chan, writeChan, readChan, newChan, forkIO)+import Control.Monad(ap,liftM,unless)+import Control.Monad.IO.Class(MonadIO(..))+import Data.ByteString(ByteString)+import Data.IORef(IORef,newIORef,modifyIORef,readIORef)+import Data.Monoid(Monoid, mempty, mappend)+import Data.Serialize(Serialize)+import System.IO(hPutStrLn,stderr)+import System.IO.Streams(InputStream, OutputStream)+import qualified Data.ByteString as BS+import qualified Data.Serialize as SE+import qualified System.IO.Streams as ST++data SplitState f b = forall a. (Monoid a) => SplitState [RequestMonadT f a] a (a -> RequestMonad b)+data RequestState f b = forall a. (Serialize a) => RequestState f (a -> RequestMonad b)+data LiftIOState b = forall a. LiftIOState (IO a) (a -> RequestMonad b)++type RequestMonad = RequestMonadT ByteString++data RequestMonadT f b+    = Split (SplitState f b)+    | Request (RequestState f b)+    | LiftIO (LiftIOState b)+    | Return b+    | Fail String++instance Functor (RequestMonadT ByteString) where+    fmap = liftM++instance Applicative (RequestMonadT ByteString) where+    pure  = return+    (<*>) = ap++instance Monad (RequestMonadT ByteString) where+    return = Return+    fail = Fail+    (>>=) = bindImpl++instance MonadIO (RequestMonadT ByteString) where+    liftIO x =  LiftIO $ LiftIOState x Return++bindImpl :: (RequestMonad a) -> (a -> RequestMonad b) -> (RequestMonad b)+bindImpl f g =+    case f of+      Split (SplitState xs z cont) -> Split (SplitState xs z (\t -> bindImpl (cont t) g))+      Request (RequestState r cont) -> Request (RequestState r (\t -> bindImpl (cont t) g))+      LiftIO (LiftIOState op cont) -> LiftIO (LiftIOState op (\t -> bindImpl (cont t) g))+      Return x -> g x+      Fail s -> Fail s++request :: (Serialize a, Serialize b) => a -> RequestMonad b+request x = Request $ RequestState (SE.encode x) Return++splitRequests :: (Monoid a) => [RequestMonad a] -> RequestMonad a+splitRequests alts = Split $ SplitState alts mempty Return++data SendQueue+    = SendQueue+    { sq_chan :: Chan (Maybe ByteString)+    , sq_sendIndex :: IORef Int+    }++queueRequests :: SendQueue -> (RequestMonad b) -> IO (RequestMonadT Int b)+queueRequests sq root =+    case root of+      LiftIO (LiftIOState op cont) -> return $ LiftIO (LiftIOState op cont)+      Request (RequestState r c) ->+          do writeChan (sq_chan sq) (Just r)+             modifyIORef (sq_sendIndex sq) (+1)+             i <- readIORef (sq_sendIndex sq)+             return $ Request (RequestState i c)+      Split (SplitState xs z cont) ->+          do xs' <- mapM (queueRequests sq) xs+             return $ Split $ SplitState xs' z cont+      Return x -> return $ Return x+      Fail s -> return $ Fail s++runRequestMonad ::+    InputStream ByteString+    -> OutputStream ByteString+    -> RequestMonad b+    -> IO b+runRequestMonad is os startMonad =+    do sendChan <- newChan+       recvIdx <- newIORef 0+       sendIdx <- newIORef 0+       _ <- forkIO $ writerThread os sendChan+       let sq = SendQueue { sq_chan = sendChan, sq_sendIndex = sendIdx }+           loop monad =+               do monad' <- receiverThread recvIdx sq is monad+                  case monad' of+                    Return x -> writeChan sendChan Nothing >> return x+                    Fail err -> fail err+                    _ -> loop monad'+       queueRequests sq startMonad >>= loop++writerThread :: OutputStream ByteString -> Chan (Maybe ByteString) -> IO ()+writerThread os chan = loop+    where+      loop =+          do mBs <- readChan chan+             ST.write mBs os+             ST.write (Just "") os+             maybe (return ()) (const loop) mBs++getFromInputStream :: (Serialize a) => InputStream ByteString -> IO a+getFromInputStream s = go (SE.Partial $ SE.runGetPartial SE.get)+    where+      go (SE.Fail err bs) = ST.unRead bs s >> fail err+      go (SE.Partial f) =+          do x <- ST.read s+             case x of+               Nothing -> (go $ f BS.empty)+               Just x' | BS.null x' -> go (SE.Partial f)+                       | otherwise -> go (f x')+      go (SE.Done r bs) = ST.unRead bs s >> return r++receiverThread ::+    IORef Int+    -> SendQueue+    -> InputStream ByteString+    -> RequestMonadT Int b+    -> IO (RequestMonadT Int b)+receiverThread recvIdx sq input root =+    case root of+      LiftIO (LiftIOState op cont) -> op >>= (queueRequests sq . cont)+      Request (RequestState i cont) ->+          do x <- getFromInputStream input+             modifyIORef recvIdx (+1)+             expected <- readIORef recvIdx+             unless (expected == i) $+                 do hPutStrLn stderr $ "Expected " ++ (show i) ++ " but got " ++ show expected+                    fail $ "Expected " ++ show i ++ " but got " ++ show expected+             queueRequests sq $ cont x+      Split (SplitState xs z cont) -> loop cont z xs []+      Return x -> return $ Return x+      Fail err -> return $ Fail err+    where+      loop cont z [] [] = queueRequests sq $ cont z+      loop cont z [] r = return $ Split $ SplitState (reverse r) z cont+      loop cont z (x:xs) r =+          do x' <- receiverThread recvIdx sq input x+             case x' of+               Return x'' -> loop cont (z `mappend` x'') xs r+               Fail s -> return $ Fail s+               other -> loop cont z xs (other:r)++
sync-mht.cabal view
@@ -1,33 +1,18 @@ -- This .cabal file was generated by src/build/hs/CabalTemplate.hs name: sync-mht-version: 0.2.0.0+version: 0.2.1.0 synopsis: -   Fast incremental file transfer using Merke-Hash-Trees-   +    Fast incremental file transfer using Merke-Hash-Trees+     description: -   A command line tool that can be used to incrementally synchronize a directory hierarchy with a-   second one. It is using a Merkle-Hash-Tree to compare the folders, such that the synchronization-   time and communication (round) complexity grows only logarithmically with the actual size of the-   directories (assuming the actual difference of the directories is small).-   -   The communication happens through standard streams between parent and child processes, which can-   easily be routed through remote command execution tools, e.g.-   -       sync-mht -s foo/ -d bar-   -   will synchronize the local folder bar/ with the local folder foo/, but-   -       sync-mht -s foo/ -d remote:/bar -r "ssh fred@example.org sync-mht"-   -   will synchronize the folder bar/ in the home directory of the user fred on the host machine-   example.org with the local folder foo/.-   -   It is also possible to use it with docker, e.g.-   -       sync-mht -s foo/ -d remote:/bar -r "docker run -i --volumes-from bar ekarayel/sync-mht sync-mht"-   -   to synchronize the folder /bar (of the container named bar) with the local folder foo/.-   +    A command line tool that can be used to incrementally synchronize a directory hierarchy with a+    second one. It is using a Merkle-Hash-Tree to compare the folders, such that the synchronization+    time and communication (round) complexity grows only logarithmically with the actual size of the+    directories (assuming the actual difference of the directories is small).+    +    The communication happens through standard streams between parent and child processes, which can+    easily be routed through remote command execution tools.+     license: MIT license-file: LICENSE author: Emin Karayel <me@eminkarayel.de>@@ -41,10 +26,19 @@     location: https://github.com/ekarayel/sync-mht source-repository this     type: git-    tag: 0.2.0.0+    tag: 0.2.1.0     location: https://github.com/ekarayel/sync-mht executable sync-mht     main-is: Main.hs+    other-modules: +        Sync.MerkleTree.Analyse+        Sync.MerkleTree.Client+        Sync.MerkleTree.CommTypes+        Sync.MerkleTree.Server+        Sync.MerkleTree.Sync+        Sync.MerkleTree.Trie+        Sync.MerkleTree.Types+        Sync.MerkleTree.Util.RequestMonad     ghc-options: -Wall     build-depends:          base >=4.7 && <4.8