diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # sync-mht [![Build Status](https://travis-ci.org/ekarayel/sync-mht.svg?branch=master)](https://travis-ci.org/ekarayel/sync-mht) - +[![codecov.io](http://codecov.io/github/ekarayel/sync-mht/coverage.svg?branch=master)](http://codecov.io/github/ekarayel/sync-mht?branch=master) ## Synopsis Fast incremental file transfer using Merkle-Hash-Trees diff --git a/dist/build/mainStub/mainStub-tmp/mainStub.hs b/dist/build/mainStub/mainStub-tmp/mainStub.hs deleted file mode 100644 --- a/dist/build/mainStub/mainStub-tmp/mainStub.hs +++ /dev/null @@ -1,5 +0,0 @@ -module Main ( main ) where -import Distribution.Simple.Test.LibV09 ( stubMain ) -import Sync.MerkleTree.Test ( tests ) -main :: IO () -main = stubMain tests diff --git a/src/Main.hs b/src/Main.hs new file mode 100644 --- /dev/null +++ b/src/Main.hs @@ -0,0 +1,9 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +import Data.Version (showVersion) +import Paths_sync_mht (version) +import System.Environment +import qualified Sync.MerkleTree.Run as S + +main :: IO () +main = getArgs >>= S.main (showVersion version) diff --git a/src/Sync/MerkleTree/Analyse.hs b/src/Sync/MerkleTree/Analyse.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Analyse.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NoMonomorphismRestriction #-} +{-# LANGUAGE OverloadedStrings #-} +module Sync.MerkleTree.Analyse + ( analyse + , isRealFile + ) where + +import Control.Monad +import Data.Maybe +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 Text.Regex (matchRegex, mkRegex, Regex) +import qualified Data.Text as T + +-- | Returns True if the given string is not "." or ".." +isRealFile :: String -> Bool +isRealFile x + | x `elem` [".", ".."] = False + | otherwise = True + +shouldIgnore :: Path -> [Regex] -> Bool +shouldIgnore p regexes = any (isJust . (`matchRegex` (toFilePath "" p))) regexes + +-- | Returns all files and directories below the given FilePath +analyse :: + FilePath -- ^ Root file path to analyse` + -> [String] -- ^ List of regular expressions to be excluded from the resulting list + -> IO [Entry] -- ^ List of file or directory entries with paths relative to the given root +analyse fp ignore = analyseSubDirectory fp (map mkRegex ignore) Root + +analyseSubDirectory :: FilePath -> [Regex] -> Path -> IO [Entry] +analyseSubDirectory fp ignore path = + do files <- getDirectoryContents fp + liftM concat $ mapM (analyseEntry fp ignore path) $ filter isRealFile files + +analyseEntry :: FilePath -> [Regex] -> Path -> String -> IO [Entry] +analyseEntry fp ignore path name + | shouldIgnore path' 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'):) $ analyseSubDirectory fp' ignore path' + | otherwise = return [] -- No support for devices, sockets yet. diff --git a/src/Sync/MerkleTree/Client.hs b/src/Sync/MerkleTree/Client.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Client.hs @@ -0,0 +1,187 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +module Sync.MerkleTree.Client where + +import Control.Monad +import Control.Monad.IO.Class +import Codec.Compression.GZip +import Data.Foldable(Foldable) +import Data.Function +import Data.Monoid(Monoid, mappend, mempty, Sum(..)) +import Data.Set(Set) +import Data.Time.Clock +import Data.List +import Data.Ratio +import Data.IORef +import Foreign.C.Types +import Sync.MerkleTree.CommTypes +import Sync.MerkleTree.Trie +import Sync.MerkleTree.Types +import System.Directory +import System.IO +import System.Posix.Files +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as BL +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 qualified Test.HUnit as H + +data Diff a = Diff (Set a) (Set a) + +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 () + +data SimpleEntry + = FileSimpleEntry Path + | DirectorySimpleEntry Path + deriving (Eq, Ord) + +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) = FileSimpleEntry $ f_name f + name (DirectoryEntry f) = DirectorySimpleEntry f + +data Progress + = Progress + { pg_size :: IORef FileSize + , pg_count :: IORef Int + , pg_last :: IORef UTCTime + } + +abstractClient :: (ClientMonad m) => ClientServerOptions -> FilePath -> Trie Entry -> m () +abstractClient cs fp trie = + do logClient $ T.concat [ "Hash of destination directory: ", showText $ t_hash trie, "\n" ] + 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, ".\n" + ] + 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 + let updateEntries = + [ e | cs_add cs, e <- newEntries ] ++ [ e | cs_update cs, e <- changedEntries ] + progressEntries <- liftIO $ newIORef $ length updateEntries + progressSize <- liftIO $ newIORef $ dataSize updateEntries + progressLast <- liftIO $ getCurrentTime >>= newIORef + let progress = + Progress + { pg_size = progressSize + , pg_count = progressEntries + , pg_last = progressLast + } + mapM_ (syncNewOrChangedEntries progress fp) + $ groupBy ((==) `on` levelOf) $ sort $ updateEntries + logClient "Done. \n" + True <- terminateReq + return () + +_CONCURRENT_FILETRANSFER_SIZE_ :: Int +_CONCURRENT_FILETRANSFER_SIZE_ = 48 + +splitEvery :: Int -> [a] -> [[a]] +splitEvery n l + | null l = [] + | (h,t) <- splitAt n l = h:(splitEvery n t) + +syncNewOrChangedEntries :: (ClientMonad m) => Progress -> FilePath -> [Entry] -> m () +syncNewOrChangedEntries pg fp entries = + forM_ (splitEvery _CONCURRENT_FILETRANSFER_SIZE_ entries) $ \entryGroup -> + split $ map (synchronizeNewOrChangedEntry pg fp) entryGroup + +showProgess :: (ClientMonad m) => Progress -> m () +showProgess pg = + do t <- liftIO getCurrentTime + l <- liftIO $ readIORef (pg_last pg) + when (diffUTCTime t l > fromRational (1 % 2)) $ + do leftSize <- liftIO $ readIORef (pg_size pg) + leftCount <- liftIO $ readIORef (pg_count pg) + logClient $ T.concat + [ "Transfering: ", showText $ unFileSize $ leftSize, " bytes and " + , showText leftCount, " files left. \r" + ] + t2 <- liftIO getCurrentTime + liftIO $ writeIORef (pg_last pg) t2 + +synchronizeNewOrChangedEntry :: (ClientMonad m) => Progress -> FilePath -> Entry -> m () +synchronizeNewOrChangedEntry pg 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 -> + do let bs = BL.toStrict $ decompress $ BL.fromStrict content + liftIO $ BS.hPut h $ bs + liftIO $ modifyIORef (pg_size pg) + (subtract $ fromIntegral $ BS.length bs) + showProgess pg + queryFileContReq contHandle >>= loop + loop firstResult + liftIO $ hClose h + liftIO $ modifyIORef (pg_count pg) (subtract 1) + let modTime = (CTime $ unModTime $ f_modtime f) + liftIO $ setFileTimes (toFilePath fp $ f_name f) modTime modTime + DirectoryEntry p -> + do liftIO $ modifyIORef (pg_count pg) (subtract 1) + 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) + +testEntry :: H.Test +testEntry = H.TestLabel "testEntry" $ H.TestList + [ False H.~=? (FileSimpleEntry p /= FileSimpleEntry p) + , False H.~=? (DirectorySimpleEntry p == FileSimpleEntry p) + ] + where + p = Path (SerText "t") Root diff --git a/src/Sync/MerkleTree/CommTypes.hs b/src/Sync/MerkleTree/CommTypes.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/CommTypes.hs @@ -0,0 +1,72 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE FlexibleInstances #-} +module Sync.MerkleTree.CommTypes where + +import GHC.Generics +import Data.Set(Set) +import Data.Serialize +import Data.ByteString(ByteString) + +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 (Generic) + +instance Serialize ContHandle + +data ClientServerOptions + = ClientServerOptions + { cs_add :: Bool + , cs_update :: Bool + , cs_delete :: Bool + , cs_ignore :: [FilePath] + } + deriving (Read, Show) + +data Request + = QuerySet TrieLocation + | QueryHash TrieLocation + | Log SerText + | QueryFile Path + | QueryFileCont ContHandle + | Terminate + deriving (Generic) + +instance Serialize Request + +data QueryFileResponse + = Final + | ToBeContinued ByteString ContHandle + deriving (Generic) + +instance Serialize QueryFileResponse + +data ProtocolVersion + = Version1 + | Version2 + deriving (Read, Show, Eq) + +thisProtocolVersion :: ProtocolVersion +thisProtocolVersion = Version2 + +data LaunchMessage + = LaunchMessage + { lm_protocolVersion :: ProtocolVersion + , lm_dir :: FilePath + , lm_side :: Side + , lm_clientServerOptions :: ClientServerOptions + } + deriving (Read, Show) + +data Side = Server | Client + deriving (Read, Show) diff --git a/src/Sync/MerkleTree/Run.hs b/src/Sync/MerkleTree/Run.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Run.hs @@ -0,0 +1,191 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +module Sync.MerkleTree.Run where + +import Control.Concurrent +import Control.Monad +import Data.List + +import System.Console.GetOpt +import System.IO +import System.IO.Error +import System.Process +import Sync.MerkleTree.CommTypes +import Sync.MerkleTree.Sync +import qualified Data.Text as T +import qualified Data.Text.IO as T + +data RemoteCmd + = RemoteCmd String + | Simulate + +data SyncOptions + = SyncOptions + { so_source :: Maybe FilePath + , so_destination :: Maybe FilePath + , so_remote :: Maybe RemoteCmd + , so_ignore :: [String] + , so_boring :: [FilePath] + , so_add :: Bool + , so_update :: Bool + , so_delete :: Bool + , so_help :: Bool + , so_nonOptions :: [String] + } + +defaultSyncOptions :: SyncOptions +defaultSyncOptions = + SyncOptions + { so_source = Nothing + , so_destination = Nothing + , so_remote = Nothing + , so_ignore = [] + , so_boring = [] + , so_add = False + , so_update = False + , so_delete = False + , so_help = False + , so_nonOptions = [] + } + +toClientServerOptions :: SyncOptions -> IO ClientServerOptions +toClientServerOptions so = + do let parseBoringFile = map T.unpack . filter noComment . map T.strip . T.lines + noComment s = not (T.null s || ("#" `T.isPrefixOf` s)) + ignoreFromBoringFiles <- + forM (so_boring so) $ liftM parseBoringFile . T.readFile + return $ + ClientServerOptions + { cs_add = so_add so + , cs_update = so_update so + , cs_delete = so_delete so + , cs_ignore = (concat ignoreFromBoringFiles) ++ (so_ignore so) + } + +optDescriptions :: [OptDescr (SyncOptions -> SyncOptions)] +optDescriptions = + [ Option ['s'] ["source"] (ReqArg (\fp so -> so { so_source = Just fp }) "DIR") + "source directory" + , Option ['d'] ["destination"] (ReqArg (\fp so -> so { so_destination = Just fp }) "DIR") + "destination directory" + , Option ['r'] ["remote-shell"] (ReqArg (\s so -> so { so_remote = Just $ RemoteCmd s }) "CMD") + "synchroize with a remote-site (see below)" + , Option ['i'] ["ignore"] (ReqArg (\fp so -> so { so_ignore = fp:(so_ignore so) }) "REGEX") + "ignore entries matching the given regex" + , Option ['b'] ["boring"] (ReqArg (\fp so -> so { so_boring = fp:(so_boring so) }) "PATH") + "ignore entries matching the regexes in the given file" + , Option ['a'] ["add"] (NoArg (\so -> so { so_add = True })) + "copy additional files from the source directory" + , Option ['u'] ["update"] (NoArg (\so -> so { so_update = True })) + "overwrite existing files" + , Option [] ["delete"] (NoArg (\so -> so { so_delete = True })) + "delete superfluos files in the destination directory" + , Option ['h'] ["help"] (NoArg (\so -> so { so_help = True })) "shows usage information" + ] + +parseNonOption :: String -> (SyncOptions -> SyncOptions) +parseNonOption s so = so { so_nonOptions = s:(so_nonOptions so) } + +toSyncOptions :: [(SyncOptions -> SyncOptions)] -> SyncOptions +toSyncOptions = foldl (flip id) defaultSyncOptions + +putError :: String -> IO () +putError = hPutStrLn stderr + +_HIDDENT_CLIENT_MODE_OPTION_ :: String +_HIDDENT_CLIENT_MODE_OPTION_ = "--hidden-client-mode-option" + +printUsageInfo :: String -> [String] -> IO () +printUsageInfo version prefix = + mapM_ putError (prefix ++ [usageInfo header optDescriptions] ++ [details]) + where + header = unlines + [ "Usage: sync-mht [OPTIONS..]" + , "" + , "Fast incremental file transfer using Merkle-Hash-Trees (Version: " ++ version ++ ")" + ] + details = unlines + [ "Note: The argument to the --remote-shell option should be a CMD running sync-mht" + , "with a remote command execution tool (like ssh or docker). If given exactly one of" + , "the directories must be prepended with 'remote:' - indicating a folder on the site," + , "accessible with the provided remote shell command." + ] + +data Location + = Remote FilePath + | Local FilePath + +parseFilePath :: FilePath -> Location +parseFilePath fp + | Just rest <- stripPrefix "remote:" fp = Remote rest + | otherwise = Local fp + +main :: String -> [String] -> IO () +main version args = flip catchIOError (putError . show) $ + do let parsedOpts = getOpt (ReturnInOrder parseNonOption) optDescriptions args + case () of + () | [_HIDDENT_CLIENT_MODE_OPTION_] == args -> runChild + | (options,[],[]) <- parsedOpts -> run version $ toSyncOptions options + | (_,_,errs) <- parsedOpts -> printUsageInfo version errs + +run :: String -> SyncOptions -> IO () +run version so + | so_help so = usage [] + | not (null (so_nonOptions so)) = + usage ["Unrecognized options: " ++ intercalate ", " (so_nonOptions so)] + | Just source <- so_source so, Just destination <- so_destination so = + do cs <- toClientServerOptions so + case (parseFilePath source, parseFilePath destination) of + (Remote _, Remote _) -> usage [doubleRemote] + (Local source', Local destination') + | Just _ <- so_remote so -> usage [missingRemote] + | otherwise -> local cs source' destination' + (Remote source', Local destination') + | Just remoteCmd <- so_remote so -> + runParent cs remoteCmd source' destination' FromRemote + | otherwise -> usage [missingRemoteCmd] + (Local source', Remote destination') + | Just remoteCmd <- so_remote so -> + runParent cs remoteCmd source' destination' ToRemote + | otherwise -> usage [missingRemoteCmd] + | otherwise = + do let missingOpts = + intercalate ", " $ map snd $ filter ((== Nothing) . ($ so) . fst) + [(so_source, "--source"),(so_destination, "--destination")] + usage ["The options " ++ missingOpts ++ " are required."] + where + usage = printUsageInfo version + doubleRemote = "Either the directory given in --source or --destination must be local." + missingRemote = concat + [ "The --remote-shell options requires that either the directory given at " + , "--source or --destination is at remote site. (Indicated by the prefix: 'remote:')" + ] + missingRemoteCmd = "The --remote-shell is required when the prefix 'remote:' is used." + +runChild :: IO () +runChild = + do streams <- openStreams stdin stdout + child streams + +runParent :: ClientServerOptions -> RemoteCmd -> FilePath -> FilePath -> Direction -> IO () +runParent clientServerOpts mRemoteCmd source destination dir = + do parentStreams <- + case mRemoteCmd of + RemoteCmd remoteCmd -> + do let remoteCmd' = remoteCmd ++ " " ++ _HIDDENT_CLIENT_MODE_OPTION_ + (Just hIn, Just hOut, Nothing, _ph) <- + createProcess $ (shell remoteCmd') + { std_in = CreatePipe + , std_out = CreatePipe + } + openStreams hOut hIn + Simulate -> + do (parentInStream, childOutStream) <- mkChanStreams + (childInStream, parentOutStream) <- mkChanStreams + _ <- forkIO $ child $ + StreamPair + { sp_in = childInStream + , sp_out = childOutStream + } + return $ StreamPair { sp_in = parentInStream, sp_out = parentOutStream } + parent parentStreams source destination dir clientServerOpts diff --git a/src/Sync/MerkleTree/Server.hs b/src/Sync/MerkleTree/Server.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Server.hs @@ -0,0 +1,69 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE TypeSynonymInstances #-} +{-# LANGUAGE FlexibleInstances #-} +module Sync.MerkleTree.Server where + +import Codec.Compression.GZip +import Control.Monad.State +import Sync.MerkleTree.CommTypes +import Sync.MerkleTree.Trie +import qualified Data.Text as T +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 qualified Data.ByteString.Lazy as BL +import System.IO + +data ServerState + = ServerState + { st_handles :: Map Int Handle -- ^ Map of open file handles with their ids + , st_nextHandle :: Int -- ^ Next available id + , st_trie :: Trie Entry -- ^ Merkle Hash Tree of server file hierarchy + , st_path :: FilePath -- ^ path of the root of the file hierarchy + } + +type ServerMonad = StateT ServerState IO + +startServerState :: FilePath -> Trie Entry -> IO ServerState +startServerState fp trie = + do T.hPutStr stderr $ T.pack $ + concat [ "Hash of source directory: ", show $ t_hash trie, "\n" ] + return $ + ServerState + { st_handles = M.empty + , st_nextHandle = 0 + , st_trie = trie + , st_path = fp + } + +instance Protocol ServerMonad where + querySetReq l = get >>= (\s -> querySet (st_trie s) l) + queryHashReq l = get >>= (\s -> queryHash (st_trie s) l) + logReq (SerText msg) = liftIO (T.hPutStr 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 + +-- | Respond to a queryFile or queryFileCont request for a given file handle and id +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 (BL.toStrict $ compress $ BL.fromStrict bs) $ ContHandle n diff --git a/src/Sync/MerkleTree/Sync.hs b/src/Sync/MerkleTree/Sync.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Sync.hs @@ -0,0 +1,175 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE NoMonomorphismRestriction #-} +{-# LANGUAGE OverloadedStrings #-} +module Sync.MerkleTree.Sync + ( child + , local + , parent + , openStreams + , mkChanStreams + , StreamPair(..) + , Direction(..) + , tests + ) where + +import Control.Concurrent(newChan) +import Control.Monad +import Control.Monad.State +import Data.Monoid +import System.FilePath +import Prelude hiding (lookup) +import Sync.MerkleTree.Trie hiding (tests) +import Sync.MerkleTree.Types +import System.IO +import System.IO.Error +import System.IO.Temp +import System.IO.Streams(InputStream, OutputStream, connect) +import Data.ByteString(ByteString) +import qualified Data.Serialize as SE +import qualified Data.ByteString as BS +import qualified System.IO.Streams as ST +import qualified System.IO.Streams.Concurrent as ST +import qualified Test.HUnit as H + +import Sync.MerkleTree.Analyse +import Sync.MerkleTree.CommTypes +import Sync.MerkleTree.Client +import Sync.MerkleTree.Server +import Sync.MerkleTree.Util.RequestMonad +import Sync.MerkleTree.Util.GetFromInputStream + +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 } + +mkChanStreams :: IO (InputStream ByteString, OutputStream ByteString) +mkChanStreams = + do chan <- newChan + liftM2 (,) (ST.chanToInput chan) (ST.chanToOutput chan) + +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 :: StreamPair -> IO () +child streams = + do launchMessage <- getFromInputStream (sp_in streams) + serverOrClient (read launchMessage) streams + +parent :: StreamPair -> FilePath -> FilePath -> Direction -> ClientServerOptions -> IO () +parent streams source destination direction clientServerOpts = + case direction of + FromRemote -> + do respond (sp_out streams) $ show $ mkLaunchMessage Server source + serverOrClient (mkLaunchMessage Client destination) streams + ToRemote -> + do respond (sp_out streams) $ show $ mkLaunchMessage Client destination + serverOrClient (mkLaunchMessage Server source) streams + where + mkLaunchMessage side dir = + LaunchMessage + { lm_dir = dir + , lm_clientServerOptions = clientServerOpts + , lm_protocolVersion = thisProtocolVersion + , lm_side = side + } + +respond :: (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) $ analyse source (cs_ignore cs) + destinationDir <- liftM (mkTrie 0) $ analyse destination (cs_ignore cs) + serverState <- startServerState source sourceDir + evalStateT (abstractClient cs destination destinationDir) serverState + +serverOrClient :: LaunchMessage -> StreamPair -> IO () +serverOrClient lm streams + | lm_protocolVersion lm == thisProtocolVersion = + let side = + case lm_side lm of + Server -> server + Client -> client (lm_clientServerOptions lm) + in do entries <- analyse (lm_dir lm) (cs_ignore $ lm_clientServerOptions lm) + side entries (lm_dir lm) streams + | otherwise = fail "Incompatible sync-mht versions." + +server :: [Entry] -> FilePath -> StreamPair -> IO () +server entries fp streams = (startServerState fp $ mkTrie 0 entries) >>= evalStateT loop + 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 + +client :: ClientServerOptions -> [Entry] -> FilePath -> StreamPair -> IO () +client cs entries fp streams = + runRequestMonad (sp_in streams) (sp_out streams) $ abstractClient cs fp $ mkTrie 0 entries + +tests :: H.Test +tests = H.TestList $ + [ H.TestLabel "testOpenStreams" $ H.TestCase $ + withSystemTempDirectory "testStreams" $ \dir -> + do let testStr = "31456" + writeFile (dir "read.in") testStr + hIn <- openFile (dir "read.in") ReadMode + hOut <- openFile (dir "write.out") WriteMode + st <- openStreams hIn hOut + connect (sp_in st) (sp_out st) + hClose hIn + hClose hOut + got <- readFile (dir "write.out") + testStr H.@=? got + , H.TestLabel "testProtocolVersion" $ H.TestCase $ + withSystemTempDirectory "testProtocolVersion" $ \dir -> + do r <- flip catchIOError (\_ -> return True) $ + do inst <- + ST.fromByteString $ SE.encode $ show $ + LaunchMessage + { lm_protocolVersion = Version1 + , lm_dir = dir + , lm_side = Client + , lm_clientServerOptions = + ClientServerOptions + { cs_add = False + , cs_update = False + , cs_delete = False + , cs_ignore = [] + } + } + out <- ST.nullOutput + child $ StreamPair { sp_in = inst, sp_out = out } + return False + True H.@=? r + ] diff --git a/src/Sync/MerkleTree/Test.hs b/src/Sync/MerkleTree/Test.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Test.hs @@ -0,0 +1,309 @@ +module Sync.MerkleTree.Test where + +import Control.Concurrent +import Control.Monad +import Data.Ix +import Data.List +import Data.Time.Calendar +import Data.Time.Clock +import Sync.MerkleTree.Analyse +import Sync.MerkleTree.Run +import Sync.MerkleTree.Client +import System.Directory +import System.FilePath +import System.IO +import System.IO.Error +import System.IO.Temp +import System.Posix.Files +import System.Random +import qualified Test.HUnit as H + +tests :: H.Test +tests = H.TestList $ + [ testOptions + , testBigFile + , testHelp + , testCmdLine + , testCmdLineFail + , testExit + , testEntry + , testIgnoreBoring + , testSync + ] + +testIgnoreBoring :: H.Test +testIgnoreBoring = + let testCase ign bor = withSystemTempDirectory "sync-mht" $ \fp -> + do let srcDir = fp "src" + destDir = fp "dest" + createDirectory srcDir + createDirectory $ srcDir "a" + createDirectory $ srcDir "with-foo" + createDirectory destDir + writeFile (fp ".boring") $ unlines $ [ "#baz", "", "foo", "^bar" ] + writeFile (srcDir "added.txt") "testA" + writeFile (srcDir "added-bar.txt") "testB" + writeFile (srcDir "baz") "testC" + writeFile (srcDir "bar-ignore.txt") "testD" + writeFile (srcDir "a" "bar") "testE" + writeFile (srcDir "some-foo.txt") "testF" + writeFile (srcDir "a" "foo") "testG" + runSync $ + defaultSyncOptions + { so_source = Just $ srcDir + , so_destination = Just $ destDir + , so_ignore = ign + , so_boring = map (const $ fp ".boring") bor + , so_add = True + , so_update = True + , so_delete = True + } + readFile (destDir "added.txt") >>= ("testA" H.@=?) + readFile (destDir "added-bar.txt") >>= ("testB" H.@=?) + readFile (destDir "baz") >>= ("testC" H.@=?) + doesFileExist (destDir "bar-ignore.txt") >>= (False H.@=?) + readFile (destDir "a" "bar") >>= ("testE" H.@=?) + doesFileExist (destDir "some-foo.txt") >>= (False H.@=?) + doesFileExist (destDir "a" "foo") >>= (False H.@=?) + (doesDirectoryExist $ destDir "with-foo") >>= (False H.@=?) + in H.TestList + [ H.TestLabel "testIgnore" $ H.TestCase $ testCase ["foo","^bar"] [] + , H.TestLabel "testBoring" $ H.TestCase $ testCase [] [()] + ] + +testBigFile :: H.Test +testBigFile = H.TestLabel "testBigFile" $ H.TestCase $ + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + data_ = show [1..(2^17)] + createDirectory srcDir + createDirectory destDir + forM_ [1..400] $ \i -> writeFile (srcDir ("new.txt"++show i)) data_ + runSync $ + defaultSyncOptions + { so_source = Just $ srcDir + , so_destination = Just $ destDir + , so_add = True + } + dataNew1 <- readFile $ destDir "new.txt123" + dataNew2 <- readFile $ destDir "new.txt234" + dataNew3 <- readFile $ destDir "new.txt345" + data_ H.@=? dataNew1 + data_ H.@=? dataNew2 + data_ H.@=? dataNew3 + +testCmdLine :: H.Test +testCmdLine = H.TestLabel "testCmdLine" $ H.TestCase $ + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + boringFile = testDir ".boring" + expected = show [1..(2^17)] + createDirectory srcDir + createDirectory destDir + writeFile boringFile "$bar^" + writeFile (srcDir "new.txt") expected + runMain ["-s",srcDir,"-d",destDir,"-a","-u","--delete","-i","$foo^","-b",boringFile] + got <- readFile $ destDir "new.txt" + expected H.@=? got + +testCmdLineFail :: H.Test +testCmdLineFail = H.TestLabel "testCmdLineFail" $ H.TestCase $ + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + expected = "test" + createDirectory srcDir + createDirectory destDir + writeFile (srcDir "new.txt") expected + runMain ["-s",srcDir,"-d",destDir,"foo","bar"] + runMain ["-s",srcDir,"-d","remote:"++destDir,"-r","exit 0"] + runMain ["-foobar"] + runMain ["--help"] + runMain ["-s",srcDir,"-d",destDir] + (doesFileExist $ destDir "new.txt") >>= (False H.@=?) + +runMain :: [String] -> IO () +runMain = main "" + +runSync :: SyncOptions -> IO () +runSync = run "" + +testHelp :: H.Test +testHelp = H.TestLabel "testHelp" $ H.TestCase $ + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + data_ = "d" + syncOpts = + defaultSyncOptions + { so_add = True + , so_source = Just $ srcDir + , so_destination = Just $ destDir + } + createDirectory srcDir + createDirectory destDir + writeFile (srcDir "new.txt") data_ + mapM (runSync) $ + [ syncOpts { so_nonOptions = ["foo", "bar"] } + , syncOpts { so_source = Just $ "remote:" ++ srcDir } + , syncOpts { so_source = Nothing } + , syncOpts { so_destination = Nothing } + , syncOpts { so_source = Nothing, so_destination = Nothing } + , defaultSyncOptions + , syncOpts { so_remote = Just $ RemoteCmd "/bin/true" } + , syncOpts { so_destination = Just $ "remote:" ++ destDir } + , syncOpts { so_help = True } + , syncOpts + { so_source = Just $ "remote:" ++ srcDir + , so_destination = Just $ "remote:" ++ destDir + , so_remote = Just $ RemoteCmd "/bin/true" + } ] + (doesFileExist $ destDir "new.txt") >>= (False H.@=?) + +testExit :: H.Test +testExit = H.TestLabel "testExit" $ H.TestCase $ + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + data_ = "d" + createDirectory srcDir + createDirectory destDir + writeFile (srcDir "new.txt") data_ + res <- (flip catchIOError) (\_ -> return True) $ + do runSync $ defaultSyncOptions + { so_source = Just $ "remote:" ++ srcDir + , so_destination = Just $ destDir + , so_remote = Just $ RemoteCmd "exit" + , so_add = True + } + return False + True H.@=? res + +testOptions :: H.Test +testOptions = H.TestLabel "testOptions" $ H.TestCase $ + let prepare add update delete go = + withSystemTempDirectory "sync-mht" $ \testDir -> + do let srcDir = testDir "src" + destDir = testDir "dest" + createDirectory srcDir + createDirectory destDir + writeFile (srcDir "same.txt") "test" + copyFile (srcDir "same.txt") (destDir "same.txt") + writeFile (srcDir "changed.txt") "test" + writeFile (destDir "changed.txt") "testB" + writeFile (srcDir "added.txt") "test" + writeFile (destDir "deleted.txt") "testB" + runSync $ + defaultSyncOptions + { so_source = Just $ srcDir + , so_destination = Just $ destDir + , so_add = add + , so_update = update + , so_delete = delete + } + True <- liftM (=="test") $ readFile (srcDir "same.txt") + True <- liftM (=="test") $ readFile (srcDir "changed.txt") + True <- liftM (=="test") $ readFile (srcDir "added.txt") + go destDir + in do prepare True False False $ \destDir -> + do readFile (destDir "same.txt") >>= ("test" H.@=?) + readFile (destDir "changed.txt") >>= ("testB" H.@=?) + readFile (destDir "added.txt") >>= ("test" H.@=?) + readFile (destDir "deleted.txt") >>= ("testB" H.@=?) + prepare False True False $ \destDir -> + do readFile (destDir "same.txt") >>= ("test" H.@=?) + readFile (destDir "changed.txt") >>= ("test" H.@=?) + readFile (destDir "deleted.txt") >>= ("testB" H.@=?) + prepare False False True $ \destDir -> + do readFile (destDir "same.txt") >>= ("test" H.@=?) + readFile (destDir "changed.txt") >>= ("testB" H.@=?) + doesFileExist (destDir "deleted.txt") >>= (False H.@=?) + +testSync :: H.Test +testSync = + H.TestList + $ flip map [0,1,2] + $ \simulate -> H.TestLabel ("testSync"++ show simulate) + $ H.TestCase + $ forM_ [1..20] $ \_ -> + withSystemTempDirectory "sync-mht" $ \testDir -> + do mkRandomDir 4 [testDir "src", testDir "src-backup"] + mkRandomDir 4 [testDir "target"] + let sourcePrefix + | simulate == 1 = "remote:" + | otherwise = "" + targetPrefix + | simulate == 2 = "remote:" + | otherwise = "" + let cmd = runSync $ + defaultSyncOptions + { so_source = Just $ (sourcePrefix ++) testDir "src" + , so_destination = Just $ (targetPrefix ++) testDir "target" + , so_remote = + case simulate of + 0 -> Nothing + _ -> Just Simulate + , so_add = True + , so_update = True + , so_delete = True + } + cmd + areDirsEqual (testDir "src") (testDir "target") + areDirsEqual (testDir "src") (testDir "src-backup") + cmd + areDirsEqual (testDir "target") (testDir "src-backup") + +utcTimeFrom :: Integer -> UTCTime +utcTimeFrom x = UTCTime (fromGregorian 2015 07 29) (fromInteger x) + +mkRandomDir :: Integer -> [FilePath] -> IO () +mkRandomDir md fps = + do forM fps createDirectory + names <- distinctNames 6 + forM_ names $ \n -> + do genDir <- randomIO + case () of + () | genDir, md > 0 -> mkRandomDir (md - 1) $ map ( n) fps + | genDir -> return () + | otherwise -> + do d <- randomRIO (1,4) + forM_ fps $ \fp -> + do writeFile (fp n) (show d) + setModificationTime (fp n) (utcTimeFrom d) + +areDirsEqual :: FilePath -> FilePath -> IO () +areDirsEqual fp1 fp2 = + do files1 <- liftM (sort . filter isRealFile) $ getDirectoryContents fp1 + files2 <- liftM (sort . filter isRealFile) $ getDirectoryContents fp2 + case () of + () | files1 == files2 -> forM_ files1 $ \f -> areEntriesEqual (fp1 f) (fp2 f) + | otherwise -> fail $ "Unequal: " ++ show (fp1, fp2, files1, files2) + +areEntriesEqual :: FilePath -> FilePath -> IO () +areEntriesEqual f1 f2 = + do s1 <- getFileStatus f1 + s2 <- getFileStatus f2 + case () of + () | isDirectory s1, isDirectory s2 -> areDirsEqual f1 f2 + | isRegularFile s1, isRegularFile s2 -> + do c1 <- readFile f1 + c2 <- readFile f2 + unless (c1 == c2) $ fail $ "Unequal files: " ++ show (f1, f2, c1, c2) + | otherwise -> + fail $ show + (f1, f2, isDirectory s1, isDirectory s2, isRegularFile s1, isRegularFile s2) + +-- | @distinctNames k@ creates k distinct file names +distinctNames :: Integer -> IO [String] +distinctNames k = retry + where + isDistinct names = null $ filter hasDuplicate $ group $ sort names + hasDuplicate = (/= 1) . length + mkName _ = liftM (:[]) $ randomRIO ('a','m') + retry = + do result <- forM [1..k] mkName + if isDistinct result + then return result + else retry diff --git a/src/Sync/MerkleTree/Trie.hs b/src/Sync/MerkleTree/Trie.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Trie.hs @@ -0,0 +1,171 @@ +{-# LANGUAGE DeriveGeneric #-} +module Sync.MerkleTree.Trie where + +import Prelude hiding (lookup) +import Control.Monad +import Control.Arrow hiding (arr, loop) +import Crypto.Hash +import Data.Array.IArray +import Data.Byteable +import Data.Set(Set) +import GHC.Generics +import qualified Data.ByteString as BS +import qualified Data.ByteString.Base16 as B16 +import qualified Data.List as L +import qualified Data.Serialize as SE +import qualified Data.Set as S +import qualified Data.Text as T +import qualified Data.Text.Encoding as TE +import qualified Test.HUnit as H + +data Hash = Hash { unHash :: BS.ByteString } + deriving (Eq, Generic) + +instance Show Hash where + showsPrec _ x = ((T.unpack $ TE.decodeUtf8 $ B16.encode $ unHash x) ++) + +instance SE.Serialize Hash + +-- Abstract Merkle Hash Trie +data Trie a + = Trie + { t_hash :: !Hash + , t_node :: !(TrieNode a) + } + deriving (Show, Eq) + +data TrieNode a + = Node !(Array Int (Trie a)) + | Leave !(Set a) + deriving (Show, Eq) + +data NodeType = NodeType | LeaveType + deriving (Eq, Generic) +instance SE.Serialize NodeType + +-- Location in the Merkle Hash Trie +data TrieLocation + = TrieLocation + { tl_level :: Int -- ^ Must be nonnegative + , tl_index :: Int -- ^ Must be between nonnegative and smaller than (degree^tl_level) + } + deriving (Generic) + +instance SE.Serialize TrieLocation + +degree :: Int +degree = 64 + +class HasDigest a where + digest :: a -> Digest SHA256 + +-- | Fingerprint of a Merkle-Hash-Tree node +-- We asssume the Tree below a node is identical while synchronizing if its FingerPrint is +data Fingerprint + = Fingerprint + { f_hash :: Hash + , f_nodeType :: NodeType + } + deriving (Eq, Generic) + +instance SE.Serialize Fingerprint + +toFingerprint :: Trie a -> Fingerprint +toFingerprint (Trie h node) = Fingerprint h nodeType + where + nodeType = + case node of + Node _ -> NodeType + Leave _ -> LeaveType + +-- | Creates a Merkle-Hash-Tree for a list of elements +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 + +-- | The function @groupOf x@ eeturns a value between 0 to degree-1 for a digest with the property +-- that @groupOf@ forms an approximate unviversal hash familiy. +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 :: (Monad m) => Trie a -> TrieLocation -> m (Trie a) +lookup trie (TrieLocation { tl_level = l, tl_index = i }) + | l < 0 || i < 0 || i >= degree^l = fail "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 = return trie + | otherwise = fail "index pair to deep" + +queryHash :: (Monad m) => Trie a -> TrieLocation -> m Fingerprint +queryHash trie = liftM toFingerprint . lookup trie + +querySet :: (Ord a, Monad m) => Trie a -> TrieLocation -> m (Set a) +querySet trie = liftM 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 + ) + +newtype TestDigest = TestDigest { unTestDigest :: T.Text } + deriving (Eq, Ord, Show) + +instance HasDigest TestDigest where + digest = hashSHA256 . TE.encodeUtf8 . unTestDigest + +tests :: H.Test +tests = H.TestList $ + [ H.TestLabel "trieLookup" + $ (Nothing H.~=? (lookup t (TrieLocation { tl_level = -1, tl_index = 0 }))) + , H.TestLabel "trieLookupTooDeep" + $ (Nothing H.~=? (lookup t (TrieLocation { tl_level = 4, tl_index = 0 }))) + ] + where + t = mkTrie 0 $ map (TestDigest . T.pack . show) [0..(13+2*degree*degree)] diff --git a/src/Sync/MerkleTree/Types.hs b/src/Sync/MerkleTree/Types.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Types.hs @@ -0,0 +1,87 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +module Sync.MerkleTree.Types where + +import System.FilePath +import Data.Int +import Crypto.Hash +import Data.Ord +import GHC.Generics +import qualified Data.Text.Encoding as TE +import qualified Data.Text as T +import Sync.MerkleTree.Trie +import qualified Data.Serialize as SE + +-- | Information about a file that we expect to change, when the contents change. +data File + = File + { f_name :: Path + , f_size :: FileSize + , f_modtime :: FileModTime + } + deriving (Eq, Ord, Generic) + +instance SE.Serialize File + +data Entry + = FileEntry File + | DirectoryEntry Path + deriving (Eq, Generic) + +instance SE.Serialize Entry + +newtype FileSize = FileSize { unFileSize :: Int64 } + deriving (Eq, Ord, Generic, Num) + +instance SE.Serialize FileSize + +data FileModTime = FileModTime { unModTime :: !Int64 } + deriving (Eq, Ord, Generic) + +instance SE.Serialize FileModTime + +-- | Representation for paths below the synchronization root directory +data Path + = Root + | Path SerText Path + deriving (Eq, Ord, Generic) + +-- | Returns the string representation of a path +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 + +-- | Entries are sorted first according to their depth in the path which is useful for directory +-- operations +instance Ord Entry where + compare = comparing withLevel + where + withLevel entry = (levelOf entry, toEither entry) + toEither entry = + case entry of + DirectoryEntry p -> Right p + FileEntry f -> Left f + +-- | Return the depth of an entries path +levelOf :: Entry -> Int +levelOf e = + case e of + DirectoryEntry p -> level p + FileEntry f -> level $ f_name f + where + level Root = 0 + level (Path _ p) = 1 + level p + +instance HasDigest Entry where + digest = hash . SE.encode + +data SerText = SerText { unSerText :: !T.Text } + deriving (Ord, Eq) + +instance SE.Serialize SerText where + get = SE.get >>= either (fail . show) (return . SerText) . TE.decodeUtf8' + put = SE.put . TE.encodeUtf8 . unSerText diff --git a/src/Sync/MerkleTree/Util/GetFromInputStream.hs b/src/Sync/MerkleTree/Util/GetFromInputStream.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Util/GetFromInputStream.hs @@ -0,0 +1,23 @@ +module Sync.MerkleTree.Util.GetFromInputStream + ( getFromInputStream + ) where + +import Data.ByteString(ByteString) +import Data.Serialize(Serialize) +import System.IO.Streams(InputStream) +import qualified Data.ByteString as BS +import qualified Data.Serialize as SE +import qualified System.IO.Streams as ST + +-- | Deserialize value from inputstream +getFromInputStream :: (Serialize a) => InputStream ByteString -> IO a +getFromInputStream s = go (SE.Partial $ SE.runGetPartial SE.get) + where + go (SE.Fail err _bs) = 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 diff --git a/src/Sync/MerkleTree/Util/RequestMonad.hs b/src/Sync/MerkleTree/Util/RequestMonad.hs new file mode 100644 --- /dev/null +++ b/src/Sync/MerkleTree/Util/RequestMonad.hs @@ -0,0 +1,175 @@ +-- | 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. +-- +-- Considering the example +-- +-- @ +-- foo = splitRequests [bar, baz] +-- bar = do x <- request (GetSumOf 1 2) +-- liftM Sum request (GetSumOf x 3) +-- baz = liftM Sum request (GetSumOf 4 5) +-- @ +-- +-- running @foo@ in the @RequestMonad@: +-- +-- @ +-- 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 #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +module Sync.MerkleTree.Util.RequestMonad + ( RequestMonad + , request + , runRequestMonad + , splitRequests + ) 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.Streams(InputStream, OutputStream) +import Sync.MerkleTree.Util.GetFromInputStream +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 -> RequestMonadT ByteString b) +data RequestState f b = forall a. (Serialize a) => RequestState f (a -> RequestMonadT ByteString b) +data LiftIOState b = forall a. LiftIOState (IO a) (a -> RequestMonadT ByteString b) + +newtype RequestMonad b = RequestMonad { unReqMonad :: RequestMonadT ByteString b } + deriving (Monad, Functor, Applicative, MonadIO) + +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 :: + (RequestMonadT ByteString a) + -> (a -> RequestMonadT ByteString b) + -> (RequestMonadT ByteString 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 = RequestMonad $ Request $ RequestState (SE.encode x) Return + +-- | Combine results in the monad non-deterministically +-- (it is required that the monoid is commutative) +splitRequests :: (Monoid a) => [RequestMonad a] -> RequestMonad a +splitRequests alts = RequestMonad $ Split $ SplitState (map unReqMonad alts) mempty Return + +data SendQueue + = SendQueue + { sq_chan :: Chan (Maybe ByteString) + , sq_sendIndex :: IORef Int + } + +queueRequests :: SendQueue -> (RequestMonadT ByteString 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 + +-- | Run the provided request monad using the given communication channels +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 (unReqMonad 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 + + +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) $ 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) diff --git a/src/Test.hs b/src/Test.hs new file mode 100644 --- /dev/null +++ b/src/Test.hs @@ -0,0 +1,12 @@ +import qualified Sync.MerkleTree.Sync as S +import qualified Sync.MerkleTree.Test as T +import qualified Sync.MerkleTree.Trie as TR +import Control.Monad +import qualified Test.HUnit as H + +main = + do let allTests = H.TestList $ [T.tests, TR.tests, S.tests] + putStrLn "Running tests:" + forM (H.testCasePaths allTests) $ putStrLn . (" " ++) . show + counts <- H.runTestTT allTests + unless (H.errors counts == 0 && H.failures counts == 0) $ fail $ show counts diff --git a/src/main/hs/Main.hs b/src/main/hs/Main.hs deleted file mode 100644 --- a/src/main/hs/Main.hs +++ /dev/null @@ -1,16 +0,0 @@ -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE OverloadedStrings #-} -import System.Environment -import System.IO.Error -import System.Console.GetOpt - -import Sync.MerkleTree.Run - -main :: IO () -main = flip catchIOError (putError . show) $ - do args <- getArgs - let parsedOpts = getOpt (ReturnInOrder parseNonOption) optDescriptions args - case () of - () | [_HIDDENT_CLIENT_MODE_OPTION_] == args -> runChild - | (options,[],[]) <- parsedOpts -> run $ toSyncOptions options - | (_,_,errs) <- parsedOpts -> printUsageInfo errs diff --git a/src/main/hs/Sync/MerkleTree/Analyse.hs b/src/main/hs/Sync/MerkleTree/Analyse.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Analyse.hs +++ /dev/null @@ -1,63 +0,0 @@ -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE NoMonomorphismRestriction #-} -{-# LANGUAGE OverloadedStrings #-} -module Sync.MerkleTree.Analyse - ( analyseDirectory - , isRealFile - ) where - -import Control.Monad -import Data.Maybe -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 Text.Regex (matchRegex, mkRegex, Regex) -import qualified Data.Text as T - -isRealFile :: String -> Bool -isRealFile x - | x `elem` [".", ".."] = False - | otherwise = True - -shouldIgnore :: Path -> [Regex] -> Bool -shouldIgnore p regexes = any (isJust . (`matchRegex` (toFilePath "" p))) regexes - -analyseDirectory :: FilePath -> [String] -> Path -> IO [Entry] -analyseDirectory fp ignore path = analyseDir fp (map mkRegex ignore) path - -analyseDir :: FilePath -> [Regex] -> Path -> IO [Entry] -analyseDir fp ignore path - | shouldIgnore path ignore = return [] - | otherwise = - do files <- getDirectoryContents fp - liftM concat $ mapM (analyse fp ignore path) $ filter isRealFile files - -analyse :: FilePath -> [Regex] -> Path -> String -> IO [Entry] -analyse fp ignore path name - | shouldIgnore path' 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'):) $ analyseDir fp' ignore path' - | otherwise = return [] -- No support for devices, sockets yet. - diff --git a/src/main/hs/Sync/MerkleTree/Client.hs b/src/main/hs/Sync/MerkleTree/Client.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Client.hs +++ /dev/null @@ -1,137 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -module Sync.MerkleTree.Client where - -import Control.Monad -import Control.Monad.IO.Class -import Codec.Compression.GZip -import Data.Foldable(Foldable) -import Data.Function -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.ByteString.Lazy as BL -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 () - -data SimpleEntry - = FileSimpleEntry Path - | DirectorySimpleEntry Path - deriving (Eq, Ord) - -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) = FileSimpleEntry $ f_name f - name (DirectoryEntry f) = DirectorySimpleEntry 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_ (synchronizeNewOrChangedEntries fp) - $ groupBy ((==) `on` levelOf) - $ sort $ [ e | cs_add cs, e <- newEntries ] ++ [ e | cs_update cs, e <- changedEntries ] - True <- terminateReq - return () - -_CONCURRENT_FILETRANSFER_SIZE_ :: Int -_CONCURRENT_FILETRANSFER_SIZE_ = 48 - -splitEvery :: Int -> [a] -> [[a]] -splitEvery n l - | null l = [] - | (h,t) <- splitAt n l = h:(splitEvery n t) - -synchronizeNewOrChangedEntries :: (ClientMonad m) => FilePath -> [Entry] -> m () -synchronizeNewOrChangedEntries fp entries = - forM_ (splitEvery _CONCURRENT_FILETRANSFER_SIZE_ entries) $ \entryGroup -> - split $ map (synchronizeNewOrChangedEntry fp) entryGroup - -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 -> - do liftIO $ BS.hPut h $ BL.toStrict $ decompress $ BL.fromStrict 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) diff --git a/src/main/hs/Sync/MerkleTree/CommTypes.hs b/src/main/hs/Sync/MerkleTree/CommTypes.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/CommTypes.hs +++ /dev/null @@ -1,74 +0,0 @@ -{-# 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 (Read, Show) - -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 ProtocolVersion - = Version1 - | Version2 - deriving (Read, Show, Eq) - -thisProtocolVersion :: ProtocolVersion -thisProtocolVersion = Version2 - -data LaunchMessage - = LaunchMessage - { lm_protocolVersion :: ProtocolVersion - , lm_dir :: FilePath - , lm_side :: Side - , lm_clientServerOptions :: ClientServerOptions - } - deriving (Read, Show) - -data Side = Server | Client - deriving (Read, Show) diff --git a/src/main/hs/Sync/MerkleTree/Run.hs b/src/main/hs/Sync/MerkleTree/Run.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Run.hs +++ /dev/null @@ -1,179 +0,0 @@ -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE OverloadedStrings #-} -module Sync.MerkleTree.Run where - -import Control.Concurrent -import Control.Monad -import Data.List -import Data.Version (showVersion) -import Paths_sync_mht (version) -import System.Console.GetOpt -import System.IO -import System.Process -import Sync.MerkleTree.CommTypes -import Sync.MerkleTree.Sync -import qualified Data.Text as T -import qualified Data.Text.IO as T - -data RemoteCmd - = RemoteCmd String - | Simulate - -data SyncOptions - = SyncOptions - { so_source :: Maybe FilePath - , so_destination :: Maybe FilePath - , so_remote :: Maybe RemoteCmd - , so_ignore :: [String] - , so_boring :: [FilePath] - , so_add :: Bool - , so_update :: Bool - , so_delete :: Bool - , so_help :: Bool - , so_nonOptions :: [String] - } - -defaultSyncOptions :: SyncOptions -defaultSyncOptions = - SyncOptions - { so_source = Nothing - , so_destination = Nothing - , so_remote = Nothing - , so_ignore = [] - , so_boring = [] - , so_add = False - , so_update = False - , so_delete = False - , so_help = False - , so_nonOptions = [] - } - -toClientServerOptions :: SyncOptions -> IO ClientServerOptions -toClientServerOptions so = - do let parseBoringFile = map T.unpack . filter noComment . map T.strip . T.lines - noComment s = not (T.null s || ("#" `T.isPrefixOf` s)) - ignoreFromBoringFiles <- - forM (so_boring so) $ liftM parseBoringFile . T.readFile - return $ - ClientServerOptions - { cs_add = so_add so - , cs_update = so_update so - , cs_delete = so_delete so - , cs_ignore = (concat ignoreFromBoringFiles) ++ (so_ignore so) - } - -optDescriptions :: [OptDescr (SyncOptions -> SyncOptions)] -optDescriptions = - [ Option ['s'] ["source"] (ReqArg (\fp so -> so { so_source = Just fp }) "DIR") - "source directory" - , Option ['d'] ["destination"] (ReqArg (\fp so -> so { so_destination = Just fp }) "DIR") - "destination directory" - , Option ['r'] ["remote-shell"] (ReqArg (\s so -> so { so_remote = Just $ RemoteCmd s }) "CMD") - "synchroize with a remote-site (see below)" - , Option ['i'] ["ignore"] (ReqArg (\fp so -> so { so_ignore = fp:(so_ignore so) }) "REGEX") - "ignore entries matching the given regex" - , Option ['b'] ["boring"] (ReqArg (\fp so -> so { so_boring = fp:(so_boring so) }) "PATH") - "ignore entries matching the regexes in the given file" - , Option ['a'] ["add"] (NoArg (\so -> so { so_add = True })) - "copy additional files from the source directory" - , Option ['u'] ["update"] (NoArg (\so -> so { so_update = True })) - "overwrite existing files" - , Option [] ["delete"] (NoArg (\so -> so { so_delete = True })) - "delete superfluos files in the destination directory" - , Option ['h'] ["help"] (NoArg (\so -> so { so_help = True })) "shows usage information" - ] - -parseNonOption :: String -> (SyncOptions -> SyncOptions) -parseNonOption s so = so { so_nonOptions = s:(so_nonOptions so) } - -toSyncOptions :: [(SyncOptions -> SyncOptions)] -> SyncOptions -toSyncOptions = foldl (flip id) defaultSyncOptions - -putError :: String -> IO () -putError = hPutStrLn stderr - -_HIDDENT_CLIENT_MODE_OPTION_ :: String -_HIDDENT_CLIENT_MODE_OPTION_ = "--hidden-client-mode-option" - -printUsageInfo :: [String] -> IO () -printUsageInfo prefix = mapM_ putError (prefix ++ [usageInfo header optDescriptions] ++ [details]) - where - header = unlines - [ "Usage: sync-mht [OPTIONS..]" - , "" - , "Fast incremental file transfer using Merkle-Hash-Trees (Version: " - ++ showVersion version - ++ ")" - ] - details = unlines - [ "Note: The argument to the --remote-shell option should be a CMD running sync-mht" - , "with a remote command execution tool (like ssh or docker). If given exactly one of" - , "the directories must be prepended with 'remote:' - indicating a folder on the site," - , "accessible with the provided remote shell command." - ] - -data Location - = Remote FilePath - | Local FilePath - -parseFilePath :: FilePath -> Location -parseFilePath fp - | Just rest <- stripPrefix "remote:" fp = Remote rest - | otherwise = Local fp - -run :: SyncOptions -> IO () -run so - | so_help so = - printUsageInfo [] - | not (null (so_nonOptions so)) = - printUsageInfo ["Unrecognized options: " ++ intercalate ", " (so_nonOptions so)] - | Just source <- so_source so, Just destination <- so_destination so = - do cs <- toClientServerOptions so - case (parseFilePath source, parseFilePath destination) of - (Remote _, Remote _) -> printUsageInfo [doubleRemote] - (Local source', Local destination') - | Just _ <- so_remote so -> printUsageInfo [missingRemote] - | otherwise -> local cs source' destination' - (Remote source', Local destination') - | Just remoteCmd <- so_remote so -> - runParent cs remoteCmd source' destination' FromRemote - | otherwise -> printUsageInfo [missingRemoteCmd] - (Local source', Remote destination') - | Just remoteCmd <- so_remote so -> - runParent cs remoteCmd source' destination' ToRemote - | otherwise -> printUsageInfo [missingRemoteCmd] - | otherwise = - do let missingOpts = - intercalate ", " $ map snd $ filter ((== Nothing) . ($ so) . fst) - [(so_source, "--source"),(so_destination, "--destination")] - printUsageInfo ["The options " ++ missingOpts ++ " are required."] - where - doubleRemote = "Either the directory given in --source or --destination must be local." - missingRemote = concat - [ "The --remote-shell options requires that either the directory given at " - , "--source or --destination is at remote site. (Indicated by the prefix: 'remote:')" - ] - missingRemoteCmd = "The --remote-shell is required when the prefix 'remote:' is used." - -runChild :: IO () -runChild = - do streams <- openStreams stdin stdout - child streams - -runParent :: ClientServerOptions -> RemoteCmd -> FilePath -> FilePath -> Direction -> IO () -runParent clientServerOpts mRemoteCmd source destination dir = - do parentStreams <- - case mRemoteCmd of - RemoteCmd remoteCmd -> - do let remoteCmd' = remoteCmd ++ " " ++ _HIDDENT_CLIENT_MODE_OPTION_ - handles <- - createProcess $ (shell remoteCmd') { std_in = CreatePipe, std_out = CreatePipe } - case handles of - (Just hIn, Just hOut, Nothing, _ph) -> openStreams hOut hIn - _ -> fail "createProcess did not return the correct set of handles." - Simulate -> - do (parentInStream, childOutStream) <- mkChanStreams - (childInStream, parentOutStream) <- mkChanStreams - _ <- forkIO $ child $ StreamPair { sp_in = childInStream, sp_out = childOutStream } - return $ StreamPair { sp_in = parentInStream, sp_out = parentOutStream } - parent parentStreams source destination dir clientServerOpts diff --git a/src/main/hs/Sync/MerkleTree/Server.hs b/src/main/hs/Sync/MerkleTree/Server.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Server.hs +++ /dev/null @@ -1,64 +0,0 @@ -{-# LANGUAGE GADTs #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE TypeSynonymInstances #-} -{-# LANGUAGE FlexibleInstances #-} -module Sync.MerkleTree.Server where - -import Codec.Compression.GZip -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 qualified Data.ByteString.Lazy as BL -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 (BL.toStrict $ compress $ BL.fromStrict bs) $ ContHandle n diff --git a/src/main/hs/Sync/MerkleTree/Sync.hs b/src/main/hs/Sync/MerkleTree/Sync.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Sync.hs +++ /dev/null @@ -1,132 +0,0 @@ -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE NoMonomorphismRestriction #-} -{-# LANGUAGE OverloadedStrings #-} -module Sync.MerkleTree.Sync - ( child - , local - , parent - , openStreams - , mkChanStreams - , StreamPair(..) - , Direction(..) - ) where - -import Control.Concurrent(newChan) -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 qualified System.IO.Streams.Concurrent 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 } - -mkChanStreams :: IO (InputStream ByteString, OutputStream ByteString) -mkChanStreams = - do chan <- newChan - liftM2 (,) (ST.chanToInput chan) (ST.chanToOutput chan) - -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 :: StreamPair -> IO () -child streams = - do launchMessage <- getFromInputStream (sp_in streams) - serverOrClient (read launchMessage) streams - -parent :: StreamPair -> FilePath -> FilePath -> Direction -> ClientServerOptions -> IO () -parent streams source destination direction clientServerOpts = - case direction of - FromRemote -> - do respond (sp_out streams) $ show $ mkLaunchMessage Server source - serverOrClient (mkLaunchMessage Client destination) streams - ToRemote -> - do respond (sp_out streams) $ show $ mkLaunchMessage Client destination - serverOrClient (mkLaunchMessage Server source) streams - where - mkLaunchMessage side dir = - LaunchMessage - { lm_dir = dir - , lm_clientServerOptions = clientServerOpts - , lm_protocolVersion = thisProtocolVersion - , lm_side = side - } - -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) - -serverOrClient :: LaunchMessage -> StreamPair -> IO () -serverOrClient lm streams - | lm_protocolVersion lm == thisProtocolVersion = - let side = - case lm_side lm of - Server -> server - Client -> client (lm_clientServerOptions lm) - in do entries <- analyseDirectory (lm_dir lm) (cs_ignore $ lm_clientServerOptions lm) Root - side entries (lm_dir lm) streams - | otherwise = fail "Incompatible sync-mht versions." - -server :: [Entry] -> FilePath -> StreamPair -> IO () -server entries fp streams = evalStateT loop (startServerState fp $ mkTrie 0 entries) - 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 :: ClientServerOptions -> [Entry] -> FilePath -> StreamPair -> IO () -client cs entries fp streams = - runRequestMonad (sp_in streams) (sp_out streams) $ abstractClient cs fp $ mkTrie 0 entries diff --git a/src/main/hs/Sync/MerkleTree/Test.hs b/src/main/hs/Sync/MerkleTree/Test.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Test.hs +++ /dev/null @@ -1,232 +0,0 @@ -module Sync.MerkleTree.Test where - -import Control.Concurrent -import Control.Monad -import Data.Ix -import Data.List -import Data.Time.Calendar -import Data.Time.Clock -import Sync.MerkleTree.Analyse -import Sync.MerkleTree.Run -import System.Directory -import System.FilePath -import System.IO -import System.IO.Error -import System.IO.Temp -import System.Posix.Files -import System.Random -import qualified Distribution.TestSuite as TS - -mkTestInstance :: String -> IO TS.Result -> TS.TestInstance -mkTestInstance name run = ti - where - ti = - TS.TestInstance - { TS.run = liftM TS.Finished (catchIOError run $ return . TS.Fail . show) - , TS.name = name - , TS.tags = [] - , TS.options = [] - , TS.setOption = setOpt - } - setOpt _ _ = Right ti - -tests :: IO [TS.Test] -tests = return $ - [ TS.testGroup "all" $ map TS.Test $ - [ testOptions ] - ++ testIgnoreBoring - ++ testSync - ] - -testIgnoreBoring :: [TS.TestInstance] -testIgnoreBoring = - let testCase ign bor = withSystemTempDirectory "sync-mht" $ \fp -> - do let srcDir = fp "src" - destDir = fp "dest" - createDirectory srcDir - createDirectory $ srcDir "a" - createDirectory destDir - writeFile (fp ".boring") $ unlines $ - [ "#baz" - , "" - , "foo" - , "^bar" - ] - writeFile (srcDir "added.txt") "testA" - writeFile (srcDir "added-bar.txt") "testB" - writeFile (srcDir "baz") "testC" - writeFile (srcDir "bar-ignore.txt") "testD" - writeFile (srcDir "a" "bar") "testE" - writeFile (srcDir "some-foo.txt") "testF" - writeFile (srcDir "a" "foo") "testG" - run $ - SyncOptions - { so_source = Just $ srcDir - , so_destination = Just $ destDir - , so_remote = Nothing - , so_ignore = ign - , so_boring = map (const $ fp ".boring") bor - , so_add = True - , so_update = True - , so_delete = True - , so_help = False - , so_nonOptions = [] - } - True <- liftM (=="testA") $ readFile (destDir "added.txt") - True <- liftM (=="testB") $ readFile (destDir "added-bar.txt") - True <- liftM (=="testC") $ readFile (destDir "baz") - False <- doesFileExist (destDir "bar-ignore.txt") - True <- liftM (=="testE") $ readFile (destDir "a" "bar") - False <- doesFileExist (destDir "some-foo.txt") - False <- doesFileExist (destDir "a" "foo") - return TS.Pass - in [ mkTestInstance "testIgnore" $ testCase ["foo","^bar"] [] - , mkTestInstance "testBoring" $ testCase [] [()] - ] - -testOptions :: TS.TestInstance -testOptions = mkTestInstance "testOptions" $ - let prepare add update delete go = - withSystemTempDirectory "sync-mht" $ \testDir -> - do let srcDir = testDir "src" - destDir = testDir "dest" - createDirectory srcDir - createDirectory destDir - writeFile (srcDir "same.txt") "test" - copyFile (srcDir "same.txt") (destDir "same.txt") - writeFile (srcDir "changed.txt") "test" - writeFile (destDir "changed.txt") "testB" - writeFile (srcDir "added.txt") "test" - writeFile (destDir "deleted.txt") "testB" - run $ - SyncOptions - { so_source = Just $ srcDir - , so_destination = Just $ destDir - , so_remote = Nothing - , so_ignore = [] - , so_boring = [] - , so_add = add - , so_update = update - , so_delete = delete - , so_help = False - , so_nonOptions = [] - } - True <- liftM (=="test") $ readFile (srcDir "same.txt") - True <- liftM (=="test") $ readFile (srcDir "changed.txt") - True <- liftM (=="test") $ readFile (srcDir "added.txt") - go srcDir destDir - in do prepare True False False $ \srcDir destDir -> - do True <- liftM (=="test") $ readFile (destDir "same.txt") - True <- liftM (=="testB") $ readFile (destDir "changed.txt") - True <- liftM (=="test") $ readFile (destDir "added.txt") - True <- liftM (=="testB") $ readFile (destDir "deleted.txt") - return () - prepare False True False $ \srcDir destDir -> - do True <- liftM (=="test") $ readFile (destDir "same.txt") - True <- liftM (=="test") $ readFile (destDir "changed.txt") - True <- liftM (=="testB") $ readFile (destDir "deleted.txt") - return () - prepare False False True $ \srcDir destDir -> - do True <- liftM (=="test") $ readFile (destDir "same.txt") - True <- liftM (=="testB") $ readFile (destDir "changed.txt") - False <- doesFileExist (destDir "deleted.txt") - return () - return TS.Pass - -testSync :: [TS.TestInstance] -testSync = - flip map [0,1,2] $ \simulate -> mkTestInstance ("testSync"++ show simulate) $ - do forM [1..10] $ \_ -> - withSystemTempDirectory "sync-mht" $ \testDir -> - do mkRandomDir 3 [testDir "src", testDir "src-backup"] - mkRandomDir 3 [testDir "target"] - let sourcePrefix - | simulate == 1 = "remote:" - | otherwise = "" - targetPrefix - | simulate == 2 = "remote:" - | otherwise = "" - run $ - SyncOptions - { so_source = Just $ (sourcePrefix ++) testDir "src" - , so_destination = Just $ (targetPrefix ++) testDir "target" - , so_remote = - case simulate of - 0 -> Nothing - _ -> Just Simulate - , so_ignore = [] - , so_boring = [] - , so_add = True - , so_update = True - , so_delete = True - , so_help = False - , so_nonOptions = [] - } - areDirsEqual (testDir "src") (testDir "target") - areDirsEqual (testDir "src") (testDir "src-backup") - return TS.Pass - -data FileOrDir - = File - | Dir - deriving (Eq, Ix, Enum, Bounded, Ord, Show) - -instance Random FileOrDir where - random = randomR (minBound, maxBound) - randomR (l, u) g = - let (i, g) = randomR (fromEnum l, fromEnum u) g - in (toEnum i, g) - -utcTimeFrom :: Integer -> UTCTime -utcTimeFrom x = UTCTime (fromGregorian 2015 07 29) (fromInteger x) - -mkRandomDir :: Integer -> [FilePath] -> IO () -mkRandomDir md fps = - do forM fps createDirectory - names <- distinctNames 6 - forM_ names $ \n -> - do m <- randomRIO (0,1) - case toEnum m of - Dir | md > 0 -> mkRandomDir (md - 1) $ map ( n) fps - | otherwise -> return () - File -> - do d <- randomRIO (1,4) - forM_ fps $ \fp -> - do writeFile (fp n) (show d) - setModificationTime (fp n) (utcTimeFrom d) - - -areDirsEqual :: FilePath -> FilePath -> IO () -areDirsEqual fp1 fp2 = - do files1 <- liftM (sort . filter isRealFile) $ getDirectoryContents fp1 - files2 <- liftM (sort . filter isRealFile) $ getDirectoryContents fp2 - case () of - () | files1 == files2 -> forM_ files1 $ \f -> areEntriesEqual (fp1 f) (fp2 f) - | otherwise -> fail $ "Unequal: " ++ show (fp1, fp2, files1, files2) - -areEntriesEqual :: FilePath -> FilePath -> IO () -areEntriesEqual f1 f2 = - do s1 <- getFileStatus f1 - s2 <- getFileStatus f2 - case () of - () | isDirectory s1, isDirectory s2 -> areDirsEqual f1 f2 - | isRegularFile s1, isRegularFile s2 -> - do c1 <- readFile f1 - c2 <- readFile f2 - unless (c1 == c2) $fail $ "Unequal files: " ++ show (f1, f2, c1, c2) - | otherwise -> - fail $ show - (f1, f2, isDirectory s1, isDirectory s2, isRegularFile s1, isRegularFile s2) - --- | @distinctNames k@ creates k distinct file names -distinctNames :: Integer -> IO [String] -distinctNames k = retry - where - isDistinct names = null $ filter hasDuplicate $ group $ sort names - hasDuplicate = (/= 1) . length - mkName _ = liftM (:[]) $ randomRIO ('a','m') - retry = - do result <- forM [1..k] mkName - if isDistinct result - then return result - else retry diff --git a/src/main/hs/Sync/MerkleTree/Trie.hs b/src/main/hs/Sync/MerkleTree/Trie.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Trie.hs +++ /dev/null @@ -1,154 +0,0 @@ -{-# 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 - ) diff --git a/src/main/hs/Sync/MerkleTree/Types.hs b/src/main/hs/Sync/MerkleTree/Types.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Types.hs +++ /dev/null @@ -1,86 +0,0 @@ -{-# 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 - withLevel entry = (levelOf entry, toEither entry) - toEither entry = - case entry of - DirectoryEntry p -> Right p - FileEntry f -> Left f - -levelOf :: Entry -> Int -levelOf e = - case e of - DirectoryEntry p -> level p - FileEntry f -> level $ f_name f - where - level Root = 0 - level (Path _ p) = 1 + level p - -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 diff --git a/src/main/hs/Sync/MerkleTree/Util/RequestMonad.hs b/src/main/hs/Sync/MerkleTree/Util/RequestMonad.hs deleted file mode 100644 --- a/src/main/hs/Sync/MerkleTree/Util/RequestMonad.hs +++ /dev/null @@ -1,179 +0,0 @@ --- | 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.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) $ 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) - - diff --git a/sync-mht.cabal b/sync-mht.cabal --- a/sync-mht.cabal +++ b/sync-mht.cabal @@ -6,9 +6,9 @@ maintainer: Emin Karayel category: Utility extra-doc-files: README.md -cabal-version: >= 1.22 +cabal-version: >= 1.18 build-type: Simple -version: 0.3.5.0 +version: 0.3.6.0 homepage: https://github.com/ekarayel/sync-mht bug-reports: https://github.com/ekarayel/sync-mht/issues package-url: https://github.com/ekarayel/sync-mht @@ -29,11 +29,11 @@ location: https://github.com/ekarayel/sync-mht --recursive source-repository this type: git - tag: 0.3.5.0 + tag: 0.3.6.0 location: https://github.com/ekarayel/sync-mht --recursive test-suite main - type: detailed-0.9 - test-module: Sync.MerkleTree.Test + type: exitcode-stdio-1.0 + main-is: Test.hs other-modules: Sync.MerkleTree.Analyse Sync.MerkleTree.Client @@ -44,33 +44,34 @@ Sync.MerkleTree.Trie Sync.MerkleTree.Types Sync.MerkleTree.Util.RequestMonad - Paths_sync_mht - hs-source-dirs: src/main/hs - default-language: Haskell2010 + Sync.MerkleTree.Util.GetFromInputStream + Sync.MerkleTree.Test build-depends: - base >=4.7 && <4.8 + base >=4.7 && <4.9 , unix >=2.7 && <2.8 , directory >=1.2.3 && <1.3 - , filepath >=1.3 && <1.4 + , filepath >=1.3 && <1.5 , process == 1.2.3.0 , cryptohash >=0.11 && <0.12 - , exceptions >=0.7 && <0.8 + , exceptions >=0.7 && <0.9 , byteable >=0.1 && <0.2 , array >=0.5 && <0.6 , containers >=0.5 && <0.6 - , text >=1.1 && <1.2 + , text >=1.1 && <1.3 , bytestring >=0.10 && <0.11 , base16-bytestring >=0.1 && <0.2 , cereal >= 0.4 && < 0.5 - , io-streams >= 1.2 && <1.3 - , transformers >= 0.3 && < 0.4 + , io-streams >= 1.2 && <1.4 + , transformers >= 0.3 && < 0.5 , regex-compat >= 0.95 && < 0.96 - , mtl >= 2.1 && < 2.2 - , zlib >= 0.6 && < 0.7 - , Cabal >= 1.22 && < 1.23 - , time >= 1.4 && < 1.5 - , random >= 1.0 && < 1.1 + , mtl >= 2.1 && < 2.3 + , zlib >= 0.5 && < 0.7 + , time >= 1.4 && < 1.6 + , random >= 1.0 && < 1.2 + , HUnit >= 1.2 && < 1.3 , temporary >= 1.2 && < 1.3 + hs-source-dirs: src + default-language: Haskell2010 executable sync-mht main-is: Main.hs other-modules: @@ -83,30 +84,70 @@ Sync.MerkleTree.Trie Sync.MerkleTree.Types Sync.MerkleTree.Util.RequestMonad + Sync.MerkleTree.Util.GetFromInputStream ghc-options: -Wall build-depends: - base >=4.7 && <4.8 + base >=4.7 && <4.9 , unix >=2.7 && <2.8 , directory >=1.2.3 && <1.3 - , filepath >=1.3 && <1.4 + , filepath >=1.3 && <1.5 , process == 1.2.3.0 , cryptohash >=0.11 && <0.12 - , exceptions >=0.7 && <0.8 + , exceptions >=0.7 && <0.9 , byteable >=0.1 && <0.2 , array >=0.5 && <0.6 , containers >=0.5 && <0.6 - , text >=1.1 && <1.2 + , text >=1.1 && <1.3 , bytestring >=0.10 && <0.11 , base16-bytestring >=0.1 && <0.2 , cereal >= 0.4 && < 0.5 - , io-streams >= 1.2 && <1.3 - , transformers >= 0.3 && < 0.4 + , io-streams >= 1.2 && <1.4 + , transformers >= 0.3 && < 0.5 , regex-compat >= 0.95 && < 0.96 - , mtl >= 2.1 && < 2.2 - , zlib >= 0.6 && < 0.7 - , Cabal >= 1.22 && < 1.23 - , time >= 1.4 && < 1.5 - , random >= 1.0 && < 1.1 + , mtl >= 2.1 && < 2.3 + , zlib >= 0.5 && < 0.7 + , time >= 1.4 && < 1.6 + , random >= 1.0 && < 1.2 + , HUnit >= 1.2 && < 1.3 , temporary >= 1.2 && < 1.3 - hs-source-dirs: src/main/hs + hs-source-dirs: src + default-language: Haskell2010 +library + exposed-modules: + Sync.MerkleTree.Analyse + Sync.MerkleTree.Client + Sync.MerkleTree.CommTypes + Sync.MerkleTree.Server + Sync.MerkleTree.Sync + Sync.MerkleTree.Run + Sync.MerkleTree.Trie + Sync.MerkleTree.Types + Sync.MerkleTree.Util.RequestMonad + Sync.MerkleTree.Util.GetFromInputStream + ghc-options: -Wall + build-depends: + base >=4.7 && <4.9 + , unix >=2.7 && <2.8 + , directory >=1.2.3 && <1.3 + , filepath >=1.3 && <1.5 + , process == 1.2.3.0 + , cryptohash >=0.11 && <0.12 + , exceptions >=0.7 && <0.9 + , byteable >=0.1 && <0.2 + , array >=0.5 && <0.6 + , containers >=0.5 && <0.6 + , text >=1.1 && <1.3 + , bytestring >=0.10 && <0.11 + , base16-bytestring >=0.1 && <0.2 + , cereal >= 0.4 && < 0.5 + , io-streams >= 1.2 && <1.4 + , transformers >= 0.3 && < 0.5 + , regex-compat >= 0.95 && < 0.96 + , mtl >= 2.1 && < 2.3 + , zlib >= 0.5 && < 0.7 + , time >= 1.4 && < 1.6 + , random >= 1.0 && < 1.2 + , HUnit >= 1.2 && < 1.3 + , temporary >= 1.2 && < 1.3 + hs-source-dirs: src default-language: Haskell2010