HaskellTorrent (empty) → 0.0

raw patch · 30 files changed

+3922/−0 lines, 30 filesdep +HTTPdep +HsOpenSSLdep +basesetup-changed

Dependencies added: HTTP, HsOpenSSL, base, bytestring, cereal, cml, containers, directory, mtl, network, parsec, pretty, random, random-shuffle, time

Files

• AUTHORS +5/−0
• HaskellTorrent.cabal +74/−0
• LICENSE +25/−0
• README.md +92/−0
• Setup.lhs +7/−0
• configure +21/−0
• src/Data/Queue.hs +42/−0
• src/Digest.hs +32/−0
• src/FS.hs +226/−0
• src/HaskellTorrent.hs +85/−0
• src/Logging.hs +40/−0
• src/LoggingTypes.hs +36/−0
• src/PeerTypes.hs +30/−0
• src/Process.hs +211/−0
• src/Process/ChokeMgr.hs +358/−0
• src/Process/Console.hs +81/−0
• src/Process/FS.hs +112/−0
• src/Process/Listen.hs +34/−0
• src/Process/Peer.hs +460/−0
• src/Process/PeerMgr.hs +95/−0
• src/Process/PieceMgr.hs +432/−0
• src/Process/Status.hs +116/−0
• src/Process/Timer.hs +25/−0
• src/Process/Tracker.hs +269/−0
• src/Protocol/BCode.hs +365/−0
• src/Protocol/Wire.hs +227/−0
• src/RateCalc.hs +76/−0
• src/Supervisor.hs +177/−0
• src/Torrent.hs +161/−0
• src/Version.hs.in +8/−0

@@ -0,0 +1,5 @@+Alex Mason <axman6@gmail.com>+Astro <astro@spaceboyz.net>+Jesper Louis Andersen <jesper.louis.andersen@gmail.com>+John Gunderman <johngunderman@gmail.com>+Thomas Christensen <thomas@chrstnsn.dk>

@@ -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+++

@@ -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.

@@ -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.+++

@@ -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

@@ -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++

@@ -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)

@@ -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+

@@ -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 "/"+++++

@@ -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 <file.torrent>"+        (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 ()

@@ -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

@@ -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

@@ -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

@@ -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+

@@ -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

@@ -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+

@@ -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)+

@@ -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 ()+-}+

@@ -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 ()+-}

@@ -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

@@ -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")++

@@ -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 }))+		   

@@ -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

@@ -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

@@ -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++++

@@ -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+

@@ -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 })+

@@ -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)+

@@ -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 ++ "-"

@@ -0,0 +1,8 @@+-- | Haskell-torrent version+module Version (version) where++githead :: String+githead = "@GITHEAD@"++version :: String+version = githead