Combinatorrent 0.2.1 → 0.2.2
raw patch · 16 files changed
+66/−55 lines, 16 files
Files
- Combinatorrent.cabal +2/−2
- configure +1/−1
- src/Process/ChokeMgr.hs +2/−1
- src/Process/Console.hs +3/−2
- src/Process/DirWatcher.hs +2/−1
- src/Process/FS.hs +2/−1
- src/Process/Peer.hs +14/−2
- src/Process/Peer/Receiver.hs +2/−1
- src/Process/Peer/Sender.hs +14/−27
- src/Process/Peer/SenderQ.hs +6/−4
- src/Process/PeerMgr.hs +2/−1
- src/Process/PieceMgr.hs +8/−5
- src/Process/Status.hs +2/−1
- src/Process/TorrentManager.hs +2/−2
- src/Process/Tracker.hs +2/−1
- src/Protocol/Wire.hs +2/−3
Combinatorrent.cabal view
@@ -1,6 +1,6 @@ name: Combinatorrent category: Network-version: 0.2.1+version: 0.2.2 category: Network description: Combinatorrent provides a BitTorrent client, based on STM for concurrency. This is an early preview release which is capable of@@ -78,7 +78,7 @@ time ghc-prof-options: -auto-all -caf-all- ghc-options: -Wall -fno-warn-orphans -threaded -feager-blackholing -O2+ ghc-options: -Wall -fno-warn-orphans -threaded if !flag(debug) cpp-options: "-DNDEBUG"
configure view
@@ -3,7 +3,7 @@ # Simple configuration script. ## Git version-GITHEAD="0.2.1 Hackage"+GITHEAD="0.2.2 Hackage" if test -d "${GIT_DIR:-.git}" ; then GITHEAD=`git describe 2>/dev/null`
src/Process/ChokeMgr.hs view
@@ -69,7 +69,7 @@ start ch rtv ur supC = do _ <- registerSTM roundTickSecs ch Tick spawnP (CF ch rtv) (initPeerDB $ calcUploadSlots ur Nothing)- (catchP (forever pgm)+ (catchP pgm (defaultStopHandler supC)) where initPeerDB slots = PeerDB 2 slots S.empty M.empty []@@ -84,6 +84,7 @@ BlockComplete ih pn blk -> informBlockComplete ih pn blk PieceDone ih pn -> informDone ih pn TorrentComplete ih -> modify (\s -> s { seeding = S.insert ih $ seeding s })+ pgm tick = do debugP "Ticked" c <- asks mgrCh _ <- registerSTM roundTickSecs c Tick
src/Process/Console.hs view
@@ -39,9 +39,9 @@ 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))+ spawnP (CF cmdC wrtC) () (catchP eventLoop (defaultStopHandler supC)) where- lp = do+ eventLoop = do c <- asks cmdCh o <- asks wrtCh m <- liftIO . atomically $ readTChan c@@ -54,6 +54,7 @@ liftIO . atomically $ writeTChan statusC (St.RequestAllTorrents v) sts <- liftIO . atomically $ takeTMVar v liftIO . atomically $ writeTChan o (show sts)+ eventLoop helpMessage :: String helpMessage = concat
src/Process/DirWatcher.hs view
@@ -37,13 +37,14 @@ -> IO ThreadId start fp chan supC = do spawnP (CF chan fp) S.empty- (catchP (forever pgm) (defaultStopHandler supC))+ (catchP pgm (defaultStopHandler supC)) where pgm = do q <- liftIO $ registerDelay (5 * 1000000) liftIO . atomically $ do b <- readTVar q if b then return () else retry processDirectory+ pgm processDirectory :: Process CF ST () processDirectory = do
src/Process/FS.hs view
@@ -47,7 +47,7 @@ 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))+ spawnP (CF fspC) (ST handles pm) (catchP lp (defaultStopHandler supC)) where lp = do c <- asks fspCh@@ -70,3 +70,4 @@ fh <- gets fileHandles pmap <- gets pieceMap liftIO $ FS.writeBlock fh pn blk pmap bs+ lp
src/Process/Peer.hs view
@@ -99,6 +99,7 @@ , upRate :: !Rate -- ^ Upload rate towards the peer (estimated) , downRate :: !Rate -- ^ Download rate from the peer (estimated) , runningEndgame :: !Bool -- ^ True if we are in endgame+ , lastMessage :: !Int } @@ -116,7 +117,7 @@ pieceSet <- PS.new nPieces 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)+ (ST True False S.empty True False pieceSet nPieces (RC.new ct) (RC.new ct) False 0) (cleanupP (startup nPieces) (defaultStopHandler supC) cleanup) startup :: Int -> Process CF ST ()@@ -131,7 +132,7 @@ -- Install the StatusP timer c <- asks timerCh _ <- registerSTM 5 c ()- forever eventLoop+ eventLoop cleanup :: Process CF ST () cleanup = do@@ -162,6 +163,7 @@ ChokeMgrEvt m -> chokeMsg m UpRateEvent up -> modify (\s -> s { upRate = RC.update up $ upRate s}) TimerEvent -> timerTick+ eventLoop data Operation = PeerMsgEvt (Message, Integer) | ChokeMgrEvt PeerMessage@@ -198,6 +200,14 @@ modify (\s -> s { blockQueue = S.delete (pn, blk) $ blockQueue s }) outChan $ SenderQ.SenderQRequestPrune pn blk +processLastMessage :: Process CF ST ()+processLastMessage = do+ lm <- gets lastMessage+ if lm >= 24+ then do outChan $ SenderQ.SenderQM KeepAlive+ else let inc = succ lm+ in inc `seq` modify (\st -> st { lastMessage = inc })+ -- 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@@ -205,6 +215,7 @@ timerTick :: Process CF ST () timerTick = do mTid <- liftIO myThreadId+ processLastMessage debugP "TimerEvent" tch <- asks timerCh _ <- registerSTM 5 tch ()@@ -437,5 +448,6 @@ -- | Send a message on a chan from the process queue outChan :: SenderQ.SenderQMsg -> Process CF ST () outChan qm = do+ modify (\st -> st { lastMessage = 0 }) ch <- asks outCh liftIO . atomically $ writeTChan ch qm
src/Process/Peer/Receiver.hs view
@@ -33,7 +33,7 @@ start h ch supC = do hSetBuffering h NoBuffering spawnP (CF ch) h- (catchP (forever readSend)+ (catchP readSend (defaultStopHandler supC)) readSend :: Process CF Handle ()@@ -49,6 +49,7 @@ Left _ -> do warningP "Incorrect parse in receiver, dying!" stopP Right msg -> liftIO . atomically $ writeTChan c (msg, fromIntegral l)+ readSend conv :: B.ByteString -> Process CF Handle Word32 conv bs = do
src/Process/Peer/Sender.hs view
@@ -2,51 +2,38 @@ ( start ) where -import Control.Applicative import Control.Concurrent import Control.Concurrent.STM import Control.Monad.Reader-import Control.Monad.State -import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as L import System.IO import Process-import Protocol.Wire import Supervisor -data CF = CF { chan :: TMVar B.ByteString }+data CF = CF { chan :: TMVar L.ByteString+ , hndl :: Handle } instance Logging CF where logName _ = "Process.Peer.Sender" -- | The raw sender process, it does nothing but send out what it syncs on.-start :: Handle -> TMVar B.ByteString -> SupervisorChannel -> IO ThreadId-start h ch supC = spawnP (CF ch) h (catchP (forever pgm)+start :: Handle -> TMVar L.ByteString -> SupervisorChannel -> IO ThreadId+start h ch supC = spawnP (CF ch h) () (catchP pgm (do t <- liftIO $ myThreadId liftIO . atomically $ writeTChan supC $ IAmDying t liftIO $ hClose h)) -pgm :: Process CF Handle ()-pgm = {-# SCC "Peer.Sender" #-} do- ch <- asks chan- tout <- liftIO $ registerDelay defaultTimeout- r <- liftIO . atomically $ do- t <- readTVar tout- if t- then return Nothing- else Just <$> takeTMVar ch- h <- get- case r of- Nothing -> putMsg (encodePacket KeepAlive)- Just m -> putMsg m- liftIO $ hFlush h- where- putMsg m = do- h <- get- liftIO $ B.hPut h m+pgm :: Process CF () ()+pgm = do+ ch <- asks chan+ h <- asks hndl+ liftIO $ do+ r <- atomically $ takeTMVar ch+ L.hPut h r+ hFlush h+ pgm -defaultTimeout :: Int-defaultTimeout = 120 * 1000000
src/Process/Peer/SenderQ.hs view
@@ -13,6 +13,7 @@ import Prelude hiding (catch, log) import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as L import Channels import Process@@ -30,7 +31,7 @@ | SenderQRequestPrune PieceNum Block -- ^ Prune SendQueue of this (pn, blk) pair data CF = CF { sqIn :: TChan SenderQMsg- , sqOut :: TMVar B.ByteString+ , sqOut :: TMVar L.ByteString , bandwidthCh :: BandwidthChannel , readBlockTV :: TMVar B.ByteString , fsCh :: FSPChannel@@ -45,12 +46,12 @@ -- | sendQueue Process, simple version. -- TODO: Split into fast and slow.-start :: TChan SenderQMsg -> TMVar B.ByteString -> BandwidthChannel+start :: TChan SenderQMsg -> TMVar L.ByteString -> BandwidthChannel -> 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)+ (catchP pgm (defaultStopHandler supC)) pgm :: Process CF ST ()@@ -68,7 +69,7 @@ Right (pn, blk) -> do d <- readBlock pn blk return $ Piece pn (blockOffset blk) d let bs = encodePacket p- sz = fromIntegral $ B.length bs+ sz = fromIntegral $ L.length bs liftIO . atomically $ (putTMVar ov bs >> return (Left sz)) `orElse` (readTChan ic >>= return . Right)@@ -85,6 +86,7 @@ modifyQ (Q.push $ Left Choke) SenderQRequestPrune n blk -> modifyQ (Q.filter (filterRequest n blk))+ pgm rateUpdateEvent :: Process CF ST () rateUpdateEvent = {-# SCC "Peer.SendQ.rateUpd" #-} do
src/Process/PeerMgr.hs view
@@ -82,7 +82,7 @@ fakeChan <- newTChanIO pool <- liftM snd $ oneForOne "PeerPool" [] fakeChan spawnP (CF ch mgrC pool chokeMgrC rtv)- (ST [] M.empty pid cmap) (catchP (forever lp)+ (ST [] M.empty pid cmap) (catchP lp (defaultStopHandler supC)) where cmap = M.empty@@ -96,6 +96,7 @@ Left msg -> incomingPeers msg Right msg -> peerEvent msg fillPeers+ lp incomingPeers :: PeerMgrMsg -> Process CF ST () incomingPeers msg =
src/Process/PieceMgr.hs view
@@ -140,12 +140,15 @@ start mgrC fspC chokeC statC db ih supC = {-# SCC "PieceMgr" #-} spawnP (CF mgrC fspC chokeC statC ih) db- (catchP (forever pgm)+ (catchP eventLoop (defaultStopHandler supC))- where pgm = do- assertST- rpcMessage- drainSend++eventLoop :: Process CF ST ()+eventLoop = do+ assertST+ rpcMessage+ drainSend+ eventLoop drainSend :: Process CF ST () drainSend = do
src/Process/Status.hs view
@@ -95,7 +95,7 @@ start fp statusC tv supC = do r <- newIORef (0,0) spawnP (CF statusC tv) M.empty- (cleanupP (forever (pgm r)) (defaultStopHandler supC) (cleanup r))+ (cleanupP (pgm r) (defaultStopHandler supC) (cleanup r)) where cleanup r = do st <- liftIO $ readIORef r@@ -114,6 +114,7 @@ case x of Nothing -> return () Just msg -> recvMsg msg+ pgm r newMap :: Integer -> TrackerChannel -> StatusState newMap l trackerMsgC =
src/Process/TorrentManager.hs view
@@ -59,8 +59,8 @@ -> IO ThreadId start chan statusC stv chokeC pid peerC supC = spawnP (CF chan statusC stv pid peerC chokeC) (ST [])- (catchP (forever pgm) (defaultStopHandler supC))- where pgm = do startStop >> dirMsg+ (catchP pgm (defaultStopHandler supC))+ where pgm = startStop >> dirMsg >> pgm dirMsg = do c <- asks tCh ls <- liftIO . atomically $ readTChan c
src/Process/Tracker.hs view
@@ -97,7 +97,7 @@ -> 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)+ (cleanupP loop (defaultStopHandler supC) stopEvent) where@@ -120,6 +120,7 @@ modify (\s -> s { state = Started }) >> talkTracker Complete -> modify (\s -> s { state = Completed }) >> talkTracker+ loop talkTracker = pokeTracker >>= timerUpdate eventTransition :: Process CF ST ()
src/Protocol/Wire.hs view
@@ -22,7 +22,6 @@ import Control.Applicative hiding (empty) import Control.Monad -import Data.Monoid import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L @@ -87,8 +86,8 @@ decodeMsg :: Get Message decodeMsg = {-# SCC "decodeMsg" #-} get -encodePacket :: Message -> B.ByteString-encodePacket m = mconcat [szEnc, mEnc]+encodePacket :: Message -> L.ByteString+encodePacket m = L.fromChunks [szEnc, mEnc] where mEnc = encode m sz = B.length mEnc szEnc = runPut . p32be $ sz