diff --git a/HaskellTorrent.cabal b/HaskellTorrent.cabal --- a/HaskellTorrent.cabal +++ b/HaskellTorrent.cabal @@ -1,6 +1,6 @@ name: HaskellTorrent category: Network -version: 0.0 +version: 0.1 category: Network description: HaskellTorrent provides a BitTorrent client, based on the CML library for concurrency. This is an early preview release which is capable of @@ -28,6 +28,10 @@ description: Enable debug support default: False +flag threadscope + description: Enable the eventlog necessary for ThreadScope + default: False + executable HaskellTorrent hs-source-dirs: src main-is: HaskellTorrent.hs @@ -36,7 +40,7 @@ Process.ChokeMgr, Process.Console, Process.FS, Process.Listen, Process.PeerMgr, Process.Peer, Process.PieceMgr, Process.Status, Process.Timer, Process.Tracker, - Digest, FS, Logging, LoggingTypes, PeerTypes, Process, RateCalc, + Digest, FS, PeerTypes, Process, RateCalc, Supervisor, Torrent extensions: CPP @@ -54,16 +58,23 @@ network, cml, HTTP, - HsOpenSSL, + hopenssl, time, random-shuffle, + QuickCheck >= 2, + HUnit, + test-framework, + test-framework-quickcheck2, + test-framework-hunit, + hslogger, directory - ghc-options: -threaded -fwarn-unused-imports - if flag(debug) - cpp-options: "-DDEBUG" - else + ghc-options: -threaded -fwarn-unused-imports -fwarn-unrecognised-pragmas -fwarn-warnings-deprecations + if !flag(debug) cpp-options: "-DNDEBUG" + + if flag(threadscope) + ghc-options: -eventlog source-repository head type: git diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2009, Jesper Louis Andersen +Copyright (c) 2009, 2010, Jesper Louis Andersen All rights reserved. Redistribution and use in source and binary forms, with or without diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -14,7 +14,8 @@ cabal install --prefix=$HOME --user -Since we are using the magnificient cabal, this is enough to install haskell torrent in our $HOME/bin directory. +Since we are using the magnificient cabal, this is enough to install haskell +torrent in our $HOME/bin directory. Usage ----------------- @@ -23,7 +24,9 @@ HaskellTorrent foo.torrent -then it will begin downloading the file in foo.torrent to the current directory via the Bittorrent protocol. *Note:* Currently we have no support for multifile torrents. +then it will begin downloading the file in foo.torrent to the current +directory via the Bittorrent protocol. *Note:* Currently we have no support +for multifile torrents. Protocol support ---------------- @@ -43,6 +46,59 @@ - 016, 017, 019 +Debugging +--------- + +For debugging, jlouis tends to use the following: + + make conf build test + +This builds HaskellTorrent with the *Debug* flag set and also builds the software with profiling +by default so it is easy to hunt down performance regressions. It also runs the internal test-suite +for various values. + +Reading material for hacking HaskellTorrent: +-------------------------------------------- + + - [Protocol specification - BEP0003](http://www.bittorrent.org/beps/bep_0003.html): + This is the original protocol specification, tracked into the BEP + process. It is worth reading because it explains the general overview + and the precision with which the original protocol was written down. + + - [Bittorrent Enhancement Process - BEP0000](http://www.bittorrent.org/beps/bep_0000.html) + The BEP process is an official process for adding extensions on top of + the BitTorrent protocol. It allows implementors to mix and match the + extensions making sense for their client and it allows people to + discuss extensions publicly in a forum. It also provisions for the + deprecation of certain features in the long run as they prove to be of + less value. + + - [wiki.theory.org](http://wiki.theory.org/Main_Page) + An alternative description of the protocol. This description is in + general much more detailed than the BEP structure. It is worth a read + because it acts somewhat as a historic remark and a side channel. Note + that there are some commentary on these pages which can be disputed + quite a lot. + + - ["Concurrent Programming in ML" - John. H. Reppy](http://www.amazon.com/Concurrent-Programming-ML-John-Reppy/dp/0521480892) + This book describes the CML system which HaskellTorrent uses. The basic + idea of CML is that of higher order events. That is, events which can + be abstractly manipulated before being synchronized on. For instance, + CML events are a Functor instance, though the code currently does not + reflect that. This also provisions for fair selective receives + with dynamic choice of which events may fire. + + - ["A Concurrent ML library in Concurrent Haskell" - Avik Chaudhuri, ICFP 2009](http://www.cs.umd.edu/~avik/projects/cmllch/) + The paper behind the CML library which HaskellTorrent uses. It + describes a way to transform the first-order MVar and ForkIO structures + of concurrent haskell into a higher order CML language by use of a + clever matchmaking algorithm. + + - ["Supervisor Behaviour"](http://erlang.org/doc/design_principles/sup_princ.html) + From the Erlang documentation. How the Erlang Supervisor behaviour + works. The Supervisor and process structure of HaskellTorrent is + somewhat inspired by the Erlang ditto. + Source code Hierarchy --------------------- @@ -77,16 +133,17 @@ - **PeerTypes**: Types used by peers. - **Process**: Code for Erlang-inspired processes. - **RateCalc**: Rate calculations for a network socket. We use this to keep track of the - current speed of a peer in one direction. + current speed of a peer in one direction. - **Supervisor**: Erlang-inspired Supervisor processes. - **Torrent**: Various helpers and types for Torrents. - **Version.hs.in**: Generates **Version.hs** via the configure script. + - **Test.hs**: Code for test-framework + - **TestInstance.hs**: Various helper instances not present in the test framework by default Known bugs ---------- - - I have seen the system violating an assertion in the Piece Manager. I think I got it, - nailed, but it may show up again. - - + "PieceMgrP"(Fatal): Process exiting due to ex: user error (P/Blk (655,Block {blockOffset = 81920, blockSize = 16384}) is in the HaveBlocks set) + "ConsoleP"(Info): Process Terminated by Supervisor +# vim: filetype=none tw=76 expandtab diff --git a/Setup.lhs b/Setup.lhs --- a/Setup.lhs +++ b/Setup.lhs @@ -3,5 +3,21 @@ In principle, we could do with a lot less than autoconfUserhooks, but simpleUserHooks is not running 'configure'. ->import Distribution.Simple ->main = defaultMainWithHooks autoconfUserHooks +We will need Distribution.Simple to do the heavyweight lifting, and we will need some filePath magic. + +> import System.FilePath +> import System.Process +> import Distribution.Simple +> import Distribution.Simple.LocalBuildInfo +> import Distribution.PackageDescription + +The main program is just to make Cabal lift it. But we will override testing. + +> main = defaultMainWithHooks hooks +> where hooks = autoconfUserHooks { runTests = runTests' } + +Running tests is to call HaskellTorrent with its parameters for tests: + +> runTests' :: Args -> Bool -> PackageDescription -> LocalBuildInfo -> IO () +> runTests' _ _ _ lbi = system testprog >> return () +> where testprog = (buildDir lbi) "HaskellTorrent" "HaskellTorrent --tests" diff --git a/configure b/configure --- a/configure +++ b/configure @@ -3,7 +3,7 @@ # Simple configuration script. ## Git version -GITHEAD="0.0 Hackage" +GITHEAD="0.1 Hackage" if test -d "${GIT_DIR:-.git}" ; then GITHEAD=`git describe 2>/dev/null` diff --git a/src/Digest.hs b/src/Digest.hs --- a/src/Digest.hs +++ b/src/Digest.hs @@ -5,28 +5,34 @@ ) where -import Control.Concurrent -import Control.Monad - -import Data.Maybe +import Data.Char +import Data.Word +import Control.Monad.State +import Foreign.Ptr +import qualified Data.ByteString as B +import Data.ByteString.Unsafe import qualified Data.ByteString.Lazy as L -import OpenSSL -import qualified OpenSSL.EVP.Digest as D +import qualified OpenSSL.Digest as SSL -- Consider newtyping this type Digest = String -sha1Digest :: IO D.Digest -sha1Digest = do dig <- D.getDigestByName "SHA1" - case dig of - Nothing -> fail "No such digest, SHA1" - Just d -> return d - digest :: L.ByteString -> IO Digest -digest bs = - runInBoundThread $ -- I don't think this is needed strictly - withOpenSSL $ - do sha1 <- sha1Digest - return $ D.digestLBS sha1 bs +digest bs = do + upack <- digestLBS SSL.SHA1 bs + return $ map (chr . fromIntegral) upack +digestLBS :: SSL.MessageDigest -> L.ByteString -> IO [Word8] +digestLBS mdType xs = + SSL.mkDigest mdType $ evalStateT (updateLBS xs >> SSL.final) + +updateBS :: B.ByteString -> SSL.Digest () +updateBS bs = do + SSL.DST ctx <- get + liftIO $ unsafeUseAsCStringLen bs $ + \(ptr, len) -> SSL.digestUpdate ctx (castPtr ptr) (fromIntegral len) + return () + +updateLBS :: L.ByteString -> SSL.Digest () +updateLBS lbs = mapM_ updateBS $ L.toChunks lbs diff --git a/src/FS.hs b/src/FS.hs --- a/src/FS.hs +++ b/src/FS.hs @@ -1,29 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -- | Filesystem routines. These are used for working with and -- manipulating files in the filesystem. module FS (PieceInfo(..), @@ -44,7 +18,7 @@ import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as L import qualified Data.Map as M -import Data.Maybe + import System.IO import System.Directory (createDirectoryIfMissing) import Data.List (intercalate) @@ -179,10 +153,10 @@ where fetchData = do pLen <- infoPieceLength bc pieceData <- infoPieces bc tLen <- infoLength bc - let pm = M.fromList . zip [0..] . extract pLen tLen 0 $ pieceData - when ( tLen /= (sum $ map len $ M.elems pm) ) - (error "PieceMap construction size assertion failed") - return pm + let pm = M.fromList . zip [0..] . extract pLen tLen 0 $ pieceData + when ( tLen /= (sum $ map len $ M.elems pm) ) + (error "PieceMap construction size assertion failed") + return pm extract :: Integer -> Integer -> Integer -> [B.ByteString] -> [PieceInfo] extract _ 0 _ [] = [] extract plen tlen offst (p : ps) | tlen < plen = PieceInfo { offset = offst, diff --git a/src/HaskellTorrent.hs b/src/HaskellTorrent.hs --- a/src/HaskellTorrent.hs +++ b/src/HaskellTorrent.hs @@ -1,14 +1,20 @@ module Main (main) where +import Control.Concurrent import Control.Concurrent.CML import Control.Monad import qualified Data.ByteString as B +import Data.List import System.Environment import System.Random +import System.Console.GetOpt +import System.Log.Logger +import System.Log.Handler.Simple +import System.IO as SIO import qualified Protocol.BCode as BCode import qualified Process.Console as Console @@ -18,68 +24,113 @@ import qualified Process.ChokeMgr as ChokeMgr (start) import qualified Process.Status as Status import qualified Process.Tracker as Tracker -import Logging +import qualified Process.Listen as Listen import FS import Supervisor import Torrent import Version +import qualified Test main :: IO () -main = getArgs >>= run +main = do args <- getArgs + if "--tests" `elem` args + then Test.runTests + else progOpts args >>= run +-- COMMAND LINE PARSING -run :: [String] -> IO () -run args = do - putStrLn $ "This is Haskell-torrent version " ++ version - case args of - [] -> putStrLn "*** Usage: haskellTorrent " - (name:_) -> download name +data Flag = Version | Debug | LogFile String + deriving (Eq, Show) -download :: String -> IO () -download name = do +options :: [OptDescr Flag] +options = + [ Option ['V','?'] ["version"] (NoArg Version) "Show version number" + , Option ['D'] ["debug"] (NoArg Debug) "Spew extra debug information" + , Option [] ["logfile"] (ReqArg LogFile "FILE") "Choose a filepath on which to log" + ] + +progOpts :: [String] -> IO ([Flag], [String]) +progOpts args = do + case getOpt Permute options args of + (o,n,[] ) -> return (o, n) + (_,_,errs) -> ioError (userError (concat errs ++ usageInfo header options)) + where header = "Usage: HaskellTorrent [OPTION...] file" + +run :: ([Flag], [String]) -> IO () +run (flags, files) = do + if Version `elem` flags + then progHeader + else case files of + [] -> putStrLn "No torrentfile input" + [name] -> progHeader >> download flags name + _ -> putStrLn "More than one torrent file given" + +progHeader :: IO () +progHeader = putStrLn $ "This is Haskell-torrent version " ++ version ++ "\n" ++ + " For help type 'help'\n" + +setupLogging :: [Flag] -> IO () +setupLogging flags = do + rootL <- getRootLogger + fLog <- case logFlag flags of + Nothing -> streamHandler SIO.stdout DEBUG + Just (LogFile fp) -> fileHandler fp DEBUG + when (Debug `elem` flags) + (updateGlobalLogger rootLoggerName + (setHandlers [fLog] . (setLevel DEBUG))) + where logFlag = find (\e -> case e of + LogFile _ -> True + _ -> False) + +download :: [Flag] -> String -> IO () +download flags name = do torrent <- B.readFile name let bcoded = BCode.decode torrent case bcoded of Left pe -> print pe - Right bc -> - do print bc - (handles, haveMap, pieceMap) <- openAndCheckFile bc - logC <- channel - Logging.startLogger logC + Right bc -> do + setupLogging flags + debugM "Main" (show bc) + (handles, haveMap, pieceMap) <- openAndCheckFile bc -- setup channels trackerC <- channel statusC <- channel waitC <- channel pieceMgrC <- channel - supC <- channel - fspC <- channel + supC <- channel + fspC <- channel statInC <- channel pmC <- channel - chokeC <- channel - chokeInfoC <- channel - putStrLn "Created channels" - -- setup StdGen and Peer data + chokeC <- channel + chokeInfoC <- channel + debugM "Main" "Created channels" + -- setup StdGen and Peer data gen <- getStdGen - ti <- mkTorrentInfo bc + ti <- mkTorrentInfo bc let pid = mkPeerId gen - left = bytesLeft haveMap pieceMap - clientState = determineState haveMap - -- Create main supervisor process - allForOne "MainSup" - [ Worker $ Console.start logC waitC - , Worker $ FSP.start handles logC pieceMap fspC - , Worker $ PeerMgr.start pmC pid (infoHash ti) - pieceMap pieceMgrC fspC logC chokeC statInC (pieceCount ti) - , Worker $ PieceMgr.start logC pieceMgrC fspC chokeInfoC statInC - (PieceMgr.createPieceDb haveMap pieceMap) - , Worker $ Status.start logC left clientState statusC statInC trackerC - , Worker $ Tracker.start ti pid defaultPort logC statusC statInC - trackerC pmC - , Worker $ ChokeMgr.start logC chokeC chokeInfoC 100 -- 100 is upload rate in KB - (case clientState of - Seeding -> True - Leeching -> False) - ] logC supC - sync $ transmit trackerC Status.Start + left = bytesLeft haveMap pieceMap + clientState = determineState haveMap + -- Create main supervisor process + tid <- allForOne "MainSup" + [ Worker $ Console.start waitC statusC + , Worker $ FSP.start handles pieceMap fspC + , Worker $ PeerMgr.start pmC pid (infoHash ti) + pieceMap pieceMgrC fspC chokeC statInC (pieceCount ti) + , Worker $ PieceMgr.start pieceMgrC fspC chokeInfoC statInC + (PieceMgr.createPieceDb haveMap pieceMap) + , Worker $ Status.start left clientState statusC statInC trackerC + , Worker $ Tracker.start ti pid defaultPort statusC statInC + trackerC pmC + , Worker $ ChokeMgr.start chokeC chokeInfoC 100 -- 100 is upload rate in KB + (case clientState of + Seeding -> True + Leeching -> False) + , Worker $ Listen.start defaultPort pmC + ] supC + sync $ transmit trackerC Status.Start sync $ receive waitC (const True) + infoM "Main" "Closing down, giving processes 10 seconds to cool off" + sync $ transmit supC (PleaseDie tid) + threadDelay $ 10*1000000 + infoM "Main" "Done..." return () diff --git a/src/Logging.hs b/src/Logging.hs deleted file mode 100644 --- a/src/Logging.hs +++ /dev/null @@ -1,40 +0,0 @@ --- | Logging primitives -{-# LANGUAGE TypeSynonymInstances #-} -module Logging ( - -- * Classes - Logging(..) - -- * Types - , LogChannel - , LogPriority(..) - -- * Spawning - , startLogger - -- * Interface (deprecated) - , logMsg - , logMsg' - ) -where - -import Control.Concurrent -import Control.Concurrent.CML -import Control.Monad.Reader - -import LoggingTypes -import Prelude hiding (log) - -import Process - -startLogger :: LogChannel -> IO ThreadId -startLogger logC = - spawnP logC () (forever lp) - where - lp = do m <- syncP =<< logEv - liftIO $ print m - logEv = recvP logC (const True) - --- | Log a message to a channel -logMsg :: LogChannel -> String -> IO () -logMsg c m = logMsg' c "Unknown" Info m - --- | Log a message to a channel with a priority -logMsg' :: LogChannel -> String -> LogPriority -> String -> IO () -logMsg' c name pri = sync . transmit c . Mes pri name diff --git a/src/LoggingTypes.hs b/src/LoggingTypes.hs deleted file mode 100644 --- a/src/LoggingTypes.hs +++ /dev/null @@ -1,36 +0,0 @@ -{-# LANGUAGE TypeSynonymInstances #-} -module LoggingTypes - ( Logging(..) - , LogPriority(..) - , LogMsg(..) - , LogChannel - ) -where - -import Control.Concurrent.CML - - -data LogPriority = Debug -- ^ Fine grained debug info - | Warn -- ^ Potentially harmful situations - | Info -- ^ Informational messages, progress reports - | Error -- ^ Errors that are continuable - | Fatal -- ^ Severe errors. Will probably make the application abort - | None -- ^ No logging at all - deriving (Show, Eq, Ord) --- --- | The class of types where we have a logger inside them somewhere -class Logging a where - -- | Returns a channel for logging and an Identifying string to use - getLogger :: a -> (String, LogChannel) - -instance Logging LogChannel where - getLogger ch = ("Unknown", ch) - - --- TODO: Consider generalizing this to any member of Show -data LogMsg = Mes LogPriority String String - -instance Show LogMsg where - show (Mes pri name str) = show name ++ "(" ++ show pri ++ "):\t" ++ str - -type LogChannel = Channel LogMsg diff --git a/src/PeerTypes.hs b/src/PeerTypes.hs --- a/src/PeerTypes.hs +++ b/src/PeerTypes.hs @@ -1,4 +1,11 @@ module PeerTypes + ( Peer(..) + , PeerMessage(..) + , PeerChannel + , MgrMessage(..) + , MgrChannel + , BandwidthChannel + ) where import Control.Concurrent @@ -15,8 +22,8 @@ data PeerMessage = ChokePeer | UnchokePeer | PeerStats UTCTime (Channel (Double, Double, Bool)) -- Up/Down/Interested - | PieceCompleted PieceNum - | CancelBlock PieceNum Block + | PieceCompleted PieceNum + | CancelBlock PieceNum Block type PeerChannel = Channel PeerMessage diff --git a/src/Process.hs b/src/Process.hs --- a/src/Process.hs +++ b/src/Process.hs @@ -25,16 +25,16 @@ , stopP , ignoreProcessBlock -- This ought to be renamed -- * Log Interface - , logInfo - , logDebug - , logWarn - , logFatal - , logError + , Logging(..) + , logP + , infoP + , debugP + , warningP + , criticalP + , errorP ) where -import Data.Monoid - import Control.Concurrent import Control.Concurrent.CML import Control.Exception @@ -44,13 +44,9 @@ import Data.Typeable -import LoggingTypes import Prelude hiding (catch, log) -import System.IO - --- import Supervisor - +import System.Log.Logger -- | A @Process a b c@ is the type of processes with access to configuration data @a@, state @b@ -- returning values of type @c@. Usually, the read-only data are configuration parameters and @@ -92,16 +88,16 @@ st <- get c <- ask (a, s') <- liftIO $ runP c st proc `catches` - [ Handler (\ThreadKilled -> do - runP c st ( do logInfo $ "Process Terminated by Supervisor" - cleanupH )) - , Handler (\StopException -> - runP c st (do logInfo $ "Process Terminating gracefully" - cleanupH >> stopH)) -- This one is ok - , Handler (\(ex :: SomeException) -> - runP c st (do logFatal $ "Process exiting due to ex: " ++ show ex - cleanupH >> stopH)) - ] + [ Handler (\ThreadKilled -> do + runP c st ( do infoP $ "Process Terminated by Supervisor" + cleanupH )) + , Handler (\StopException -> + runP c st (do infoP $ "Process Terminating gracefully" + cleanupH >> stopH)) -- This one is ok + , Handler (\(ex :: SomeException) -> + runP c st (do criticalP $ "Process exiting due to ex: " ++ show ex + cleanupH >> stopH)) + ] put s' return a @@ -111,14 +107,14 @@ syncP :: Event (c, b) -> Process a b c syncP ev = do (a, s) <- liftIO $ sync ev - put s - return a + put s + return a sendP :: Channel c -> c -> Process a b (Event ((), b)) sendP ch v = do s <- get return $ (wrap (transmit ch v) - (\() -> return ((), s))) + (\() -> return ((), s))) sendPC :: (a -> Channel c) -> c -> Process a b (Event ((), b)) sendPC sel v = asks sel >>= flip sendP v @@ -127,7 +123,7 @@ recvP ch pred = do s <- get return (wrap (receive ch pred) - (\v -> return (v, s))) + (\v -> return (v, s))) recvPC :: (a -> Channel c) -> Process a b (Event (c, b)) recvPC sel = asks sel >>= flip recvP (const True) @@ -159,7 +155,7 @@ #if (__GLASGOW_HASKELL__ == 610) [ Handler (\BlockedOnDeadMVar -> return (err, st)) ] #elif (__GLASGOW_HASKELL__ == 612) - [ Handler (\BlockedIndefinitelyOnMVar -> return (err, st)) ] + [ Handler (\BlockedIndefinitelyOnMVar -> return (err, st)) ] #else #error Unknown GHC version #endif @@ -168,44 +164,21 @@ ------ LOGGING --- | If a process has access to a logging channel, it is able to log messages to the world -log :: Logging a => LogPriority -> String -> Process a b () -log prio msg = do - (name, logC) <- asks getLogger - when (prio >= logLevel name) - (liftIO $ logMsg' logC name prio msg) - where logMsg' c name pri = sync . transmit c . Mes pri name - -logInfo, logDebug, logFatal, logWarn, logError :: Logging a => String -> Process a b () -logInfo = log Info -logDebug = log Debug -logFatal = log Fatal -logWarn = log Warn -logError = log Error - --- Logging filters -type LogFilter = String -> LogPriority - -matchP :: String -> LogPriority -> LogFilter -matchP process prio = \s -> if s == process then prio else None - -matchAny :: LogPriority -> LogFilter -matchAny prio = const prio - -matchNone :: LogFilter -matchNone = const None +-- +-- | The class of types where we have a logger inside them somewhere +class Logging a where + -- | Returns a channel for logging and an Identifying string to use + logName :: a -> String -instance Monoid LogPriority where - mempty = None - mappend None g = g - mappend f g = f +logP :: Logging a => Priority -> String -> Process a b () +logP prio msg = do + n <- asks logName + liftIO $ logM n prio msg --- | The level by which we log -logLevel :: LogFilter -#ifdef DEBUG -logLevel = mconcat [matchP "TrackerP" Debug, - matchAny Info] -#else -logLevel = matchAny Info -#endif +infoP, debugP, criticalP, warningP, errorP :: Logging a => String -> Process a b () +infoP = logP INFO +debugP = logP DEBUG +criticalP = logP CRITICAL +warningP = logP WARNING +errorP = logP ERROR diff --git a/src/Process/ChokeMgr.hs b/src/Process/ChokeMgr.hs --- a/src/Process/ChokeMgr.hs +++ b/src/Process/ChokeMgr.hs @@ -26,7 +26,6 @@ import PeerTypes import Process.PieceMgr hiding (start) import Process -import Logging import Supervisor import Torrent hiding (infoHash) import Process.Timer as Timer @@ -35,60 +34,59 @@ ---------------------------------------------------------------------- data ChokeMgrMsg = Tick - | RemovePeer PeerPid - | AddPeer PeerPid PeerChannel + | RemovePeer PeerPid + | AddPeer PeerPid PeerChannel type ChokeMgrChannel = Channel ChokeMgrMsg -data CF = CF { logCh :: LogChannel - , mgrCh :: ChokeMgrChannel - , infoCh :: ChokeInfoChannel - } +data CF = CF { mgrCh :: ChokeMgrChannel + , infoCh :: ChokeInfoChannel + } instance Logging CF where - getLogger cf = ("ChokeMgrP", logCh cf) + logName _ = "Process.ChokeMgr" type ChokeMgrProcess a = Process CF PeerDB a -- INTERFACE ---------------------------------------------------------------------- -start :: LogChannel -> ChokeMgrChannel -> ChokeInfoChannel -> Int -> Bool -> SupervisorChan +start :: ChokeMgrChannel -> ChokeInfoChannel -> Int -> Bool -> SupervisorChan -> IO ThreadId -start logC ch infoC ur weSeed supC = do +start ch infoC ur weSeed supC = do Timer.register 10 Tick ch - spawnP (CF logC ch infoC) (initPeerDB $ calcUploadSlots ur Nothing) - (catchP (forever pgm) - (defaultStopHandler supC)) + spawnP (CF ch infoC) (initPeerDB $ calcUploadSlots ur Nothing) + (catchP (forever pgm) + (defaultStopHandler supC)) where initPeerDB slots = PeerDB 2 weSeed slots M.empty [] pgm = do chooseP [mgrEvent, infoEvent] >>= syncP mgrEvent = recvWrapPC mgrCh - (\msg -> case msg of - Tick -> tick - RemovePeer t -> removePeer t - AddPeer t pCh -> do - logDebug $ "Adding peer " ++ show t - weSeed <- gets seeding - addPeer pCh weSeed t) + (\msg -> case msg of + Tick -> tick + RemovePeer t -> removePeer t + AddPeer t pCh -> do + debugP $ "Adding peer " ++ show t + weSeed <- gets seeding + addPeer pCh weSeed t) infoEvent = - recvWrapPC infoCh - (\m -> case m of - BlockComplete pn blk -> informBlockComplete pn blk - PieceDone pn -> informDone pn - TorrentComplete -> do - modify (\s -> s { seeding = True - , peerMap = - M.map (\pi -> pi { pAreSeeding = True }) - $ peerMap s})) - tick = do logDebug "Ticked" - ch <- asks mgrCh - Timer.register 10 Tick ch - updateDB - runRechokeRound - removePeer tid = do logDebug $ "Removing peer " ++ show tid - modify (\db -> db { peerMap = M.delete tid (peerMap db), - peerChain = (peerChain db) \\ [tid] }) + recvWrapPC infoCh + (\m -> case m of + BlockComplete pn blk -> informBlockComplete pn blk + PieceDone pn -> informDone pn + TorrentComplete -> do + modify (\s -> s { seeding = True + , peerMap = + M.map (\pi -> pi { pAreSeeding = True }) + $ peerMap s})) + tick = do debugP "Ticked" + ch <- asks mgrCh + Timer.register 10 Tick ch + updateDB + runRechokeRound + removePeer tid = do debugP $ "Removing peer " ++ show tid + modify (\db -> db { peerMap = M.delete tid (peerMap db), + peerChain = (peerChain db) \\ [tid] }) -- INTERNAL FUNCTIONS ---------------------------------------------------------------------- @@ -101,9 +99,9 @@ -- far we are in the process of wandering the optimistic unchoke chain. data PeerDB = PeerDB { chokeRound :: Int -- ^ Counted down by one from 2. If 0 then we should - -- advance the peer chain. + -- advance the peer chain. , seeding :: Bool -- ^ True if we are seeding the torrent. - -- In a multi-torrent world, this has to change. + -- In a multi-torrent world, this has to change. , uploadSlots :: Int -- ^ Current number of upload slots , peerMap :: PeerMap -- ^ Map of peers , peerChain :: [PeerPid] -- ^ The order in which peers are optimistically unchoked @@ -129,9 +127,9 @@ compareInv :: Ord a => a -> a -> Ordering compareInv x y = case compare x y of - LT -> GT - EQ -> EQ - GT -> LT + LT -> GT + EQ -> EQ + GT -> LT comparingWith :: Ord a => (a -> a -> Ordering) -> (b -> a) -> b -> b -> Ordering comparingWith comp project x y = @@ -159,8 +157,8 @@ return $ map fst $ back ++ front where lookupPeer mp peer = case M.lookup peer mp of - Nothing -> fail "Could not look up peer in map" - Just p -> return (peer, p) + Nothing -> fail "Could not look up peer in map" + Just p -> return (peer, p) -- | Add a peer to the Peer Database addPeer :: PeerChannel -> Bool -> PeerPid -> ChokeMgrProcess () @@ -169,13 +167,13 @@ modify (\db -> db { peerMap = M.insert tid initialPeerInfo (peerMap db)}) where initialPeerInfo = PeerInfo { pChokingUs = True - , pDownRate = 0.0 - , pUpRate = 0.0 - , pChannel = pCh - , pInterestedInUs = False - , pAreSeeding = weSeeding - , pIsASeeder = False -- May get updated quickly - } + , pDownRate = 0.0 + , pUpRate = 0.0 + , pChannel = pCh + , pInterestedInUs = False + , pAreSeeding = weSeeding + , pIsASeeder = False -- May get updated quickly + } -- | Insert a Peer randomly into the Peer chain. Threads the random number generator -- through. @@ -237,29 +235,30 @@ -- peers @s@. The value of @upSlots@ defines the number of upload slots available selectPeers :: Int -> [RechokeData] -> [RechokeData] -> ChokeMgrProcess (S.Set PeerPid) selectPeers uploadSlots downPeers seedPeers = do - let (nDownSlots, nSeedSlots) = assignUploadSlots uploadSlots downPeers seedPeers - downPids = S.fromList $ map fst $ take nDownSlots $ sortLeech downPeers - seedPids = S.fromList $ map fst $ take nSeedSlots $ sortSeeds seedPeers - logDebug $ "Slots: " ++ show nDownSlots ++ " downloads, " ++ show nSeedSlots ++ " seeders" - when (uploadSlots < nDownSlots + nSeedSlots) - (fail "Wrong calculation of slots") - logDebug $ "Electing peers - leechers: " ++ show downPids ++ "; seeders: " ++ show seedPids - rm <- rateMap - logDebug $ "Peer rates: " ++ rm - return $ S.union downPids seedPids + let (nDownSlots, nSeedSlots) = assignUploadSlots uploadSlots downPeers seedPeers + downPids = S.fromList $ map fst $ take nDownSlots $ sortLeech downPeers + seedPids = S.fromList $ map fst $ take nSeedSlots $ sortSeeds seedPeers + debugP $ "Slots: " ++ show nDownSlots ++ " downloads, " ++ show nSeedSlots ++ " seeders" + when (uploadSlots < nDownSlots + nSeedSlots) + (fail "Wrong calculation of slots") + debugP $ "Electing peers - leechers: " ++ show downPids ++ "; seeders: " ++ show seedPids + rm <- rateMap + debugP $ "Peer rates: " ++ rm + return $ S.union downPids seedPids where - rateMap :: ChokeMgrProcess String - rateMap = do - pm <- gets peerMap - rts <- return $ map (\(pid, pi) -> - show pid ++ " Up: " ++ show (pUpRate pi) ++ " Down: " ++ show (pDownRate pi)) - $ M.toList pm - return $ show rts + rateMap :: ChokeMgrProcess String + rateMap = do + pm <- gets peerMap + rts <- return $ map (\(pid, pi) -> + show pid ++ " Up: " ++ show (pUpRate pi) ++ " Down: " ++ show (pDownRate pi)) + $ M.toList pm + return $ show rts -- | Send a message to the peer process at PeerChannel. May raise -- exceptions if the peer is not running anymore. -msgPeer :: PeerMessage -> PeerChannel -> ChokeMgrProcess () -msgPeer msg ch = syncP =<< sendP ch msg +msgPeer :: PeerChannel -> PeerMessage -> ChokeMgrProcess ThreadId +msgPeer ch msg = liftIO $ spawn proc + where proc = sync $ transmit ch msg -- | This function carries out the choking and unchoking of peers in a round. performChokingUnchoking :: S.Set PeerPid -> [RechokeData] -> ChokeMgrProcess () @@ -273,8 +272,8 @@ -- If we block on the sync, it means that the process in the other end must -- be dead. Thus we can just skip it. We will eventually receive this knowledge -- through another channel. - unchoke pi = ignoreProcessBlock () $ unchokePeer (pChannel pi) - choke pi = ignoreProcessBlock () $ chokePeer (pChannel pi) + unchoke pi = unchokePeer (pChannel pi) + choke pi = chokePeer (pChannel pi) -- If we have k optimistic slots, @optChoke k peers@ will unchoke the first @k@ interested -- in us. The rest will either be unchoked if they are not interested (ensuring fast start -- should they become interested); or they will be choked to avoid TCP/IP congestion. @@ -286,8 +285,8 @@ optChoke k ((_, pi) : ps) = if pInterestedInUs pi then unchokePeer (pChannel pi) >> optChoke (k-1) ps else unchokePeer (pChannel pi) >> optChoke k ps - chokePeer = msgPeer ChokePeer - unchokePeer = msgPeer UnchokePeer + chokePeer = flip msgPeer ChokePeer + unchokePeer = flip msgPeer UnchokePeer -- | Function to split peers into those where we are seeding and those were we are leeching. -- also prunes the list for peers which are not interesting. @@ -305,8 +304,8 @@ pm <- gets peerMap T.mapM (cPeer pm) chain where cPeer pm pid = case M.lookup pid pm of - Nothing -> fail "buildRechokeData: Couldn't lookup pid" - Just x -> return (pid, x) + Nothing -> fail "buildRechokeData: Couldn't lookup pid" + Just x -> return (pid, x) rechoke :: ChokeMgrProcess () rechoke = do @@ -322,14 +321,12 @@ informDone :: PieceNum -> ChokeMgrProcess () informDone pn = traversePeers sendDone >> return () where - sendDone pi = ignoreProcessBlock () - $ (sendP (pChannel pi) $ PieceCompleted pn) >>= syncP + sendDone pi = msgPeer (pChannel pi) (PieceCompleted pn) informBlockComplete :: PieceNum -> Block -> ChokeMgrProcess () informBlockComplete pn blk = traversePeers sendComp >> return () where - sendComp pi = ignoreProcessBlock () - $ (sendP (pChannel pi) $ CancelBlock pn blk) >>= syncP + sendComp pi = msgPeer (pChannel pi) (CancelBlock pn blk) updateDB :: ChokeMgrProcess () updateDB = do @@ -337,22 +334,22 @@ modify (\db -> db { peerMap = nmp }) where gatherRate pi = do - ch <- liftIO channel - t <- liftIO getCurrentTime - ignoreProcessBlock pi (gather t ch pi) + ch <- liftIO channel + t <- liftIO getCurrentTime + ignoreProcessBlock pi (gather t ch pi) gather t ch pi = do - (sendP (pChannel pi) $ PeerStats t ch) >>= syncP - (uprt, downrt, interested) <- recvP ch (const True) >>= syncP - return pi { pDownRate = downrt, - pUpRate = uprt, - pInterestedInUs = interested } -- TODO: Seeder state + (sendP (pChannel pi) $ PeerStats t ch) >>= syncP + (uprt, downrt, interested) <- recvP ch (const True) >>= syncP + return pi { pDownRate = downrt, + pUpRate = uprt, + pInterestedInUs = interested } -- TODO: Seeder state runRechokeRound :: ChokeMgrProcess () runRechokeRound = do cRound <- gets chokeRound if (cRound == 0) - then do nChain <- advancePeerChain - modify (\db -> db { chokeRound = 2, - peerChain = nChain }) - else modify (\db -> db { chokeRound = (chokeRound db) - 1 }) + then do nChain <- advancePeerChain + modify (\db -> db { chokeRound = 2, + peerChain = nChain }) + else modify (\db -> db { chokeRound = (chokeRound db) - 1 }) rechoke diff --git a/src/Process/Console.hs b/src/Process/Console.hs --- a/src/Process/Console.hs +++ b/src/Process/Console.hs @@ -1,29 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -- | The Console process has two main purposes. It is a telnet-like -- interface with the user and it is our first simple logging device -- for what happens inside the system. @@ -38,44 +12,92 @@ import Control.Monad.Reader import Prelude hiding (catch) -import Process +import System.Log.Logger -import Logging + +import Process +import qualified Process.Status as St import Supervisor -data Cmd = Quit -- Quit the program + +data Cmd = Quit -- ^ Quit the program + | Show -- ^ Show current state + | Help -- ^ Print Help message + | Unknown String -- ^ Unknown command deriving (Eq, Show) type CmdChannel = Channel Cmd data CF = CF { cmdCh :: CmdChannel - , logCh :: LogChannel } + , wrtCh :: Channel String } instance Logging CF where - getLogger cf = ("ConsoleP", logCh cf) + logName _ = "Process.Console" -- | Start the logging process and return a channel to it. Sending on this -- Channel means writing stuff out on stdOut -start :: LogChannel -> Channel () -> SupervisorChan -> IO ThreadId -start logC waitC supC = do - cmdC <- readerP logC -- We shouldn't be doing this in the long run - spawnP (CF cmdC logC) () (catchP (forever lp) (defaultStopHandler supC)) +start :: Channel () -> Channel St.ST -> SupervisorChan -> IO ThreadId +start waitC statusC supC = do + cmdC <- readerP -- We shouldn't be doing this in the long run + wrtC <- writerP + spawnP (CF cmdC wrtC) () (catchP (forever lp) (defaultStopHandler supC)) where - lp = syncP =<< quitEvent + lp = syncP =<< chooseP [quitEvent, helpEvent, unknownEvent, showEvent] quitEvent = do - ch <- asks cmdCh - ev <- recvP ch (==Quit) - wrapP ev - (\_ -> syncP =<< sendP waitC ()) - + ch <- asks cmdCh + ev <- recvP ch (==Quit) + wrapP ev + (\_ -> syncP =<< sendP waitC ()) + helpEvent = do + ch <- asks cmdCh + wrtC <- asks wrtCh + ev <- recvP ch (==Help) + wrapP ev + (\_ -> syncP =<< sendPC wrtCh helpMessage) + unknownEvent = do + ch <- asks cmdCh + wrtC <- asks wrtCh + ev <- recvP ch (\m -> case m of + Unknown _ -> True + _ -> False) + wrapP ev + (\(Unknown cmd) -> syncP =<< (sendPC wrtCh $ "Unknown command: " ++ cmd)) + showEvent = do + ch <- asks cmdCh + wrtC <- asks wrtCh + ev <- recvP ch (==Show) + wrapP ev + (\_ -> do + st <- syncP =<< recvP statusC (const True) + syncP =<< sendPC wrtCh (show st)) -readerP :: LogChannel -> IO CmdChannel -readerP logCh = do cmdCh <- channel - spawn $ lp cmdCh - return cmdCh - where lp cmdCh = do c <- getLine - case c of - "quit" -> sync $ transmit cmdCh Quit - cmd -> do logMsg' logCh "Console" Info $ "Unrecognized command: " ++ show cmd - lp cmdCh +helpMessage :: String +helpMessage = concat + [ "Command Help:\n" + , "\n" + , " help - Show this help\n" + , " quit - Quit the program\n" + , " show - Show the current downloading status\n" + ] + +writerP :: IO (Channel String) +writerP = do wrtCh <- channel + spawn $ lp wrtCh + return wrtCh + where lp wCh = forever (do m <- sync $ receive wCh (const True) + putStrLn m) + +readerP :: IO CmdChannel +readerP = do cmdCh <- channel + spawn $ lp cmdCh + return cmdCh + where lp cmdCh = forever $ + do c <- getLine + case c of + "help" -> sync $ transmit cmdCh Help + "quit" -> sync $ transmit cmdCh Quit + "show" -> sync $ transmit cmdCh Show + cmd -> do logM "Process.Console.readerP" INFO $ + "Unrecognized command: " ++ show cmd + sync $ transmit cmdCh (Unknown cmd) diff --git a/src/Process/FS.hs b/src/Process/FS.hs --- a/src/Process/FS.hs +++ b/src/Process/FS.hs @@ -1,29 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -- | File system process. Acts as a maintainer for the filesystem in -- question and can only do single-file torrents. It should be -- fairly easy to add Multi-file torrents by altering this file and @@ -39,15 +13,10 @@ import Control.Concurrent import Control.Concurrent.CML import Control.Monad.State -import Control.Monad.Reader -import System.IO - - import qualified Data.ByteString as B import qualified Data.Map as M -import Logging import Process import Torrent import qualified FS @@ -61,11 +30,10 @@ data CF = CF { fspCh :: FSPChannel -- ^ Channel on which to receive messages - , logCh :: LogChannel -- ^ Channel for logging } instance Logging CF where - getLogger cf = ("FSP", logCh cf) + logName _ = "Process.FS" data ST = ST { fileHandles :: FS.Handles -- ^ The file we are working on @@ -76,33 +44,33 @@ -- INTERFACE ---------------------------------------------------------------------- -start :: FS.Handles -> LogChannel -> FS.PieceMap -> FSPChannel -> SupervisorChan-> IO ThreadId -start handles logC pm fspC supC = - spawnP (CF fspC logC) (ST handles pm) (catchP (forever lp) (defaultStopHandler supC)) +start :: FS.Handles -> FS.PieceMap -> FSPChannel -> SupervisorChan-> IO ThreadId +start handles pm fspC supC = + spawnP (CF fspC) (ST handles pm) (catchP (forever lp) (defaultStopHandler supC)) where lp = msgEvent >>= syncP msgEvent = do - ev <- recvPC fspCh - wrapP ev (\msg -> - case msg of - CheckPiece n ch -> do - pm <- gets pieceMap - case M.lookup n pm of - Nothing -> sendP ch Nothing >>= syncP - Just pi -> do r <- gets fileHandles >>= (liftIO . FS.checkPiece pi) - sendP ch (Just r) >>= syncP - ReadBlock n blk ch -> do - logDebug $ "Reading block #" ++ show n - ++ "(" ++ show (blockOffset blk) ++ ", " ++ show (blockSize blk) ++ ")" - -- TODO: Protection, either here or in the Peer code - h <- gets fileHandles - bs <- gets pieceMap >>= (liftIO . FS.readBlock n blk h) - sendP ch bs >>= syncP - WriteBlock pn blk bs -> do + ev <- recvPC fspCh + wrapP ev (\msg -> + case msg of + CheckPiece n ch -> do + pm <- gets pieceMap + case M.lookup n pm of + Nothing -> sendP ch Nothing >>= syncP + Just pi -> do r <- gets fileHandles >>= (liftIO . FS.checkPiece pi) + sendP ch (Just r) >>= syncP + ReadBlock n blk ch -> do + debugP $ "Reading block #" ++ show n + ++ "(" ++ show (blockOffset blk) ++ ", " ++ show (blockSize blk) ++ ")" -- TODO: Protection, either here or in the Peer code - fh <- gets fileHandles - pm <- gets pieceMap - liftIO $ FS.writeBlock fh pn blk pm bs) + h <- gets fileHandles + bs <- gets pieceMap >>= (liftIO . FS.readBlock n blk h) + sendP ch bs >>= syncP + WriteBlock pn blk bs -> do + -- TODO: Protection, either here or in the Peer code + fh <- gets fileHandles + pm <- gets pieceMap + liftIO $ FS.writeBlock fh pn blk pm bs) checkPiece :: FSPChannel -> PieceNum -> IO (Maybe Bool) checkPiece fspC n = do diff --git a/src/Process/Listen.hs b/src/Process/Listen.hs --- a/src/Process/Listen.hs +++ b/src/Process/Listen.hs @@ -1,34 +1,31 @@ -module ListenP +module Process.Listen + ( start + ) where --- TODO: Create the OMBox here rather than taking it as a parameter -{- -import Control.Concurrent.CML +import Control.Concurrent +import Control.Monad +import Control.Monad.Trans import Network -import ConsoleP -import PeerP -import Torrent -import FSP - - +import Process +import Process.PeerMgr hiding (start) +import Supervisor -start :: PortID -> PeerId -> InfoHash -> FSPChannel -> LogChannel -> IO () -start port pid ih fsC logC = - do sock <- listenOn port - spawn $ acceptor sock pid ih fsC logC - return () +data CF = CF { peerMgrCh :: PeerMgrChannel + } +instance Logging CF where + logName _ = "Process.Listen" -acceptor :: Socket -> PeerId -> InfoHash -> FSPChannel -> LogChannel -> IO () -acceptor sock pid ih fsC logC = - do (h, _, _) <- accept sock - spawn $ acceptor sock pid ih fsC logC - r <- PeerP.listenHandshake h pid ih fsC logC - case r of - Left err -> do logMsg logC $ "Incoming peer Accept error" ++ err - return () - Right () -> return () --} +start :: PortID -> PeerMgrChannel -> SupervisorChan -> IO ThreadId +start port peerMgrC supC = do + spawnP (CF peerMgrC) () (catchP (openListen >>= pgm) + (defaultStopHandler supC)) -- TODO: Close socket resource! + where openListen = liftIO $ listenOn port + pgm sock = forever lp + where lp = do + conn <- liftIO $ accept sock + syncP =<< sendPC peerMgrCh (NewIncoming conn) diff --git a/src/Process/Peer.hs b/src/Process/Peer.hs --- a/src/Process/Peer.hs +++ b/src/Process/Peer.hs @@ -4,7 +4,7 @@ -- * Types PeerMessage(..) -- * Interface - , connect + , peerChildren ) where @@ -21,19 +21,18 @@ import qualified Data.Serialize.Get as G import qualified Data.Map as M +import qualified Data.IntSet as IS import Data.Maybe import Data.Set as S hiding (map) import Data.Time.Clock import Data.Word -import Network - import System.IO +import System.Timeout import PeerTypes import Process -import Logging import Process.FS import Process.PieceMgr import qualified Data.Queue as Q @@ -47,387 +46,361 @@ -- INTERFACE ---------------------------------------------------------------------- - -type ConnectRecord = (HostName, PortID, PeerId, InfoHash, PieceMap) - -connect :: ConnectRecord -> SupervisorChan -> PieceMgrChannel -> FSPChannel -> LogChannel - -> StatusChan - -> MgrChannel -> Int - -> IO ThreadId -connect (host, port, pid, ih, pm) pool pieceMgrC fsC logC statC mgrC nPieces = - spawn (connector >> return ()) - where connector = - do logMsg logC $ "Connecting to " ++ show host ++ " (" ++ showPort port ++ ")" - h <- connectTo host port - logMsg logC "Connected, initiating handShake" - r <- initiateHandshake logC h pid ih - logMsg logC "Handshake run" - case r of - Left err -> do logMsg logC ("Peer handshake failure at host " ++ host - ++ " with error " ++ err) - return () - Right (_caps, _rpid) -> - do logMsg logC "entering peerP loop code" - supC <- channel -- TODO: Should be linked later on - children <- peerChildren logC h mgrC pieceMgrC fsC statC pm nPieces - sync $ transmit pool $ SpawnNew (Supervisor $ allForOne "PeerSup" children logC) - return () +peerChildren :: Handle -> MgrChannel -> PieceMgrChannel + -> FSPChannel -> StatusChan -> PieceMap -> Int -> IO Children +peerChildren handle pMgrC pieceMgrC fsC statusC pm nPieces = do + queueC <- channel + senderC <- channel + receiverC <- channel + sendBWC <- channel + return [Worker $ senderP handle senderC, + Worker $ sendQueueP queueC senderC sendBWC, + Worker $ receiverP handle receiverC, + Worker $ peerP pMgrC pieceMgrC fsC pm nPieces handle + queueC receiverC sendBWC statusC] -- INTERNAL FUNCTIONS ---------------------------------------------------------------------- - -data SPCF = SPCF { spLogCh :: LogChannel - , spMsgCh :: Channel B.ByteString - } +data SPCF = SPCF { spMsgCh :: Channel B.ByteString + } instance Logging SPCF where - getLogger cf = ("SenderP", spLogCh cf) + logName _ = "Process.Peer.Sender" -- | The raw sender process, it does nothing but send out what it syncs on. -senderP :: LogChannel -> Handle -> Channel B.ByteString -> SupervisorChan -> IO ThreadId -senderP logC h ch supC = spawnP (SPCF logC ch) h (catchP (foreverP pgm) - (do t <- liftIO $ myThreadId - syncP =<< (sendP supC $ IAmDying t) - liftIO $ hClose h)) +senderP :: Handle -> Channel B.ByteString -> SupervisorChan -> IO ThreadId +senderP h ch supC = spawnP (SPCF ch) h (catchP (foreverP pgm) + (do t <- liftIO $ myThreadId + syncP =<< (sendP supC $ IAmDying t) + liftIO $ hClose h)) where pgm :: Process SPCF Handle () pgm = do - c <- ask - m <- syncP =<< recvPC spMsgCh - h <- get - liftIO $ do B.hPut h m - hFlush h + m <- liftIO $ timeout defaultTimeout s + h <- get + case m of + Nothing -> putMsg (encodePacket KeepAlive) + Just m -> putMsg m + liftIO $ hFlush h + defaultTimeout = 120 * 10^6 + putMsg m = liftIO $ B.hPut h m + s = sync $ receive ch (const True) -- | Messages we can send to the Send Queue data SendQueueMessage = SendQCancel PieceNum Block -- ^ Peer requested that we cancel a piece | SendQMsg Message -- ^ We want to send the Message to the peer | SendOChoke -- ^ We want to choke the peer - | SendQRequestPrune PieceNum Block -- ^ Prune SendQueue of this (pn, blk) pair + | SendQRequestPrune PieceNum Block -- ^ Prune SendQueue of this (pn, blk) pair -data SQCF = SQCF { sqLogC :: LogChannel - , sqInCh :: Channel SendQueueMessage - , sqOutCh :: Channel B.ByteString - , bandwidthCh :: BandwidthChannel - } +data SQCF = SQCF { sqInCh :: Channel SendQueueMessage + , sqOutCh :: Channel B.ByteString + , bandwidthCh :: BandwidthChannel + } data SQST = SQST { outQueue :: Q.Queue Message - , bytesTransferred :: Integer - } + , bytesTransferred :: Integer + } instance Logging SQCF where - getLogger cf = ("SendQueueP", sqLogC cf) + logName _ = "Process.Peer.SendQueue" -- | sendQueue Process, simple version. -- TODO: Split into fast and slow. -sendQueueP :: LogChannel -> Channel SendQueueMessage -> Channel B.ByteString -> BandwidthChannel - -> SupervisorChan - -> IO ThreadId -sendQueueP logC inC outC bandwC supC = spawnP (SQCF logC inC outC bandwC) (SQST Q.empty 0) - (catchP (foreverP pgm) - (defaultStopHandler supC)) +sendQueueP :: Channel SendQueueMessage -> Channel B.ByteString -> BandwidthChannel + -> SupervisorChan + -> IO ThreadId +sendQueueP inC outC bandwC supC = spawnP (SQCF inC outC bandwC) (SQST Q.empty 0) + (catchP (foreverP pgm) + (defaultStopHandler supC)) where pgm :: Process SQCF SQST () pgm = do - q <- gets outQueue - l <- gets bytesTransferred - case (Q.isEmpty q, l > 0) of - (True, False) -> queueEvent >>= syncP - (True, True ) -> chooseP [queueEvent, rateUpdateEvent] >>= syncP - (False, False) -> chooseP [queueEvent, sendEvent] >>= syncP - (False, True) -> chooseP [queueEvent, sendEvent, rateUpdateEvent] >>= syncP + q <- gets outQueue + l <- gets bytesTransferred + -- Gather together events which may trigger + syncP =<< (chooseP $ + concat [if Q.isEmpty q then [] else [sendEvent], + if l > 0 then [rateUpdateEvent] else [], + [queueEvent]]) rateUpdateEvent = do - l <- gets bytesTransferred - ev <- sendPC bandwidthCh l - wrapP ev (\() -> - modify (\s -> s { bytesTransferred = 0 })) + l <- gets bytesTransferred + ev <- sendPC bandwidthCh l + wrapP ev (\() -> + modify (\s -> s { bytesTransferred = 0 })) queueEvent = do - ev <- recvPC sqInCh - wrapP ev (\m -> case m of - SendQMsg msg -> do logDebug "Queueing event for sending" - modifyQ (Q.push msg) - SendQCancel n blk -> modifyQ (Q.filter (filterPiece n (blockOffset blk))) - SendOChoke -> do modifyQ (Q.filter filterAllPiece) - modifyQ (Q.push Choke) - SendQRequestPrune n blk -> - modifyQ (Q.filter (filterRequest n blk))) + recvWrapPC sqInCh + (\m -> case m of + SendQMsg msg -> do debugP "Queueing event for sending" + modifyQ (Q.push msg) + SendQCancel n blk -> modifyQ (Q.filter (filterPiece n (blockOffset blk))) + SendOChoke -> do modifyQ (Q.filter filterAllPiece) + modifyQ (Q.push Choke) + SendQRequestPrune n blk -> + modifyQ (Q.filter (filterRequest n blk))) modifyQ :: (Q.Queue Message -> Q.Queue Message) -> Process SQCF SQST () modifyQ f = modify (\s -> s { outQueue = f (outQueue s) }) sendEvent = do - Just (e, r) <- gets (Q.pop . outQueue) - let bs = encodePacket e - tEvt <- sendPC sqOutCh bs - wrapP tEvt (\() -> do logDebug "Dequeued event" - modify (\s -> s { outQueue = r, - bytesTransferred = - bytesTransferred s + fromIntegral (B.length bs)})) + Just (e, r) <- gets (Q.pop . outQueue) + let bs = encodePacket e + tEvt <- sendPC sqOutCh bs + wrapP tEvt (\() -> do debugP "Dequeued event" + modify (\s -> s { outQueue = r, + bytesTransferred = + bytesTransferred s + fromIntegral (B.length bs)})) filterAllPiece (Piece _ _ _) = True filterAllPiece _ = False filterPiece n off m = case m of Piece n off _ -> False _ -> True filterRequest n blk m = - case m of Request n blk -> False - _ -> True - -peerChildren :: LogChannel -> Handle -> MgrChannel -> PieceMgrChannel - -> FSPChannel -> StatusChan -> PieceMap -> Int -> IO Children -peerChildren logC handle pMgrC pieceMgrC fsC statusC pm nPieces = do - queueC <- channel - senderC <- channel - receiverC <- channel - sendBWC <- channel - return [Worker $ senderP logC handle senderC, - Worker $ sendQueueP logC queueC senderC sendBWC, - Worker $ receiverP logC handle receiverC, - Worker $ peerP pMgrC pieceMgrC fsC pm logC nPieces handle - queueC receiverC sendBWC statusC] + case m of Request n blk -> False + _ -> True -data RPCF = RPCF { rpLogC :: LogChannel - , rpMsgC :: Channel (Message, Integer) } +data RPCF = RPCF { rpMsgC :: Channel (Message, Integer) } instance Logging RPCF where - getLogger cf = ("ReceiverP", rpLogC cf) + logName _ = "Process.Peer.Receiver" -receiverP :: LogChannel -> Handle -> Channel (Message, Integer) -> SupervisorChan -> IO ThreadId -receiverP logC h ch supC = spawnP (RPCF logC ch) h - (catchP (foreverP pgm) - (defaultStopHandler supC)) +receiverP :: Handle -> Channel (Message, Integer) + -> SupervisorChan -> IO ThreadId +receiverP h ch supC = spawnP (RPCF ch) h + (catchP (foreverP pgm) + (defaultStopHandler supC)) where - pgm = do logDebug "Peer waiting for input" + pgm = do debugP "Peer waiting for input" readHeader ch readHeader ch = do h <- get - feof <- liftIO $ hIsEOF h - if feof - then do logDebug "Handle Closed" - stopP - else do bs' <- liftIO $ B.hGet h 4 - l <- conv bs' - readMessage l ch + feof <- liftIO $ hIsEOF h + if feof + then do debugP "Handle Closed" + stopP + else do bs' <- liftIO $ B.hGet h 4 + l <- conv bs' + readMessage l ch readMessage l ch = do if (l == 0) - then return () - else do logDebug $ "Reading off " ++ show l ++ " bytes" - h <- get - bs <- liftIO $ B.hGet h (fromIntegral l) - case G.runGet decodeMsg bs of - Left _ -> do logWarn "Incorrect parse in receiver, dying!" + then return () + else do debugP $ "Reading off " ++ show l ++ " bytes" + h <- get + bs <- liftIO $ B.hGet h (fromIntegral l) + case G.runGet decodeMsg bs of + Left _ -> do warningP "Incorrect parse in receiver, dying!" stopP Right msg -> do sendPC rpMsgC (msg, fromIntegral l) >>= syncP conv bs = do case G.runGet G.getWord32be bs of - Left err -> do logWarn $ "Incorrent parse in receiver, dying: " ++ show err + Left err -> do warningP $ "Incorrent parse in receiver, dying: " ++ show err stopP Right i -> return i data PCF = PCF { inCh :: Channel (Message, Integer) - , outCh :: Channel SendQueueMessage - , peerMgrCh :: MgrChannel - , pieceMgrCh :: PieceMgrChannel - , logCh :: LogChannel - , fsCh :: FSPChannel - , peerCh :: PeerChannel - , sendBWCh :: BandwidthChannel - , timerCh :: Channel () - , statCh :: StatusChan - , pieceMap :: PieceMap - } + , outCh :: Channel SendQueueMessage + , peerMgrCh :: MgrChannel + , pieceMgrCh :: PieceMgrChannel + , fsCh :: FSPChannel + , peerCh :: PeerChannel + , sendBWCh :: BandwidthChannel + , timerCh :: Channel () + , statCh :: StatusChan + , pieceMap :: PieceMap + } instance Logging PCF where - getLogger cf = ("PeerP", logCh cf) + logName _ = "Process.Peer" data PST = PST { weChoke :: Bool -- ^ True if we are choking the peer - , weInterested :: Bool -- ^ True if we are interested in the peer - , blockQueue :: S.Set (PieceNum, Block) -- ^ Blocks queued at the peer - , peerChoke :: Bool -- ^ Is the peer choking us? True if yes - , peerInterested :: Bool -- ^ True if the peer is interested - , peerPieces :: [PieceNum] -- ^ List of pieces the peer has access to - , upRate :: Rate -- ^ Upload rate towards the peer (estimated) - , downRate :: Rate -- ^ Download rate from the peer (estimated) - , runningEndgame :: Bool -- ^ True if we are in endgame - } + , weInterested :: Bool -- ^ True if we are interested in the peer + , blockQueue :: S.Set (PieceNum, Block) -- ^ Blocks queued at the peer + , peerChoke :: Bool -- ^ Is the peer choking us? True if yes + , peerInterested :: Bool -- ^ True if the peer is interested + , peerPieces :: IS.IntSet -- ^ List of pieces the peer has access to + , upRate :: Rate -- ^ Upload rate towards the peer (estimated) + , downRate :: Rate -- ^ Download rate from the peer (estimated) + , runningEndgame :: Bool -- ^ True if we are in endgame + } -peerP :: MgrChannel -> PieceMgrChannel -> FSPChannel -> PieceMap -> LogChannel -> Int -> Handle +peerP :: MgrChannel -> PieceMgrChannel -> FSPChannel -> PieceMap -> Int -> Handle -> Channel SendQueueMessage -> Channel (Message, Integer) -> BandwidthChannel - -> StatusChan - -> SupervisorChan -> IO ThreadId -peerP pMgrC pieceMgrC fsC pm logC nPieces h outBound inBound sendBWC statC supC = do + -> StatusChan + -> SupervisorChan -> IO ThreadId +peerP pMgrC pieceMgrC fsC pm nPieces h outBound inBound sendBWC statC supC = do ch <- channel tch <- channel ct <- getCurrentTime - spawnP (PCF inBound outBound pMgrC pieceMgrC logC fsC ch sendBWC tch statC pm) - (PST True False S.empty True False [] (RC.new ct) (RC.new ct) False) - (cleanupP startup (defaultStopHandler supC) cleanup) + spawnP (PCF inBound outBound pMgrC pieceMgrC fsC ch sendBWC tch statC pm) + (PST True False S.empty True False IS.empty (RC.new ct) (RC.new ct) False) + (cleanupP startup (defaultStopHandler supC) cleanup) where startup = do - tid <- liftIO $ myThreadId - logDebug "Syncing a connectBack" - asks peerCh >>= (\ch -> sendPC peerMgrCh $ Connect tid ch) >>= syncP - pieces <- getPiecesDone - syncP =<< (sendPC outCh $ SendQMsg $ BitField (constructBitField nPieces pieces)) - -- Install the StatusP timer - c <- asks timerCh - Timer.register 30 () c - foreverP (recvEvt) - cleanup = do - t <- liftIO myThreadId - syncP =<< sendPC peerMgrCh (Disconnect t) + tid <- liftIO $ myThreadId + debugP "Syncing a connectBack" + asks peerCh >>= (\ch -> sendPC peerMgrCh $ Connect tid ch) >>= syncP + pieces <- getPiecesDone + syncP =<< (sendPC outCh $ SendQMsg $ BitField (constructBitField nPieces pieces)) + -- Install the StatusP timer + c <- asks timerCh + Timer.register 30 () c + foreverP (eventLoop) + cleanup = do + t <- liftIO myThreadId + syncP =<< sendPC peerMgrCh (Disconnect t) getPiecesDone = do - c <- liftIO $ channel - syncP =<< (sendPC pieceMgrCh $ GetDone c) - recvP c (const True) >>= syncP - recvEvt = do - syncP =<< chooseP [peerMsgEvent, chokeMgrEvent, upRateEvent, timerEvent] - chokeMgrEvent = do - evt <- recvPC peerCh - wrapP evt (\msg -> do - logDebug "ChokeMgrEvent" - case msg of - PieceCompleted pn -> do - syncP =<< (sendPC outCh $ SendQMsg $ Have pn) - ChokePeer -> do syncP =<< sendPC outCh SendOChoke - logDebug "Choke Peer" - modify (\s -> s {weChoke = True}) - UnchokePeer -> do syncP =<< (sendPC outCh $ SendQMsg Unchoke) - logDebug "UnchokePeer" - modify (\s -> s {weChoke = False}) - PeerStats t retCh -> do - i <- gets peerInterested - ur <- gets upRate - dr <- gets downRate - let (up, nur) = RC.extractRate t ur - (down, ndr) = RC.extractRate t dr - logInfo $ "Peer has rates up/down: " ++ show up ++ "/" ++ show down - sendP retCh (up, down, i) >>= syncP - modify (\s -> s { upRate = nur , downRate = ndr }) - CancelBlock pn blk -> do - modify (\s -> s { blockQueue = S.delete (pn, blk) $ blockQueue s }) - syncP =<< (sendPC outCh $ SendQRequestPrune pn blk)) - timerEvent = do - evt <- recvPC timerCh - wrapP evt (\() -> do - logDebug "TimerEvent" - tch <- asks timerCh - Timer.register 30 () tch - ur <- gets upRate - dr <- gets downRate - let (upCnt, nuRate) = RC.extractCount $ ur - (downCnt, ndRate) = RC.extractCount $ dr - logDebug $ "Sending peerStats: " ++ show upCnt ++ ", " ++ show downCnt - (sendPC statCh $ PeerStat { peerUploaded = upCnt - , peerDownloaded = downCnt }) >>= syncP - modify (\s -> s { upRate = nuRate, downRate = ndRate })) - upRateEvent = do - evt <- recvPC sendBWCh - wrapP evt (\uploaded -> do - modify (\s -> s { upRate = RC.update uploaded $ upRate s})) - peerMsgEvent = do - evt <- recvPC inCh - wrapP evt (\(msg, sz) -> do - modify (\s -> s { downRate = RC.update sz $ downRate s}) - case msg of - KeepAlive -> return () - Choke -> do putbackBlocks - modify (\s -> s { peerChoke = True }) - Unchoke -> do modify (\s -> s { peerChoke = False }) - fillBlocks + c <- liftIO $ channel + syncP =<< (sendPC pieceMgrCh $ GetDone c) + recvP c (const True) >>= syncP + eventLoop = do + syncP =<< chooseP [peerMsgEvent, chokeMgrEvent, upRateEvent, timerEvent] + chokeMgrEvent = do + evt <- recvPC peerCh + wrapP evt (\msg -> do + debugP "ChokeMgrEvent" + case msg of + PieceCompleted pn -> do + syncP =<< (sendPC outCh $ SendQMsg $ Have pn) + ChokePeer -> do syncP =<< sendPC outCh SendOChoke + debugP "Choke Peer" + modify (\s -> s {weChoke = True}) + UnchokePeer -> do syncP =<< (sendPC outCh $ SendQMsg Unchoke) + debugP "UnchokePeer" + modify (\s -> s {weChoke = False}) + PeerStats t retCh -> do + i <- gets peerInterested + ur <- gets upRate + dr <- gets downRate + let (up, nur) = RC.extractRate t ur + (down, ndr) = RC.extractRate t dr + infoP $ "Peer has rates up/down: " ++ show up ++ "/" ++ show down + sendP retCh (up, down, i) >>= syncP + modify (\s -> s { upRate = nur , downRate = ndr }) + CancelBlock pn blk -> do + modify (\s -> s { blockQueue = S.delete (pn, blk) $ blockQueue s }) + syncP =<< (sendPC outCh $ SendQRequestPrune pn blk)) + timerEvent = do + evt <- recvPC timerCh + wrapP evt (\() -> do + debugP "TimerEvent" + tch <- asks timerCh + Timer.register 30 () tch + ur <- gets upRate + dr <- gets downRate + let (upCnt, nuRate) = RC.extractCount $ ur + (downCnt, ndRate) = RC.extractCount $ dr + debugP $ "Sending peerStats: " ++ show upCnt ++ ", " ++ show downCnt + (sendPC statCh $ PeerStat { peerUploaded = upCnt + , peerDownloaded = downCnt }) >>= syncP + modify (\s -> s { upRate = nuRate, downRate = ndRate })) + upRateEvent = do + evt <- recvPC sendBWCh + wrapP evt (\uploaded -> do + modify (\s -> s { upRate = RC.update uploaded $ upRate s})) + peerMsgEvent = do + evt <- recvPC inCh + wrapP evt (\(msg, sz) -> do + modify (\s -> s { downRate = RC.update sz $ downRate s}) + case msg of + KeepAlive -> return () + Choke -> do putbackBlocks + modify (\s -> s { peerChoke = True }) + Unchoke -> do modify (\s -> s { peerChoke = False }) + fillBlocks Interested -> modify (\s -> s { peerInterested = True }) - NotInterested -> modify (\s -> s { peerInterested = False }) - Have pn -> haveMsg pn - BitField bf -> bitfieldMsg bf - Request pn blk -> requestMsg pn blk - Piece n os bs -> do pieceMsg n os bs - fillBlocks - Cancel pn blk -> cancelMsg pn blk - Port _ -> return ()) -- No DHT yet, silently ignore - putbackBlocks = do - blks <- gets blockQueue - syncP =<< sendPC pieceMgrCh (PutbackBlocks (S.toList blks)) - modify (\s -> s { blockQueue = S.empty }) - haveMsg :: PieceNum -> Process PCF PST () - haveMsg pn = do - pm <- asks pieceMap - if M.member pn pm - then do modify (\s -> s { peerPieces = pn : peerPieces s}) - considerInterest - else do logWarn "Unknown Piece" - stopP - bitfieldMsg bf = do - pieces <- gets peerPieces - case pieces of - -- TODO: Don't trust the bitfield - [] -> do modify (\s -> s { peerPieces = createPeerPieces bf}) - considerInterest - _ -> do logInfo "Got out of band Bitfield request, dying" - stopP - requestMsg :: PieceNum -> Block -> Process PCF PST () - requestMsg pn blk = do - choking <- gets weChoke - unless (choking) - (do - bs <- readBlock pn blk -- TODO: Pushdown to send process - syncP =<< sendPC outCh (SendQMsg $ Piece pn (blockOffset blk) bs)) - readBlock :: PieceNum -> Block -> Process PCF PST B.ByteString - readBlock pn blk = do - c <- liftIO $ channel - syncP =<< sendPC fsCh (ReadBlock pn blk c) - syncP =<< recvP c (const True) - pieceMsg :: PieceNum -> Int -> B.ByteString -> Process PCF PST () - pieceMsg n os bs = do - let sz = B.length bs - blk = Block os sz - e = (n, blk) - q <- gets blockQueue - when (S.member e q) - (do storeBlock n (Block os sz) bs - modify (\s -> s { blockQueue = S.delete e (blockQueue s)})) - -- When e is not a member, the piece may be stray, so ignore it. - -- Perhaps print something here. - cancelMsg n blk = - syncP =<< sendPC outCh (SendQCancel n blk) - considerInterest = do - c <- liftIO channel - pcs <- gets peerPieces - syncP =<< sendPC pieceMgrCh (AskInterested pcs c) - interested <- syncP =<< recvP c (const True) - if interested - then do modify (\s -> s { weInterested = True }) - syncP =<< sendPC outCh (SendQMsg Interested) - else modify (\s -> s { weInterested = False}) + NotInterested -> modify (\s -> s { peerInterested = False }) + Have pn -> haveMsg pn + BitField bf -> bitfieldMsg bf + Request pn blk -> requestMsg pn blk + Piece n os bs -> do pieceMsg n os bs + fillBlocks + Cancel pn blk -> cancelMsg pn blk + Port _ -> return ()) -- No DHT yet, silently ignore + putbackBlocks = do + blks <- gets blockQueue + syncP =<< sendPC pieceMgrCh (PutbackBlocks (S.toList blks)) + modify (\s -> s { blockQueue = S.empty }) + haveMsg :: PieceNum -> Process PCF PST () + haveMsg pn = do + pm <- asks pieceMap + if M.member pn pm + then do modify (\s -> s { peerPieces = IS.insert pn $ peerPieces s}) + considerInterest + else do warningP "Unknown Piece" + stopP + bitfieldMsg bf = do + pieces <- gets peerPieces + if IS.null pieces + -- TODO: Don't trust the bitfield + then do modify (\s -> s { peerPieces = createPeerPieces bf}) + considerInterest + else do infoP "Got out of band Bitfield request, dying" + stopP + requestMsg :: PieceNum -> Block -> Process PCF PST () + requestMsg pn blk = do + choking <- gets weChoke + unless (choking) + (do + bs <- readBlock pn blk -- TODO: Pushdown to send process + syncP =<< sendPC outCh (SendQMsg $ Piece pn (blockOffset blk) bs)) + readBlock :: PieceNum -> Block -> Process PCF PST B.ByteString + readBlock pn blk = do + c <- liftIO $ channel + syncP =<< sendPC fsCh (ReadBlock pn blk c) + syncP =<< recvP c (const True) + pieceMsg :: PieceNum -> Int -> B.ByteString -> Process PCF PST () + pieceMsg n os bs = do + let sz = B.length bs + blk = Block os sz + e = (n, blk) + q <- gets blockQueue + when (S.member e q) + (do storeBlock n (Block os sz) bs + modify (\s -> s { blockQueue = S.delete e (blockQueue s)})) + -- When e is not a member, the piece may be stray, so ignore it. + -- Perhaps print something here. + cancelMsg n blk = + syncP =<< sendPC outCh (SendQCancel n blk) + considerInterest = do + c <- liftIO channel + pcs <- gets peerPieces + syncP =<< sendPC pieceMgrCh (AskInterested pcs c) + interested <- syncP =<< recvP c (const True) + if interested + then do modify (\s -> s { weInterested = True }) + syncP =<< sendPC outCh (SendQMsg Interested) + else modify (\s -> s { weInterested = False}) fillBlocks = do - choked <- gets peerChoke - unless choked checkWatermark + choked <- gets peerChoke + unless choked checkWatermark checkWatermark = do - q <- gets blockQueue - eg <- gets runningEndgame - let sz = S.size q - mark = if eg then endgameLoMark else loMark - when (sz < mark) - (do - toQueue <- grabBlocks (hiMark - sz) - logDebug $ "Got " ++ show (length toQueue) ++ " blocks: " ++ show toQueue + q <- gets blockQueue + eg <- gets runningEndgame + let sz = S.size q + mark = if eg then endgameLoMark else loMark + when (sz < mark) + (do + toQueue <- grabBlocks (hiMark - sz) + debugP $ "Got " ++ show (length toQueue) ++ " blocks: " ++ show toQueue queuePieces toQueue) - queuePieces toQueue = do - mapM_ pushPiece toQueue - modify (\s -> s { blockQueue = S.union (blockQueue s) (S.fromList toQueue) }) - pushPiece (pn, blk) = - syncP =<< sendPC outCh (SendQMsg $ Request pn blk) - storeBlock n blk bs = - syncP =<< sendPC pieceMgrCh (StoreBlock n blk bs) - grabBlocks n = do - c <- liftIO $ channel - ps <- gets peerPieces - syncP =<< sendPC pieceMgrCh (GrabBlocks n ps c) - blks <- syncP =<< recvP c (const True) - case blks of - Leech blks -> return blks - Endgame blks -> - modify (\s -> s { runningEndgame = True }) >> return blks + queuePieces toQueue = do + mapM_ pushPiece toQueue + modify (\s -> s { blockQueue = S.union (blockQueue s) (S.fromList toQueue) }) + pushPiece (pn, blk) = + syncP =<< sendPC outCh (SendQMsg $ Request pn blk) + storeBlock n blk bs = + syncP =<< sendPC pieceMgrCh (StoreBlock n blk bs) + grabBlocks n = do + c <- liftIO $ channel + ps <- gets peerPieces + syncP =<< sendPC pieceMgrCh (GrabBlocks n ps c) + blks <- syncP =<< recvP c (const True) + case blks of + Leech blks -> return blks + Endgame blks -> + modify (\s -> s { runningEndgame = True }) >> return blks loMark = 10 - endgameLoMark = 1 - hiMark = 15 -- These two values are chosen rather arbitrarily at the moment. + endgameLoMark = 1 + hiMark = 15 -- These three values are chosen rather arbitrarily at the moment. -createPeerPieces :: L.ByteString -> [PieceNum] -createPeerPieces = map fromIntegral . concat . decodeBytes 0 . L.unpack +createPeerPieces :: L.ByteString -> IS.IntSet +createPeerPieces = IS.fromList . map fromIntegral . concat . decodeBytes 0 . L.unpack where decodeByte :: Int -> Word8 -> [Maybe Int] decodeByte soFar w = let dBit n = if testBit w (7-n) @@ -437,24 +410,5 @@ decodeBytes _ [] = [] decodeBytes soFar (w : ws) = catMaybes (decodeByte soFar w) : decodeBytes (soFar + 8) ws - -showPort :: PortID -> String -showPort (PortNumber pn) = show pn -showPort _ = "N/A" - disconnectPeer :: MgrChannel -> ThreadId -> IO () disconnectPeer c t = sync $ transmit c $ Disconnect t - - --- TODO: Consider if this code is correct with what we did to [connect] -{- -listenHandshake :: Handle -> PeerId -> InfoHash -> FSPChannel -> LogChannel - -> MgrChannel - -> IO (Either String ()) -listenHandshake h pid ih fsC logC mgrC = - do r <- initiateHandshake logC h pid ih - case r of - Left err -> return $ Left err - Right (_caps, _rpid) -> do peerP mgrC fsC logC h -- TODO: Coerce with connect - return $ Right () --} diff --git a/src/Process/PeerMgr.hs b/src/Process/PeerMgr.hs --- a/src/Process/PeerMgr.hs +++ b/src/Process/PeerMgr.hs @@ -1,95 +1,167 @@ module Process.PeerMgr ( -- * Types Peer(..) + , PeerMgrMsg(..) + , PeerMgrChannel -- * Interface , start ) where -import Data.List import qualified Data.Map as M -import Data.Ord import Control.Concurrent import Control.Concurrent.CML import Control.Monad.State import Control.Monad.Reader +import Network +import System.IO +import System.Log.Logger + import PeerTypes import Process -import Logging import Process.Peer as Peer import Process.ChokeMgr hiding (start) import Process.FS hiding (start) import Process.PieceMgr hiding (start) import Process.Status hiding (start) +import Protocol.Wire + import Supervisor import Torrent hiding (infoHash) +data PeerMgrMsg = PeersFromTracker [Peer] + | NewIncoming (Handle, HostName, PortNumber) -data CF = CF { peerCh :: Channel [Peer] - , pieceMgrCh :: PieceMgrChannel - , mgrCh :: Channel MgrMessage - , fsCh :: FSPChannel - , peerPool :: SupervisorChan - , chokeMgrCh :: ChokeMgrChannel - , logCh :: LogChannel - } +type PeerMgrChannel = Channel PeerMgrMsg +data CF = CF { peerCh :: PeerMgrChannel + , pieceMgrCh :: PieceMgrChannel + , mgrCh :: Channel MgrMessage + , fsCh :: FSPChannel + , peerPool :: SupervisorChan + , chokeMgrCh :: ChokeMgrChannel + } + instance Logging CF where - getLogger cf = ("PeerMgrP", logCh cf) + logName _ = "Process.PeerMgr" data ST = ST { peersInQueue :: [Peer] , peers :: M.Map ThreadId (Channel PeerMessage) , peerId :: PeerId , infoHash :: InfoHash - } + } -start :: Channel [Peer] -> PeerId -> InfoHash -> PieceMap -> PieceMgrChannel -> FSPChannel - -> LogChannel -> ChokeMgrChannel -> StatusChan -> Int -> SupervisorChan +start :: PeerMgrChannel -> PeerId -> InfoHash -> PieceMap -> PieceMgrChannel -> FSPChannel + -> ChokeMgrChannel -> StatusChan -> Int -> SupervisorChan -> IO ThreadId -start ch pid ih pm pieceMgrC fsC logC chokeMgrC statC nPieces supC = +start ch pid ih pm pieceMgrC fsC chokeMgrC statC nPieces supC = do mgrC <- channel fakeChan <- channel - pool <- liftM snd $ oneForOne "PeerPool" [] logC fakeChan - spawnP (CF ch pieceMgrC mgrC fsC pool chokeMgrC logC) + pool <- liftM snd $ oneForOne "PeerPool" [] fakeChan + spawnP (CF ch pieceMgrC mgrC fsC pool chokeMgrC) (ST [] M.empty pid ih) (catchP (forever lp) - (defaultStopHandler supC)) + (defaultStopHandler supC)) where - lp = do chooseP [trackerPeers, peerEvent] >>= syncP - fillPeers - trackerPeers = do - ev <- recvPC peerCh - wrapP ev (\ps -> - do logDebug "Adding peers to queue" - modify (\s -> s { peersInQueue = ps ++ peersInQueue s })) - peerEvent = do - ev <- recvPC mgrCh - wrapP ev (\msg -> do - case msg of - Connect tid c -> newPeer tid c - Disconnect tid -> removePeer tid) - newPeer tid c = do logDebug $ "Adding new peer " ++ show tid - sendPC chokeMgrCh (AddPeer tid c) >>= syncP - modify (\s -> s { peers = M.insert tid c (peers s)}) - removePeer tid = do logDebug $ "Removing peer " ++ show tid - sendPC chokeMgrCh (RemovePeer tid) >>= syncP - modify (\s -> s { peers = M.delete tid (peers s)}) + lp = do chooseP [incomingPeers, peerEvent] >>= syncP + fillPeers + incomingPeers = + recvWrapPC peerCh (\msg -> + case msg of + PeersFromTracker ps -> do + debugP "Adding peers to queue" + modify (\s -> s { peersInQueue = ps ++ peersInQueue s }) + NewIncoming conn@(h, _, _) -> do + sz <- liftM M.size $ gets peers + if sz < numPeers + then do debugP "New incoming peer, handling" + addIncoming conn + return () + else do debugP "Already too many peers, closing!" + liftIO $ hClose h) + peerEvent = + recvWrapPC mgrCh (\msg -> case msg of + Connect tid c -> newPeer tid c + Disconnect tid -> removePeer tid) + newPeer tid c = do debugP $ "Adding new peer " ++ show tid + sendPC chokeMgrCh (AddPeer tid c) >>= syncP + modify (\s -> s { peers = M.insert tid c (peers s)}) + removePeer tid = do debugP $ "Removing peer " ++ show tid + sendPC chokeMgrCh (RemovePeer tid) >>= syncP + modify (\s -> s { peers = M.delete tid (peers s)}) numPeers = 40 fillPeers = do - sz <- liftM M.size $ gets peers - when (sz < numPeers) - (do q <- gets peersInQueue - let (toAdd, rest) = splitAt (numPeers - sz) q - logDebug $ "Filling with up to " ++ show (numPeers - sz) ++ " peers" - mapM_ addPeer toAdd - modify (\s -> s { peersInQueue = rest })) + sz <- liftM M.size $ gets peers + when (sz < numPeers) + (do q <- gets peersInQueue + let (toAdd, rest) = splitAt (numPeers - sz) q + debugP $ "Filling with up to " ++ show (numPeers - sz) ++ " peers" + mapM_ addPeer toAdd + modify (\s -> s { peersInQueue = rest })) addPeer (Peer hn prt) = do - pid <- gets peerId - ih <- gets infoHash - pool <- asks peerPool - pmC <- asks pieceMgrCh - fsC <- asks fsCh - mgrC <- asks mgrCh - logC <- asks logCh - liftIO $ Peer.connect (hn, prt, pid, ih, pm) pool pmC fsC logC statC mgrC nPieces + pid <- gets peerId + ih <- gets infoHash + pool <- asks peerPool + pmC <- asks pieceMgrCh + fsC <- asks fsCh + mgrC <- asks mgrCh + liftIO $ connect (hn, prt, pid, ih, pm) pool pmC fsC statC mgrC nPieces + addIncoming conn = do + pid <- gets peerId + ih <- gets infoHash + pool <- asks peerPool + pieceMgrC <- asks pieceMgrCh + fsC <- asks fsCh + mgrC <- asks mgrCh + let ihTst = (== ih) + liftIO $ acceptor conn ih ihTst pool pid mgrC pieceMgrC fsC statC pm nPieces + +type ConnectRecord = (HostName, PortID, PeerId, InfoHash, PieceMap) + +connect :: ConnectRecord -> SupervisorChan -> PieceMgrChannel -> FSPChannel + -> StatusChan + -> MgrChannel -> Int + -> IO ThreadId +connect (host, port, pid, ih, pm) pool pieceMgrC fsC statC mgrC nPieces = + spawn (connector >> return ()) + where connector = + do debugM "Process.PeerMgr.connect" $ + "Connecting to " ++ show host ++ " (" ++ showPort port ++ ")" + h <- connectTo host port + debugM "Process.PeerMgr.connect" "Connected, initiating handShake" + r <- initiateHandshake h pid ih + debugM "Process.PeerMgr.connect" "Handshake run" + case r of + Left err -> do debugM "Process.PeerMgr.connect" + ("Peer handshake failure at host " ++ host + ++ " with error " ++ err) + return () + Right (_caps, _rpid) -> + do debugM "Process.PeerMgr.connect" "entering peerP loop code" + supC <- channel -- TODO: Should be linked later on + children <- peerChildren h mgrC pieceMgrC fsC statC pm nPieces + sync $ transmit pool $ SpawnNew (Supervisor $ allForOne "PeerSup" children) + return () + +acceptor (h,hn,pn) ih ihTst pool pid mgrC pieceMgrC fsC statC pm nPieces = + spawn (connector >> return ()) + where connector = do + debugLog "Handling incoming connection" + r <- receiveHandshake h pid ihTst ih + debugLog "RecvHandshake run" + case r of + Left err -> do debugLog ("Incoming Peer handshake failure with " ++ show hn ++ "(" + ++ show pn ++ "), error: " ++ err) + return () + Right (_caps, _rpid) -> + do debugLog "entering peerP loop code" + supC <- channel -- TODO: Should be linked later on + children <- peerChildren h mgrC pieceMgrC fsC statC pm nPieces + sync $ transmit pool $ SpawnNew (Supervisor $ allForOne "PeerSup" children) + return () + debugLog = debugM "Process.PeerMgr.acceptor" + +showPort :: PortID -> String +showPort (PortNumber pn) = show pn +showPort _ = "N/A" diff --git a/src/Process/PieceMgr.hs b/src/Process/PieceMgr.hs --- a/src/Process/PieceMgr.hs +++ b/src/Process/PieceMgr.hs @@ -17,13 +17,13 @@ import qualified Data.ByteString as B import qualified Data.Map as M import qualified Data.Set as S +import qualified Data.IntSet as IS import Prelude hiding (log) import System.Random import System.Random.Shuffle -import Logging import Process.FS hiding (start) import Process.Status as STP hiding (start) import Supervisor @@ -41,8 +41,8 @@ -- Better implementations for selecting among the pending Pieces is probably crucial -- to an effective client, but we keep it simple for now. data PieceDB = PieceDB - { pendingPieces :: [PieceNum] -- ^ Pieces currently pending download - , donePiece :: [PieceNum] -- ^ Pieces that are done + { pendingPieces :: IS.IntSet -- ^ Pieces currently pending download + , donePiece :: IS.IntSet -- ^ Pieces that are done , donePush :: [ChokeInfoMsg] -- ^ Pieces that should be pushed to the Choke Mgr. , inProgress :: M.Map PieceNum InProgressPiece -- ^ Pieces in progress , downloading :: [(PieceNum, Block)] -- ^ Blocks we are currently downloading @@ -71,22 +71,22 @@ -- leeching like normal. The "Endgame mode" is when the client is entering the -- endgame. This means that the Peer should act differently to the blocks. data Blocks = Leech [(PieceNum, Block)] - | Endgame [(PieceNum, Block)] + | Endgame [(PieceNum, Block)] -- | Messages for RPC towards the PieceMgr. -data PieceMgrMsg = GrabBlocks Int [PieceNum] (Channel Blocks) +data PieceMgrMsg = GrabBlocks Int IS.IntSet (Channel Blocks) -- ^ Ask for grabbing some blocks | StoreBlock PieceNum Block B.ByteString -- ^ Ask for storing a block on the file system | PutbackBlocks [(PieceNum, Block)] -- ^ Put these blocks back for retrieval - | AskInterested [PieceNum] (Channel Bool) - -- ^ Ask if any of these pieces are interesting + | AskInterested IS.IntSet (Channel Bool) + -- ^ Ask if any of these pieces are interesting | GetDone (Channel [PieceNum]) - -- ^ Get the pieces which are already done + -- ^ Get the pieces which are already done data ChokeInfoMsg = PieceDone PieceNum - | BlockComplete PieceNum Block + | BlockComplete PieceNum Block | TorrentComplete deriving (Eq, Show) @@ -94,132 +94,132 @@ type ChokeInfoChannel = Channel ChokeInfoMsg data PieceMgrCfg = PieceMgrCfg - { logCh :: LogChannel - , pieceMgrCh :: PieceMgrChannel + { pieceMgrCh :: PieceMgrChannel , fspCh :: FSPChannel , chokeCh :: ChokeInfoChannel , statusCh :: StatusChan } instance Logging PieceMgrCfg where - getLogger cf = ("PieceMgrP", logCh cf) + logName _ = "Process.PieceMgr" type PieceMgrProcess v = Process PieceMgrCfg PieceDB v -start :: LogChannel -> PieceMgrChannel -> FSPChannel -> ChokeInfoChannel -> StatusChan -> PieceDB +start :: PieceMgrChannel -> FSPChannel -> ChokeInfoChannel -> StatusChan -> PieceDB -> SupervisorChan -> IO ThreadId -start logC mgrC fspC chokeC statC db supC = - spawnP (PieceMgrCfg logC mgrC fspC chokeC statC) db - (catchP (forever pgm) - (defaultStopHandler supC)) +start mgrC fspC chokeC statC db supC = + spawnP (PieceMgrCfg mgrC fspC chokeC statC) db + (catchP (forever pgm) + (defaultStopHandler supC)) where pgm = do - assertPieceDB - dl <- gets donePush - (if dl == [] - then receiveEvt - else chooseP [receiveEvt, sendEvt (head dl)]) >>= syncP - sendEvt elem = do - ev <- sendPC chokeCh elem - wrapP ev remDone - remDone :: () -> Process PieceMgrCfg PieceDB () - remDone () = modify (\db -> db { donePush = tail (donePush db) }) + assertPieceDB + dl <- gets donePush + (if dl == [] + then receiveEvt + else chooseP [receiveEvt, sendEvt (head dl)]) >>= syncP + sendEvt elem = do + ev <- sendPC chokeCh elem + wrapP ev remDone + remDone :: () -> Process PieceMgrCfg PieceDB () + remDone () = modify (\db -> db { donePush = tail (donePush db) }) receiveEvt = do - ev <- recvPC pieceMgrCh - wrapP ev (\msg -> - case msg of - GrabBlocks n eligible c -> - do logDebug "Grabbing Blocks" - blocks <- grabBlocks' n eligible - logDebug "Grabbed..." - syncP =<< sendP c blocks - StoreBlock pn blk d -> - do logDebug $ "Storing block: " ++ show (pn, blk) - storeBlock pn blk d - modify (\s -> s { downloading = downloading s \\ [(pn, blk)] }) - endgameBroadcast pn blk - done <- updateProgress pn blk - when done - (do assertPieceComplete pn - pend <- gets pendingPieces - iprog <- gets inProgress - logInfo $ "Piece #" ++ show pn - ++ " completed, there are " - ++ (show $ length pend) ++ " pending " - ++ (show $ M.size iprog) ++ " in progress" - l <- gets infoMap >>= - (\pm -> case M.lookup pn pm of - Nothing -> fail "Storeblock: M.lookup" - Just x -> return $ len x) - sendPC statusCh (CompletedPiece l) >>= syncP - pieceOk <- checkPiece pn - case pieceOk of - Nothing -> - do fail "PieceMgrP: Piece Nonexisting!" - Just True -> do completePiece pn - markDone pn - checkFullCompletion - Just False -> putbackPiece pn) - PutbackBlocks blks -> - mapM_ putbackBlock blks - GetDone c -> do done <- gets donePiece - syncP =<< sendP c done - AskInterested pieces retC -> do - inProg <- liftM (S.fromList . M.keys) $ gets inProgress - pend <- liftM S.fromList $ gets pendingPieces - -- @i@ is the intersection with with we need and the peer has. - let i = S.null $ S.intersection (S.fromList pieces) - $ S.union inProg pend - syncP =<< sendP retC (not i)) - storeBlock n blk contents = syncP =<< (sendPC fspCh $ WriteBlock n blk contents) - endgameBroadcast pn blk = - gets endGaming >>= - flip when (modify (\db -> db { donePush = (BlockComplete pn blk) : donePush db })) - markDone pn = do - modify (\db -> db { donePush = (PieceDone pn) : donePush db }) - checkPiece n = do - ch <- liftIO channel - syncP =<< (sendPC fspCh $ CheckPiece n ch) - syncP =<< recvP ch (const True) + ev <- recvPC pieceMgrCh + wrapP ev (\msg -> + case msg of + GrabBlocks n eligible c -> + do debugP "Grabbing Blocks" + blocks <- grabBlocks' n eligible + debugP "Grabbed..." + syncP =<< sendP c blocks + StoreBlock pn blk d -> + do debugP $ "Storing block: " ++ show (pn, blk) + storeBlock pn blk d + modify (\s -> s { downloading = downloading s \\ [(pn, blk)] }) + endgameBroadcast pn blk + done <- updateProgress pn blk + when done + (do assertPieceComplete pn + pend <- gets pendingPieces + iprog <- gets inProgress + infoP $ "Piece #" ++ show pn + ++ " completed, there are " + ++ (show $ IS.size pend) ++ " pending " + ++ (show $ M.size iprog) ++ " in progress" + l <- gets infoMap >>= + (\pm -> case M.lookup pn pm of + Nothing -> fail "Storeblock: M.lookup" + Just x -> return $ len x) + sendPC statusCh (CompletedPiece l) >>= syncP + pieceOk <- checkPiece pn + case pieceOk of + Nothing -> + do fail "PieceMgrP: Piece Nonexisting!" + Just True -> do completePiece pn + markDone pn + checkFullCompletion + Just False -> putbackPiece pn) + PutbackBlocks blks -> + mapM_ putbackBlock blks + GetDone c -> do done <- liftM IS.toList $ gets donePiece + syncP =<< sendP c done + AskInterested pieces retC -> do + inProg <- liftM (IS.fromList . M.keys) $ gets inProgress + pend <- gets pendingPieces + -- @i@ is the intersection with with we need and the peer has. + let i = IS.null $ IS.intersection pieces + $ IS.union inProg pend + syncP =<< sendP retC (not i)) + storeBlock n blk contents = syncP =<< (sendPC fspCh $ WriteBlock n blk contents) + endgameBroadcast pn blk = + gets endGaming >>= + flip when (modify (\db -> db { donePush = (BlockComplete pn blk) : donePush db })) + markDone pn = do + modify (\db -> db { donePush = (PieceDone pn) : donePush db }) + checkPiece n = do + ch <- liftIO channel + syncP =<< (sendPC fspCh $ CheckPiece n ch) + syncP =<< recvP ch (const True) -- HELPERS ---------------------------------------------------------------------- createPieceDb :: PiecesDoneMap -> PieceMap -> PieceDB createPieceDb mmap pmap = PieceDB pending done [] M.empty [] pmap False - where pending = M.keys $ M.filter (==False) mmap - done = M.keys $ M.filter (==True) mmap + where pending = filt (==False) + done = filt (==True) + filt f = IS.fromList . M.keys $ M.filter f mmap ---------------------------------------------------------------------- -- | The call @completePiece db pn@ will mark that the piece @pn@ is completed completePiece :: PieceNum -> PieceMgrProcess () completePiece pn = modify (\db -> db { inProgress = M.delete pn (inProgress db), - donePiece = pn : donePiece db }) + donePiece = IS.insert pn $ donePiece db }) -- | Handle torrent completion checkFullCompletion :: PieceMgrProcess () checkFullCompletion = do doneP <- gets donePiece im <- gets infoMap - when (M.size im == length doneP) - (do logInfo "Torrent Completed" - sendPC statusCh STP.TorrentCompleted >>= syncP - sendPC chokeCh TorrentComplete >>= syncP) + when (M.size im == IS.size doneP) + (do liftIO $ putStrLn "Torrent Completed" + sendPC statusCh STP.TorrentCompleted >>= syncP + sendPC chokeCh TorrentComplete >>= syncP) -- | The call @putBackPiece db pn@ will mark the piece @pn@ as not being complete -- and put it back into the download queue again. putbackPiece :: PieceNum -> PieceMgrProcess () putbackPiece pn = modify (\db -> db { inProgress = M.delete pn (inProgress db), - pendingPieces = pn : pendingPieces db }) + pendingPieces = IS.insert pn $ pendingPieces db }) -- | Put back a block for downloading. -- TODO: This is rather slow, due to the (\\) call, but hopefully happens rarely. putbackBlock :: (PieceNum, Block) -> PieceMgrProcess () putbackBlock (pn, blk) = do done <- gets donePiece - unless (pn `elem` done) -- Happens at endgame, stray block + unless (IS.member pn done) -- Happens at endgame, stray block $ modify (\db -> db { inProgress = ndb (inProgress db) - , downloading = downloading db \\ [(pn, blk)]}) + , downloading = downloading db \\ [(pn, blk)]}) where ndb db = M.alter f pn db -- The first of these might happen in the endgame f Nothing = fail "The 'impossible' happened" @@ -231,18 +231,18 @@ assertPieceComplete pn = do inprog <- gets inProgress ipp <- case M.lookup pn inprog of - Nothing -> fail "assertPieceComplete: Could not lookup piece number" - Just x -> return x + Nothing -> fail "assertPieceComplete: Could not lookup piece number" + Just x -> return x dl <- gets downloading pm <- gets infoMap sz <- case M.lookup pn pm of - Nothing -> fail "assertPieceComplete: Could not lookup piece in piecemap" - Just x -> return $ len x + Nothing -> fail "assertPieceComplete: Could not lookup piece in piecemap" + Just x -> return $ len x unless (assertAllDownloaded dl pn) (fail "Could not assert that all pieces were downloaded when completing a piece") unless (assertComplete ipp sz) (fail $ "Could not assert completion of the piece #" ++ show pn - ++ " with block state " ++ show ipp) + ++ " with block state " ++ show ipp) where assertComplete ip sz = checkContents 0 (fromIntegral sz) (S.toAscList (ipHaveBlocks ip)) -- Check a single block under assumptions of a cursor at offs checkBlock (offs, left, state) blk = (offs + blockSize blk, @@ -251,7 +251,7 @@ checkContents os l blks = case foldl checkBlock (os, l, True) blks of (_, 0, True) -> True _ -> False - assertAllDownloaded blocks pn = all (\(pn', _) -> pn /= pn') blocks + assertAllDownloaded blocks pn = all (\(pn', _) -> pn /= pn') blocks -- | Update the progress on a Piece. When we get a block from the piece, we will -- track this in the Piece Database. This function returns a pair @(complete, nDb)@ @@ -261,8 +261,8 @@ updateProgress pn blk = do ipdb <- gets inProgress case M.lookup pn ipdb of - Nothing -> do logDebug "updateProgress can't find progress block, error?" - return False + Nothing -> do debugP "updateProgress can't find progress block, error?" + return False Just pg -> let blkSet = ipHaveBlocks pg in if blk `S.member` blkSet @@ -271,9 +271,9 @@ -- at times else checkComplete pg { ipHaveBlocks = S.insert blk blkSet } where checkComplete pg = do - modify (\db -> db { inProgress = M.adjust (const pg) pn (inProgress db) }) - logDebug $ "Iphave : " ++ show (ipHave pg) ++ " ipDone: " ++ show (ipDone pg) - return (ipHave pg == ipDone pg) + modify (\db -> db { inProgress = M.adjust (const pg) pn (inProgress db) }) + debugP $ "Iphave : " ++ show (ipHave pg) ++ " ipDone: " ++ show (ipDone pg) + return (ipHave pg == ipDone pg) ipHave = S.size . ipHaveBlocks blockPiece :: BlockSize -> PieceSize -> [Block] @@ -289,144 +289,129 @@ -- the @n@. In doing so, it will only consider pieces in @eligible@. It returns a -- pair @(blocks, db')@, where @blocks@ are the blocks it picked and @db'@ is the resulting -- db with these blocks removed. -grabBlocks' :: Int -> [PieceNum] -> PieceMgrProcess Blocks +grabBlocks' :: Int -> IS.IntSet -> PieceMgrProcess Blocks grabBlocks' k eligible = do blocks <- tryGrabProgress k eligible [] pend <- gets pendingPieces - if blocks == [] && pend == [] - then do blks <- grabEndGame k (S.fromList eligible) - modify (\db -> db { endGaming = True }) - logDebug $ "PieceMgr entered endgame." - return $ Endgame blks - else do modify (\s -> s { downloading = blocks ++ (downloading s) }) - return $ Leech blocks + if blocks == [] && IS.null pend + then do blks <- grabEndGame k eligible + modify (\db -> db { endGaming = True }) + debugP $ "PieceMgr entered endgame." + return $ Endgame blks + else do modify (\s -> s { downloading = blocks ++ (downloading s) }) + return $ Leech blocks where -- Grabbing blocks is a state machine implemented by tail calls -- Try grabbing pieces from the pieces in progress first tryGrabProgress 0 _ captured = return captured tryGrabProgress n ps captured = do - inprog <- gets inProgress - case ps `intersect` fmap fst (M.toList inprog) of - [] -> tryGrabPending n ps captured - (h:_) -> grabFromProgress n ps h captured + inProg <- gets inProgress + let is = IS.intersection ps (IS.fromList $ M.keys inProg) + case IS.null is of + True -> tryGrabPending n ps captured + False -> grabFromProgress n ps (head $ IS.elems is) captured -- The Piece @p@ was found, grab it grabFromProgress n ps p captured = do inprog <- gets inProgress - ipp <- case M.lookup p inprog of - Nothing -> fail "grabFromProgress: could not lookup piece" - Just x -> return x + ipp <- case M.lookup p inprog of + Nothing -> fail "grabFromProgress: could not lookup piece" + Just x -> return x let (grabbed, rest) = splitAt n (ipPendingBlocks ipp) nIpp = ipp { ipPendingBlocks = rest } -- This rather ugly piece of code should be substituted with something better if grabbed == [] -- All pieces are taken, try the next one. - then tryGrabProgress n (ps \\ [p]) captured + then tryGrabProgress n (IS.delete p ps) captured else do modify (\db -> db { inProgress = M.insert p nIpp inprog }) - tryGrabProgress (n - length grabbed) ps ([(p,g) | g <- grabbed] ++ captured) + tryGrabProgress (n - length grabbed) ps ([(p,g) | g <- grabbed] ++ captured) -- Try grabbing pieces from the pending blocks tryGrabPending n ps captured = do - pending <- gets pendingPieces - case ps `intersect` pending of - [] -> return $ captured -- No (more) pieces to download, return - ls -> do - h <- pickRandom ls - infMap <- gets infoMap - inProg <- gets inProgress + pending <- gets pendingPieces + let isn = IS.intersection ps pending + case IS.null isn of + True -> return $ captured -- No (more) pieces to download, return + False -> do + h <- pickRandom (IS.toList isn) + infMap <- gets infoMap + inProg <- gets inProgress blockList <- createBlock h let sz = length blockList - ipp = InProgressPiece sz S.empty blockList - modify (\db -> db { pendingPieces = pendingPieces db \\ [h], + ipp = InProgressPiece sz S.empty blockList + modify (\db -> db { pendingPieces = IS.delete h (pendingPieces db), inProgress = M.insert h ipp inProg }) - tryGrabProgress n ps captured + tryGrabProgress n ps captured grabEndGame n ps = do -- In endgame we are allowed to grab from the downloaders - dls <- liftM (filter (\(p, _) -> S.member p ps)) $ gets downloading - g <- liftIO newStdGen + dls <- liftM (filter (\(p, _) -> IS.member p ps)) $ gets downloading + g <- liftIO newStdGen let shuffled = shuffle' dls (length dls) g - return $ take n shuffled + return $ take n shuffled pickRandom pieces = do - n <- liftIO $ getStdRandom (\gen -> randomR (0, length pieces - 1) gen) - return $ pieces !! n + n <- liftIO $ getStdRandom (\gen -> randomR (0, length pieces - 1) gen) + return $ pieces !! n createBlock :: Int -> PieceMgrProcess [Block] createBlock pn = do - gets infoMap >>= (\im -> case M.lookup pn im of - Nothing -> fail "createBlock: could not lookup piece" - Just ipp -> return $ cBlock ipp) + gets infoMap >>= (\im -> case M.lookup pn im of + Nothing -> fail "createBlock: could not lookup piece" + Just ipp -> return $ cBlock ipp) where cBlock = blockPiece defaultBlockSize . fromInteger . len assertPieceDB :: PieceMgrProcess () -assertPieceDB = assertPending >> assertDone >> assertInProgress >> assertDownloading +assertPieceDB = assertSets >> assertInProgress >> assertDownloading where -- If a piece is pending in the database, we have the following rules: -- - -- - It is not finished. + -- - It is not done. -- - It is not being downloaded -- - It is not in progresss. - assertPending = do - pending <- gets pendingPieces - mapM_ checkPending pending - checkPending pn = do - done <- gets donePiece - when (pn `elem` done) - (fail $ "Pending piece " ++ show pn ++ " is in the done list") - down <- gets downloading - when (pn `elem` map fst down) - (fail $ "Pending piece " ++ show pn ++ " is in the downloading list") - inProg <- gets inProgress - when (case M.lookup pn inProg of - Nothing -> False - Just _ -> True) - (fail $ "Pending piece " ++ show pn ++ " is in the progress map") + -- -- If a piece is done, we have the following rules: -- - -- - It is not pending. -- - It is not in progress. -- - There are no more downloading blocks. - assertDone = do - done <- gets donePiece - mapM_ checkDone done - checkDone pn = do - pending <- gets pendingPieces - when (pn `elem` pending) - (fail $ "Done piece " ++ show pn ++ " is in the pending list") - down <- gets downloading - when (pn `elem` map fst down) - (fail $ "Done piece " ++ show pn ++ " is in the downloading list") - inProg <- gets inProgress - when (case M.lookup pn inProg of - Nothing -> False - Just _ -> True) - (fail $ "Done piece " ++ show pn ++ " is in the progress map") + assertSets = do + pending <- gets pendingPieces + done <- gets donePiece + down <- liftM (IS.fromList . map fst) $ gets downloading + iprog <- liftM (IS.fromList . M.keys) $ gets inProgress + let pdis = IS.intersection pending done + pdownis = IS.intersection pending down + piprogis = IS.intersection pending iprog + doneprogis = IS.intersection done iprog + donedownis = IS.intersection done down + unless (IS.null pdis) + (fail $ "Pending/Done violation of pieces: " ++ show pdis) + unless (IS.null pdownis) + (fail $ "Pending/Downloading violation of pieces: " ++ show pdownis) + unless (IS.null piprogis) + (fail $ "Pending/InProgress violation of pieces: " ++ show piprogis) + unless (IS.null doneprogis) + (fail $ "Done/InProgress violation of pieces: " ++ show doneprogis) + unless (IS.null donedownis) + (fail $ "Done/Downloading violation of pieces: " ++ show donedownis) + -- If a piece is in Progress, we have: -- - -- - The piece is not Done - -- - The piece is not pending -- - There is a relationship with what pieces are downloading -- - If a block is ipPending, it is not in the downloading list -- - If a block is ipHave, it is not in the downloading list assertInProgress = do - inProg <- gets inProgress - mapM_ checkInProgress $ M.toList inProg + inProg <- gets inProgress + mapM_ checkInProgress $ M.toList inProg checkInProgress (pn, ipp) = do - when ( (S.size $ ipHaveBlocks ipp) >= ipDone ipp) - (fail $ "Piece in progress " ++ show pn - ++ " has downloaded more blocks than the piece has") - done <- gets donePiece - when (pn `elem` done) - (fail $ "Piece in progress " ++ show pn ++ " is in the done list") - pending <- gets pendingPieces - when (pn `elem` pending) - (fail $ "Piece in progress " ++ show pn ++ " is in the pending list") + when ( (S.size $ ipHaveBlocks ipp) >= ipDone ipp) + (fail $ "Piece in progress " ++ show pn + ++ " has downloaded more blocks than the piece has") assertDownloading = do - down <- gets downloading - mapM_ checkDownloading down + down <- gets downloading + mapM_ checkDownloading down checkDownloading (pn, blk) = do - prog <- gets inProgress - case M.lookup pn prog of - Nothing -> fail $ "Piece " ++ show pn ++ " not in progress while We think it was" - Just ipp -> do - when (blk `elem` ipPendingBlocks ipp) - (fail $ "P/Blk " ++ show (pn, blk) ++ " is in the Pending Block list") - when (S.member blk $ ipHaveBlocks ipp) - (fail $ "P/Blk " ++ show (pn, blk) ++ " is in the HaveBlocks set") + prog <- gets inProgress + case M.lookup pn prog of + Nothing -> fail $ "Piece " ++ show pn ++ " not in progress while We think it was" + Just ipp -> do + when (blk `elem` ipPendingBlocks ipp) + (fail $ "P/Blk " ++ show (pn, blk) ++ " is in the Pending Block list") + when (S.member blk $ ipHaveBlocks ipp) + (fail $ "P/Blk " ++ show (pn, blk) ++ " is in the HaveBlocks set") diff --git a/src/Process/Status.hs b/src/Process/Status.hs --- a/src/Process/Status.hs +++ b/src/Process/Status.hs @@ -1,29 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -- | The status code runs a Status Process. This process keeps track -- of a number of interval valies for a given torrent file and it -- periodically updates the tracker process with the relevant @@ -47,33 +21,30 @@ import Control.Concurrent.CML import Control.Monad.State -import Control.Monad.Reader import Prelude hiding (log) -import Logging import Process import Supervisor import Torrent data StatusMsg = TrackerStat { trackIncomplete :: Maybe Integer - , trackComplete :: Maybe Integer } - | CompletedPiece Integer - | PeerStat { peerUploaded :: Integer - , peerDownloaded :: Integer } - | TorrentCompleted + , trackComplete :: Maybe Integer } + | CompletedPiece Integer + | PeerStat { peerUploaded :: Integer + , peerDownloaded :: Integer } + | TorrentCompleted type StatusChan = Channel StatusMsg -- | TrackerChannel is the channel of the tracker data TrackerMsg = Stop | TrackerTick Integer | Start | Complete -data CF = CF { logCh :: LogChannel - , statusCh :: Channel StatusMsg - , trackerCh1 :: Channel TrackerMsg - , trackerCh :: Channel ST } +data CF = CF { statusCh :: Channel StatusMsg + , trackerCh1 :: Channel TrackerMsg + , trackerCh :: Channel ST } instance Logging CF where - getLogger cf = ("StatusP", logCh cf) + logName _ = "Process.Status" data ST = ST { uploaded :: Integer, downloaded :: Integer, @@ -81,36 +52,35 @@ incomplete :: Maybe Integer, complete :: Maybe Integer, state :: TorrentState } + deriving Show -- | Start a new Status process with an initial torrent state and a -- channel on which to transmit status updates to the tracker. -start :: LogChannel -> Integer -> TorrentState -> Channel ST +start :: Integer -> TorrentState -> Channel ST -> Channel StatusMsg -> Channel TrackerMsg -> SupervisorChan -> IO ThreadId -start logC l tState trackerC statusC trackerC1 supC = do - spawnP (CF logC statusC trackerC1 trackerC) (ST 0 0 l Nothing Nothing tState) - (catchP (foreverP pgm) (defaultStopHandler supC)) +start l tState trackerC statusC trackerC1 supC = do + spawnP (CF statusC trackerC1 trackerC) (ST 0 0 l Nothing Nothing tState) + (catchP (foreverP pgm) (defaultStopHandler supC)) where - pgm = do ev <- chooseP [sendEvent, recvEvent] - syncP ev + pgm = syncP =<< chooseP [sendEvent, recvEvent] sendEvent = get >>= sendPC trackerCh recvEvent = do evt <- recvPC statusCh - wrapP evt (\m -> - case m of - TrackerStat ic c -> - modify (\s -> s { incomplete = ic, complete = c }) - CompletedPiece bytes -> do - logDebug "StatusProcess updated left" - modify (\s -> s { left = (left s) - bytes }) - PeerStat up down -> do - modify (\s -> s { uploaded = (uploaded s) + up, - downloaded = (downloaded s) + down }) - u <- gets uploaded - d <- gets downloaded - logDebug $ "StatusProcess up/down count: " ++ show u ++ ", " ++ show d - TorrentCompleted -> do - logDebug "TorrentCompletion at StatusP" - l <- gets left - when (l /= 0) (fail "Warning: Left is not 0 upon Torrent Completion") - syncP =<< sendPC trackerCh1 Complete - modify (\s -> s { state = Seeding })) - + wrapP evt (\m -> + case m of + TrackerStat ic c -> + modify (\s -> s { incomplete = ic, complete = c }) + CompletedPiece bytes -> do + debugP "StatusProcess updated left" + modify (\s -> s { left = (left s) - bytes }) + PeerStat up down -> do + modify (\s -> s { uploaded = (uploaded s) + up, + downloaded = (downloaded s) + down }) + u <- gets uploaded + d <- gets downloaded + debugP $ "StatusProcess up/down count: " ++ show u ++ ", " ++ show d + TorrentCompleted -> do + debugP "TorrentCompletion at StatusP" + l <- gets left + when (l /= 0) (fail "Warning: Left is not 0 upon Torrent Completion") + syncP =<< sendPC trackerCh1 Complete + modify (\s -> s { state = Seeding })) diff --git a/src/Process/Tracker.hs b/src/Process/Tracker.hs --- a/src/Process/Tracker.hs +++ b/src/Process/Tracker.hs @@ -8,6 +8,8 @@ -- it to the user. -- module Process.Tracker + ( start + ) where import Control.Applicative @@ -29,7 +31,6 @@ import Protocol.BCode as BCode hiding (encode) -import Logging import Process import Supervisor import Torrent @@ -75,15 +76,14 @@ -- | Configuration of the tracker process data CF = CF { - logCh :: LogChannel - , statusCh :: Channel Status.ST + statusCh :: Channel Status.ST , statusPCh :: Channel Status.StatusMsg , trackerMsgCh :: Channel Status.TrackerMsg - , peerMgrCh :: Channel [PeerMgr.Peer] + , peerMgrCh :: PeerMgr.PeerMgrChannel } instance Logging CF where - getLogger cf = ("TrackerP", logCh cf) + logName _ = "Process.Tracker" -- | Internal state of the tracker CHP process data ST = ST { @@ -95,30 +95,35 @@ , nextTick :: Integer } -start :: TorrentInfo -> PeerId -> PortID -> LogChannel -> Channel Status.ST - -> Channel Status.StatusMsg -> Channel Status.TrackerMsg -> Channel [PeerMgr.Peer] +start :: TorrentInfo -> PeerId -> PortID -> Channel Status.ST + -> Channel Status.StatusMsg -> Channel Status.TrackerMsg -> PeerMgr.PeerMgrChannel -> SupervisorChan -> IO ThreadId -start ti pid port logC sc statusC msgC pc supC = +start ti pid port sc statusC msgC pc supC = do tm <- getPOSIXTime - spawnP (CF logC sc statusC msgC pc) (ST ti pid Stopped port tm 0) - (catchP (forever loop) - (defaultStopHandler supC)) -- TODO: Gracefully close down here! - -loop :: Process CF ST () -loop = do msg <- recvPC trackerMsgCh >>= syncP - logDebug $ "Got tracker event" - case msg of - Status.TrackerTick x -> - do t <- gets nextTick - when (x+1 == t) talkTracker - Status.Stop -> - modify (\s -> s { state = Stopped }) >> talkTracker - Status.Start -> - modify (\s -> s { state = Started }) >> talkTracker - Status.Complete -> - modify (\s -> s { state = Completed }) >> talkTracker + spawnP (CF sc statusC msgC pc) (ST ti pid Stopped port tm 0) + (cleanupP (forever loop) + (defaultStopHandler supC) + stopEvent) where - talkTracker = pokeTracker >>= timerUpdate + stopEvent :: Process CF ST () + stopEvent = do + debugP "Stopping... telling tracker" + modify (\s -> s { state = Stopped }) >> talkTracker + loop :: Process CF ST () + loop = do + msg <- recvPC trackerMsgCh >>= syncP + debugP $ "Got tracker event" + case msg of + Status.TrackerTick x -> + do t <- gets nextTick + when (x+1 == t) talkTracker + Status.Stop -> + modify (\s -> s { state = Stopped }) >> talkTracker + Status.Start -> + modify (\s -> s { state = Started }) >> talkTracker + Status.Complete -> + modify (\s -> s { state = Completed }) >> talkTracker + talkTracker = pokeTracker >>= timerUpdate eventTransition :: Process CF ST () eventTransition = do @@ -126,48 +131,48 @@ modify (\s -> s { state = newS st}) where newS st = case st of - Running -> Running - Stopped -> Stopped - Completed -> Running - Started -> Running + Running -> Running + Stopped -> Stopped + Completed -> Running + Started -> Running -- | Poke the tracker. It returns the new timer intervals to use pokeTracker :: Process CF ST (Integer, Maybe Integer) pokeTracker = do upDownLeft <- syncP =<< recvPC statusCh url <- buildRequestURL upDownLeft - logDebug $ "Request URL: " ++ url + debugP $ "Request URL: " ++ url uri <- case parseURI url of - Nothing -> fail $ "Could not parse the url " ++ url - Just u -> return u + Nothing -> fail $ "Could not parse the url " ++ url + Just u -> return u resp <- trackerRequest uri case resp of - Left err -> do logInfo $ "Tracker HTTP Error: " ++ err - return (failTimerInterval, Just failTimerInterval) - Right (ResponseWarning wrn) -> - do logInfo $ "Tracker Warning Response: " ++ fromBS wrn - return (failTimerInterval, Just failTimerInterval) + Left err -> do infoP $ "Tracker HTTP Error: " ++ err + return (failTimerInterval, Just failTimerInterval) + Right (ResponseWarning wrn) -> + do infoP $ "Tracker Warning Response: " ++ fromBS wrn + return (failTimerInterval, Just failTimerInterval) Right (ResponseError err) -> - do logInfo $ "Tracker Error Response: " ++ fromBS err - return (failTimerInterval, Just failTimerInterval) + do infoP $ "Tracker Error Response: " ++ fromBS err + return (failTimerInterval, Just failTimerInterval) Right (ResponseDecodeError err) -> - do logInfo $ "Response Decode error: " ++ fromBS err - return (failTimerInterval, Just failTimerInterval) - Right bc -> do sendPC peerMgrCh (newPeers bc) >>= syncP - let trackerStats = Status.TrackerStat { Status.trackComplete = completeR bc, - Status.trackIncomplete = incompleteR bc } - sendPC statusPCh trackerStats >>= syncP - eventTransition - return (timeoutInterval bc, timeoutMinInterval bc) + do infoP $ "Response Decode error: " ++ fromBS err + return (failTimerInterval, Just failTimerInterval) + Right bc -> do sendPC peerMgrCh (PeerMgr.PeersFromTracker $ newPeers bc) >>= syncP + let trackerStats = Status.TrackerStat { Status.trackComplete = completeR bc, + Status.trackIncomplete = incompleteR bc } + sendPC statusPCh trackerStats >>= syncP + eventTransition + return (timeoutInterval bc, timeoutMinInterval bc) timerUpdate :: (Integer, Maybe Integer) -> Process CF ST () timerUpdate (timeout, minTimeout) = do st <- gets state when (st == Running) - (do t <- tick - ch <- asks trackerMsgCh + (do t <- tick + ch <- asks trackerMsgCh Timer.register timeout (Status.TrackerTick t) ch - logDebug $ "Set timer to: " ++ show timeout) + debugP $ "Set timer to: " ++ show timeout) where tick = do t <- gets nextTick modify (\s -> s { nextTick = t + 1 }) return t @@ -211,14 +216,14 @@ (2,_,_) -> case BCode.decode . toBS . rspBody $ r of Left pe -> return $ Left (show pe) - Right bc -> do logDebug $ "Response: " ++ BCode.prettyPrint bc + Right bc -> do debugP $ "Response: " ++ BCode.prettyPrint bc return $ Right $ processResultDict bc (3,_,_) -> case findHeader HdrLocation r of Nothing -> return $ Left (show r) Just newURL -> case parseURI newURL of - Nothing -> return $ Left (show newURL) - Just uri -> trackerRequest uri + Nothing -> return $ Left (show newURL) + Just uri -> trackerRequest uri _ -> return $ Left (show r) where request = Request {rqURI = uri, rqMethod = GET, @@ -228,32 +233,32 @@ -- Construct a new request URL. Perhaps this ought to be done with the HTTP client library buildRequestURL :: Status.ST -> Process CF ST String buildRequestURL ss = do ti <- gets torrentInfo - hdrs <- headers - let hl = concat $ hlist hdrs - return $ concat [fromBS $ announceURL ti, "?", hl] + hdrs <- headers + let hl = concat $ hlist hdrs + return $ concat [fromBS $ announceURL ti, "?", hl] where hlist x = intersperse "&" $ map (\(k,v) -> k ++ "=" ++ v) x headers = do - s <- get - p <- prt - return $ [("info_hash", rfc1738Encode $ infoHash $ torrentInfo s), + s <- get + p <- prt + return $ [("info_hash", rfc1738Encode $ infoHash $ torrentInfo s), ("peer_id", rfc1738Encode $ peerId s), ("uploaded", show $ Status.uploaded ss), ("downloaded", show $ Status.downloaded ss), ("left", show $ Status.left ss), ("port", show p), ("compact", "1")] ++ - (trackerfyEvent $ state s) + (trackerfyEvent $ state s) prt :: Process CF ST Integer prt = do lp <- gets localPort - case lp of - PortNumber pnum -> return $ fromIntegral pnum + case lp of + PortNumber pnum -> return $ fromIntegral pnum _ -> do fail "Unknown port type" - trackerfyEvent ev = - case ev of - Running -> [] - Completed -> [("event", "completed")] - Started -> [("event", "started")] - Stopped -> [("event", "stopped")] + trackerfyEvent ev = + case ev of + Running -> [] + Completed -> [("event", "completed")] + Started -> [("event", "started")] + Stopped -> [("event", "stopped")] -- Carry out URL-encoding of a string. Note that the clients seems to do it the wrong way -- so we explicitly code it up here in the same wrong way, jlouis. diff --git a/src/Protocol/BCode.hs b/src/Protocol/BCode.hs --- a/src/Protocol/BCode.hs +++ b/src/Protocol/BCode.hs @@ -1,30 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - - -- | Add a module description here -- also add descriptions to each function. module Protocol.BCode @@ -56,7 +29,9 @@ trackerWarning, trackerError, toBS, - fromBS ) + fromBS, + --Tests + testSuite) where import Control.Monad @@ -64,9 +39,8 @@ import qualified Data.ByteString.Lazy as L import qualified Data.ByteString as B import Data.Char -import Data.Int + import Data.List -import Data.Maybe import qualified Data.Map as M import Text.PrettyPrint.HughesPJ hiding (char) @@ -74,7 +48,14 @@ import Data.Serialize.Put import Data.Serialize.Get +import Test.QuickCheck +import Test.Framework +import Test.Framework.Providers.QuickCheck2 +import Test.Framework.Providers.HUnit +import Test.HUnit hiding (Path) + import Digest +import TestInstance() -- for instances only -- | BCode represents the structure of a bencoded file data BCode = BInt Integer -- ^ An integer @@ -83,6 +64,19 @@ | BDict (M.Map B.ByteString BCode) -- ^ A key, value map deriving (Show, Eq) +instance Arbitrary BCode where + arbitrary = sized bc' + where bc' :: Int -> Gen BCode + bc' 0 = oneof [BInt <$> arbitrary, + BString <$> arbitrary] + bc' n | n > 0 = + oneof [BInt <$> arbitrary, + BString <$> arbitrary, + BArray <$> sequence (replicate n $ bc' (n `div` 8)), + do keys <- vectorOf n arbitrary + values <- sequence (replicate n $ bc' (n `div` 8)) + return $ BDict $ M.fromList $ zip keys values] + data Path = PString B.ByteString | PInt Int @@ -202,8 +196,8 @@ hashInfoDict :: BCode -> IO Digest hashInfoDict bc = do ih <- case info bc of - Nothing -> fail "Could not find infoHash" - Just x -> return x + Nothing -> fail "Could not find infoHash" + Just x -> return x let encoded = encode ih digest $ L.fromChunks $ [encoded] @@ -329,14 +323,33 @@ prettyPrint = render . pp -testDecodeEncodeProp1 :: BCode -> Bool -testDecodeEncodeProp1 m = - let encoded = encode m +toBDict :: [(String,BCode)] -> BCode +toBDict = BDict . M.fromList . map (\(k,v) -> ((toBS k),v)) + +toBString :: String -> BCode +toBString = BString . toBS + + +-- TESTS + + +testSuite = testGroup "Protocol/BCode" + [ testProperty "QC encode-decode/id" propEncodeDecodeId, + testCase "HUnit encode-decode/id" testDecodeEncodeProp1 ] + +propEncodeDecodeId :: BCode -> Bool +propEncodeDecodeId bc = + let encoded = encode bc decoded = decode encoded - in case decoded of - Left _ -> False - Right m' -> m == m' + in + Right bc == decoded +testDecodeEncodeProp1 = + let encoded = encode testData + decoded = decode encoded + in + assertEqual "for encode/decode identify" (Right testData) decoded + testData = [BInt 123, BInt (-123), BString (toBS "Hello"), @@ -353,13 +366,4 @@ ]) ] ] - -toBDict :: [(String,BCode)] -> BCode -toBDict = BDict . M.fromList . map (\(k,v) -> ((toBS k),v)) - -toBString :: String -> BCode -toBString = BString . toBS - - - diff --git a/src/Protocol/Wire.hs b/src/Protocol/Wire.hs --- a/src/Protocol/Wire.hs +++ b/src/Protocol/Wire.hs @@ -5,6 +5,16 @@ module Protocol.Wire + ( Message(..) + , encodePacket + , decodeMsg + , constructBitField + -- Handshaking + , initiateHandshake + , receiveHandshake + -- Tests + , testSuite + ) where import Control.Applicative hiding (empty) @@ -18,11 +28,14 @@ import Data.Serialize.Put import Data.Serialize.Get -import Data.Word import Data.Char import System.IO +import System.Log.Logger -import Logging +import Test.Framework +import Test.Framework.Providers.QuickCheck2 +import Test.QuickCheck + import Torrent ------------------------------------------------------------ @@ -43,7 +56,17 @@ | Port Integer deriving (Eq, Show) +instance Arbitrary Message where + arbitrary = oneof [return KeepAlive, return Choke, return Unchoke, return Interested, + return NotInterested, + Have <$> arbitrary, + BitField <$> arbitrary, + Request <$> arbitrary <*> arbitrary, + Piece <$> arbitrary <*> arbitrary <*> arbitrary, + Cancel <$> arbitrary <*> arbitrary, + Port <$> choose (0,16383)] + data HandShake = HandShake String -- Protocol Header Word64 -- Extension Bias @@ -70,8 +93,8 @@ encodePacket :: Message -> B.ByteString encodePacket m = mconcat [szEnc, mEnc] where mEnc = encode m - sz = B.length mEnc - szEnc = runPut . p32be $ sz + sz = B.length mEnc + szEnc = runPut . p32be $ sz instance Serialize Message where put KeepAlive = return () @@ -138,24 +161,28 @@ toLBS :: String -> L.ByteString toLBS = L.pack . map toW8 +fromLBS :: L.ByteString -> String +fromLBS = map (chr . fromIntegral) . L.unpack + toW8 :: Char -> Word8 toW8 = fromIntegral . ord + -- | Receive the header parts from the other end -receiveHeader :: Handle -> Int -> InfoHash +receiveHeader :: Handle -> Int -> (InfoHash -> Bool) -> IO (Either String ([Capabilities], L.ByteString)) -receiveHeader h sz ih = parseHeader `fmap` B.hGet h sz - where parseHeader = runGet (headerParser ih) +receiveHeader h sz ihTst = parseHeader `fmap` B.hGet h sz + where parseHeader = runGet (headerParser ihTst) -headerParser :: InfoHash -> Get ([Capabilities], L.ByteString) -headerParser ih = do +headerParser :: (InfoHash -> Bool) -> Get ([Capabilities], L.ByteString) +headerParser ihTst = do hdSz <- getWord8 when (fromIntegral hdSz /= protocolHeaderSize) $ fail "Wrong header size" protoString <- getByteString protocolHeaderSize when (protoString /= toBS protocolHeader) $ fail "Wrong protocol header" caps <- getWord64be - ihR <- getLazyByteString 20 - when (ihR /= toLBS ih) $ fail "Wrong InfoHash" + ihR <- liftM fromLBS $ getLazyByteString 20 + unless (ihTst ihR) $ fail "Wrong InfoHash" pid <- getLazyByteString 20 return (decodeCapabilities caps, pid) @@ -165,28 +192,41 @@ decodeCapabilities _ = [] -- | Initiate a handshake on a socket -initiateHandshake :: LogChannel -> Handle -> PeerId -> InfoHash +initiateHandshake :: Handle -> PeerId -> InfoHash -> IO (Either String ([Capabilities], L.ByteString)) -initiateHandshake logC handle peerid infohash = do - logMsg logC "Sending off handshake message" +initiateHandshake handle peerid infohash = do + debugM "Protocol.Wire" "Sending off handshake message" L.hPut handle msg hFlush handle - logMsg logC "Receiving handshake from other end" - receiveHeader handle sz infohash -- TODO: Exceptions - where msg = L.fromChunks . map runPut $ [putLazyByteString protocolHandshake, - putLazyByteString $ toLBS infohash, - putByteString . toBS $ peerid] + debugM "Protocol.Wire" "Receiving handshake from other end" + receiveHeader handle sz (== infohash) -- TODO: Exceptions ? + where msg = handShakeMessage peerid infohash sz = fromIntegral (L.length msg) --- --- -- TESTS -testDecodeEncodeProp1 :: Message -> Bool -testDecodeEncodeProp1 m = - let encoded = encode m - decoded = decode encoded - in case decoded of - Left _ -> False - Right m' -> m == m' +-- | Construct a default handshake message from a PeerId and an InfoHash +handShakeMessage :: PeerId -> InfoHash -> L.ByteString +handShakeMessage pid ih = + L.fromChunks . map runPut $ [putLazyByteString protocolHandshake, + putLazyByteString $ toLBS ih, + putByteString . toBS $ pid] + +-- | Receive a handshake on a socket +receiveHandshake :: Handle -> PeerId -> (InfoHash -> Bool) -> InfoHash + -> IO (Either String ([Capabilities], L.ByteString)) +receiveHandshake h pid ihTst ih = do + debugM "Protocol.Wire" "Receiving handshake from other end" + r <- receiveHeader h sz ihTst -- TODO: Exceptions ? + case r of + Left err -> return $ Left err + Right (caps, rpid) -> + do debugM "Protocol.Wire" "Sending back handshake message" + L.hPut h msg + hFlush h + return $ Right (caps, rpid) + where msg = handShakeMessage pid ih + sz = fromIntegral (L.length msg) + + -- | The call @constructBitField pieces@ will return the a ByteString suitable for inclusion in a -- BITFIELD message to a peer. constructBitField :: Int -> [PieceNum] -> L.ByteString @@ -198,30 +238,27 @@ in if length first /= 8 then error "Wront bitfield" else bytify first : build rest - bytify [b7,b6,b5,b4,b3,b2,b1,b0] = sum [if b0 then 1 else 0, - if b1 then 2 else 0, - if b2 then 4 else 0, - if b3 then 8 else 0, - if b4 then 16 else 0, - if b5 then 32 else 0, - if b6 then 64 else 0, + bytify [b7,b6,b5,b4,b3,b2,b1,b0] = sum [if b0 then 1 else 0, + if b1 then 2 else 0, + if b2 then 4 else 0, + if b3 then 8 else 0, + if b4 then 16 else 0, + if b5 then 32 else 0, + if b6 then 64 else 0, if b7 then 128 else 0] --- Prelude.map testDecodeEncodeProp1 -testData = [KeepAlive, - Choke, - Unchoke, - Interested, - NotInterested, - Have 0, - Have 1, - Have 1934729, - BitField (L.pack [1,2,3]), - Request 123 (Block 4 7), - Piece 5 7 (B.pack [1,2,3,4,5,6,7,8,9,0]), - Piece 5 7 (B.pack (concat . replicate 30 $ [minBound..maxBound])), - Cancel 5 (Block 6 7), - Port 123 - ] --- Currently returns all True +-- +-- -- TESTS + +testSuite = testGroup "Protocol/Wire" + [ testProperty "QC encode-decode/id" propEncodeDecodeId] + + +propEncodeDecodeId :: Message -> Bool +propEncodeDecodeId m = + let encoded = encode m + decoded = decode encoded + in + Right m == decoded + diff --git a/src/RateCalc.hs b/src/RateCalc.hs --- a/src/RateCalc.hs +++ b/src/RateCalc.hs @@ -35,16 +35,16 @@ new :: UTCTime -> Rate new t = Rate { rate = 0.0 , bytes = 0 - , count = 0 - , nextExpected = addUTCTime fudge t - , lastExt = addUTCTime (-fudge) t - , rateSince = addUTCTime (-fudge) t - } + , count = 0 + , nextExpected = addUTCTime fudge t + , lastExt = addUTCTime (-fudge) t + , rateSince = addUTCTime (-fudge) t + } -- | The call @update n rt@ updates the rate structure @rt@ with @n@ new bytes update :: Integer -> Rate -> Rate update n rt = rt { bytes = bytes rt + n - , count = count rt + n} + , count = count rt + n} -- | The call @extractRate t rt@ extracts the current rate from the rate structure and updates the rate -- structures internal book-keeping @@ -63,12 +63,12 @@ -- in time. But it might come earlier if the rate is high. -- Last is updated with the current time. Finally, we move the windows earliest value -- forward if it is more than 20 seconds from now. - (r, rt { rate = r - , bytes = 0 - , nextExpected = addUTCTime (fromInteger expectN) t - , lastExt = t - , rateSince = max (rateSince rt) (addUTCTime (-maxRatePeriod) t) - }) + (r, rt { rate = r + , bytes = 0 + , nextExpected = addUTCTime (fromInteger expectN) t + , lastExt = t + , rateSince = max (rateSince rt) (addUTCTime (-maxRatePeriod) t) + }) -- | The call @extractCount rt@ extract the bytes transferred since last extraction extractCount :: Rate -> (Integer, Rate) diff --git a/src/Supervisor.hs b/src/Supervisor.hs --- a/src/Supervisor.hs +++ b/src/Supervisor.hs @@ -26,21 +26,20 @@ import Prelude hiding (catch) -import Logging import Process data Child = Supervisor (SupervisorChan -> IO ThreadId) | Worker (SupervisorChan -> IO ThreadId) data SupervisorMsg = IAmDying ThreadId - | PleaseDie ThreadId - | SpawnNew Child + | PleaseDie ThreadId + | SpawnNew Child type SupervisorChan = Channel SupervisorMsg type Children = [Child] data ChildInfo = HSupervisor ThreadId - | HWorker ThreadId + | HWorker ThreadId pDie :: SupervisorChan -> IO () @@ -53,62 +52,62 @@ getChan :: a -> SupervisorChan data CFOFA = CFOFA { name :: String - , chan :: SupervisorChan - , parent :: SupervisorChan - , logCh :: LogChannel } + , chan :: SupervisorChan + , parent :: SupervisorChan + } instance SupervisorConf CFOFA where getParent = parent getChan = chan instance Logging CFOFA where - getLogger cf = (name cf, logCh cf) -- TODO: Better naming! + logName = name data STOFA = STOFA { childInfo :: [ChildInfo] } -- | Run a set of processes and do it once in the sense that if someone dies, -- no restart is attempted. We will just kill off everybody without any kind -- of prejudice. -allForOne :: String -> Children -> LogChannel -> SupervisorChan -> IO ThreadId -allForOne name children logC parentC = do +allForOne :: String -> Children -> SupervisorChan -> IO ThreadId +allForOne name children parentC = do c <- channel - spawnP (CFOFA name c parentC logC) (STOFA []) (catchP startup - (defaultStopHandler parentC)) + spawnP (CFOFA name c parentC) (STOFA []) (catchP startup + (defaultStopHandler parentC)) where startup = do childs <- mapM spawnChild children - modify (\_ -> STOFA childs) - forever eventLoop + modify (\_ -> STOFA (reverse childs)) + forever eventLoop eventLoop = do - mTid <- liftIO myThreadId - syncP =<< chooseP [childEvent, parentEvent mTid] + mTid <- liftIO myThreadId + syncP =<< chooseP [childEvent, parentEvent mTid] childEvent = do - ev <- recvPC chan - wrapP ev (\msg -> case msg of - IAmDying tid -> do gets childInfo >>= mapM_ finChild - t <- liftIO myThreadId - sendPC parent (IAmDying t) >>= syncP - SpawnNew chld -> do n <- spawnChild chld - modify (\(STOFA cs) -> STOFA (n : cs))) + ev <- recvPC chan + wrapP ev (\msg -> case msg of + IAmDying tid -> do gets childInfo >>= mapM_ finChild + t <- liftIO myThreadId + sendPC parent (IAmDying t) >>= syncP + SpawnNew chld -> do n <- spawnChild chld + modify (\(STOFA cs) -> STOFA (n : cs))) parentEvent mTid = do - ev <- recvP parentC (\m -> case m of - PleaseDie tid | tid == mTid -> True - _ -> False) + ev <- recvP parentC (\m -> case m of + PleaseDie tid | tid == mTid -> True + _ -> False) wrapP ev (\msg -> case msg of - PleaseDie _ -> gets childInfo >>= mapM_ finChild - _ -> return ()) + PleaseDie _ -> gets childInfo >>= mapM_ finChild + _ -> return ()) data CFOFO = CFOFO { oName :: String - , oChan :: SupervisorChan - , oparent :: SupervisorChan - , ologCh :: LogChannel } + , oChan :: SupervisorChan + , oparent :: SupervisorChan + } instance SupervisorConf CFOFO where getParent = oparent getChan = oChan instance Logging CFOFO where - getLogger cf = (oName cf, ologCh cf) -- TODO: Better naming! + logName = oName data STOFO = STOFO [ChildInfo] @@ -120,38 +119,38 @@ -- the death and let the other processes keep running. -- -- TODO: Restart policies. -oneForOne :: String -> Children -> LogChannel -> SupervisorChan -> IO (ThreadId, SupervisorChan) -oneForOne name children logC parentC = do +oneForOne :: String -> Children -> SupervisorChan -> IO (ThreadId, SupervisorChan) +oneForOne name children parentC = do c <- channel - t <- spawnP (CFOFO name c parentC logC) (STOFO []) (catchP startup - (defaultStopHandler parentC)) + t <- spawnP (CFOFO name c parentC) (STOFO []) (catchP startup + (defaultStopHandler parentC)) return (t, c) where startup :: Process CFOFO STOFO () startup = do - childs <- mapM spawnChild children - modify (\_ -> STOFO childs) - forever eventLoop + childs <- mapM spawnChild children + modify (\_ -> STOFO (reverse childs)) + forever eventLoop eventLoop :: Process CFOFO STOFO () eventLoop = do - mTid <- liftIO myThreadId - syncP =<< chooseP [childEvent, parentEvent mTid] + mTid <- liftIO myThreadId + syncP =<< chooseP [childEvent, parentEvent mTid] childEvent = do - ev <- recvPC oChan - wrapP ev (\msg -> case msg of - IAmDying tid -> pruneChild tid - SpawnNew chld -> do n <- spawnChild chld - modify (\(STOFO cs) -> STOFO (n : cs))) + ev <- recvPC oChan + wrapP ev (\msg -> case msg of + IAmDying tid -> pruneChild tid + SpawnNew chld -> do n <- spawnChild chld + modify (\(STOFO cs) -> STOFO (n : cs))) parentEvent mTid = do - ev <- recvP parentC (\m -> case m of - PleaseDie tid | tid == mTid -> True - _ -> False) - wrapP ev (\msg -> do (STOFO cs) <- get - mapM_ finChild cs - stopP) + ev <- recvP parentC (\m -> case m of + PleaseDie tid | tid == mTid -> True + _ -> False) + wrapP ev (\msg -> do (STOFO cs) <- get + mapM_ finChild cs + stopP) pruneChild tid = modify (\(STOFO cs) -> STOFO (filter check cs)) - where check (HSupervisor t) = t == tid - check (HWorker t) = t == tid + where check (HSupervisor t) = t == tid + check (HWorker t) = t == tid finChild :: SupervisorConf a => ChildInfo -> Process a b () diff --git a/src/Torrent.hs b/src/Torrent.hs --- a/src/Torrent.hs +++ b/src/Torrent.hs @@ -1,29 +1,3 @@ --- Haskell Torrent --- Copyright (c) 2009, Jesper Louis Andersen, --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are --- met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- * Redistributions in binary form must reproduce the above copyright --- notice, this list of conditions and the following disclaimer in the --- documentation and/or other materials provided with the distribution. --- --- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS --- IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, --- THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR --- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR --- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, --- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, --- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR --- PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF --- LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING --- NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS --- SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - -- | The following module is responsible for general types used -- throughout the system. module Torrent ( @@ -52,6 +26,7 @@ ) where +import Control.Applicative import qualified Data.Foldable as F import qualified Data.ByteString as B import Data.List @@ -61,6 +36,7 @@ import Numeric import System.Random +import Test.QuickCheck import Protocol.BCode import Digest @@ -83,6 +59,7 @@ announceURL :: AnnounceURL } deriving Show data TorrentState = Seeding | Leeching + deriving Show -- PIECES ---------------------------------------------------------------------- @@ -110,9 +87,9 @@ bytesLeft :: PiecesDoneMap -> PieceMap -> Integer bytesLeft done pm = M.foldWithKey (\k v accu -> - case M.lookup k done of - Just False -> (len v) + accu - _ -> accu) 0 pm + case M.lookup k done of + Just False -> (len v) + accu + _ -> accu) 0 pm -- BLOCKS ---------------------------------------------------------------------- @@ -122,6 +99,10 @@ , blockSize :: BlockSize -- ^ size of this block within the piece } deriving (Eq, Ord, Show) +instance Arbitrary Block where + arbitrary = Block <$> arbitrary <*> arbitrary + + defaultBlockSize :: BlockSize defaultBlockSize = 16384 -- Bytes @@ -142,7 +123,7 @@ mkTorrentInfo :: BCode -> IO TorrentInfo mkTorrentInfo bc = do (ann, np) <- case queryInfo bc of Nothing -> fail "Could not create torrent info" - Just x -> return x + Just x -> return x ih <- hashInfoDict bc return TorrentInfo { infoHash = ih, announceURL = ann, pieceCount = np } where