HaskellTorrent 0.0 → 0.1
raw patch · 26 files changed
+1391/−1381 lines, 26 filesdep +HUnitdep +QuickCheckdep +hopenssldep −HsOpenSSLsetup-changed
Dependencies added: HUnit, QuickCheck, hopenssl, hslogger, test-framework, test-framework-hunit, test-framework-quickcheck2
Dependencies removed: HsOpenSSL
Files
- HaskellTorrent.cabal +18/−7
- LICENSE +1/−1
- README.md +64/−7
- Setup.lhs +18/−2
- configure +1/−1
- src/Digest.hs +23/−17
- src/FS.hs +5/−31
- src/HaskellTorrent.hs +92/−41
- src/Logging.hs +0/−40
- src/LoggingTypes.hs +0/−36
- src/PeerTypes.hs +9/−2
- src/Process.hs +38/−65
- src/Process/ChokeMgr.hs +89/−92
- src/Process/Console.hs +72/−50
- src/Process/FS.hs +24/−56
- src/Process/Listen.hs +22/−25
- src/Process/Peer.hs +286/−332
- src/Process/PeerMgr.hs +125/−53
- src/Process/PieceMgr.hs +180/−195
- src/Process/Status.hs +34/−64
- src/Process/Tracker.hs +74/−69
- src/Protocol/BCode.hs +51/−47
- src/Protocol/Wire.hs +89/−52
- src/RateCalc.hs +12/−12
- src/Supervisor.hs +53/−54
- src/Torrent.hs +11/−30
HaskellTorrent.cabal view
@@ -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
LICENSE view
@@ -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
README.md view
@@ -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
Setup.lhs view
@@ -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"
configure view
@@ -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`
src/Digest.hs view
@@ -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
src/FS.hs view
@@ -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,
src/HaskellTorrent.hs view
@@ -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 <file.torrent>"- (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 ()
− src/Logging.hs
@@ -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
− src/LoggingTypes.hs
@@ -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
src/PeerTypes.hs view
@@ -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
src/Process.hs view
@@ -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
src/Process/ChokeMgr.hs view
@@ -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
src/Process/Console.hs view
@@ -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)
src/Process/FS.hs view
@@ -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
src/Process/Listen.hs view
@@ -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)
src/Process/Peer.hs view
@@ -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 ()--}
src/Process/PeerMgr.hs view
@@ -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"
src/Process/PieceMgr.hs view
@@ -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")
src/Process/Status.hs view
@@ -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 }))
src/Process/Tracker.hs view
@@ -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.
src/Protocol/BCode.hs view
@@ -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---
src/Protocol/Wire.hs view
@@ -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+
src/RateCalc.hs view
@@ -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)
src/Supervisor.hs view
@@ -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 ()
src/Torrent.hs view
@@ -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