Combinatorrent 0.2.0 → 0.2.1
raw patch · 26 files changed
+443/−391 lines, 26 filesdep +deepseq
Dependencies added: deepseq
Files
- Combinatorrent.cabal +5/−4
- configure +1/−1
- src/Channels.hs +5/−0
- src/Combinatorrent.hs +1/−1
- src/Digest.hs +6/−7
- src/FS.hs +11/−9
- src/Process.hs +3/−9
- src/Process/ChokeMgr.hs +6/−2
- src/Process/Console.hs +1/−1
- src/Process/DirWatcher.hs +2/−2
- src/Process/FS.hs +7/−8
- src/Process/Listen.hs +1/−1
- src/Process/Peer.hs +136/−117
- src/Process/Peer/Receiver.hs +5/−3
- src/Process/Peer/Sender.hs +2/−2
- src/Process/Peer/SenderQ.hs +43/−22
- src/Process/PeerMgr.hs +26/−18
- src/Process/PieceMgr.hs +37/−30
- src/Process/Status.hs +15/−7
- src/Process/TorrentManager.hs +2/−2
- src/Process/Tracker.hs +4/−3
- src/Protocol/BCode.hs +10/−9
- src/Protocol/Wire.hs +3/−9
- src/RateCalc.hs +8/−5
- src/Supervisor.hs +82/−106
- src/Torrent.hs +21/−13
Combinatorrent.cabal view
@@ -1,6 +1,6 @@ name: Combinatorrent category: Network-version: 0.2.0+version: 0.2.1 category: Network description: Combinatorrent provides a BitTorrent client, based on STM for concurrency. This is an early preview release which is capable of@@ -13,7 +13,7 @@ license: BSD3 license-file: LICENSE-copyright: (c) 2009 Jesper Louis Andersen+copyright: (c) 2009,2010 Jesper Louis Andersen author: Jesper Louis Andersen maintainer: jesper.louis.andersen@gmail.com stability: experimental@@ -56,6 +56,7 @@ bytestring, cereal, containers,+ deepseq, directory, filepath, hopenssl,@@ -76,8 +77,8 @@ test-framework-quickcheck2, time - ghc-prof-options: -auto-all -ignore-scc -caf-all- ghc-options: -Wall -fno-warn-orphans -threaded+ ghc-prof-options: -auto-all -caf-all+ ghc-options: -Wall -fno-warn-orphans -threaded -feager-blackholing -O2 if !flag(debug) cpp-options: "-DNDEBUG"
configure view
@@ -3,7 +3,7 @@ # Simple configuration script. ## Git version-GITHEAD="0.2 Hackage"+GITHEAD="0.2.1 Hackage" if test -d "${GIT_DIR:-.git}" ; then GITHEAD=`git describe 2>/dev/null`
src/Channels.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE TypeSynonymInstances #-} module Channels ( Peer(..) , PeerMessage(..)@@ -12,6 +13,7 @@ import Control.Concurrent import Control.Concurrent.STM+import Control.DeepSeq import Network import Torrent@@ -25,6 +27,9 @@ | CancelBlock PieceNum Block type PeerChannel = TChan PeerMessage++instance NFData PeerChannel where+ rnf pc = pc `seq` () ---- TRACKER
src/Combinatorrent.hs view
@@ -128,7 +128,7 @@ stv <- atomically $ newTVar [] debugM "Main" "Created channels" pid <- generatePeerId- tid <- allForOne "MainSup"+ (tid, _) <- allForOne "MainSup" (workersWatch ++ [ Worker $ Console.start waitC statusC , Worker $ TorrentManager.start watchC statusC stv chokeC pid pmC
src/Digest.hs view
@@ -5,10 +5,11 @@ ) where -import Data.Char-import Data.Word+import Control.Applicative import Control.Monad.State +import Data.Word+ import Foreign.Ptr import qualified Data.ByteString as B import Data.ByteString.Unsafe@@ -16,12 +17,10 @@ import qualified OpenSSL.Digest as SSL -- Consider newtyping this-type Digest = String+type Digest = B.ByteString -digest :: L.ByteString -> IO Digest-digest bs = {-# SCC "sha1_digest" #-} do- upack <- digestLBS SSL.SHA1 bs- return $ map (chr . fromIntegral) upack+digest :: L.ByteString -> IO B.ByteString+digest bs = {-# SCC "sha1_digest" #-} B.pack <$> digestLBS SSL.SHA1 bs digestLBS :: SSL.MessageDigest -> L.ByteString -> IO [Word8] digestLBS mdType xs = {-# SCC "sha1_digestLBS" #-}
src/FS.hs view
@@ -15,6 +15,7 @@ import Control.Monad.State +import Data.Array import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as L import qualified Data.Map as M@@ -70,9 +71,7 @@ projectHandles (Handles handles') 0 (size - size1) pInfoLookup :: PieceNum -> PieceMap -> IO PieceInfo-pInfoLookup pn mp = case M.lookup pn mp of- Nothing -> fail "FS: Error lookup in PieceMap"- Just i -> return i+pInfoLookup pn mp = return $ mp ! pn -- | FIXME: minor code duplication with @readBlock@ readPiece :: PieceNum -> Handles -> PieceMap -> IO L.ByteString@@ -144,7 +143,7 @@ checkFile :: Handles -> PieceMap -> IO PiecesDoneMap checkFile handles pm = do l <- mapM checkP pieces return $ M.fromList l- where pieces = M.toAscList pm+ where pieces = assocs pm checkP :: (PieceNum, PieceInfo) -> IO (PieceNum, Bool) checkP (pn, pInfo) = do b <- checkPiece pInfo handles return (pn, b)@@ -156,19 +155,22 @@ 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) )+ let pis = extract pLen tLen 0 pieceData+ l = length pis+ pm = array (0, l-1) (zip [0..] pis)+ when ( tLen /= (sum $ map len $ 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,+ extract plen tlen offst (p : ps) | tlen < plen = PieceInfo {+ offset = offst, len = tlen,- digest = B.unpack p } : extract plen 0 (offst + plen) ps+ digest = p } : extract plen 0 (offst + plen) ps | otherwise = inf : extract plen (tlen - plen) (offst + plen) ps where inf = PieceInfo { offset = offst, len = plen,- digest = B.unpack p }+ digest = p } extract _ _ _ _ = error "mkPieceMap: the impossible happened!" -- | Predicate function. True if nothing is missing from the map.
src/Process.hs view
@@ -13,7 +13,6 @@ , spawnP , catchP , cleanupP- , foreverP , stopP -- * Log Interface , Logging(..)@@ -30,7 +29,7 @@ import Control.Exception import Control.Monad.Reader-import Control.Monad.State+import Control.Monad.State.Strict import Data.Typeable @@ -44,7 +43,7 @@ -- stack on top of IO. newtype Process a b c = Process (ReaderT a (StateT b IO) c) #ifndef __HADDOCK__- deriving (Functor, Monad, MonadIO, MonadState b, MonadReader a, Typeable)+ deriving (Functor, Monad, MonadIO, MonadState b, MonadReader a) #endif data StopException = StopException@@ -62,8 +61,7 @@ -- | Spawn and run a process monad spawnP :: a -> b -> Process a b () -> IO ThreadId spawnP c st p = forkIO proc- where proc = do _ <- runP c st p- return ()+ where proc = runP c st p >> return () -- | Run the process monad for its side effect, with a stopHandler if exceptions -- are raised in the process@@ -90,10 +88,6 @@ ] put s' return a---- | Run a process forever in a loop-foreverP :: Process a b c -> Process a b c-foreverP p = p >> foreverP p ------ LOGGING
src/Process/ChokeMgr.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE FlexibleContexts, TupleSections #-}+{-# LANGUAGE TypeSynonymInstances, FlexibleContexts, TupleSections #-} module Process.ChokeMgr ( -- * Types, Channels ChokeMgrChannel@@ -11,6 +11,7 @@ import Control.Concurrent import Control.Concurrent.STM+import Control.DeepSeq import Control.Exception (assert) import Control.Monad.Reader import Control.Monad.State@@ -42,6 +43,9 @@ | BlockComplete InfoHash PieceNum Block -- ^ Note that a block is complete (endgame) | TorrentComplete InfoHash -- ^ Note that the torrent in question is complete +instance NFData ChokeMgrMsg++ type ChokeMgrChannel = TChan ChokeMgrMsg type RateTVar = TVar [(ThreadId, (Double, Double, Bool, Bool, Bool))] @@ -60,7 +64,7 @@ roundTickSecs :: Int roundTickSecs = 11 -start :: ChokeMgrChannel -> RateTVar -> Int -> SupervisorChan+start :: ChokeMgrChannel -> RateTVar -> Int -> SupervisorChannel -> IO ThreadId start ch rtv ur supC = do _ <- registerSTM roundTickSecs ch Tick
src/Process/Console.hs view
@@ -35,7 +35,7 @@ -- | Start the logging process and return a channel to it. Sending on this -- Channel means writing stuff out on stdOut-start :: TMVar () -> St.StatusChannel -> SupervisorChan -> IO ThreadId+start :: TMVar () -> St.StatusChannel -> SupervisorChannel -> IO ThreadId start waitC statusC supC = do cmdC <- readerP -- We shouldn't be doing this in the long run wrtC <- writerP
src/Process/DirWatcher.hs view
@@ -33,11 +33,11 @@ start :: FilePath -- ^ Path to watch -> TorrentMgrChan -- ^ Channel to return answers on- -> SupervisorChan+ -> SupervisorChannel -> IO ThreadId start fp chan supC = do spawnP (CF chan fp) S.empty- (catchP (foreverP pgm) (defaultStopHandler supC))+ (catchP (forever pgm) (defaultStopHandler supC)) where pgm = do q <- liftIO $ registerDelay (5 * 1000000) liftIO . atomically $ do
src/Process/FS.hs view
@@ -15,8 +15,8 @@ import Control.Monad.Reader import Control.Monad.State +import Data.Array import qualified Data.ByteString as B-import qualified Data.Map as M import Process import Torrent@@ -37,15 +37,15 @@ logName _ = "Process.FS" data ST = ST- { fileHandles :: !FS.Handles -- ^ The file we are working on- , pieceMap :: FS.PieceMap -- ^ Map of where the pieces reside+ { fileHandles :: !FS.Handles -- ^ The file we are working on+ , pieceMap :: !FS.PieceMap -- ^ Map of where the pieces reside } -- INTERFACE ---------------------------------------------------------------------- -start :: FS.Handles -> FS.PieceMap -> FSPChannel -> SupervisorChan-> IO ThreadId+start :: FS.Handles -> FS.PieceMap -> FSPChannel -> SupervisorChannel -> IO ThreadId start handles pm fspC supC = spawnP (CF fspC) (ST handles pm) (catchP (forever lp) (defaultStopHandler supC)) where@@ -55,10 +55,9 @@ case msg of CheckPiece n v -> do pmap <- gets pieceMap- case M.lookup n pmap of- Nothing -> liftIO . atomically $ putTMVar v Nothing- Just p -> do r <- gets fileHandles >>= (liftIO . FS.checkPiece p)- liftIO . atomically $ putTMVar v (Just r)+ let p = pmap ! n+ r <- gets fileHandles >>= (liftIO . FS.checkPiece p)+ liftIO . atomically $ putTMVar v (Just r) ReadBlock n blk v -> do debugP $ "Reading block #" ++ show n ++ "(" ++ show (blockOffset blk) ++ ", " ++ show (blockSize blk) ++ ")"
src/Process/Listen.hs view
@@ -19,7 +19,7 @@ instance Logging CF where logName _ = "Process.Listen" -start :: PortID -> PeerMgrChannel -> SupervisorChan -> IO ThreadId+start :: PortID -> PeerMgrChannel -> SupervisorChannel -> IO ThreadId start port peerMgrC supC = do spawnP (CF peerMgrC) () (catchP (openListen >>= pgm) (defaultStopHandler supC)) -- TODO: Close socket resource!
src/Process/Peer.hs view
@@ -11,17 +11,20 @@ import Control.Applicative import Control.Concurrent import Control.Concurrent.STM+import Control.DeepSeq+import Control.Exception + import Control.Monad.State import Control.Monad.Reader import Prelude hiding (catch, log) +import Data.Array import Data.Bits import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L -import qualified Data.Map as M import qualified Data.PieceSet as PS import Data.Maybe @@ -53,119 +56,130 @@ start :: Handle -> MgrChannel -> RateTVar -> PieceMgrChannel -> FSPChannel -> TVar [PStat] -> PieceMap -> Int -> InfoHash -> IO Children-start handle pMgrC rtv pieceMgrC fsC stv pm nPieces ih = do+start h pMgrC rtv pieceMgrC fsC stv pm nPieces ih = do queueC <- newTChanIO senderMV <- newEmptyTMVarIO receiverC <- newTChanIO sendBWC <- newTChanIO- return [Worker $ Sender.start handle senderMV,- Worker $ SenderQ.start queueC senderMV sendBWC,- Worker $ Receiver.start handle receiverC,- Worker $ peerP pMgrC rtv pieceMgrC fsC pm nPieces+ return [Worker $ Sender.start h senderMV,+ Worker $ SenderQ.start queueC senderMV sendBWC fsC,+ Worker $ Receiver.start h receiverC,+ Worker $ peerP pMgrC rtv pieceMgrC pm nPieces queueC receiverC sendBWC stv ih] -- INTERNAL FUNCTIONS ---------------------------------------------------------------------- -data PCF = PCF { inCh :: TChan (Message, Integer)- , outCh :: TChan SenderQ.SenderQMsg- , peerMgrCh :: MgrChannel- , pieceMgrCh :: PieceMgrChannel- , fsCh :: FSPChannel- , peerCh :: PeerChannel- , sendBWCh :: BandwidthChannel- , timerCh :: TChan ()- , statTV :: TVar [PStat]- , rateTV :: RateTVar- , pcInfoHash :: InfoHash- , pieceMap :: PieceMap- }+data CF = CF { inCh :: TChan (Message, Integer)+ , outCh :: TChan SenderQ.SenderQMsg+ , peerMgrCh :: MgrChannel+ , pieceMgrCh :: PieceMgrChannel+ , peerCh :: PeerChannel+ , sendBWCh :: BandwidthChannel+ , timerCh :: TChan ()+ , statTV :: TVar [PStat]+ , rateTV :: RateTVar+ , pcInfoHash :: InfoHash+ , pieceMap :: !PieceMap+ , piecesDoneTV :: TMVar [PieceNum]+ , interestTV :: TMVar Bool+ , grabBlockTV :: TMVar Blocks+ } -instance Logging PCF where+instance Logging CF where 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 :: !(PS.PieceSet) -- ^ 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- }+data ST = ST { 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 :: !(PS.PieceSet) -- ^ List of pieces the peer has access to+ , missingPieces :: !Int -- ^ Tracks the number of pieces the peer misses before seeding+ , 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 -> RateTVar -> PieceMgrChannel -> FSPChannel -> PieceMap -> Int+peerP :: MgrChannel -> RateTVar -> PieceMgrChannel -> PieceMap -> Int -> TChan SenderQ.SenderQMsg -> TChan (Message, Integer) -> BandwidthChannel -> TVar [PStat] -> InfoHash- -> SupervisorChan -> IO ThreadId-peerP pMgrC rtv pieceMgrC fsC pm nPieces outBound inBound sendBWC stv ih supC = do+ -> SupervisorChannel -> IO ThreadId+peerP pMgrC rtv pieceMgrC pm nPieces outBound inBound sendBWC stv ih supC = do ch <- newTChanIO tch <- newTChanIO ct <- getCurrentTime+ pdtmv <- newEmptyTMVarIO+ intmv <- newEmptyTMVarIO+ gbtmv <- newEmptyTMVarIO pieceSet <- PS.new nPieces- spawnP (PCF inBound outBound pMgrC pieceMgrC fsC ch sendBWC tch stv rtv ih pm)- (PST True False S.empty True False pieceSet (RC.new ct) (RC.new ct) False)- (cleanupP startup (defaultStopHandler supC) cleanup)- where startup = do- tid <- liftIO $ myThreadId- debugP "Syncing a connectBack"- asks peerCh >>= (\ch -> do- c <- asks peerMgrCh- liftIO . atomically $ writeTChan c $ Connect ih tid ch)- pieces <- getPiecesDone- outChan $ SenderQ.SenderQM $ BitField (constructBitField nPieces pieces)- -- Install the StatusP timer- c <- asks timerCh- _ <- registerSTM 5 c ()- foreverP eventLoop+ spawnP (CF inBound outBound pMgrC pieceMgrC ch sendBWC tch stv rtv ih pm+ pdtmv intmv gbtmv)+ (ST True False S.empty True False pieceSet nPieces (RC.new ct) (RC.new ct) False)+ (cleanupP (startup nPieces) (defaultStopHandler supC) cleanup) - cleanup = do- t <- liftIO myThreadId- pieces <- gets peerPieces >>= PS.toList- ch <- asks pieceMgrCh- ch2 <- asks peerMgrCh- liftIO . atomically $ writeTChan ch (PeerUnhave pieces)- liftIO . atomically $ writeTChan ch2 (Disconnect t)+startup :: Int -> Process CF ST ()+startup nPieces = do+ tid <- liftIO $ myThreadId+ ch <- asks peerCh+ pmc <- asks peerMgrCh+ ih <- asks pcInfoHash+ liftIO . atomically $ writeTChan pmc $ Connect ih tid ch+ pieces <- getPiecesDone+ outChan $ SenderQ.SenderQM $ BitField (constructBitField nPieces pieces)+ -- Install the StatusP timer+ c <- asks timerCh+ _ <- registerSTM 5 c ()+ forever eventLoop - readOp :: Process PCF PST Operation- readOp = do- inb <- asks inCh- chk <- asks peerCh- tch <- asks timerCh- bwc <- asks sendBWCh- liftIO . atomically $- (readTChan inb >>= return . PeerMsgEvt) `orElse`- (readTChan chk >>= return . ChokeMgrEvt) `orElse`- (readTChan tch >> return TimerEvent) `orElse`- (readTChan bwc >>= return . UpRateEvent)+cleanup :: Process CF ST ()+cleanup = do+ t <- liftIO myThreadId+ pieces <- gets peerPieces >>= PS.toList+ ch <- asks pieceMgrCh+ ch2 <- asks peerMgrCh+ liftIO . atomically $ writeTChan ch (PeerUnhave pieces)+ liftIO . atomically $ writeTChan ch2 (Disconnect t) - eventLoop = do- op <- readOp- case op of- PeerMsgEvt (m, sz) -> peerMsg m sz- ChokeMgrEvt m -> chokeMsg m- UpRateEvent up -> modify (\s -> s { upRate = RC.update up $ upRate s})- TimerEvent -> timerTick+readOp :: Process CF ST Operation+readOp = do+ inb <- asks inCh+ chk <- asks peerCh+ tch <- asks timerCh+ bwc <- asks sendBWCh+ liftIO . atomically $+ (readTChan inb >>= return . PeerMsgEvt) `orElse`+ (readTChan chk >>= return . ChokeMgrEvt) `orElse`+ (readTChan tch >> return TimerEvent) `orElse`+ (readTChan bwc >>= return . UpRateEvent) +eventLoop :: Process CF ST ()+eventLoop = do+ op <- readOp+ case op of+ PeerMsgEvt (m, sz) -> peerMsg m sz+ ChokeMgrEvt m -> chokeMsg m+ UpRateEvent up -> modify (\s -> s { upRate = RC.update up $ upRate s})+ TimerEvent -> timerTick+ data Operation = PeerMsgEvt (Message, Integer) | ChokeMgrEvt PeerMessage | TimerEvent | UpRateEvent Integer- + -- | Return a list of pieces which are currently done by us-getPiecesDone :: Process PCF PST [PieceNum]+getPiecesDone :: Process CF ST [PieceNum] getPiecesDone = do ch <- asks pieceMgrCh+ c <- asks piecesDoneTV liftIO $ do- c <- newEmptyTMVarIO atomically $ writeTChan ch (GetDone c) atomically $ takeTMVar c -- | Process an event from the Choke Manager-chokeMsg :: PeerMessage -> Process PCF PST ()+chokeMsg :: PeerMessage -> Process CF ST () chokeMsg msg = do debugP "ChokeMgrEvent" case msg of@@ -184,17 +198,11 @@ modify (\s -> s { blockQueue = S.delete (pn, blk) $ blockQueue s }) outChan $ SenderQ.SenderQRequestPrune pn blk --- True if the peer is a seeder--- Optimization: Don't calculate this all the time. It only changes once and then it keeps--- being there.-isASeeder :: Process PCF PST Bool-isASeeder = liftM2 (==) (gets peerPieces >>= PS.size) (M.size <$> asks pieceMap)- -- A Timer event handles a number of different status updates. One towards the -- Choke Manager so it has a information about whom to choke and unchoke - and -- one towards the status process to keep track of uploaded and downloaded -- stuff.-timerTick :: Process PCF PST ()+timerTick :: Process CF ST () timerTick = do mTid <- liftIO myThreadId debugP "TimerEvent"@@ -229,7 +237,7 @@ -- | Process an Message from the peer in the other end of the socket.-peerMsg :: Message -> Integer -> Process PCF PST ()+peerMsg :: Message -> Integer -> Process CF ST () peerMsg msg sz = do modify (\s -> s { downRate = RC.update sz $ downRate s}) case msg of@@ -250,7 +258,7 @@ -- | Put back blocks for other peer processes to grab. This is done whenever -- the peer chokes us, or if we die by an unknown cause.-putbackBlocks :: Process PCF PST ()+putbackBlocks :: Process CF ST () putbackBlocks = do blks <- gets blockQueue pmch <- asks pieceMgrCh@@ -258,54 +266,64 @@ modify (\s -> s { blockQueue = S.empty }) -- | Process a HAVE message from the peer. Note we also update interest as a side effect-haveMsg :: PieceNum -> Process PCF PST ()+haveMsg :: PieceNum -> Process CF ST () haveMsg pn = do pm <- asks pieceMap- if M.member pn pm+ let (lo, hi) = bounds pm+ if pn >= lo && pn <= hi then do PS.insert pn =<< gets peerPieces pmch <- asks pieceMgrCh liftIO . atomically $ writeTChan pmch (PeerHave [pn])+ decMissingCounter 1 considerInterest else do warningP "Unknown Piece" stopP +-- True if the peer is a seeder+isASeeder :: Process CF ST Bool+isASeeder = do sdr <- gets missingPieces+ sdr `deepseq` (return $! sdr == 0)++-- Decrease the counter of missing pieces for the peer+decMissingCounter :: Int -> Process CF ST ()+decMissingCounter n = do+ modify (\s -> s { missingPieces = missingPieces s - n})+ m <- gets missingPieces+ when (m == 0) assertSeeder++-- Assert that the peer is a seeder+assertSeeder :: Process CF ST ()+assertSeeder = do+ ok <- liftM2 (==) (gets peerPieces >>= PS.size) (succ . snd . bounds <$> asks pieceMap)+ assert ok (return ())+ -- | Process a BITFIELD message from the peer. Side effect: Consider Interest.-bitfieldMsg :: BitField -> Process PCF PST ()+bitfieldMsg :: BitField -> Process CF ST () bitfieldMsg bf = do pieces <- gets peerPieces piecesNull <- PS.null pieces if piecesNull -- TODO: Don't trust the bitfield- then do nPieces <- M.size <$> asks pieceMap+ then do nPieces <- succ . snd . bounds <$> asks pieceMap pp <- createPeerPieces nPieces bf modify (\s -> s { peerPieces = pp }) peerLs <- PS.toList pp pmch <- asks pieceMgrCh liftIO . atomically $ writeTChan pmch (PeerHave peerLs)+ decMissingCounter (length peerLs) considerInterest else do infoP "Got out of band Bitfield request, dying" stopP -- | Process a request message from the Peer-requestMsg :: PieceNum -> Block -> Process PCF PST ()+requestMsg :: PieceNum -> Block -> Process CF ST () requestMsg pn blk = do choking <- gets weChoke unless (choking)- (do- bs <- readBlock pn blk -- TODO: Pushdown to send process- outChan $ SenderQ.SenderQM $ Piece pn (blockOffset blk) bs)---- | Read a block from the filesystem for sending-readBlock :: PieceNum -> Block -> Process PCF PST B.ByteString-readBlock pn blk = do- v <- liftIO $ newEmptyTMVarIO- fch <- asks fsCh- liftIO $ do- atomically $ writeTChan fch (ReadBlock pn blk v)- atomically $ takeTMVar v+ (outChan $ SenderQ.SenderQPiece pn blk) -- | Handle a Piece Message incoming from the peer-pieceMsg :: PieceNum -> Int -> B.ByteString -> Process PCF PST ()+pieceMsg :: PieceNum -> Int -> B.ByteString -> Process CF ST () pieceMsg n os bs = do let sz = B.length bs blk = Block os sz@@ -314,18 +332,19 @@ -- When e is not a member, the piece may be stray, so ignore it. -- Perhaps print something here. when (S.member e q)- (do storeBlock n (Block os sz) bs- modify (\s -> s { blockQueue = S.delete e (blockQueue s)}))+ (do storeBlock n blk bs+ bq <- gets blockQueue >>= return . S.delete e+ bq `deepseq` modify (\s -> s { blockQueue = bq })) -- | Handle a cancel message from the peer-cancelMsg :: PieceNum -> Block -> Process PCF PST ()+cancelMsg :: PieceNum -> Block -> Process CF ST () cancelMsg n blk = outChan $ SenderQ.SenderQCancel n blk -- | Update our interest state based on the pieces the peer has. -- Obvious optimization: Do less work, there is no need to consider all pieces most of the time-considerInterest :: Process PCF PST ()+considerInterest :: Process CF ST () considerInterest = do- c <- liftIO newEmptyTMVarIO+ c <- asks interestTV pcs <- gets peerPieces pmch <- asks pieceMgrCh interested <- liftIO $ do@@ -338,7 +357,7 @@ -- | Try to fill up the block queue at the peer. The reason we pipeline a -- number of blocks is to get around the line delay present on the internet.-fillBlocks :: Process PCF PST ()+fillBlocks :: Process CF ST () fillBlocks = do choked <- gets peerChoke unless choked checkWatermark@@ -347,7 +366,7 @@ -- fill till the upper one. This in turn keeps the pipeline of pieces full as -- long as the peer is interested in talking to us. -- TODO: Decide on a queue size based on the current download rate.-checkWatermark :: Process PCF PST ()+checkWatermark :: Process CF ST () checkWatermark = do q <- gets blockQueue eg <- gets runningEndgame@@ -372,26 +391,26 @@ -- | Queue up pieces for retrieval at the Peer-queuePieces :: [(PieceNum, Block)] -> Process PCF PST ()+queuePieces :: [(PieceNum, Block)] -> Process CF ST () queuePieces toQueue = do mapM_ (uncurry pushRequest) toQueue modify (\s -> s { blockQueue = S.union (blockQueue s) (S.fromList toQueue) }) -- | Push a request to the peer so he can send it to us-pushRequest :: PieceNum -> Block -> Process PCF PST ()+pushRequest :: PieceNum -> Block -> Process CF ST () pushRequest pn blk = outChan $ SenderQ.SenderQM $ Request pn blk -- | Tell the PieceManager to store the given block-storeBlock :: PieceNum -> Block -> B.ByteString -> Process PCF PST ()+storeBlock :: PieceNum -> Block -> B.ByteString -> Process CF ST () storeBlock n blk bs = do pmch <- asks pieceMgrCh liftIO . atomically $ writeTChan pmch (StoreBlock n blk bs) -- | The call @grabBlocks n@ will attempt to grab (up to) @n@ blocks from the -- piece Manager for request at the peer.-grabBlocks :: Int -> Process PCF PST [(PieceNum, Block)]+grabBlocks :: Int -> Process CF ST [(PieceNum, Block)] grabBlocks n = do- c <- liftIO newEmptyTMVarIO+ c <- asks grabBlockTV ps <- gets peerPieces pmch <- asks pieceMgrCh blks <- liftIO $ do@@ -416,7 +435,7 @@ decodeBytes soFar (w : ws) = catMaybes (decodeByte soFar w) : decodeBytes (soFar + 8) ws -- | Send a message on a chan from the process queue-outChan :: SenderQ.SenderQMsg -> Process PCF PST ()+outChan :: SenderQ.SenderQMsg -> Process CF ST () outChan qm = do ch <- asks outCh liftIO . atomically $ writeTChan ch qm
src/Process/Peer/Receiver.hs view
@@ -29,9 +29,11 @@ logName _ = "Process.Peer.Receiver" start :: Handle -> TChan (Message, Integer)- -> SupervisorChan -> IO ThreadId-start h ch supC = spawnP (CF ch) h- (catchP (foreverP readSend)+ -> SupervisorChannel -> IO ThreadId+start h ch supC = do+ hSetBuffering h NoBuffering+ spawnP (CF ch) h+ (catchP (forever readSend) (defaultStopHandler supC)) readSend :: Process CF Handle ()
src/Process/Peer/Sender.hs view
@@ -23,8 +23,8 @@ logName _ = "Process.Peer.Sender" -- | The raw sender process, it does nothing but send out what it syncs on.-start :: Handle -> TMVar B.ByteString -> SupervisorChan -> IO ThreadId-start h ch supC = spawnP (CF ch) h (catchP (foreverP pgm)+start :: Handle -> TMVar B.ByteString -> SupervisorChannel -> IO ThreadId+start h ch supC = spawnP (CF ch) h (catchP (forever pgm) (do t <- liftIO $ myThreadId liftIO . atomically $ writeTChan supC $ IAmDying t liftIO $ hClose h))
src/Process/Peer/SenderQ.hs view
@@ -16,6 +16,7 @@ import Channels import Process+import Process.FS hiding (start) import qualified Data.Queue as Q import Supervisor import Torrent@@ -24,16 +25,19 @@ -- | Messages we can send to the Send Queue data SenderQMsg = SenderQCancel PieceNum Block -- ^ Peer requested that we cancel a piece | SenderQM Message -- ^ We want to send the Message to the peer+ | SenderQPiece PieceNum Block -- ^ Request for a piece transmit | SenderOChoke -- ^ We want to choke the peer | SenderQRequestPrune PieceNum Block -- ^ Prune SendQueue of this (pn, blk) pair data CF = CF { sqIn :: TChan SenderQMsg , sqOut :: TMVar B.ByteString , bandwidthCh :: BandwidthChannel+ , readBlockTV :: TMVar B.ByteString+ , fsCh :: FSPChannel } -data ST = ST { outQueue :: Q.Queue Message- , bytesTransferred :: Integer+data ST = ST { outQueue :: !(Q.Queue (Either Message (PieceNum, Block)))+ , bytesTransferred :: !Integer } instance Logging CF where@@ -42,10 +46,11 @@ -- | sendQueue Process, simple version. -- TODO: Split into fast and slow. start :: TChan SenderQMsg -> TMVar B.ByteString -> BandwidthChannel- -> SupervisorChan- -> IO ThreadId-start inC outC bandwC supC = spawnP (CF inC outC bandwC) (ST Q.empty 0)- (catchP (foreverP pgm)+ -> FSPChannel -> SupervisorChannel -> IO ThreadId+start inC outC bandwC fspC supC = do+ rbtv <- liftIO newEmptyTMVarIO+ spawnP (CF inC outC bandwC rbtv fspC) (ST Q.empty 0)+ (catchP (forever pgm) (defaultStopHandler supC)) pgm :: Process CF ST ()@@ -58,7 +63,11 @@ ov <- asks sqOut r <- case Q.first q of Nothing -> liftIO $ atomically (readTChan ic >>= return . Right)- Just p -> do let bs = encodePacket p+ Just r -> do p <- case r of+ Left m -> return m+ Right (pn, blk) -> do d <- readBlock pn blk+ return $ Piece pn (blockOffset blk) d+ let bs = encodePacket p sz = fromIntegral $ B.length bs liftIO . atomically $ (putTMVar ov bs >> return (Left sz)) `orElse`@@ -69,10 +78,11 @@ , outQueue = Q.remove (outQueue s)}) Right m -> case m of- SenderQM msg -> modifyQ (Q.push msg)- SenderQCancel n blk -> modifyQ (Q.filter (filterPiece n (blockOffset blk)))+ SenderQM msg -> modifyQ (Q.push $ Left msg)+ SenderQPiece n blk -> modifyQ (Q.push $ Right (n, blk))+ SenderQCancel n blk -> modifyQ (Q.filter (filterPiece n blk)) SenderOChoke -> do modifyQ (Q.filter filterAllPiece)- modifyQ (Q.push Choke)+ modifyQ (Q.push $ Left Choke) SenderQRequestPrune n blk -> modifyQ (Q.filter (filterRequest n blk)) @@ -83,19 +93,30 @@ liftIO . atomically $ writeTChan bwc l modify (\s -> s { bytesTransferred = 0 }) -filterAllPiece :: Message -> Bool-filterAllPiece (Piece _ _ _) = True-filterAllPiece _ = False+filterAllPiece :: Either Message (PieceNum, Block) -> Bool+filterAllPiece (Right _) = True+filterAllPiece (Left _) = False -filterPiece :: PieceNum -> Int -> Message -> Bool-filterPiece n off (Piece n1 off1 _) | n == n1 && off == off1 = False- | otherwise = True+filterPiece :: PieceNum -> Block -> Either Message (PieceNum, Block) -> Bool+filterPiece n blk (Right (n1, blk1)) | n == n1 && blk == blk1 = False+ | otherwise = True filterPiece _ _ _ = True -filterRequest :: PieceNum -> Block -> Message -> Bool-filterRequest n blk (Request n1 blk1) | n == n1 && blk == blk1 = False- | otherwise = True-filterRequest _ _ _ = True+filterRequest :: PieceNum -> Block -> Either Message (PieceNum, Block) -> Bool+filterRequest n blk (Left (Request n1 blk1)) | n == n1 && blk == blk1 = False+ | otherwise = True+filterRequest _ _ _ = True -modifyQ :: (Q.Queue Message -> Q.Queue Message) -> Process CF ST ()-modifyQ f = modify (\s -> s { outQueue = f (outQueue s) })+modifyQ :: (Q.Queue (Either Message (PieceNum, Block)) ->+ Q.Queue (Either Message (PieceNum, Block)))+ -> Process CF ST ()+modifyQ f = modify (\s -> s { outQueue = f $! outQueue s })++-- | Read a block from the filesystem for sending+readBlock :: PieceNum -> Block -> Process CF ST B.ByteString+readBlock pn blk = do+ v <- asks readBlockTV+ fch <- asks fsCh+ liftIO $ do+ atomically $ writeTChan fch (ReadBlock pn blk v)+ atomically $ takeTMVar v
src/Process/PeerMgr.hs view
@@ -10,14 +10,18 @@ ) where -import qualified Data.Map as M+import Control.Applicative import Control.Concurrent import Control.Concurrent.STM+import Control.DeepSeq import Control.Monad.State import Control.Monad.Reader +import Data.Array+import qualified Data.Map as M+ import Network import System.IO import System.Log.Logger@@ -40,19 +44,20 @@ | StopTorrent InfoHash data TorrentLocal = TorrentLocal- { tcPcMgrCh :: PieceMgrChannel- , tcFSCh :: FSPChannel- , tcStatTV :: TVar [PStat]- , tcPM :: PieceMap+ { tcPcMgrCh :: !PieceMgrChannel+ , tcFSCh :: !FSPChannel+ , tcStatTV :: !(TVar [PStat])+ , tcPM :: !PieceMap } -+instance NFData ThreadId where+ rnf x = x `seq` () type PeerMgrChannel = TChan PeerMgrMsg data CF = CF { peerCh :: PeerMgrChannel , mgrCh :: MgrChannel- , peerPool :: SupervisorChan+ , peerPool :: SupervisorChannel , chokeMgrCh :: ChokeMgrChannel , chokeRTV :: RateTVar }@@ -63,14 +68,14 @@ type ChanManageMap = M.Map InfoHash TorrentLocal -data ST = ST { peersInQueue :: [(InfoHash, Peer)]- , peers :: M.Map ThreadId PeerChannel- , peerId :: PeerId- , cmMap :: ChanManageMap+data ST = ST { peersInQueue :: ![(InfoHash, Peer)]+ , peers :: !(M.Map ThreadId PeerChannel)+ , peerId :: !PeerId+ , cmMap :: !ChanManageMap } start :: PeerMgrChannel -> PeerId- -> ChokeMgrChannel -> RateTVar -> SupervisorChan+ -> ChokeMgrChannel -> RateTVar -> SupervisorChannel -> IO ThreadId start ch pid chokeMgrC rtv supC = do mgrC <- newTChanIO@@ -119,11 +124,13 @@ newPeer ih tid c = do debugP $ "Adding new peer " ++ show tid cch <- asks chokeMgrCh liftIO . atomically $ writeTChan cch (AddPeer ih tid c)- modify (\s -> s { peers = M.insert tid c (peers s)})+ npeers <- M.insert tid c <$> gets peers+ npeers `deepseq` modify (\s -> s { peers = npeers }) removePeer tid = do debugP $ "Removing peer " ++ show tid cch <- asks chokeMgrCh liftIO . atomically $ writeTChan cch (RemovePeer tid)- modify (\s -> s { peers = M.delete tid (peers s)})+ npeers <- M.delete tid <$> gets peers+ npeers `deepseq` modify (\s -> s { peers = npeers }) numPeers :: Int numPeers = 40@@ -158,7 +165,7 @@ type ConnectRecord = (HostName, PortID, PeerId, InfoHash) -connect :: ConnectRecord -> SupervisorChan -> MgrChannel -> RateTVar -> ChanManageMap+connect :: ConnectRecord -> SupervisorChannel -> MgrChannel -> RateTVar -> ChanManageMap -> IO ThreadId connect (host, port, pid, ih) pool mgrC rtv cmap = forkIO (connector >> return ())@@ -181,12 +188,12 @@ Nothing -> error "Impossible (2), I hope" Just x -> x children <- Peer.start h mgrC rtv (tcPcMgrCh tc) (tcFSCh tc) (tcStatTV tc)- (tcPM tc) (M.size (tcPM tc)) ihsh+ (tcPM tc) (succ . snd . bounds $ tcPM tc) ihsh atomically $ writeTChan pool $ SpawnNew (Supervisor $ allForOne "PeerSup" children) return () -acceptor :: (Handle, HostName, PortNumber) -> SupervisorChan+acceptor :: (Handle, HostName, PortNumber) -> SupervisorChannel -> PeerId -> MgrChannel -> RateTVar -> ChanManageMap -> IO ThreadId acceptor (h,hn,pn) pool pid mgrC rtv cmmap =@@ -206,7 +213,8 @@ Nothing -> error "Impossible, I hope" Just x -> x children <- Peer.start h mgrC rtv (tcPcMgrCh tc) (tcFSCh tc)- (tcStatTV tc) (tcPM tc) (M.size (tcPM tc)) ih+ (tcStatTV tc) (tcPM tc)+ (succ . snd . bounds $ tcPM tc) ih atomically $ writeTChan pool $ SpawnNew (Supervisor $ allForOne "PeerSup" children) return ()
src/Process/PieceMgr.hs view
@@ -10,10 +10,12 @@ import Control.Applicative import Control.Concurrent import Control.Concurrent.STM-import Control.Exception (assert)+import Control.DeepSeq+import Control.Exception (assert, evaluate) import Control.Monad.Reader import Control.Monad.State+import Data.Array import Data.List import qualified Data.PendingSet as PendS import qualified Data.ByteString as B@@ -28,7 +30,7 @@ import Process.FS hiding (start) import Process.Status as STP hiding (start) -import Process.ChokeMgr (ChokeMgrMsg(..), ChokeMgrChannel)+import Process.ChokeMgr hiding (start) import Supervisor import Torrent import Tracer@@ -134,7 +136,7 @@ type PieceMgrProcess v = Process CF ST v start :: PieceMgrChannel -> FSPChannel -> ChokeMgrChannel -> StatusChannel -> ST -> InfoHash- -> SupervisorChan -> IO ThreadId+ -> SupervisorChannel -> IO ThreadId start mgrC fspC chokeC statC db ih supC = {-# SCC "PieceMgr" #-} spawnP (CF mgrC fspC chokeC statC ih) db@@ -159,7 +161,10 @@ modify (\db -> db { donePush = tail (donePush db) }) traceMsg :: PieceMgrMsg -> Process CF ST ()-traceMsg m = modify (\db -> db { traceBuffer = trace (show m) (traceBuffer db) })+traceMsg m = do+ tb <- gets traceBuffer+ let ntb = (trace $! show m) tb+ modify (\db -> db { traceBuffer = ntb }) rpcMessage :: Process CF ST () rpcMessage = do@@ -167,17 +172,19 @@ m <- liftIO . atomically $ readTChan ch traceMsg m case m of- GrabBlocks n eligible c ->+ GrabBlocks n eligible c -> {-# SCC "GrabBlocks" #-} do blocks <- grabBlocks n eligible liftIO . atomically $ do putTMVar c blocks -- Is never supposed to block- StoreBlock pn blk d -> storeBlock pn blk d- PutbackBlocks blks -> mapM_ putbackBlock blks- GetDone c -> do+ StoreBlock pn blk d -> {-# SCC "StoreBlock" #-}+ storeBlock pn blk d+ PutbackBlocks blks -> {-# SCC "PutbackBlocks" #-}+ mapM_ putbackBlock blks+ GetDone c -> {-# SCC "GetDone" #-} do done <- PS.toList =<< gets donePiece liftIO . atomically $ do putTMVar c done -- Is never supposed to block either PeerHave idxs -> peerHave idxs PeerUnhave idxs -> peerUnhave idxs- AskInterested pieces retC -> do+ AskInterested pieces retC -> {-# SCC "AskInterested" #-} do intr <- askInterested pieces liftIO . atomically $ do putTMVar retC intr -- And this neither too! @@ -187,7 +194,9 @@ debugP $ "Storing block: " ++ show (pn, blk) fch <- asks fspCh liftIO . atomically $ writeTChan fch $ WriteBlock pn blk d- modify (\s -> s { downloading = downloading s \\ [(pn, blk)] })+ dld <- gets downloading+ let ndl = dld \\ [(pn, blk)]+ dld `deepseq` modify (\s -> s { downloading = ndl }) endgameBroadcast pn blk done <- updateProgress pn blk when done@@ -199,10 +208,7 @@ ++ " completed, there are " ++ (show pendSz) ++ " 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)+ l <- gets infoMap >>= (\pm -> return $! len . (pm !) $ pn) ih <- asks pMgrInfoHash c <- asks statusCh liftIO . atomically $ writeTChan c (CompletedPiece ih l)@@ -236,7 +242,10 @@ endgameBroadcast pn blk = do ih <- asks pMgrInfoHash gets endGaming >>=- flip when (modify (\db -> db { donePush = (BlockComplete ih pn blk) : donePush db }))+ flip when+ (do dp <- gets donePush+ let dp' = (BlockComplete ih pn blk) : dp+ dp' `deepseq` modify (\db -> db { donePush = dp' })) markDone :: PieceNum -> Process CF ST () markDone pn = do@@ -260,7 +269,7 @@ done <- filt (==True) return $ ST pending done [] M.empty [] pmap False PendS.empty 0 (Tracer.new 20) where- filt f = PS.fromList (M.size pmap) . M.keys $ M.filter f mmap+ filt f = PS.fromList (succ . snd . bounds $ pmap) . M.keys $ M.filter f mmap ---------------------------------------------------------------------- @@ -276,7 +285,7 @@ doneP <- gets donePiece im <- gets infoMap donePSz <- PS.size doneP- when (M.size im == donePSz)+ when (succ (snd (bounds im)) == donePSz) (do liftIO $ putStrLn "Torrent Completed; to honor the torrent-gods thou must now sacrifice a goat!" ih <- asks pMgrInfoHash asks statusCh >>= (\ch -> liftIO . atomically $ writeTChan ch (STP.TorrentCompleted ih))@@ -321,9 +330,7 @@ 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+ let sz = len (pm ! pn) unless (assertAllDownloaded dl pn) (fail "Could not assert that all pieces were downloaded when completing a piece") unless (assertComplete ipp sz)@@ -357,19 +364,21 @@ -- 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) })+ ip <- gets inProgress+ ip' <- liftIO . evaluate $ M.adjust (const pg) pn ip+ modify (\db -> db { inProgress = ip' }) debugP $ "Iphave : " ++ show (ipHave pg) ++ " ipDone: " ++ show (ipDone pg) return (ipHave pg == ipDone pg) ipHave = S.size . ipHaveBlocks blockPiece :: BlockSize -> PieceSize -> [Block] blockPiece blockSz pieceSize = build pieceSize 0 []- where build 0 _os accum = reverse accum- build leftBytes os accum | leftBytes >= blockSz =+ where build 0 _os acc = reverse acc+ build leftBytes os acc | leftBytes >= blockSz = build (leftBytes - blockSz) (os + blockSz)- $ Block os blockSz : accum- | otherwise = build 0 (os + leftBytes) $ Block os leftBytes : accum+ $ Block os blockSz : acc+ | otherwise = build 0 (os + leftBytes) $ Block os leftBytes : acc -- | The call @grabBlocks n eligible@ tries to pick off up to @n@ pieces from -- to download. In doing so, it will only consider pieces in @eligible@. It@@ -458,9 +467,7 @@ -- download at peers. createBlock :: PieceNum -> 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 -> return $! cBlock $ im ! pn) where cBlock = blockPiece defaultBlockSize . fromInteger . len anyM :: Monad m => (a -> m Bool) -> [a] -> m Bool@@ -490,8 +497,8 @@ assertSets = do pending <- gets pendingPieces done <- gets donePiece- down <- return . map fst =<< gets downloading- iprog <- return . M.keys =<< gets inProgress+ down <- map fst <$> gets downloading+ iprog <- M.keys <$> gets inProgress pdownis <- anyM (flip PS.member pending) down donedownis <- anyM (flip PS.member done) down pdis <- PS.intersection pending done
src/Process/Status.hs view
@@ -20,6 +20,7 @@ import Control.Concurrent import Control.Concurrent.STM+import Control.DeepSeq import Control.Exception (assert) import Control.Monad.Reader@@ -70,6 +71,10 @@ , trackerMsgCh :: TrackerChannel } +instance NFData StatusState where+ rnf (SState up down l inc comp st _) =+ rnf up `seq` rnf down `seq` rnf l `seq` rnf inc `seq` rnf comp `seq` rnf st `seq` ()+ gatherStats :: (Integer, Integer) -> [(String, String)] gatherStats (upload, download) = [("uploaded", show upload), ("downloaded", show download),@@ -86,18 +91,18 @@ -- | Start a new Status process with an initial torrent state and a -- channel on which to transmit status updates to the tracker.-start :: Maybe FilePath -> StatusChannel -> TVar [PStat] -> SupervisorChan -> IO ThreadId+start :: Maybe FilePath -> StatusChannel -> TVar [PStat] -> SupervisorChannel -> IO ThreadId start fp statusC tv supC = do r <- newIORef (0,0) spawnP (CF statusC tv) M.empty- (cleanupP (foreverP (pgm r)) (defaultStopHandler supC) (cleanup r))+ (cleanupP (forever (pgm r)) (defaultStopHandler supC) (cleanup r)) where cleanup r = do st <- liftIO $ readIORef r case fp of Nothing -> return () Just fpath -> liftIO $ writeFile fpath (show . gatherStats $ st)- pgm r = {-# SCC "StatusP" #-} do+ pgm r = do fetchUpdates r d <- liftIO $ registerDelay (5 * 1000000) ch <- asks statusCh@@ -149,8 +154,11 @@ return updates mapM_ (\(PStat ih up down) -> do (u, d) <- liftIO $ readIORef r- liftIO $ writeIORef r (u+up, d+down)- modify (\s -> M.adjust (\st ->- st { uploaded = (uploaded st) + up- , downloaded = (downloaded st) + down }) ih s)) updates+ let nup = u + up+ ndn = d + down+ liftIO $ nup `deepseq` ndn `deepseq` writeIORef r (nup, ndn)+ s <- get+ put $! M.adjust (\st ->+ st `deepseq` st { uploaded = (uploaded st) + up+ , downloaded = (downloaded st) + down }) ih s) updates
src/Process/TorrentManager.hs view
@@ -55,7 +55,7 @@ -> ChokeMgr.ChokeMgrChannel -> PeerId -> PeerMgr.PeerMgrChannel- -> SupervisorChan+ -> SupervisorChannel -> IO ThreadId start chan statusC stv chokeC pid peerC supC = spawnP (CF chan statusC stv pid peerC chokeC) (ST [])@@ -103,7 +103,7 @@ let left = bytesLeft haveMap pieceMap ti <- liftIO $ mkTorrentInfo bc pieceDb <- PieceMgr.createPieceDb haveMap pieceMap- tid <- liftIO $ allForOne ("TorrentSup - " ++ fp)+ (tid, _) <- liftIO $ allForOne ("TorrentSup - " ++ fp) [ Worker $ FSP.start handles pieceMap fspC , Worker $ PieceMgr.start pieceMgrC fspC chokeC statusC pieceDb (infoHash ti) , Worker $ Tracker.start (infoHash ti) ti pid defaultPort statusC trackerC pmC
src/Process/Tracker.hs view
@@ -18,7 +18,7 @@ import Control.Monad.Reader import Control.Monad.State -import Data.Char (ord)+import Data.Char (ord, chr) import Data.List (intersperse) import qualified Data.ByteString as B @@ -94,7 +94,7 @@ start :: InfoHash -> TorrentInfo -> PeerId -> PortID -> Status.StatusChannel -> TrackerChannel -> PeerMgr.PeerMgrChannel- -> SupervisorChan -> IO ThreadId+ -> SupervisorChannel -> IO ThreadId start ih ti pid port statusC msgC pc supC = spawnP (CF statusC msgC pc ih) (ST ti pid Stopped port 0) (cleanupP (forever loop)@@ -244,7 +244,8 @@ headers = do s <- get p <- prt- return $ [("info_hash", rfc1738Encode $ infoHash $ torrentInfo s),+ return $ [("info_hash", rfc1738Encode $+ map (chr . fromIntegral) . B.unpack . infoHash . torrentInfo $ s), ("peer_id", rfc1738Encode $ peerId s), ("uploaded", show $ Status.uploaded ss), ("downloaded", show $ Status.downloaded ss),
src/Protocol/BCode.hs view
@@ -38,6 +38,7 @@ import Control.Applicative hiding (many) import qualified Data.ByteString.Lazy as L import qualified Data.ByteString as B+ import Data.Char import Data.List@@ -69,7 +70,7 @@ where bc' :: Int -> Gen BCode bc' 0 = oneof [BInt <$> arbitrary, BString <$> arbitrary]- bc' n | n > 0 =+ bc' n = oneof [BInt <$> arbitrary, BString <$> arbitrary, BArray <$> sequence (replicate n $ bc' (n `div` 8)),@@ -292,18 +293,18 @@ mbLength = infoLength bc mbFiles = do BArray fileList <- searchInfo "files" bc return $ do fileDict@(BDict _) <- fileList- let Just (BInt length) = search [toPS "length"] fileDict- Just (BArray path) = search [toPS "path"] fileDict- path' = map (\(BString s) -> fromBS s) path- return (path', length)+ let Just (BInt l) = search [toPS "length"] fileDict+ Just (BArray pth) = search [toPS "path"] fileDict+ pth' = map (\(BString s) -> fromBS s) pth+ return (pth', l) in case (mbFpath, mbLength, mbFiles) of (Just fpath, _, Just files) -> Just $- map (\(path, length) ->- (fpath:path, length)+ map (\(pth, l) ->+ (fpath:pth, l) ) files- (Just fpath, Just length, _) ->- Just [([fpath], length)]+ (Just fpath, Just l, _) ->+ Just [([fpath], l)] (_, _, Just files) -> Just files _ ->
src/Protocol/Wire.hs view
@@ -157,12 +157,6 @@ toBS :: String -> B.ByteString toBS = B.pack . map toW8 -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 @@ -180,7 +174,7 @@ protoString <- getByteString protocolHeaderSize when (protoString /= toBS protocolHeader) $ fail "Wrong protocol header" caps <- getWord64be- ihR <- liftM fromLBS $ getLazyByteString 20+ ihR <- getByteString 20 unless (ihTst ihR) $ fail "Unknown InfoHash" pid <- getLazyByteString 20 return (decodeCapabilities caps, pid, ihR)@@ -205,7 +199,7 @@ handShakeMessage :: PeerId -> InfoHash -> L.ByteString handShakeMessage pid ih = L.fromChunks . map runPut $ [putLazyByteString protocolHandshake,- putLazyByteString $ toLBS ih,+ putByteString ih, putByteString . toBS $ pid] -- | Receive a handshake on a socket@@ -222,7 +216,7 @@ hFlush h return $ Right (caps, rpid, ih) where msg ih = handShakeMessage pid ih- sz = fromIntegral (L.length $ msg "12345678901234567890") -- Dummy value+ sz = fromIntegral (L.length $ msg (B.pack $ replicate 20 32)) -- Dummy value -- | The call @constructBitField pieces@ will return the a ByteString suitable for inclusion in a
src/RateCalc.hs view
@@ -19,8 +19,8 @@ , bytes :: !Integer -- ^ The amount of bytes transferred since last rate extraction , count :: !Integer -- ^ The amount of bytes transferred since last count extraction , nextExpected :: !UTCTime -- ^ When is the next rate update expected- , lastExt :: !UTCTime -- ^ When was the last rate update- , rateSince :: !UTCTime -- ^ From where is the rate measured+ , lastExt :: !UTCTime -- ^ When was the last rate update+ , rateSince :: !UTCTime -- ^ From where is the rate measured } fudge :: NominalDiffTime@@ -40,9 +40,11 @@ -- | 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}+update n rt = nb `seq` nc `seq` rt { bytes = nb + n, count = nc + n}+ where nb = bytes rt+ nc = count rt + -- | The call @extractRate t rt@ extracts the current rate from the rate structure and updates the rate -- structures internal book-keeping extractRate :: UTCTime -> Rate -> (Double, Rate)@@ -69,5 +71,6 @@ -- | The call @extractCount rt@ extract the bytes transferred since last extraction extractCount :: Rate -> (Integer, Rate)-extractCount rt = (count rt, rt { count = 0 })+extractCount rt = crt `seq` (crt, rt { count = 0 })+ where crt = count rt
src/Supervisor.hs view
@@ -8,7 +8,7 @@ Child(..) , Children , SupervisorMsg(..)- , SupervisorChan+ , SupervisorChannel -- * Supervisor Initialization , allForOne , oneForOne@@ -18,7 +18,6 @@ ) where -import Control.Applicative import Control.Concurrent import Control.Concurrent.STM import Control.Monad.State@@ -28,93 +27,95 @@ import Process -data Child = Supervisor (SupervisorChan -> IO ThreadId)- | Worker (SupervisorChan -> IO ThreadId)+data Child = Supervisor (SupervisorChannel -> IO (ThreadId, SupervisorChannel))+ | Worker (SupervisorChannel -> IO ThreadId) +instance Show Child where+ show (Supervisor _) = "Supervisor"+ show (Worker _) = "Worker"+ data SupervisorMsg = IAmDying ThreadId | PleaseDie ThreadId | SpawnNew Child+ deriving Show -type SupervisorChan = TChan SupervisorMsg+type SupervisorChannel = TChan SupervisorMsg type Children = [Child] data ChildInfo = HSupervisor ThreadId | HWorker ThreadId +data RestartPolicy = AllForOne | OneForOne -pDie :: SupervisorChan -> IO ()+pDie :: SupervisorChannel -> IO () pDie supC = do tid <- myThreadId atomically $ writeTChan supC (IAmDying tid) -class SupervisorConf a where- getParent :: a -> SupervisorChan- getChan :: a -> SupervisorChan--data CFOFA = CFOFA { name :: String- , chan :: SupervisorChan- , parent :: SupervisorChan- }--instance SupervisorConf CFOFA where- getParent = parent- getChan = chan+data CF = CF { name :: String -- ^ Name of the supervisor+ , chan :: SupervisorChannel -- ^ Channel of the supervisor+ , parent :: SupervisorChannel -- ^ Channel of the parent supervisor+ , restartPolicy :: RestartPolicy } -instance Logging CFOFA where+instance Logging CF where logName = name -data STOFA = STOFA { childInfo :: [ChildInfo] }+data ST = ST { 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 -> SupervisorChan -> IO ThreadId-allForOne n children parentC = do+start :: RestartPolicy -> String -> Children -> SupervisorChannel -> IO (ThreadId,+ SupervisorChannel)+start policy n children parentC = do c <- newTChanIO- spawnP (CFOFA n c parentC) (STOFA []) (catchP startup- (defaultStopHandler parentC))- where- startup = do- childs <- mapM spawnChild children- modify (\_ -> STOFA (reverse childs))- forever eventLoop- eventLoop = do- mTid <- liftIO myThreadId- pc <- asks parent- ch <- asks chan- m <- liftIO . atomically $- (readTChan ch >>= return . Left) `orElse`- (readTChan pc >>= return . Right)- case m of- Left ev -> case ev of- IAmDying _tid -> do- gets childInfo >>= mapM_ finChild- t <- liftIO myThreadId- asks parent >>= \c -> liftIO . atomically $ writeTChan c (IAmDying t)- SpawnNew chld -> do- nc <- spawnChild chld- modify (\(STOFA cs) -> STOFA (nc : cs))- _ -> fail "Impossible"- Right ev -> case ev of- PleaseDie tid | tid == mTid -> do- gets childInfo >>= mapM_ finChild- stopP- _ -> return ()+ t <- spawnP (CF n c parentC policy) (ST []) (catchP (startup children)+ (defaultStopHandler parentC))+ return (t, c) -data CFOFO = CFOFO { oName :: String- , oChan :: SupervisorChan- , oparent :: SupervisorChan- }+startup :: [Child] -> Process CF ST ()+startup children = do+ spawnedChildren <- mapM spawnChild children+ put $ ST (reverse spawnedChildren)+ forever eventLoop -instance SupervisorConf CFOFO where- getParent = oparent- getChan = oChan+eventLoop :: Process CF ST ()+eventLoop = do+ mTid <- liftIO myThreadId+ pc <- asks parent+ ch <- asks chan+ m <- liftIO . atomically $+ (readTChan ch >>= return . Left) `orElse`+ (readTChan pc >>= return . Right)+ case m of+ Left (IAmDying tid) -> handleIAmDying tid+ Left (SpawnNew chld) -> handleSpawnNew chld+ Right (PleaseDie tid) | tid == mTid -> handlePleaseDie+ _ -> return () -- Ignore these. Since the chan is duped, we get stray messages from above -instance Logging CFOFO where- logName = oName+handleIAmDying :: ThreadId -> Process CF ST ()+handleIAmDying tid = do+ p <- asks restartPolicy+ case p of+ AllForOne -> do+ gets childInfo >>= mapM_ finChild+ stopP+ OneForOne ->+ pruneChild tid -data STOFO = STOFO { oChildInfo :: [ChildInfo] }+handleSpawnNew :: Child -> Process CF ST ()+handleSpawnNew chld = do+ nc <- spawnChild chld+ modify (\(ST cs) -> ST (nc : cs)) +handlePleaseDie :: Process CF ST ()+handlePleaseDie = do+ gets childInfo >>= mapM_ finChild+ stopP+++pruneChild :: ThreadId -> Process CF ST ()+pruneChild tid = modify (\(ST cs) -> ST (filter chk cs))+ where chk (HSupervisor t) = t == tid+ chk (HWorker t) = t == tid+ -- | A One-for-one supervisor is called with @oneForOne children parentCh@. It will spawn and run -- @children@ and be linked into the supervisor structure on @parentCh@. It returns the ThreadId -- of the supervisor itself and the Channel of which it is the controller.@@ -123,61 +124,36 @@ -- the death and let the other processes keep running. -- -- TODO: Restart policies.-oneForOne :: String -> Children -> SupervisorChan -> IO (ThreadId, SupervisorChan)-oneForOne n children parentC = do- c <- newTChanIO- t <- spawnP (CFOFO n c parentC) (STOFO []) (catchP startup- (defaultStopHandler parentC))- return (t, c)- where- startup :: Process CFOFO STOFO ()- startup = do- childs <- mapM spawnChild children- modify (\_ -> STOFO (reverse childs))- forever eventLoop- eventLoop :: Process CFOFO STOFO ()- eventLoop = do- mTid <- liftIO myThreadId- pc <- asks oparent- ch <- asks oChan- m <- liftIO . atomically $- (readTChan ch >>= return . Left) `orElse`- (readTChan pc >>= return . Right)- case m of- Left ev -> case ev of- IAmDying tid -> pruneChild tid- SpawnNew chld -> do nc <- spawnChild chld- modify (\(STOFO cs) -> STOFO (nc : cs))- _ -> fail "Impossible (2)"- Right ev -> case ev of- PleaseDie tid | tid == mTid -> do- gets oChildInfo >>= mapM_ finChild- stopP- _ -> return ()- pruneChild tid = modify (\(STOFO cs) -> STOFO (filter chk cs))- where chk (HSupervisor t) = t == tid- chk (HWorker t) = t == tid+oneForOne :: String -> Children -> SupervisorChannel -> IO (ThreadId, SupervisorChannel)+oneForOne = start OneForOne -finChild :: SupervisorConf a => ChildInfo -> Process a b ()-finChild (HWorker tid) = liftIO $ killThread tid -- Make this call killP in Process?+-- | 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 -> SupervisorChannel -> IO (ThreadId, SupervisorChannel)+allForOne = start AllForOne+++finChild :: ChildInfo -> Process CF ST ()+finChild (HWorker tid) = liftIO $ killThread tid finChild (HSupervisor tid) = do- c <- getChan <$> ask+ c <- asks chan liftIO . atomically $ writeTChan c (PleaseDie tid) -spawnChild :: SupervisorConf a => Child -> Process a b ChildInfo+spawnChild :: Child -> Process CF ST ChildInfo spawnChild (Worker proc) = do- c <- getChan <$> ask+ c <- asks chan nc <- liftIO . atomically $ dupTChan c tid <- liftIO $ proc nc return $ HWorker tid spawnChild (Supervisor proc) = do- c <- getChan <$> ask+ c <- asks chan nc <- liftIO . atomically $ dupTChan c- tid <- liftIO $ proc nc+ (tid, _) <- liftIO $ proc nc return $ HSupervisor tid -defaultStopHandler :: SupervisorChan -> Process a b ()+defaultStopHandler :: SupervisorChannel -> Process a b () defaultStopHandler supC = do t <- liftIO $ myThreadId liftIO . atomically $ writeTChan supC $ IAmDying t
src/Torrent.hs view
@@ -27,9 +27,12 @@ where import Control.Applicative+import Control.DeepSeq++import Data.Array+import Data.List import qualified Data.Foldable as F import qualified Data.ByteString as B-import Data.List import qualified Data.Map as M import Network@@ -61,18 +64,20 @@ data TorrentState = Seeding | Leeching deriving Show +instance NFData TorrentState+ -- PIECES ---------------------------------------------------------------------- type PieceNum = Int type PieceSize = Int data PieceInfo = PieceInfo {- offset :: Integer, -- ^ Offset of the piece, might be greater than Int- len :: Integer, -- ^ Length of piece; usually a small value- digest :: String -- ^ Digest of piece; taken from the .torret file+ offset :: !Integer, -- ^ Offset of the piece, might be greater than Int+ len :: !Integer, -- ^ Length of piece; usually a small value+ digest :: !B.ByteString -- ^ Digest of piece; taken from the .torret file } deriving (Eq, Show) -type PieceMap = M.Map PieceNum PieceInfo+type PieceMap = Array PieceNum PieceInfo -- | The PiecesDoneMap is a map which is true if we have the piece and false otherwise type PiecesDoneMap = M.Map PieceNum Bool@@ -86,19 +91,22 @@ -- map of the shape of the torrent in question. bytesLeft :: PiecesDoneMap -> PieceMap -> Integer bytesLeft done pm =- M.foldWithKey (\k v accu ->+ foldl' (\accu (k,v) -> case M.lookup k done of- Just False -> (len v) + accu- _ -> accu) 0 pm+ Just False -> (len v) + accu+ _ -> accu) 0 $ Data.Array.assocs pm -- BLOCKS ---------------------------------------------------------------------- type BlockSize = Int -data Block = Block { blockOffset :: Int -- ^ offset of this block within the piece- , blockSize :: BlockSize -- ^ size of this block within the piece+data Block = Block { blockOffset :: !Int -- ^ offset of this block within the piece+ , blockSize :: !BlockSize -- ^ size of this block within the piece } deriving (Eq, Ord, Show) +instance NFData Block where+ rnf (Block bo sz) = rnf bo `seq` rnf sz `seq` ()+ instance Arbitrary Block where arbitrary = Block <$> pos <*> pos where pos = choose (0, 4294967296 - 1)@@ -130,9 +138,9 @@ ih <- hashInfoDict bc return TorrentInfo { infoHash = ih, announceURL = ann, pieceCount = np } where- queryInfo bc =- do ann <- announce bc- np <- numberPieces bc+ queryInfo b =+ do ann <- announce b+ np <- numberPieces b return (ann, np) -- | Create a new PeerId for this client