diff --git a/AUTHORS b/AUTHORS new file mode 100644 --- /dev/null +++ b/AUTHORS @@ -0,0 +1,5 @@ +Alex Mason +Astro +Jesper Louis Andersen +John Gunderman +Thomas Christensen diff --git a/HaskellTorrent.cabal b/HaskellTorrent.cabal new file mode 100644 --- /dev/null +++ b/HaskellTorrent.cabal @@ -0,0 +1,74 @@ +name: HaskellTorrent +category: Network +version: 0.0 +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 + downloading files from various torrent trackers, but have not yet + demonstrated to be correct in all aspects. + + It is expected that the package currently contains numerous and even + grave bugs. Patches to fix any problem are welcome! +cabal-version: >= 1.6 + +license: BSD3 +license-file: LICENSE +copyright: (c) 2009 Jesper Louis Andersen +author: Jesper Louis Andersen +maintainer: jesper.louis.andersen@gmail.com +stability: experimental +synopsis: A concurrent bittorrent client +build-type: Configure + +extra-tmp-files: src/Version.hs +extra-source-files: src/Version.hs.in, configure +data-files: AUTHORS, README.md + +flag debug + description: Enable debug support + default: False + +executable HaskellTorrent + hs-source-dirs: src + main-is: HaskellTorrent.hs + other-modules: Protocol.BCode, Protocol.Wire, + Data.Queue, + 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, + Supervisor, Torrent + + extensions: CPP + + build-depends: + base >= 3.0, + base <= 5.0, + mtl, + cereal, + pretty, + parsec, + containers, + bytestring, + random, + network, + cml, + HTTP, + HsOpenSSL, + time, + random-shuffle, + directory + + ghc-options: -threaded -fwarn-unused-imports + if flag(debug) + cpp-options: "-DDEBUG" + else + cpp-options: "-DNDEBUG" + +source-repository head + type: git + location: git://github.com/jlouis/haskell-torrent.git + branch: master + + + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,25 @@ +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 +HOLDER 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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,92 @@ +Haskell Torrent - a haskell bittorrent client. +========== + +Introduction +---------- + +This is a Haskell bittorrent client. I am the introduction document +and I need to be written by some generous soul! + +Installation +------------ + +Here is what I do to install haskell torrrent locally on my machine: + + cabal install --prefix=$HOME --user + +Since we are using the magnificient cabal, this is enough to install haskell torrent in our $HOME/bin directory. + +Usage +----------------- + +Haskell torrent can currently only do one very simple thing. If you call it with + + 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. + +Protocol support +---------------- + +Currently haskell-torrent supports the following BEPs (See the +[BEP Process](http://www.bittorrent.org/beps/bep_0000.html) document for an +explanation of these) + + - 004, 020, + +Haskell-torrent is not supporting these BEPs, but strives to do so one day: + + - 003, 005, 006, 007, 010, 012, 015, 009, 023, 018, 021, 022, 024, 026, 027, + 028, 029, 030, 031, 032 + +Haskell-torrent will probably never support these BEPs: + + - 016, 017, 019 + +Source code Hierarchy +--------------------- + + - **Data**: Data structures. + - **Queue**: Functional queues. Standard variant with two lists. + + - **Process**: Process definitions for the different processes comprising Haskell Torrent + - **ChokeMgr**: Manages choking and unchoking of peers, based upon the current speed of the peer + and its current state. Global for multiple torrents. + - **Console**: Simple console process. Only responds to 'quit' at the moment. + - **FS**: Process managing the file system. + - **Listen**: Not used at the moment. Step towards listening sockets. + - **Peer**: Several process definitions for handling peers. Two for sending, one for receiving + and one for controlling the peer and handle the state. + - **PeerMgr**: Management of a set of peers for a single torrent. + - **PieceMgr**: Keeps track of what pieces have been downloaded and what are missing. Also hands + out blocks for downloading to the peers. + - **Status**: Keeps track of uploaded/downloaded/left bytes for a single torrent. Could be globalized. + - **Timer**: Timer events. + - **Tracker**: Communication with the tracker. + + - **Protocol**: Modules for interacting with the various bittorrent protocols. + - **BCode**: The bittorrent BCode coding. Used by several protocols. + - **Wire**: The protocol used for communication between peers. + + - **Top Level**: + - **Digest**: SHA1 digests as used in the bittorrent protocol. + - **FS**: Low level Filesystem code. Interacts with files. + - **HaskellTorrent**: Main entry point to the code. Sets up processes. + - **Logging**: Logging interface. + - **LoggingTypes**: Types and instances used by the Logging framework. + - **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. + - **Supervisor**: Erlang-inspired Supervisor processes. + - **Torrent**: Various helpers and types for Torrents. + - **Version.hs.in**: Generates **Version.hs** via the configure script. + +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. + + + diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,7 @@ +#!/usr/bin/env runhaskell + +In principle, we could do with a lot less than autoconfUserhooks, but simpleUserHooks +is not running 'configure'. + +>import Distribution.Simple +>main = defaultMainWithHooks autoconfUserHooks diff --git a/configure b/configure new file mode 100644 --- /dev/null +++ b/configure @@ -0,0 +1,21 @@ +#!/bin/sh +# +# Simple configuration script. + +## Git version +GITHEAD="0.0 Hackage" +if test -d "${GIT_DIR:-.git}" ; then + GITHEAD=`git describe 2>/dev/null` + + if test -z ${GITHEAD} ; then + GITHEAD=`git rev-parse HEAD` + fi + + if test -n "`git diff-index -m --name-only HEAD`" ; then + GITHEAD="${GITHEAD}-dirty" + fi +fi + +sed -e "s/@GITHEAD@/$GITHEAD/g" src/Version.hs.in > src/Version.hs + + diff --git a/src/Data/Queue.hs b/src/Data/Queue.hs new file mode 100644 --- /dev/null +++ b/src/Data/Queue.hs @@ -0,0 +1,42 @@ +-- | Simple Functional queues based on a double list. This usually achieves good amortized bounds +module Data.Queue ( + -- * Types + Queue, + -- * Functions + empty, + isEmpty, + push, + pop, + Data.Queue.filter) +where + +import qualified Data.List as Lst + +data Queue a = Queue [a] [a] + +-- | The definition of an empty Queue +empty :: Queue a +empty = Queue [] [] + +-- | Returns True on an empty Queue, and False otherwise. +isEmpty :: Queue a -> Bool +isEmpty (Queue [] []) = True +isEmpty _ = False + +-- | Pushes a new element to the tail of the list. +-- Operates in constant time. +push :: a -> Queue a -> Queue a +push e (Queue front back) = Queue front (e : back) + + +-- | Pops the top most element off the queue. +-- Operates in amortized constant time +pop :: Queue a -> Maybe (a, Queue a) +pop (Queue [] []) = Nothing +pop (Queue (e : es) back) = Just (e, Queue es back) +pop (Queue [] back) = pop (Queue (reverse back) []) + +-- | Generates a new Queue only containing elements for which +-- p returns true. +filter :: (a -> Bool) -> Queue a -> Queue a +filter p (Queue front back) = Queue (Lst.filter p front) (Lst.filter p back) diff --git a/src/Digest.hs b/src/Digest.hs new file mode 100644 --- /dev/null +++ b/src/Digest.hs @@ -0,0 +1,32 @@ +-- | Simple abstraction for message digests +module Digest + ( Digest + , digest + ) +where + +import Control.Concurrent +import Control.Monad + +import Data.Maybe + +import qualified Data.ByteString.Lazy as L +import OpenSSL +import qualified OpenSSL.EVP.Digest as D + +-- 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 + diff --git a/src/FS.hs b/src/FS.hs new file mode 100644 --- /dev/null +++ b/src/FS.hs @@ -0,0 +1,226 @@ +-- 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(..), + PieceMap, + Handles, + readPiece, + readBlock, + writeBlock, + mkPieceMap, + checkFile, + checkPiece, + openAndCheckFile, + canSeed) +where + +import Control.Monad.State + +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) + +import Protocol.BCode as BCode +import qualified Digest as D +import Torrent + +-- | For multi-file torrents we've got to maintain multiple file +-- handles. The data structure may as well be a Map Range Handle, +-- but that's detailto only @projectHandles@. More importantly, +-- functions operating on the files must be aware that a +-- piece/block can span multiple files. +-- +-- FIXME: Replace this with a handle cache later. Many peers & many +-- tiny files will make us overstep the fd limit (usually +-- 1024). +newtype Handles = Handles [(Handle, Integer)] -- ^[(fileHandle, fileLength)] + +projectHandles :: Handles + -> Integer -- ^Torrent offset + -> Integer -- ^Torrent size + -> [(Handle -- ^File handle + ,Integer -- ^File chunk offset + ,Integer -- ^File chunk size + )] +{- +projectHandles handles offset size = let r = projectHandles' handles offset size + in trace ("projectHandles " ++ + show handles ++ " " ++ + show offset ++ " " ++ + show size ++ " = " ++ + show r + ) $ + r +-} +projectHandles (Handles handles@((h1, length1):handles')) offset size + | size <= 0 = + fail "FS: Should have already stopped projection" + | null handles = + fail "FS: Attempt to read beyond torrent length" + | offset >= length1 = + projectHandles (Handles handles') (offset - length1) size + | otherwise = + let size1 = length1 - offset -- ^How much of h1 to take? + in if size1 >= size + then [(h1, offset, size)] + else (h1, offset, size1) : + 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 + +-- | FIXME: minor code duplication with @readBlock@ +readPiece :: PieceNum -> Handles -> PieceMap -> IO L.ByteString +readPiece pn handles mp = + do pInfo <- pInfoLookup pn mp + bs <- L.concat `fmap` + forM (projectHandles handles (offset pInfo) (len pInfo)) + (\(h, offset, size) -> + do hSeek h AbsoluteSeek offset + L.hGet h (fromInteger size) + ) + if L.length bs == (fromInteger . len $ pInfo) + then return bs + else fail "FS: Wrong number of bytes read" + +-- | FIXME: concatenating strict ByteStrings may turn out +-- expensive. Returning lazy ones may be more appropriate. +readBlock :: PieceNum -> Block -> Handles -> PieceMap -> IO B.ByteString +readBlock pn blk handles mp = + do pInfo <- pInfoLookup pn mp + B.concat `fmap` + forM (projectHandles handles (offset pInfo + (fromIntegral $ blockOffset blk)) + (fromIntegral $ blockSize blk)) + (\(h, offset, size) -> + do hSeek h AbsoluteSeek offset + B.hGet h $ fromInteger size + ) + +-- | The call @writeBlock h n blk pm blkData@ will write the contents of @blkData@ +-- to the file pointed to by handle at the correct position in the file. If the +-- block is of a wrong length, the call will fail. +writeBlock :: Handles -> PieceNum -> Block -> PieceMap -> B.ByteString -> IO () +writeBlock handles n blk pm blkData = + do when lenFail $ fail "Writing block of wrong length" + pInfo <- pInfoLookup n pm + foldM_ (\blkData (h, offset, size) -> + do let size' = fromInteger size + hSeek h AbsoluteSeek offset + B.hPut h $ B.take size' blkData + hFlush h + return $ B.drop size' blkData + ) blkData (projectHandles handles (position pInfo) (fromIntegral $ B.length blkData)) + return () + where + position :: PieceInfo -> Integer + position pinfo = (offset pinfo) + fromIntegral (blockOffset blk) + lenFail = B.length blkData /= blockSize blk + +-- | The @checkPiece h inf@ checks the file system for correctness of a given piece, namely if +-- the piece described by @inf@ is correct inside the file pointed to by @h@. +checkPiece :: PieceInfo -> Handles -> IO Bool +checkPiece inf handles = do + bs <- L.concat `fmap` + forM (projectHandles handles (offset inf) (fromInteger $ len inf)) + (\(h, offset, size) -> + do hSeek h AbsoluteSeek offset + L.hGet h (fromInteger size) + ) + dgs <- liftIO $ D.digest bs + return (dgs == digest inf) + +-- | Create a MissingMap from a file handle and a piecemap. The system will read each part of +-- the file and then check it against the digest. It will create a map of what we are missing +-- in the file as a missing map. We could alternatively choose a list of pieces missing rather +-- then creating the data structure here. This is perhaps better in the long run. +checkFile :: Handles -> PieceMap -> IO PiecesDoneMap +checkFile handles pm = do l <- mapM checkP pieces + return $ M.fromList l + where pieces = M.toAscList pm + checkP :: (PieceNum, PieceInfo) -> IO (PieceNum, Bool) + checkP (pn, pInfo) = do b <- checkPiece pInfo handles + return (pn, b) + +-- | Extract the PieceMap from a bcoded structure +-- Needs some more defense in the long run. +mkPieceMap :: BCode -> Maybe PieceMap +mkPieceMap bc = fetchData + 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 + extract :: Integer -> Integer -> Integer -> [B.ByteString] -> [PieceInfo] + extract _ 0 _ [] = [] + extract plen tlen offst (p : ps) | tlen < plen = PieceInfo { offset = offst, + len = tlen, + digest = B.unpack 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 } + extract _ _ _ _ = error "mkPieceMap: the impossible happened!" + +-- | Predicate function. True if nothing is missing from the map. +canSeed :: PiecesDoneMap -> Bool +canSeed = M.fold (&&) True + +-- | Process a BCoded torrent file. Create directories, open the files +-- in question, check it and return Handles plus a haveMap for the +-- file +openAndCheckFile :: BCode -> IO (Handles, PiecesDoneMap, PieceMap) +openAndCheckFile bc = + do + handles <- Handles `fmap` + forM files + (\(path, length) -> + do let dir = joinPath $ init path + when (dir /= "") $ + createDirectoryIfMissing True dir + let fpath = joinPath path + h <- openBinaryFile fpath ReadWriteMode + return (h, length) + ) + have <- checkFile handles pieceMap + return (handles, have, pieceMap) + where Just files = BCode.infoFiles bc + Just pieceMap = mkPieceMap bc + joinPath = intercalate "/" + + + + + diff --git a/src/HaskellTorrent.hs b/src/HaskellTorrent.hs new file mode 100644 --- /dev/null +++ b/src/HaskellTorrent.hs @@ -0,0 +1,85 @@ +module Main (main) +where + +import Control.Concurrent.CML +import Control.Monad + +import qualified Data.ByteString as B + +import System.Environment +import System.Random + + +import qualified Protocol.BCode as BCode +import qualified Process.Console as Console +import qualified Process.FS as FSP +import qualified Process.PeerMgr as PeerMgr +import qualified Process.PieceMgr as PieceMgr (start, createPieceDb) +import qualified Process.ChokeMgr as ChokeMgr (start) +import qualified Process.Status as Status +import qualified Process.Tracker as Tracker +import Logging +import FS +import Supervisor +import Torrent +import Version + +main :: IO () +main = getArgs >>= run + + +run :: [String] -> IO () +run args = do + putStrLn $ "This is Haskell-torrent version " ++ version + case args of + [] -> putStrLn "*** Usage: haskellTorrent " + (name:_) -> download name + +download :: String -> IO () +download 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 + -- setup channels + trackerC <- channel + statusC <- channel + waitC <- channel + pieceMgrC <- channel + supC <- channel + fspC <- channel + statInC <- channel + pmC <- channel + chokeC <- channel + chokeInfoC <- channel + putStrLn "Created channels" + -- setup StdGen and Peer data + gen <- getStdGen + 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 + sync $ receive waitC (const True) + return () diff --git a/src/Logging.hs b/src/Logging.hs new file mode 100644 --- /dev/null +++ b/src/Logging.hs @@ -0,0 +1,40 @@ +-- | Logging primitives +{-# LANGUAGE TypeSynonymInstances #-} +module Logging ( + -- * Classes + Logging(..) + -- * Types + , LogChannel + , LogPriority(..) + -- * Spawning + , startLogger + -- * Interface (deprecated) + , logMsg + , logMsg' + ) +where + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.Reader + +import LoggingTypes +import Prelude hiding (log) + +import Process + +startLogger :: LogChannel -> IO ThreadId +startLogger logC = + spawnP logC () (forever lp) + where + lp = do m <- syncP =<< logEv + liftIO $ print m + logEv = recvP logC (const True) + +-- | Log a message to a channel +logMsg :: LogChannel -> String -> IO () +logMsg c m = logMsg' c "Unknown" Info m + +-- | Log a message to a channel with a priority +logMsg' :: LogChannel -> String -> LogPriority -> String -> IO () +logMsg' c name pri = sync . transmit c . Mes pri name diff --git a/src/LoggingTypes.hs b/src/LoggingTypes.hs new file mode 100644 --- /dev/null +++ b/src/LoggingTypes.hs @@ -0,0 +1,36 @@ +{-# LANGUAGE TypeSynonymInstances #-} +module LoggingTypes + ( Logging(..) + , LogPriority(..) + , LogMsg(..) + , LogChannel + ) +where + +import Control.Concurrent.CML + + +data LogPriority = Debug -- ^ Fine grained debug info + | Warn -- ^ Potentially harmful situations + | Info -- ^ Informational messages, progress reports + | Error -- ^ Errors that are continuable + | Fatal -- ^ Severe errors. Will probably make the application abort + | None -- ^ No logging at all + deriving (Show, Eq, Ord) +-- +-- | The class of types where we have a logger inside them somewhere +class Logging a where + -- | Returns a channel for logging and an Identifying string to use + getLogger :: a -> (String, LogChannel) + +instance Logging LogChannel where + getLogger ch = ("Unknown", ch) + + +-- TODO: Consider generalizing this to any member of Show +data LogMsg = Mes LogPriority String String + +instance Show LogMsg where + show (Mes pri name str) = show name ++ "(" ++ show pri ++ "):\t" ++ str + +type LogChannel = Channel LogMsg diff --git a/src/PeerTypes.hs b/src/PeerTypes.hs new file mode 100644 --- /dev/null +++ b/src/PeerTypes.hs @@ -0,0 +1,30 @@ +module PeerTypes +where + +import Control.Concurrent +import Control.Concurrent.CML + +import Data.Time.Clock + +import Network +import Torrent + +data Peer = Peer { peerHost :: HostName, + peerPort :: PortID } + +data PeerMessage = ChokePeer + | UnchokePeer + | PeerStats UTCTime (Channel (Double, Double, Bool)) -- Up/Down/Interested + | PieceCompleted PieceNum + | CancelBlock PieceNum Block + +type PeerChannel = Channel PeerMessage + + +data MgrMessage = Connect ThreadId (Channel PeerMessage) + | Disconnect ThreadId + +type MgrChannel = Channel MgrMessage + +-- | A Channel type we use for transferring the amount of data we transferred +type BandwidthChannel = Channel Integer diff --git a/src/Process.hs b/src/Process.hs new file mode 100644 --- /dev/null +++ b/src/Process.hs @@ -0,0 +1,211 @@ +-- | Core Process code +{-# LANGUAGE ExistentialQuantification, FlexibleInstances, + GeneralizedNewtypeDeriving, + MultiParamTypeClasses, CPP #-} +-- required for deriving Typeable +{-# OPTIONS_GHC -fglasgow-exts #-} + +module Process ( + -- * Types + Process + -- * Interface + , runP + , spawnP + , catchP + , cleanupP + , foreverP + , syncP + , chooseP + , sendP + , sendPC + , recvP + , recvPC + , recvWrapPC + , wrapP + , stopP + , ignoreProcessBlock -- This ought to be renamed + -- * Log Interface + , logInfo + , logDebug + , logWarn + , logFatal + , logError + ) +where + +import Data.Monoid + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Exception + +import Control.Monad.Reader +import Control.Monad.State + +import Data.Typeable + +import LoggingTypes +import Prelude hiding (catch, log) + +import System.IO + +-- import Supervisor + + +-- | 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 +-- channels, and the state the internal process state. It is implemented by means of a transformer +-- 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) +#endif + +data StopException = StopException + deriving (Show, Typeable) + +instance Exception StopException + +stopP :: Process a b c +stopP = throw StopException + +-- | Run the process monad given a configuation of type @a@ and a initial state of type @b@ +runP :: a -> b -> Process a b c -> IO (c, b) +runP c st (Process p) = runStateT (runReaderT p c) st + +-- | Spawn and run a process monad +spawnP :: a -> b -> Process a b () -> IO ThreadId +spawnP c st p = spawn proc + where proc = do runP c st p + return () + +-- | Run the process monad for its side effect, with a stopHandler if exceptions +-- are raised in the process +catchP :: Logging a => Process a b () -> Process a b () -> Process a b () +catchP proc stopH = cleanupP proc stopH (return ()) + +-- | Run the process monad for its side effect. @cleanupP p sh ch@ describes to +-- run @p@. If @p@ dies by a kill from a supervisor, run @ch@. Otherwise it runs +-- @ch >> sh@ on death. +cleanupP :: Logging a => Process a b () -> Process a b () -> Process a b () -> Process a b () +cleanupP proc stopH cleanupH = do + 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)) + ] + 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 + +syncP :: Event (c, b) -> Process a b c +syncP ev = do (a, s) <- liftIO $ sync ev + 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))) + +sendPC :: (a -> Channel c) -> c -> Process a b (Event ((), b)) +sendPC sel v = asks sel >>= flip sendP v + +recvP :: Channel c -> (c -> Bool) -> Process a b (Event (c, b)) +recvP ch pred = do + s <- get + return (wrap (receive ch pred) + (\v -> return (v, s))) + +recvPC :: (a -> Channel c) -> Process a b (Event (c, b)) +recvPC sel = asks sel >>= flip recvP (const True) + +wrapP :: Event (c, b) -> (c -> Process a b y) -> Process a b (Event (y, b)) +wrapP ev p = do + c <- ask + return $ wrap ev (\(v, s) -> runP c s (p v)) + +-- Convenience function +recvWrapPC :: (a -> Channel c) -> (c -> Process a b y) -> Process a b (Event (y, b)) +recvWrapPC sel p = do + ev <- recvPC sel + wrapP ev p + +chooseP :: [Process a b (Event (c, b))] -> Process a b (Event (c, b)) +chooseP events = (sequence events) >>= (return . choose) + +-- VERSION SPECIFIC PROCESS ORIENTED FUNCTIONS + +-- | @ignoreProcessBlock err thnk@ runs a process action, ignoring blocks on dead +-- MVars. If the MVar is blocked, return the default value @err@. +ignoreProcessBlock :: c -> Process a b c -> Process a b c +ignoreProcessBlock err thnk = do + st <- get + c <- ask + (a, s') <- liftIO $ runP c st thnk `catches` + -- Peer dead, ignore +#if (__GLASGOW_HASKELL__ == 610) + [ Handler (\BlockedOnDeadMVar -> return (err, st)) ] +#elif (__GLASGOW_HASKELL__ == 612) + [ Handler (\BlockedIndefinitelyOnMVar -> return (err, st)) ] +#else +#error Unknown GHC version +#endif + put s' + return a + +------ 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 + +instance Monoid LogPriority where + mempty = None + mappend None g = g + mappend f g = f + +-- | The level by which we log +logLevel :: LogFilter +#ifdef DEBUG +logLevel = mconcat [matchP "TrackerP" Debug, + matchAny Info] +#else +logLevel = matchAny Info +#endif + diff --git a/src/Process/ChokeMgr.hs b/src/Process/ChokeMgr.hs new file mode 100644 --- /dev/null +++ b/src/Process/ChokeMgr.hs @@ -0,0 +1,358 @@ +module Process.ChokeMgr ( + -- * Types, Channels + ChokeMgrChannel + , ChokeMgrMsg(..) + -- * Interface + , start + ) +where + +import Data.Time.Clock +import Data.List +import qualified Data.Map as M +import qualified Data.Set as S +import Data.Traversable as T + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.Reader +import Control.Monad.State + + +import Prelude hiding (catch, log) + +import System.Random + +import PeerTypes +import Process.PieceMgr hiding (start) +import Process +import Logging +import Supervisor +import Torrent hiding (infoHash) +import Process.Timer as Timer + +-- DATA STRUCTURES +---------------------------------------------------------------------- + +data ChokeMgrMsg = Tick + | RemovePeer PeerPid + | AddPeer PeerPid PeerChannel +type ChokeMgrChannel = Channel ChokeMgrMsg + +data CF = CF { logCh :: LogChannel + , mgrCh :: ChokeMgrChannel + , infoCh :: ChokeInfoChannel + } + +instance Logging CF where + getLogger cf = ("ChokeMgrP", logCh cf) + +type ChokeMgrProcess a = Process CF PeerDB a + +-- INTERFACE +---------------------------------------------------------------------- + +start :: LogChannel -> ChokeMgrChannel -> ChokeInfoChannel -> Int -> Bool -> SupervisorChan + -> IO ThreadId +start logC 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)) + 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) + 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] }) + +-- INTERNAL FUNCTIONS +---------------------------------------------------------------------- + +type PeerPid = ThreadId -- For now, should probably change + + +-- | The PeerDB is the database we keep over peers. It maps all the information necessary to determine +-- which peers are interesting to keep uploading to and which are slow. It also keeps track of how +-- 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. + , seeding :: Bool -- ^ True if we are seeding the torrent. + -- 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 + } + +-- | The PeerInfo structure maps, for each peer pid, its accompanying informative data for the PeerDB +data PeerInfo = PeerInfo + { pChokingUs :: Bool -- ^ True if the peer is choking us + , pDownRate :: Double -- ^ The rate of the peer in question, bytes downloaded in last window + , pUpRate :: Double -- ^ The rate of the peer in question, bytes uploaded in last window + , pChannel :: PeerChannel -- ^ The channel on which to communicate with the peer + , pInterestedInUs :: Bool -- ^ Reflection from Peer DB + , pAreSeeding :: Bool -- ^ True if this peer is connected on a torrent we seed + , pIsASeeder :: Bool -- ^ True if the peer is a seeder + } + +type PeerMap = M.Map PeerPid PeerInfo + +-- | Auxilliary data structure. Used in the rechoking process. +type RechokeData = (PeerPid, PeerInfo) + +-- | Comparison with inverse ordering +compareInv :: Ord a => a -> a -> Ordering +compareInv x y = + case compare x y of + LT -> GT + EQ -> EQ + GT -> LT + +comparingWith :: Ord a => (a -> a -> Ordering) -> (b -> a) -> b -> b -> Ordering +comparingWith comp project x y = + comp (project x) (project y) + +-- | Leechers are sorted by their current download rate. We want to keep fast peers around. +sortLeech :: [RechokeData] -> [RechokeData] +sortLeech = sortBy (comparingWith compareInv $ pDownRate . snd) + +-- | Seeders are sorted by their current upload rate. +sortSeeds :: [RechokeData] -> [RechokeData] +sortSeeds = sortBy (comparingWith compareInv $ pUpRate . snd) + +-- | Advance the peer chain to the next peer eligible for optimistic +-- unchoking. That is, skip peers which are not interested in our pieces +-- and peers which are not choking us. The former we can't send any data to, +-- so we can't get better speeds at them. The latter are already sending us data, +-- so we know how good they are as peers. +advancePeerChain :: ChokeMgrProcess [PeerPid] +advancePeerChain = do + peers <- gets peerChain + mp <- gets peerMap + lPeers <- T.mapM (lookupPeer mp) peers + let (front, back) = break (\(_, p) -> pInterestedInUs p && pChokingUs p) lPeers + 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) + +-- | Add a peer to the Peer Database +addPeer :: PeerChannel -> Bool -> PeerPid -> ChokeMgrProcess () +addPeer pCh weSeeding tid = do + addPeerChain tid + 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 + } + +-- | Insert a Peer randomly into the Peer chain. Threads the random number generator +-- through. +addPeerChain :: PeerPid -> ChokeMgrProcess () +addPeerChain pid = do + ls <- gets peerChain + pt <- liftIO $ getStdRandom (\gen -> randomR (0, length ls - 1) gen) + let (front, back) = splitAt pt ls + modify (\db -> db { peerChain = (front ++ pid : back) }) + +-- | Calculate the amount of upload slots we have available. If the +-- number of slots is explicitly given, use that. Otherwise we +-- choose the slots based the current upload rate set. The faster +-- the rate, the more slots we allow. +calcUploadSlots :: Int -> Maybe Int -> Int +calcUploadSlots _ (Just n) = n +calcUploadSlots rate Nothing | rate <= 0 = 7 -- This is just a guess + | rate < 9 = 2 + | rate < 15 = 3 + | rate < 42 = 4 + | otherwise = round . sqrt $ fromIntegral rate * 0.6 + +-- | The call @assignUploadSlots c ds ss@ will assume that we have @c@ +-- slots for uploading at our disposal. The list @ds@ will be peers +-- that we would like to upload to among the torrents we are +-- currently downloading. The list @ss@ is the same thing but for +-- torrents that we seed. The function returns a pair @(kd,ks)@ +-- where @kd@ is the number of downloader slots and @ks@ is the +-- number of seeder slots. +-- +-- The function will move surplus slots around so all of them gets used. +assignUploadSlots :: Int -> [RechokeData] -> [RechokeData] -> (Int, Int) +assignUploadSlots slots downloaderPeers seederPeers = + -- Shuffle surplus slots around so all gets used + shuffleSeeders downloaderPeers seederPeers $ shuffleDownloaders + downloaderPeers + (downloaderSlots, seederSlots) + where + -- Calculate the slots available for the downloaders and seeders + downloaderSlots = max 1 $ round $ fromIntegral slots * 0.7 + seederSlots = max 1 $ round $ fromIntegral slots * 0.3 + + -- If there is a surplus of downloader slots, then assign them to + -- the seeder slots + shuffleDownloaders dPeers (dSlots, sSlots) = + case max 0 (dSlots - length dPeers) of + 0 -> (dSlots, sSlots) + k -> (dSlots - k, sSlots + k) + + -- If there is a surplus of seeder slots, then assign these to + -- the downloader slots. Limit the downloader slots to the number + -- of downloaders, however + shuffleSeeders dPeers sPeers (dSlots, sSlots) = + case max 0 (sSlots - length sPeers) of + 0 -> (dSlots, sSlots) + k -> (min (dSlots + k) (length dPeers), sSlots - k) + +-- | @selectPeers upSlots d s@ selects peers from a list of downloader peers @d@ and a list of seeder +-- 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 + 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 + +-- | 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 + +-- | This function carries out the choking and unchoking of peers in a round. +performChokingUnchoking :: S.Set PeerPid -> [RechokeData] -> ChokeMgrProcess () +performChokingUnchoking elected peers = + do T.mapM (unchoke . snd) unchokers + optChoke defaultOptimisticSlots chokers + where + -- Partition the peers based on they were selected or not + (unchokers, chokers) = partition (\rd -> S.member (fst rd) elected) peers + -- Choke and unchoke helpers. + -- 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) + -- 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. + optChoke _ [] = return () + optChoke 0 ((_, pi) : ps) = do if pInterestedInUs pi + then chokePeer (pChannel pi) + else unchokePeer (pChannel pi) + optChoke 0 ps + 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 + +-- | 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. +-- TODO: Snubbed peers +splitSeedLeech :: [RechokeData] -> ([RechokeData], [RechokeData]) +splitSeedLeech ps = partition (pAreSeeding . snd) $ filter picker ps + where + -- TODO: pIsASeeder is always false at the moment + picker (_, pi) = not (pIsASeeder pi) && pInterestedInUs pi + + +buildRechokeData :: ChokeMgrProcess [RechokeData] +buildRechokeData = do + chain <- gets peerChain + 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) + +rechoke :: ChokeMgrProcess () +rechoke = do + peers <- buildRechokeData + us <- gets uploadSlots + let (seed, down) = splitSeedLeech peers + electedPeers <- selectPeers us down seed + performChokingUnchoking electedPeers peers + +-- | Traverse all peers and process them with a thunk. +traversePeers thnk = T.mapM thnk =<< gets peerMap + +informDone :: PieceNum -> ChokeMgrProcess () +informDone pn = traversePeers sendDone >> return () + where + sendDone pi = ignoreProcessBlock () + $ (sendP (pChannel pi) $ PieceCompleted pn) >>= syncP + +informBlockComplete :: PieceNum -> Block -> ChokeMgrProcess () +informBlockComplete pn blk = traversePeers sendComp >> return () + where + sendComp pi = ignoreProcessBlock () + $ (sendP (pChannel pi) $ CancelBlock pn blk) >>= syncP + +updateDB :: ChokeMgrProcess () +updateDB = do + nmp <- traversePeers gatherRate + modify (\db -> db { peerMap = nmp }) + where + gatherRate pi = do + 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 + +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 }) + rechoke diff --git a/src/Process/Console.hs b/src/Process/Console.hs new file mode 100644 --- /dev/null +++ b/src/Process/Console.hs @@ -0,0 +1,81 @@ +-- 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. +{-# LANGUAGE ScopedTypeVariables #-} +module Process.Console + ( start + ) +where + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.Reader + +import Prelude hiding (catch) +import Process + +import Logging +import Supervisor + +data Cmd = Quit -- Quit the program + deriving (Eq, Show) + +type CmdChannel = Channel Cmd + +data CF = CF { cmdCh :: CmdChannel + , logCh :: LogChannel } + +instance Logging CF where + getLogger cf = ("ConsoleP", logCh cf) + +-- | 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)) + where + lp = syncP =<< quitEvent + quitEvent = do + ch <- asks cmdCh + ev <- recvP ch (==Quit) + wrapP ev + (\_ -> syncP =<< sendP waitC ()) + + +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 + diff --git a/src/Process/FS.hs b/src/Process/FS.hs new file mode 100644 --- /dev/null +++ b/src/Process/FS.hs @@ -0,0 +1,112 @@ +-- 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 +-- the FS module. +{-# LANGUAGE ScopedTypeVariables #-} +module Process.FS + ( FSPChannel + , FSPMsg(..) + , start + ) +where + +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 +import Supervisor + +data FSPMsg = CheckPiece PieceNum (Channel (Maybe Bool)) + | WriteBlock PieceNum Block B.ByteString + | ReadBlock PieceNum Block (Channel B.ByteString) + +type FSPChannel = Channel FSPMsg + +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) + +data ST = ST + { fileHandles :: FS.Handles -- ^ The file we are working on + , pieceMap :: FS.PieceMap -- ^ Map of where the pieces reside + } + + +-- 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)) + 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 + -- 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 + ch <- channel + sync . transmit fspC $ CheckPiece n ch + sync $ receive ch (const True) + diff --git a/src/Process/Listen.hs b/src/Process/Listen.hs new file mode 100644 --- /dev/null +++ b/src/Process/Listen.hs @@ -0,0 +1,34 @@ +module ListenP +where + +-- TODO: Create the OMBox here rather than taking it as a parameter +{- +import Control.Concurrent.CML + +import Network + +import ConsoleP +import PeerP +import Torrent +import FSP + + + +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 () + + +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 () +-} + diff --git a/src/Process/Peer.hs b/src/Process/Peer.hs new file mode 100644 --- /dev/null +++ b/src/Process/Peer.hs @@ -0,0 +1,460 @@ +-- | Peer proceeses +{-# LANGUAGE ScopedTypeVariables #-} +module Process.Peer ( + -- * Types + PeerMessage(..) + -- * Interface + , connect + ) +where + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.State +import Control.Monad.Reader + +import Prelude hiding (catch, log) + +import Data.Bits +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L +import qualified Data.Serialize.Get as G + +import qualified Data.Map as M +import Data.Maybe + +import Data.Set as S hiding (map) +import Data.Time.Clock +import Data.Word + +import Network + +import System.IO + +import PeerTypes +import Process +import Logging +import Process.FS +import Process.PieceMgr +import qualified Data.Queue as Q +import RateCalc as RC +import Process.Status +import Supervisor +import Process.Timer as Timer +import Torrent +import Protocol.Wire + +-- 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 () + +-- INTERNAL FUNCTIONS +---------------------------------------------------------------------- + +data SPCF = SPCF { spLogCh :: LogChannel + , spMsgCh :: Channel B.ByteString + } + +instance Logging SPCF where + getLogger cf = ("SenderP", spLogCh cf) + +-- | 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)) + where + pgm :: Process SPCF Handle () + pgm = do + c <- ask + m <- syncP =<< recvPC spMsgCh + h <- get + liftIO $ do B.hPut h m + hFlush h + +-- | 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 + +data SQCF = SQCF { sqLogC :: LogChannel + , sqInCh :: Channel SendQueueMessage + , sqOutCh :: Channel B.ByteString + , bandwidthCh :: BandwidthChannel + } + +data SQST = SQST { outQueue :: Q.Queue Message + , bytesTransferred :: Integer + } + +instance Logging SQCF where + getLogger cf = ("SendQueueP", sqLogC cf) + +-- | 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)) + 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 + rateUpdateEvent = do + 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))) + 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)})) + 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] + +data RPCF = RPCF { rpLogC :: LogChannel + , rpMsgC :: Channel (Message, Integer) } + +instance Logging RPCF where + getLogger cf = ("ReceiverP", rpLogC cf) + +receiverP :: LogChannel -> Handle -> Channel (Message, Integer) -> SupervisorChan -> IO ThreadId +receiverP logC h ch supC = spawnP (RPCF logC ch) h + (catchP (foreverP pgm) + (defaultStopHandler supC)) + where + pgm = do logDebug "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 + 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!" + 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 + 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 + } + +instance Logging PCF where + getLogger cf = ("PeerP", logCh cf) + +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 + } + +peerP :: MgrChannel -> PieceMgrChannel -> FSPChannel -> PieceMap -> LogChannel -> 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 + 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) + 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) + 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 + 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}) + fillBlocks = do + 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 + 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 + loMark = 10 + endgameLoMark = 1 + hiMark = 15 -- These two values are chosen rather arbitrarily at the moment. + +createPeerPieces :: L.ByteString -> [PieceNum] +createPeerPieces = map fromIntegral . concat . decodeBytes 0 . L.unpack + where decodeByte :: Int -> Word8 -> [Maybe Int] + decodeByte soFar w = + let dBit n = if testBit w (7-n) + then Just (n+soFar) + else Nothing + in fmap dBit [0..7] + decodeBytes _ [] = [] + decodeBytes soFar (w : ws) = catMaybes (decodeByte soFar w) : decodeBytes (soFar + 8) ws + + +showPort :: PortID -> String +showPort (PortNumber pn) = show pn +showPort _ = "N/A" + +disconnectPeer :: MgrChannel -> ThreadId -> IO () +disconnectPeer c t = sync $ transmit c $ Disconnect t + + +-- TODO: Consider if this code is correct with what we did to [connect] +{- +listenHandshake :: Handle -> PeerId -> InfoHash -> FSPChannel -> LogChannel + -> MgrChannel + -> IO (Either String ()) +listenHandshake h pid ih fsC logC mgrC = + do r <- initiateHandshake logC h pid ih + case r of + Left err -> return $ Left err + Right (_caps, _rpid) -> do peerP mgrC fsC logC h -- TODO: Coerce with connect + return $ Right () +-} diff --git a/src/Process/PeerMgr.hs b/src/Process/PeerMgr.hs new file mode 100644 --- /dev/null +++ b/src/Process/PeerMgr.hs @@ -0,0 +1,95 @@ +module Process.PeerMgr ( + -- * Types + Peer(..) + -- * 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 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 Supervisor +import Torrent hiding (infoHash) + + +data CF = CF { peerCh :: Channel [Peer] + , pieceMgrCh :: PieceMgrChannel + , mgrCh :: Channel MgrMessage + , fsCh :: FSPChannel + , peerPool :: SupervisorChan + , chokeMgrCh :: ChokeMgrChannel + , logCh :: LogChannel + } + +instance Logging CF where + getLogger cf = ("PeerMgrP", logCh cf) + +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 + -> IO ThreadId +start ch pid ih pm pieceMgrC fsC logC 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) + (ST [] M.empty pid ih) (catchP (forever lp) + (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)}) + 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 })) + 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 diff --git a/src/Process/PieceMgr.hs b/src/Process/PieceMgr.hs new file mode 100644 --- /dev/null +++ b/src/Process/PieceMgr.hs @@ -0,0 +1,432 @@ +module Process.PieceMgr + ( PieceMgrMsg(..) + , PieceMgrChannel + , ChokeInfoChannel + , ChokeInfoMsg(..) + , Blocks(..) + , start + , createPieceDb + ) +where + + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.State +import Data.List +import qualified Data.ByteString as B +import qualified Data.Map as M +import qualified Data.Set as S + +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 +import Torrent +import Process + +---------------------------------------------------------------------- + +-- | The Piece Database tracks the current state of the Torrent with respect to pieces. +-- In the database, we book-keep what pieces are missing, what are done and what are +-- currently in the process of being downloaded. The crucial moment is when we think +-- we have a full piece: we check it against its SHA1 and if it is good, we can mark +-- that piece as done. +-- +-- 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 + , 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 + , infoMap :: PieceMap -- ^ Information about pieces + , endGaming :: Bool -- ^ If we have done any endgame work this is true + } deriving Show + +-- | The InProgressPiece data type describes pieces in progress of being downloaded. +-- we keep track of blocks which are pending as well as blocks which are done. We +-- also keep track of a count of the blocks. When a block is done, we cons it unto +-- @ipHaveBlocks@. When @ipHave == ipDone@, we check the piece for correctness. The +-- field @ipHaveBlocks@ could in principle be omitted, but for now it is kept since +-- we can use it for asserting implementation correctness. We note that both the +-- check operations are then O(1) and probably fairly fast. +data InProgressPiece = InProgressPiece + { ipDone :: Int -- ^ Number of blocks when piece is done + , ipHaveBlocks :: S.Set Block -- ^ The blocks we have + , ipPendingBlocks :: [Block] -- ^ Blocks still pending + } deriving Show + +-- INTERFACE +---------------------------------------------------------------------- + +-- | When the PieceMgrP returns blocks to a peer, it will return them in either +-- "Leech Mode" or in "Endgame mode". The "Leech mode" is when the client is +-- 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)] + +-- | Messages for RPC towards the PieceMgr. +data PieceMgrMsg = GrabBlocks Int [PieceNum] (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 + | GetDone (Channel [PieceNum]) + -- ^ Get the pieces which are already done + +data ChokeInfoMsg = PieceDone PieceNum + | BlockComplete PieceNum Block + | TorrentComplete + deriving (Eq, Show) + +type PieceMgrChannel = Channel PieceMgrMsg +type ChokeInfoChannel = Channel ChokeInfoMsg + +data PieceMgrCfg = PieceMgrCfg + { logCh :: LogChannel + , pieceMgrCh :: PieceMgrChannel + , fspCh :: FSPChannel + , chokeCh :: ChokeInfoChannel + , statusCh :: StatusChan + } + +instance Logging PieceMgrCfg where + getLogger cf = ("PieceMgrP", logCh cf) + +type PieceMgrProcess v = Process PieceMgrCfg PieceDB v + +start :: LogChannel -> 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)) + 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) }) + 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) + +-- 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 + +---------------------------------------------------------------------- + +-- | 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 }) + +-- | 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) + +-- | 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 }) + +-- | 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 + $ modify (\db -> db { inProgress = ndb (inProgress db) + , 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" + f (Just ipp) = Just ipp { ipPendingBlocks = blk : ipPendingBlocks ipp } + +-- | Assert that a Piece is Complete. Can be omitted when we know it works +-- and we want a faster client. +assertPieceComplete :: PieceNum -> PieceMgrProcess () +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 + 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 + 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) + 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, + left - blockSize blk, + state && offs == blockOffset blk) + checkContents os l blks = case foldl checkBlock (os, l, True) blks of + (_, 0, True) -> True + _ -> False + 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)@ +-- where @complete@ is @True@ if the piece is percieved to be complete and @False@ +-- otherwise. +updateProgress :: PieceNum -> Block -> PieceMgrProcess Bool +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 + Just pg -> + let blkSet = ipHaveBlocks pg + in if blk `S.member` blkSet + then return False -- Stray block download. + -- Will happen without FAST extension + -- 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) + 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 = + build (leftBytes - blockSz) + (os + blockSz) + $ Block os blockSz : accum + | otherwise = build 0 (os + leftBytes) $ Block os leftBytes : accum + +-- | The call @grabBlocks' n eligible db@ tries to pick off up to @n@ pieces from +-- 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' 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 + 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 + -- 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 + 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 + else do modify (\db -> db { inProgress = M.insert p nIpp inprog }) + 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 + blockList <- createBlock h + let sz = length blockList + ipp = InProgressPiece sz S.empty blockList + modify (\db -> db { pendingPieces = pendingPieces db \\ [h], + inProgress = M.insert h ipp inProg }) + 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 + let shuffled = shuffle' dls (length dls) g + return $ take n shuffled + pickRandom pieces = do + 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) + where cBlock = blockPiece defaultBlockSize . fromInteger . len + +assertPieceDB :: PieceMgrProcess () +assertPieceDB = assertPending >> assertDone >> assertInProgress >> assertDownloading + where + -- If a piece is pending in the database, we have the following rules: + -- + -- - It is not finished. + -- - 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") + -- 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 + 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") + assertDownloading = do + 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") + + diff --git a/src/Process/Status.hs b/src/Process/Status.hs new file mode 100644 --- /dev/null +++ b/src/Process/Status.hs @@ -0,0 +1,116 @@ +-- 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 +-- information about data uploaded, downloaded and how much is +-- left. The tracker is then responsible for using this data +-- correctly to tell the tracker what to do +module Process.Status ( + -- * Types + StatusMsg(..) + , TrackerMsg(..) + -- * Channels + , StatusChan + -- * State + , ST(uploaded, downloaded, state, left) + -- * Interface + , start + ) +where + +import Control.Concurrent +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 + +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 } + +instance Logging CF where + getLogger cf = ("StatusP", logCh cf) + +data ST = ST { uploaded :: Integer, + downloaded :: Integer, + left :: Integer, + incomplete :: Maybe Integer, + complete :: Maybe Integer, + state :: TorrentState } + +-- | 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 + -> 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)) + where + pgm = do ev <- chooseP [sendEvent, recvEvent] + syncP ev + 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 })) + diff --git a/src/Process/Timer.hs b/src/Process/Timer.hs new file mode 100644 --- /dev/null +++ b/src/Process/Timer.hs @@ -0,0 +1,25 @@ +-- | The timer module is responsible for timing in the project. With +-- the timer system, we can register timers in the future and then +-- we can get a tick triggered at the point in time where we need to +-- act. This allows us to postpone events into the future at a +-- designated time. +-- +module Process.Timer (register) + +where + +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.Trans + +-- | Registers a timer tick on a channel in a number of seconds with +-- an annotated version. +registerL :: Integer -> a -> Channel a -> IO () +registerL secs v tickChan = do spawn timerProcess + return () + where timerProcess = do threadDelay $ fromInteger $ secs * 1000000 + sync $ transmit tickChan v + + +register :: MonadIO m => Integer -> a -> Channel a -> m () +register secs v c = liftIO $ registerL secs v c diff --git a/src/Process/Tracker.hs b/src/Process/Tracker.hs new file mode 100644 --- /dev/null +++ b/src/Process/Tracker.hs @@ -0,0 +1,269 @@ +-- | The TrackerP module is responsible for keeping in touch with the Tracker of a torrent. +-- The tracker is contacted periodically, and we exchange information with it. Specifically, +-- we tell the tracker how much we have downloaded, uploaded and what is left. We also +-- tell it about our current state (i.e., are we a seeder or a leecher?). +-- +-- The tracker responds to us with a new set of Peers and general information about the +-- torrent in question. It may also respond with an error in which case we should present +-- it to the user. +-- +module Process.Tracker +where + +import Control.Applicative +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.Reader +import Control.Monad.State + +import Data.Char (ord) +import Data.List (intersperse) +import Data.Time.Clock.POSIX +import qualified Data.ByteString as B + +import Network +import Network.HTTP hiding (port) +import Network.URI hiding (unreserved) + +import Numeric (showHex) + + +import Protocol.BCode as BCode hiding (encode) +import Logging +import Process +import Supervisor +import Torrent +import qualified Process.Status as Status +import qualified Process.PeerMgr as PeerMgr +import qualified Process.Timer as Timer + + + +-- | The tracker state is used to tell the tracker our current state. In order +-- to output it correctly, we override the default show instance with the +-- version below. This may be wrong to do in the long run, but for now it works +-- fine. +-- +-- The state is either started or stopped upon the client starting. The +-- tracker will create an entry for us if we tell it that we started, and it +-- will tear down this entry if we tell it that we stopped. It will know if +-- we are a seeder or a leecher based on how much data is left for us to +-- download. +-- +-- the 'Completed' entry is used once in the lifetime of a torrent. It +-- explains to the tracker that we completed the torrent in question. +data TrackerEvent = Started | Stopped | Completed | Running + deriving Eq + + + +-- | The tracker will in general respond with a BCoded dictionary. In our world, this is +-- not the data structure we would like to work with. Hence, we parse the structure into +-- the ADT below. +data TrackerResponse = ResponseOk { newPeers :: [PeerMgr.Peer], + completeR :: Maybe Integer, + incompleteR :: Maybe Integer, + timeoutInterval :: Integer, + timeoutMinInterval :: Maybe Integer } + | ResponseDecodeError B.ByteString + | ResponseWarning B.ByteString + | ResponseError B.ByteString + +-- | If we fail to contact the tracker, we will wait for 15 minutes. The number is quite arbitrarily chosen +failTimerInterval :: Integer +failTimerInterval = 15 * 60 + +-- | Configuration of the tracker process +data CF = CF { + logCh :: LogChannel + , statusCh :: Channel Status.ST + , statusPCh :: Channel Status.StatusMsg + , trackerMsgCh :: Channel Status.TrackerMsg + , peerMgrCh :: Channel [PeerMgr.Peer] + } + +instance Logging CF where + getLogger cf = ("TrackerP", logCh cf) + +-- | Internal state of the tracker CHP process +data ST = ST { + torrentInfo :: TorrentInfo + , peerId :: PeerId + , state :: TrackerEvent + , localPort :: PortID + , nextContactTime :: POSIXTime + , nextTick :: Integer + } + +start :: TorrentInfo -> PeerId -> PortID -> LogChannel -> Channel Status.ST + -> Channel Status.StatusMsg -> Channel Status.TrackerMsg -> Channel [PeerMgr.Peer] + -> SupervisorChan -> IO ThreadId +start ti pid port logC 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 + where + talkTracker = pokeTracker >>= timerUpdate + +eventTransition :: Process CF ST () +eventTransition = do + st <- gets state + modify (\s -> s { state = newS st}) + where newS st = + case st of + 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 + uri <- case parseURI url of + 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) + Right (ResponseError err) -> + do logInfo $ "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) + +timerUpdate :: (Integer, Maybe Integer) -> Process CF ST () +timerUpdate (timeout, minTimeout) = do + st <- gets state + when (st == Running) + (do t <- tick + ch <- asks trackerMsgCh + Timer.register timeout (Status.TrackerTick t) ch + logDebug $ "Set timer to: " ++ show timeout) + where tick = do t <- gets nextTick + modify (\s -> s { nextTick = t + 1 }) + return t + +-- Process a result dict into a tracker response object. +processResultDict :: BCode -> TrackerResponse +processResultDict d = + case BCode.trackerError d of + Just err -> ResponseError err + Nothing -> case BCode.trackerWarning d of + Just warn -> ResponseWarning warn + Nothing -> case decodeOk of + Nothing -> ResponseDecodeError . toBS $ "Could not decode response properly" + Just rok -> rok + where decodeOk = + ResponseOk <$> (decodeIps <$> BCode.trackerPeers d) + <*> (pure $ BCode.trackerComplete d) + <*> (pure $ BCode.trackerIncomplete d) + <*> (BCode.trackerInterval d) + <*> (pure $ BCode.trackerMinInterval d) + + +decodeIps :: B.ByteString -> [PeerMgr.Peer] +decodeIps = decodeIps' . fromBS + +-- Decode a list of IP addresses. We expect these to be a compact response by default. +decodeIps' :: String -> [PeerMgr.Peer] +decodeIps' [] = [] +decodeIps' (b1 : b2 : b3 : b4 : p1 : p2 : rest) = PeerMgr.Peer ip port : decodeIps' rest + where ip = concat . intersperse "." . map (show . ord) $ [b1, b2, b3, b4] + port = PortNumber . fromIntegral $ ord p1 * 256 + ord p2 +decodeIps' xs = error $ "decodeIps': invalid IPs: " ++ xs -- Quench all other cases + +trackerRequest :: URI -> Process CF ST (Either String TrackerResponse) +trackerRequest uri = + do resp <- liftIO $ simpleHTTP request + case resp of + Left x -> return $ Left ("Error connecting: " ++ show x) + Right r -> + case rspCode r of + (2,_,_) -> + case BCode.decode . toBS . rspBody $ r of + Left pe -> return $ Left (show pe) + Right bc -> do logDebug $ "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 + _ -> return $ Left (show r) + where request = Request {rqURI = uri, + rqMethod = GET, + rqHeaders = [], + rqBody = ""} + +-- 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] + where hlist x = intersperse "&" $ map (\(k,v) -> k ++ "=" ++ v) x + headers = do + 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) + prt :: Process CF ST Integer + prt = do lp <- gets localPort + 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")] + +-- 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. +rfc1738Encode :: String -> String +rfc1738Encode = concatMap (\c -> if unreserved c then [c] else encode c) + where unreserved = (`elem` chars) + -- I killed ~ from this list as the Mainline client doesn't announce it - jlouis + chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_./" + encode :: Char -> String + encode c = '%' : pHex c + pHex c = + let p = (showHex . ord $ c) "" + in if length p == 1 then '0' : p else p diff --git a/src/Protocol/BCode.hs b/src/Protocol/BCode.hs new file mode 100644 --- /dev/null +++ b/src/Protocol/BCode.hs @@ -0,0 +1,365 @@ +-- 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 +( + BCode, + Path(..), + encode, + -- encodeBS, + decode, + search, + announce, + comment, + creationDate, + info, + hashInfoDict, + infoLength, + infoName, + infoPieceLength, + infoPieces, + numberPieces, + infoFiles, + prettyPrint, + + trackerComplete, + trackerIncomplete, + trackerInterval, + trackerMinInterval, + trackerPeers, + trackerWarning, + trackerError, + toBS, + fromBS ) +where + +import Control.Monad +import Control.Applicative hiding (many) +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) + +import Data.Serialize +import Data.Serialize.Put +import Data.Serialize.Get + +import Digest + +-- | BCode represents the structure of a bencoded file +data BCode = BInt Integer -- ^ An integer + | BString B.ByteString -- ^ A string of bytes + | BArray [BCode] -- ^ An array + | BDict (M.Map B.ByteString BCode) -- ^ A key, value map + deriving (Show, Eq) + +data Path = PString B.ByteString + | PInt Int + +toW8 :: Char -> Word8 +toW8 = fromIntegral . ord + +fromW8 :: Word8 -> Char +fromW8 = chr . fromIntegral + +toBS :: String -> B.ByteString +toBS = B.pack . map toW8 + +fromBS :: B.ByteString -> String +fromBS = map fromW8 . B.unpack + + +instance Serialize BCode where + put (BInt i) = wrap 'i' 'e' $ putShow i + put (BArray arr) = wrap 'l' 'e' . mapM_ put $ arr + put (BDict mp) = wrap 'd' 'e' dict + where dict = mapM_ encPair . M.toList $ mp + encPair (k, v) = put (BString k) >> put v + put (BString s) = do + putShow (B.length s) + putWord8 (toW8 ':') + putByteString s + + get = getBInt <|> getBArray <|> getBDict <|> getBString + +-- | Get something wrapped in two Chars +getWrapped :: Char -> Char -> Get a -> Get a +getWrapped a b p = char a *> p <* char b + +-- | Parses a BInt +getBInt :: Get BCode +getBInt = BInt . read <$> getWrapped 'i' 'e' intP + where intP = ((:) <$> char '-' <*> getDigits) <|> getDigits + +-- | Parses a BArray +getBArray :: Get BCode +getBArray = BArray <$> getWrapped 'l' 'e' (many get) + + +-- | Parses a BDict +getBDict :: Get BCode +getBDict = BDict . M.fromList <$> getWrapped 'd' 'e' (many getPairs) + where getPairs = do + (BString s) <- getBString + x <- get + return (s,x) + +-- | Parses a BString +getBString :: Get BCode +getBString = do + count <- getDigits + BString <$> ( char ':' *> getStr (read count :: Integer)) + where maxInt = fromIntegral (maxBound :: Int) :: Integer + + getStr n | n >= 0 = B.concat <$> (sequence $ getStr' n) + | otherwise = fail $ "read a negative length string, length: " ++ show n + + getStr' n | n > maxInt = getByteString maxBound : getStr' (n-maxInt) + | otherwise = [getByteString . fromIntegral $ n] + + +-- | Get one or more digit characters +getDigits :: Get String +getDigits = many1 digit + +-- | Returns a character if it is a digit, fails otherwise. uses isDigit. +digit :: Get Char +digit = do + x <- getCharG + if isDigit x + then return x + else fail $ "Expected digit, got: " ++ show x + + +-- * Put helper functions + +-- | Put an element, wrapped by two characters +wrap :: Char -> Char -> Put -> Put +wrap a b m = do + putWord8 (toW8 a) + m + putWord8 (toW8 b) + +-- | Put something as it is shown using @show@ +putShow :: Show a => a -> Put +putShow = mapM_ put . show + +-- * Get Helper functions + +-- | Parse zero or items using a given parser +many :: Get a -> Get [a] +many p = many1 p `mplus` return [] + +-- | Parse one or more items using a given parser +many1 :: Get a -> Get [a] +many1 p = (:) <$> p <*> many p + +-- | Parse a given character +char :: Char -> Get Char +char c = do + x <- getCharG + if x == c + then return c + else fail $ "Expected char: '" ++ c:"' got: '" ++ [x,'\''] + +-- | Get a Char. Only works with single byte characters +getCharG :: Get Char +getCharG = fromW8 <$> getWord8 + +-- BCode helper functions + +-- | Return the hash of the info-dict in a torrent file +hashInfoDict :: BCode -> IO Digest +hashInfoDict bc = + do ih <- case info bc of + Nothing -> fail "Could not find infoHash" + Just x -> return x + let encoded = encode ih + digest $ L.fromChunks $ [encoded] + + +toPS :: String -> Path +toPS = PString . toBS + +{- Simple search function over BCoded data structures, general case. In practice, we + will prefer some simpler mnemonics -} +search :: [Path] -> BCode -> Maybe BCode +search [] bc = Just bc +search (PInt i : rest) (BArray bs) | i < 0 || i > length bs = Nothing + | otherwise = search rest (bs!!i) +search (PString s : rest) (BDict mp) = M.lookup s mp >>= search rest +search _ _ = Nothing + +search' :: String -> BCode -> Maybe B.ByteString +search' str b = case search [toPS str] b of + Nothing -> Nothing + Just (BString s) -> Just s + _ -> Nothing + +searchStr :: String -> BCode -> Maybe B.ByteString +searchStr = search' + +searchInt :: String -> BCode -> Maybe Integer +searchInt str b = case search [toPS str] b of + Just (BInt i) -> Just i + _ -> Nothing + +searchInfo :: String -> BCode -> Maybe BCode +searchInfo str = search [toPS "info", toPS str] + +{- Various accessors -} +announce, comment, creationDate :: BCode -> Maybe B.ByteString +announce = search' "announce" +comment = search' "comment" +creationDate = search' "creation date" + +{- Tracker accessors -} +trackerComplete, trackerIncomplete, trackerInterval :: BCode -> Maybe Integer +trackerMinInterval :: BCode -> Maybe Integer +trackerComplete = searchInt "complete" +trackerIncomplete = searchInt "incomplete" +trackerInterval = searchInt "interval" +trackerMinInterval = searchInt "min interval" + +trackerError, trackerWarning :: BCode -> Maybe B.ByteString +trackerError = searchStr "failure reason" +trackerWarning = searchStr "warning mesage" + +trackerPeers :: BCode -> Maybe B.ByteString +trackerPeers = searchStr "peers" + +info :: BCode -> Maybe BCode +info = search [toPS "info"] + +infoName :: BCode -> Maybe B.ByteString +infoName bc = case search [toPS "info", toPS "name"] bc of + Just (BString s) -> Just s + _ -> Nothing + +infoPieceLength ::BCode -> Maybe Integer +infoPieceLength bc = do BInt i <- search [toPS "info", toPS "piece length"] bc + return i + +infoLength :: BCode -> Maybe Integer +infoLength bc = maybe length2 Just length1 + where + -- |info/length key for single-file torrent + length1 = do BInt i <- search [toPS "info", toPS "length"] bc + return i + -- |length summed from files of multi-file torrent + length2 = sum `fmap` + map snd `fmap` + infoFiles bc + +infoPieces :: BCode -> Maybe [B.ByteString] +infoPieces b = do t <- searchInfo "pieces" b + case t of + BString str -> return $ sha1Split str + _ -> mzero + where sha1Split r | r == B.empty = [] + | otherwise = block : sha1Split rest + where (block, rest) = B.splitAt 20 r + +numberPieces :: BCode -> Maybe Int +numberPieces = fmap length . infoPieces + +infoFiles :: BCode -> Maybe [([String], Integer)] -- ^[(filePath, fileLength)] +infoFiles bc = let mbFpath = fromBS `fmap` infoName bc + 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) + in case (mbFpath, mbLength, mbFiles) of + (Just fpath, _, Just files) -> + Just $ + map (\(path, length) -> + (fpath:path, length) + ) files + (Just fpath, Just length, _) -> + Just [([fpath], length)] + (_, _, Just files) -> + Just files + _ -> + Nothing + +pp :: BCode -> Doc +pp bc = + case bc of + BInt i -> integer i + BString s -> text (show s) + BArray arr -> text "[" <+> (cat $ intersperse comma al) <+> text "]" + where al = map pp arr + BDict mp -> text "{" <+> cat (intersperse comma mpl) <+> text "}" + where mpl = map (\(s, bc') -> text (fromBS s) <+> text "->" <+> pp bc') $ M.toList mp + +prettyPrint :: BCode -> String +prettyPrint = render . pp + + +testDecodeEncodeProp1 :: BCode -> Bool +testDecodeEncodeProp1 m = + let encoded = encode m + decoded = decode encoded + in case decoded of + Left _ -> False + Right m' -> m == m' + +testData = [BInt 123, + BInt (-123), + BString (toBS "Hello"), + BString (toBS ['\NUL'..'\255']), + BArray [BInt 1234567890 + ,toBString "a longer string with eieldei stuff to mess things up" + ], + toBDict [ + ("hello",BInt 3) + ,("a key",toBString "and a value") + ,("a sub dict",toBDict [ + ("some stuff",BInt 1) + ,("some more stuff", toBString "with a string") + ]) + ] + ] + +toBDict :: [(String,BCode)] -> BCode +toBDict = BDict . M.fromList . map (\(k,v) -> ((toBS k),v)) + +toBString :: String -> BCode +toBString = BString . toBS + + + + diff --git a/src/Protocol/Wire.hs b/src/Protocol/Wire.hs new file mode 100644 --- /dev/null +++ b/src/Protocol/Wire.hs @@ -0,0 +1,227 @@ +{- + A parser and encoder for the BitTorrent wire protocol using the + cereal package. +-} + + +module Protocol.Wire +where + +import Control.Applicative hiding (empty) +import Control.Monad + +import Data.Monoid +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L + +import Data.Serialize +import Data.Serialize.Put +import Data.Serialize.Get + +import Data.Word +import Data.Char +import System.IO + +import Logging +import Torrent + +------------------------------------------------------------ + +type BitField = L.ByteString +type PieceLength = Int + +data Message = KeepAlive + | Choke + | Unchoke + | Interested + | NotInterested + | Have PieceNum -- Int + | BitField BitField + | Request PieceNum Block + | Piece PieceNum Int B.ByteString + | Cancel PieceNum Block + | Port Integer + deriving (Eq, Show) + + +data HandShake = HandShake + String -- Protocol Header + Word64 -- Extension Bias + +-- | The Protocol header for the Peer Wire Protocol +protocolHeader :: String +protocolHeader = "BitTorrent protocol" + +extensionBasis :: Word64 +extensionBasis = 0 + +extensionFast :: Word64 +extensionFast = 4 + +p8 :: Word8 -> Put +p8 = putWord8 + +p32be :: Integral a => a -> Put +p32be = putWord32be . fromIntegral + +decodeMsg :: Get Message +decodeMsg = get + +encodePacket :: Message -> B.ByteString +encodePacket m = mconcat [szEnc, mEnc] + where mEnc = encode m + sz = B.length mEnc + szEnc = runPut . p32be $ sz + +instance Serialize Message where + put KeepAlive = return () + put Choke = p8 0 + put Unchoke = p8 1 + put Interested = p8 2 + put NotInterested = p8 3 + put (Have pn) = p8 4 *> p32be pn + put (BitField bf) = p8 5 *> putLazyByteString bf + put (Request pn (Block os sz)) + = p8 6 *> mapM_ p32be [pn,os,sz] + put (Piece pn os c) = p8 7 *> mapM_ p32be [pn,os] *> putByteString c + put (Cancel pn (Block os sz)) + = p8 8 *> mapM_ p32be [pn,os,sz] + put (Port p) = p8 9 *> (putWord16be . fromIntegral $ p) + + get = getKA <|> getChoke + <|> getUnchoke <|> getIntr + <|> getNI <|> getHave + <|> getBF <|> getReq + <|> getPiece <|> getCancel + <|> getPort + +getChoke = byte 0 *> return Choke +getUnchoke = byte 1 *> return Unchoke +getIntr = byte 2 *> return Interested +getNI = byte 3 *> return NotInterested +getHave = byte 4 *> (Have <$> gw32) +getBF = byte 5 *> (BitField <$> (remaining >>= getLazyByteString . fromIntegral)) +getReq = byte 6 *> (Request <$> gw32 <*> (Block <$> gw32 <*> gw32)) +getPiece = byte 7 *> (Piece <$> gw32 <*> gw32 <*> (remaining >>= getByteString)) +getCancel = byte 8 *> (Cancel <$> gw32 <*> (Block <$> gw32 <*> gw32)) +getPort = byte 9 *> (Port . fromIntegral <$> getWord16be) +getKA = do + empty <- isEmpty + if empty + then return KeepAlive + else fail "Non empty message - not a KeepAlive" + +gw32 :: Integral a => Get a +gw32 = fromIntegral <$> getWord32be + +byte :: Word8 -> Get Word8 +byte w = do + x <- lookAhead getWord8 + if x == w + then getWord8 + else fail $ "Expected byte: " ++ show w ++ " got: " ++ show x + +-- | Size of the protocol header +protocolHeaderSize :: Int +protocolHeaderSize = length protocolHeader + +-- | Protocol handshake code. This encodes the protocol handshake part +protocolHandshake :: L.ByteString +protocolHandshake = L.fromChunks . map runPut $ + [putWord8 $ fromIntegral protocolHeaderSize, + mapM_ (putWord8 . fromIntegral . ord) protocolHeader, + putWord64be extensionBasis] + +toBS :: String -> B.ByteString +toBS = B.pack . map toW8 + +toLBS :: String -> L.ByteString +toLBS = L.pack . map toW8 + +toW8 :: Char -> Word8 +toW8 = fromIntegral . ord + +-- | Receive the header parts from the other end +receiveHeader :: Handle -> Int -> InfoHash + -> IO (Either String ([Capabilities], L.ByteString)) +receiveHeader h sz ih = parseHeader `fmap` B.hGet h sz + where parseHeader = runGet (headerParser ih) + +headerParser :: InfoHash -> Get ([Capabilities], L.ByteString) +headerParser ih = 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" + pid <- getLazyByteString 20 + return (decodeCapabilities caps, pid) + + +data Capabilities = Fast +decodeCapabilities :: Word64 -> [Capabilities] +decodeCapabilities _ = [] + +-- | Initiate a handshake on a socket +initiateHandshake :: LogChannel -> Handle -> PeerId -> InfoHash + -> IO (Either String ([Capabilities], L.ByteString)) +initiateHandshake logC handle peerid infohash = do + logMsg logC "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] + 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' + +-- | The call @constructBitField pieces@ will return the a ByteString suitable for inclusion in a +-- BITFIELD message to a peer. +constructBitField :: Int -> [PieceNum] -> L.ByteString +constructBitField sz pieces = L.pack . build $ m + where m = map (`elem` pieces) [0..sz-1 + pad] + pad = 8 - (sz `mod` 8) + build [] = [] + build l = let (first, rest) = splitAt 8 l + 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, + 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 + diff --git a/src/RateCalc.hs b/src/RateCalc.hs new file mode 100644 --- /dev/null +++ b/src/RateCalc.hs @@ -0,0 +1,76 @@ +-- | Rate calculation. +module RateCalc ( + -- * Types + Rate + -- * Interface + , new + , update + , extractCount + , extractRate + ) + +where + +import Data.Time.Clock + +-- | A Rate is a record of information used for calculating the rate +data Rate = Rate + { rate :: Double -- ^ The current rate + , 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 + } + +fudge :: NominalDiffTime +fudge = fromInteger 5 -- Seconds + +rateUpdate :: Integer +rateUpdate = 5 * 1000 -- Millisecs + +maxRatePeriod :: NominalDiffTime +maxRatePeriod = fromInteger 20 -- Seconds + +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 + } + +-- | 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} + +-- | 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) +extractRate t rt = + let oldWindow :: Double + oldWindow = realToFrac $ diffUTCTime (lastExt rt) (rateSince rt) + newWindow :: Double + newWindow = realToFrac $ diffUTCTime t (rateSince rt) + n = bytes rt + r = (rate rt * oldWindow + (fromIntegral n)) / newWindow + expectN = min 5 (round $ (fromIntegral n / (max r 0.0001))) + in + -- Update the rate and book-keep the missing pieces. The total is simply a built-in + -- counter. The point where we expect the next update is pushed at most 5 seconds ahead + -- 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) + }) + +-- | The call @extractCount rt@ extract the bytes transferred since last extraction +extractCount :: Rate -> (Integer, Rate) +extractCount rt = (count rt, rt { count = 0 }) + diff --git a/src/Supervisor.hs b/src/Supervisor.hs new file mode 100644 --- /dev/null +++ b/src/Supervisor.hs @@ -0,0 +1,177 @@ +-- | Erlang style supervisors for Haskell. +-- Note that yet, these are not really good enough for using in other projects. +-- are currently subject to change until I figure out how a nice interface will +-- look like. At that moment they could be split off into their own package. +{-# LANGUAGE ScopedTypeVariables #-} +module Supervisor ( + -- * Types + Child(..) + , Children + , SupervisorMsg(..) + , SupervisorChan + -- * Supervisor Initialization + , allForOne + , oneForOne + -- * helper calls + , pDie + , defaultStopHandler + ) +where + +import Control.Applicative +import Control.Concurrent +import Control.Concurrent.CML +import Control.Monad.State +import Control.Monad.Reader + +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 + +type SupervisorChan = Channel SupervisorMsg +type Children = [Child] + +data ChildInfo = HSupervisor ThreadId + | HWorker ThreadId + + +pDie :: SupervisorChan -> IO () +pDie supC = do + tid <- myThreadId + sync $ transmit supC $ IAmDying tid + +class SupervisorConf a where + getParent :: a -> SupervisorChan + getChan :: a -> SupervisorChan + +data CFOFA = CFOFA { name :: String + , chan :: SupervisorChan + , parent :: SupervisorChan + , logCh :: LogChannel } + +instance SupervisorConf CFOFA where + getParent = parent + getChan = chan + +instance Logging CFOFA where + getLogger cf = (name cf, logCh cf) -- TODO: Better naming! + +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 + c <- channel + spawnP (CFOFA name c parentC logC) (STOFA []) (catchP startup + (defaultStopHandler parentC)) + where + startup = do + childs <- mapM spawnChild children + modify (\_ -> STOFA childs) + forever eventLoop + eventLoop = do + 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))) + parentEvent mTid = do + 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 ()) + +data CFOFO = CFOFO { oName :: String + , oChan :: SupervisorChan + , oparent :: SupervisorChan + , ologCh :: LogChannel } + +instance SupervisorConf CFOFO where + getParent = oparent + getChan = oChan + +instance Logging CFOFO where + getLogger cf = (oName cf, ologCh cf) -- TODO: Better naming! + +data STOFO = STOFO [ChildInfo] + +-- | 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. +-- +-- Should a process die, the one-for-one supervisor will do nothing about it. It will just record +-- 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 + c <- channel + t <- spawnP (CFOFO name c parentC logC) (STOFO []) (catchP startup + (defaultStopHandler parentC)) + return (t, c) + where + startup :: Process CFOFO STOFO () + startup = do + childs <- mapM spawnChild children + modify (\_ -> STOFO childs) + forever eventLoop + eventLoop :: Process CFOFO STOFO () + eventLoop = do + 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))) + 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) + pruneChild tid = modify (\(STOFO cs) -> STOFO (filter check cs)) + where check (HSupervisor t) = t == tid + check (HWorker t) = t == tid + + +finChild :: SupervisorConf a => ChildInfo -> Process a b () +finChild (HWorker tid) = liftIO $ killThread tid -- Make this call killP in Process? +finChild (HSupervisor tid) = do + c <- getChan <$> ask + syncP =<< sendP c (PleaseDie tid) + +spawnChild :: SupervisorConf a => Child -> Process a b ChildInfo +spawnChild (Worker proc) = do + c <- getChan <$> ask + tid <- liftIO $ proc c + return $ HWorker tid +spawnChild (Supervisor proc) = do + c <- getChan <$> ask + tid <- liftIO $ proc c + return $ HSupervisor tid + +defaultStopHandler :: SupervisorChan -> Process a b () +defaultStopHandler supC = do + t <- liftIO $ myThreadId + syncP =<< (sendP supC $ IAmDying t) + diff --git a/src/Torrent.hs b/src/Torrent.hs new file mode 100644 --- /dev/null +++ b/src/Torrent.hs @@ -0,0 +1,161 @@ +-- 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 ( + -- * Types + InfoHash + , PeerId + , AnnounceURL + , TorrentState(..) + , TorrentInfo(..) + , PieceNum + , PieceSize + , PieceMap + , PiecesDoneMap + , PieceInfo(..) + , BlockSize + , Block(..) + -- * Interface + , determineState + , bytesLeft + , defaultBlockSize + , defaultOptimisticSlots + , defaultPort + , haskellTorrentVersion + , mkPeerId + , mkTorrentInfo + ) +where + +import qualified Data.Foldable as F +import qualified Data.ByteString as B +import Data.List +import qualified Data.Map as M + +import Network +import Numeric + +import System.Random + +import Protocol.BCode +import Digest + +-- | The type of Infohashes as used in torrents. These are identifiers +-- of torrents +type InfoHash = Digest + +-- | The peerId is the ID of a client. It is used to identify clients +-- from each other +type PeerId = String + +-- | The internal type of Announce URLs +type AnnounceURL = B.ByteString + +-- | Internal type for a torrent. It identifies a torrent in various places of the system. +data TorrentInfo = TorrentInfo { + infoHash :: InfoHash, + pieceCount :: Int, -- Number of pieces in torrent + announceURL :: AnnounceURL } deriving Show + +data TorrentState = Seeding | Leeching + +-- 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 + } deriving (Eq, Show) + +type PieceMap = M.Map PieceNum PieceInfo + +-- | The PiecesDoneMap is a map which is true if we have the piece and false otherwise +type PiecesDoneMap = M.Map PieceNum Bool + +-- | Given what pieces that are done, return the current state of the client. +determineState :: PiecesDoneMap -> TorrentState +determineState pd | F.all (==True) pd = Seeding + | otherwise = Leeching + +-- | Return the amount of bytes left on a torrent given what pieces are done and the +-- map of the shape of the torrent in question. +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 + +-- 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 + } deriving (Eq, Ord, Show) + +defaultBlockSize :: BlockSize +defaultBlockSize = 16384 -- Bytes + +-- | Default number of optimistic slots +defaultOptimisticSlots :: Int +defaultOptimisticSlots = 2 + +-- | Default port to communicate on +defaultPort :: PortID +defaultPort = PortNumber $ fromInteger 1579 + +-- | The current version of Haskell-Torrent. It should be be here. +haskellTorrentVersion :: String +haskellTorrentVersion = "d001" + +-- | Convert a BCode block into its corresponding TorrentInfo block, perhaps +-- failing in the process. +mkTorrentInfo :: BCode -> IO TorrentInfo +mkTorrentInfo bc = do + (ann, np) <- case queryInfo bc of Nothing -> fail "Could not create torrent info" + Just x -> return x + ih <- hashInfoDict bc + return TorrentInfo { infoHash = ih, announceURL = ann, pieceCount = np } + where + queryInfo bc = + do ann <- announce bc + np <- numberPieces bc + return (ann, np) + +-- | Create a new PeerId for this client +mkPeerId :: StdGen -> PeerId +mkPeerId gen = header ++ take (20 - length header) ranString + where randomList :: Int -> StdGen -> [Int] + randomList n = take n . unfoldr (Just . random) + rs = randomList 10 gen + ranString = concatMap (\i -> showHex (abs i) "") rs + header = "-HT" ++ haskellTorrentVersion ++ "-" diff --git a/src/Version.hs.in b/src/Version.hs.in new file mode 100644 --- /dev/null +++ b/src/Version.hs.in @@ -0,0 +1,8 @@ +-- | Haskell-torrent version +module Version (version) where + +githead :: String +githead = "@GITHEAD@" + +version :: String +version = githead