summaryrefslogtreecommitdiff
path: root/src/Network/BitTorrent
diff options
context:
space:
mode:
Diffstat (limited to 'src/Network/BitTorrent')
-rw-r--r--src/Network/BitTorrent/Client.hs195
-rw-r--r--src/Network/BitTorrent/Client/Handle.hs188
-rw-r--r--src/Network/BitTorrent/Client/Types.hs163
-rw-r--r--src/Network/BitTorrent/Exchange.hs35
-rw-r--r--src/Network/BitTorrent/Exchange/Bitfield.hs399
-rw-r--r--src/Network/BitTorrent/Exchange/Block.hs369
-rw-r--r--src/Network/BitTorrent/Exchange/Connection.hs1012
-rw-r--r--src/Network/BitTorrent/Exchange/Download.hs296
-rw-r--r--src/Network/BitTorrent/Exchange/Manager.hs62
-rw-r--r--src/Network/BitTorrent/Exchange/Message.hs1232
-rw-r--r--src/Network/BitTorrent/Exchange/Session.hs586
-rw-r--r--src/Network/BitTorrent/Internal/Cache.hs169
-rw-r--r--src/Network/BitTorrent/Internal/Progress.hs154
-rw-r--r--src/Network/BitTorrent/Internal/Types.hs10
-rw-r--r--src/Network/BitTorrent/Readme.md10
-rw-r--r--src/Network/BitTorrent/Tracker.hs50
-rw-r--r--src/Network/BitTorrent/Tracker/List.hs193
-rw-r--r--src/Network/BitTorrent/Tracker/Message.hs920
-rw-r--r--src/Network/BitTorrent/Tracker/RPC.hs175
-rw-r--r--src/Network/BitTorrent/Tracker/RPC/HTTP.hs191
-rw-r--r--src/Network/BitTorrent/Tracker/RPC/UDP.hs454
-rw-r--r--src/Network/BitTorrent/Tracker/Session.hs306
22 files changed, 0 insertions, 7169 deletions
diff --git a/src/Network/BitTorrent/Client.hs b/src/Network/BitTorrent/Client.hs
deleted file mode 100644
index c84290dd..00000000
--- a/src/Network/BitTorrent/Client.hs
+++ /dev/null
@@ -1,195 +0,0 @@
1{-# LANGUAGE FlexibleInstances #-}
2{-# LANGUAGE TypeSynonymInstances #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving #-}
4{-# LANGUAGE TemplateHaskell #-}
5module Network.BitTorrent.Client
6 ( -- * Options
7 Options (..)
8
9 -- * Client session
10 , Client
11
12 -- ** Session data
13 , clientPeerId
14 , clientListenerPort
15 , allowedExtensions
16
17 -- ** Session initialization
18 , LogFun
19 , newClient
20 , closeClient
21 , withClient
22 , simpleClient
23
24 -- * BitTorrent monad
25 , MonadBitTorrent (..)
26 , BitTorrent
27 , runBitTorrent
28 , getClient
29
30 -- * Handle
31 , Handle
32 , handleTopic
33 , handleTrackers
34 , handleExchange
35
36 -- ** Construction
37 , TorrentSource (..)
38 , closeHandle
39
40 -- ** Query
41 , getHandle
42 , getIndex
43
44 -- ** Management
45 , start
46 , pause
47 , stop
48 ) where
49
50import Control.Applicative
51import Control.Exception
52import Control.Concurrent
53import Control.Concurrent.Chan.Split as CS
54import Control.Monad.Logger
55import Control.Monad.Trans
56import Control.Monad.Trans.Resource
57
58import Data.Default
59import Data.HashMap.Strict as HM
60import Data.Text
61import Network
62
63import Data.Torrent
64import Network.Address
65import Network.BitTorrent.Client.Types
66import Network.BitTorrent.Client.Handle
67import Network.BitTorrent.DHT as DHT hiding (Options)
68import Network.BitTorrent.Tracker as Tracker hiding (Options)
69import Network.BitTorrent.Exchange as Exchange hiding (Options)
70import qualified Network.BitTorrent.Exchange as Exchange (Options(..))
71
72
73data Options = Options
74 { optFingerprint :: Fingerprint
75 , optName :: Text
76 , optPort :: PortNumber
77 , optExtensions :: [Extension]
78 , optNodeAddr :: NodeAddr IPv4
79 , optBootNode :: Maybe (NodeAddr IPv4)
80 }
81
82instance Default Options where
83 def = Options
84 { optFingerprint = def
85 , optName = "hs-bittorrent"
86 , optPort = 6882
87 , optExtensions = []
88 , optNodeAddr = "0.0.0.0:6882"
89 , optBootNode = Nothing
90 }
91
92exchangeOptions :: PeerId -> Options -> Exchange.Options
93exchangeOptions pid Options {..} = Exchange.Options
94 { optPeerAddr = PeerAddr (Just pid) (peerHost def) optPort
95 , optBacklog = optBacklog def
96 }
97
98connHandler :: MVar (HashMap InfoHash Handle) -> Exchange.Handler
99connHandler tmap ih = do
100 m <- readMVar tmap
101 case HM.lookup ih m of
102 Nothing -> error "torrent not found"
103 Just (Handle {..}) -> return handleExchange
104
105initClient :: Options -> LogFun -> ResIO Client
106initClient opts @ Options {..} logFun = do
107 pid <- liftIO genPeerId
108 tmap <- liftIO $ newMVar HM.empty
109
110 let peerInfo = PeerInfo pid Nothing optPort
111 let mkTracker = Tracker.newManager def peerInfo
112 (_, tmgr) <- allocate mkTracker Tracker.closeManager
113
114 let mkEx = Exchange.newManager (exchangeOptions pid opts) (connHandler tmap)
115 (_, emgr) <- allocate mkEx Exchange.closeManager
116
117 let mkNode = DHT.newNode defaultHandlers def optNodeAddr logFun Nothing
118 (_, node) <- allocate mkNode DHT.closeNode
119
120 resourceMap <- getInternalState
121 eventStream <- liftIO newSendPort
122
123 return Client
124 { clientPeerId = pid
125 , clientListenerPort = optPort
126 , allowedExtensions = toCaps optExtensions
127 , clientResources = resourceMap
128 , trackerManager = tmgr
129 , exchangeManager = emgr
130 , clientNode = node
131 , clientTorrents = tmap
132 , clientLogger = logFun
133 , clientEvents = eventStream
134 }
135
136newClient :: Options -> LogFun -> IO Client
137newClient opts logFun = do
138 s <- createInternalState
139 runInternalState (initClient opts logFun) s
140 `onException` closeInternalState s
141
142closeClient :: Client -> IO ()
143closeClient Client {..} = closeInternalState clientResources
144
145withClient :: Options -> LogFun -> (Client -> IO a) -> IO a
146withClient opts lf action = bracket (newClient opts lf) closeClient action
147
148-- do not perform IO in 'initClient', do it in the 'boot'
149--boot :: BitTorrent ()
150--boot = do
151-- Options {..} <- asks options
152-- liftDHT $ bootstrap (maybeToList optBootNode)
153
154-- | Run bittorrent client with default options and log to @stderr@.
155--
156-- For testing purposes only.
157--
158simpleClient :: BitTorrent () -> IO ()
159simpleClient m = do
160 runStderrLoggingT $ LoggingT $ \ logger -> do
161 withClient def logger (`runBitTorrent` m)
162
163{-----------------------------------------------------------------------
164-- Torrent identifiers
165-----------------------------------------------------------------------}
166
167class TorrentSource s where
168 openHandle :: FilePath -> s -> BitTorrent Handle
169
170instance TorrentSource InfoHash where
171 openHandle path ih = openMagnet path (nullMagnet ih)
172 {-# INLINE openHandle #-}
173
174instance TorrentSource Magnet where
175 openHandle = openMagnet
176 {-# INLINE openHandle #-}
177
178instance TorrentSource InfoDict where
179 openHandle path dict = openTorrent path (nullTorrent dict)
180 {-# INLINE openHandle #-}
181
182instance TorrentSource Torrent where
183 openHandle = openTorrent
184 {-# INLINE openHandle #-}
185
186instance TorrentSource FilePath where
187 openHandle contentDir torrentPath = do
188 t <- liftIO $ fromFile torrentPath
189 openTorrent contentDir t
190 {-# INLINE openHandle #-}
191
192getIndex :: BitTorrent [Handle]
193getIndex = do
194 Client {..} <- getClient
195 elems <$> liftIO (readMVar clientTorrents)
diff --git a/src/Network/BitTorrent/Client/Handle.hs b/src/Network/BitTorrent/Client/Handle.hs
deleted file mode 100644
index 66baac48..00000000
--- a/src/Network/BitTorrent/Client/Handle.hs
+++ /dev/null
@@ -1,188 +0,0 @@
1module Network.BitTorrent.Client.Handle
2 ( -- * Handle
3 Handle
4
5 -- * Initialization
6 , openTorrent
7 , openMagnet
8 , closeHandle
9
10 -- * Control
11 , start
12 , pause
13 , stop
14
15 -- * Query
16 , getHandle
17 , getStatus
18 ) where
19
20import Control.Concurrent.Chan.Split
21import Control.Concurrent.Lifted as L
22import Control.Monad
23import Control.Monad.Trans
24import Data.Default
25import Data.List as L
26import Data.HashMap.Strict as HM
27
28import Data.Torrent
29import Network.BitTorrent.Client.Types as Types
30import Network.BitTorrent.DHT as DHT
31import Network.BitTorrent.Exchange as Exchange
32import Network.BitTorrent.Tracker as Tracker
33
34{-----------------------------------------------------------------------
35-- Safe handle set manupulation
36-----------------------------------------------------------------------}
37
38allocHandle :: InfoHash -> BitTorrent Handle -> BitTorrent Handle
39allocHandle ih m = do
40 Client {..} <- getClient
41
42 (h, added) <- modifyMVar clientTorrents $ \ handles -> do
43 case HM.lookup ih handles of
44 Just h -> return (handles, (h, False))
45 Nothing -> do
46 h <- m
47 return (HM.insert ih h handles, (h, True))
48
49 when added $ do
50 liftIO $ send clientEvents (TorrentAdded ih)
51
52 return h
53
54freeHandle :: InfoHash -> BitTorrent () -> BitTorrent ()
55freeHandle ih finalizer = do
56 Client {..} <- getClient
57
58 modifyMVar_ clientTorrents $ \ handles -> do
59 case HM.lookup ih handles of
60 Nothing -> return handles
61 Just _ -> do
62 finalizer
63 return (HM.delete ih handles)
64
65lookupHandle :: InfoHash -> BitTorrent (Maybe Handle)
66lookupHandle ih = do
67 Client {..} <- getClient
68 handles <- readMVar clientTorrents
69 return (HM.lookup ih handles)
70
71{-----------------------------------------------------------------------
72-- Initialization
73-----------------------------------------------------------------------}
74
75newExchangeSession :: FilePath -> Either InfoHash InfoDict -> BitTorrent Exchange.Session
76newExchangeSession rootPath source = do
77 c @ Client {..} <- getClient
78 liftIO $ Exchange.newSession clientLogger (externalAddr c) rootPath source
79
80-- | Open a torrent in 'stop'ed state. Use 'nullTorrent' to open
81-- handle from 'InfoDict'. This operation do not block.
82openTorrent :: FilePath -> Torrent -> BitTorrent Handle
83openTorrent rootPath t @ Torrent {..} = do
84 let ih = idInfoHash tInfoDict
85 allocHandle ih $ do
86 statusVar <- newMVar Types.Stopped
87 tses <- liftIO $ Tracker.newSession ih (trackerList t)
88 eses <- newExchangeSession rootPath (Right tInfoDict)
89 eventStream <- liftIO newSendPort
90 return $ Handle
91 { handleTopic = ih
92 , handlePrivate = idPrivate tInfoDict
93 , handleStatus = statusVar
94 , handleTrackers = tses
95 , handleExchange = eses
96 , handleEvents = eventStream
97 }
98
99-- | Use 'nullMagnet' to open handle from 'InfoHash'.
100openMagnet :: FilePath -> Magnet -> BitTorrent Handle
101openMagnet rootPath Magnet {..} = do
102 allocHandle exactTopic $ do
103 statusVar <- newMVar Types.Stopped
104 tses <- liftIO $ Tracker.newSession exactTopic def
105 eses <- newExchangeSession rootPath (Left exactTopic)
106 eventStream <- liftIO newSendPort
107 return $ Handle
108 { handleTopic = exactTopic
109 , handlePrivate = False
110 , handleStatus = statusVar
111 , handleTrackers = tses
112 , handleExchange = eses
113 , handleEvents = eventStream
114 }
115
116-- | Stop torrent and destroy all sessions. You don't need to close
117-- handles at application exit, all handles will be automatically
118-- closed at 'Network.BitTorrent.Client.closeClient'. This operation
119-- may block.
120closeHandle :: Handle -> BitTorrent ()
121closeHandle h @ Handle {..} = do
122 freeHandle handleTopic $ do
123 Client {..} <- getClient
124 stop h
125 liftIO $ Exchange.closeSession handleExchange
126 liftIO $ Tracker.closeSession trackerManager handleTrackers
127
128{-----------------------------------------------------------------------
129-- Control
130-----------------------------------------------------------------------}
131
132modifyStatus :: HandleStatus -> Handle -> (HandleStatus -> BitTorrent ()) -> BitTorrent ()
133modifyStatus targetStatus Handle {..} targetAction = do
134 modifyMVar_ handleStatus $ \ actualStatus -> do
135 unless (actualStatus == targetStatus) $ do
136 targetAction actualStatus
137 return targetStatus
138 liftIO $ send handleEvents (StatusChanged targetStatus)
139
140-- | Start downloading, uploading and announcing this torrent.
141--
142-- This operation is blocking, use
143-- 'Control.Concurrent.Async.Lifted.async' if needed.
144start :: Handle -> BitTorrent ()
145start h @ Handle {..} = do
146 modifyStatus Types.Running h $ \ status -> do
147 case status of
148 Types.Running -> return ()
149 Types.Stopped -> do
150 Client {..} <- getClient
151 liftIO $ Tracker.notify trackerManager handleTrackers Tracker.Started
152 unless handlePrivate $ do
153 liftDHT $ DHT.insert handleTopic (error "start")
154 liftIO $ do
155 peers <- askPeers trackerManager handleTrackers
156 print $ "got: " ++ show (L.length peers) ++ " peers"
157 forM_ peers $ \ peer -> do
158 Exchange.connect peer handleExchange
159
160-- | Stop downloading this torrent.
161pause :: Handle -> BitTorrent ()
162pause _ = return ()
163
164-- | Stop downloading, uploading and announcing this torrent.
165stop :: Handle -> BitTorrent ()
166stop h @ Handle {..} = do
167 modifyStatus Types.Stopped h $ \ status -> do
168 case status of
169 Types.Stopped -> return ()
170 Types.Running -> do
171 Client {..} <- getClient
172 unless handlePrivate $ do
173 liftDHT $ DHT.delete handleTopic (error "stop")
174 liftIO $ Tracker.notify trackerManager handleTrackers Tracker.Stopped
175
176{-----------------------------------------------------------------------
177-- Query
178-----------------------------------------------------------------------}
179
180getHandle :: InfoHash -> BitTorrent Handle
181getHandle ih = do
182 mhandle <- lookupHandle ih
183 case mhandle of
184 Nothing -> error "should we throw some exception?"
185 Just h -> return h
186
187getStatus :: Handle -> IO HandleStatus
188getStatus Handle {..} = readMVar handleStatus
diff --git a/src/Network/BitTorrent/Client/Types.hs b/src/Network/BitTorrent/Client/Types.hs
deleted file mode 100644
index e2ad858f..00000000
--- a/src/Network/BitTorrent/Client/Types.hs
+++ /dev/null
@@ -1,163 +0,0 @@
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE TypeFamilies #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving #-}
6module Network.BitTorrent.Client.Types
7 ( -- * Core types
8 HandleStatus (..)
9 , Handle (..)
10 , Client (..)
11 , externalAddr
12
13 -- * Monad BitTorrent
14 , BitTorrent (..)
15 , runBitTorrent
16 , getClient
17
18 , MonadBitTorrent (..)
19
20 -- * Events
21 , Types.Event (..)
22 ) where
23
24import Control.Applicative
25import Control.Concurrent
26import Control.Concurrent.Chan.Split as CS
27import Control.Monad.Base
28import Control.Monad.Logger
29import Control.Monad.Reader
30import Control.Monad.Trans.Control
31import Control.Monad.Trans.Resource
32import Data.Function
33import Data.HashMap.Strict as HM
34import Data.Ord
35import Network
36import System.Log.FastLogger
37
38import Data.Torrent
39import Network.Address
40import Network.BitTorrent.Internal.Types as Types
41import Network.BitTorrent.DHT as DHT
42import Network.BitTorrent.Exchange as Exchange
43import Network.BitTorrent.Tracker as Tracker hiding (Event)
44
45data HandleStatus
46 = Running
47 | Stopped
48 deriving (Show, Eq)
49
50data Handle = Handle
51 { handleTopic :: !InfoHash
52 , handlePrivate :: !Bool
53
54 , handleStatus :: !(MVar HandleStatus)
55 , handleTrackers :: !Tracker.Session
56 , handleExchange :: !Exchange.Session
57 , handleEvents :: !(SendPort (Event Handle))
58 }
59
60instance EventSource Handle where
61 data Event Handle = StatusChanged HandleStatus
62 listen Handle {..} = CS.listen undefined
63
64data Client = Client
65 { clientPeerId :: !PeerId
66 , clientListenerPort :: !PortNumber
67 , allowedExtensions :: !Caps
68 , clientResources :: !InternalState
69 , trackerManager :: !Tracker.Manager
70 , exchangeManager :: !Exchange.Manager
71 , clientNode :: !(Node IPv4)
72 , clientTorrents :: !(MVar (HashMap InfoHash Handle))
73 , clientLogger :: !LogFun
74 , clientEvents :: !(SendPort (Event Client))
75 }
76
77instance Eq Client where
78 (==) = (==) `on` clientPeerId
79
80instance Ord Client where
81 compare = comparing clientPeerId
82
83instance EventSource Client where
84 data Event Client = TorrentAdded InfoHash
85 listen Client {..} = CS.listen clientEvents
86
87-- | External IP address of a host running a bittorrent client
88-- software may be used to acknowledge remote peer the host connected
89-- to. See 'Network.BitTorrent.Exchange.Message.ExtendedHandshake'.
90externalAddr :: Client -> PeerAddr (Maybe IP)
91externalAddr Client {..} = PeerAddr
92 { peerId = Just clientPeerId
93 , peerHost = Nothing -- TODO return external IP address, if known
94 , peerPort = clientListenerPort
95 }
96
97{-----------------------------------------------------------------------
98-- BitTorrent monad
99-----------------------------------------------------------------------}
100
101newtype BitTorrent a = BitTorrent
102 { unBitTorrent :: ReaderT Client IO a
103 } deriving ( Functor, Applicative, Monad
104 , MonadIO, MonadThrow, MonadBase IO
105 )
106
107class MonadBitTorrent m where
108 liftBT :: BitTorrent a -> m a
109
110#if MIN_VERSION_monad_control(1,0,0)
111newtype BTStM a = BTStM { unBTSt :: StM (ReaderT Client IO) a }
112
113instance MonadBaseControl IO BitTorrent where
114 type StM BitTorrent a = BTStM a
115 liftBaseWith cc = BitTorrent $ liftBaseWith $ \ cc' ->
116 cc $ \ (BitTorrent m) -> BTStM <$> cc' m
117 {-# INLINE liftBaseWith #-}
118
119 restoreM = BitTorrent . restoreM . unBTSt
120 {-# INLINE restoreM #-}
121#else
122instance MonadBaseControl IO BitTorrent where
123 newtype StM BitTorrent a = StM { unSt :: StM (ReaderT Client IO) a }
124 liftBaseWith cc = BitTorrent $ liftBaseWith $ \ cc' ->
125 cc $ \ (BitTorrent m) -> StM <$> cc' m
126 {-# INLINE liftBaseWith #-}
127
128 restoreM = BitTorrent . restoreM . unSt
129 {-# INLINE restoreM #-}
130#endif
131
132-- | NOP.
133instance MonadBitTorrent BitTorrent where
134 liftBT = id
135
136instance MonadTrans t => MonadBitTorrent (t BitTorrent) where
137 liftBT = lift
138
139-- | Registered but not closed manually resources will be
140-- automatically closed at 'Network.BitTorrent.Client.closeClient'
141instance MonadResource BitTorrent where
142 liftResourceT m = BitTorrent $ do
143 s <- asks clientResources
144 liftIO $ runInternalState m s
145
146-- | Run DHT operation, only if the client node is running.
147instance MonadDHT BitTorrent where
148 liftDHT action = BitTorrent $ do
149 node <- asks clientNode
150 liftIO $ runDHT node action
151
152instance MonadLogger BitTorrent where
153 monadLoggerLog loc src lvl msg = BitTorrent $ do
154 logger <- asks clientLogger
155 liftIO $ logger loc src lvl (toLogStr msg)
156
157runBitTorrent :: Client -> BitTorrent a -> IO a
158runBitTorrent client action = runReaderT (unBitTorrent action) client
159{-# INLINE runBitTorrent #-}
160
161getClient :: BitTorrent Client
162getClient = BitTorrent ask
163{-# INLINE getClient #-}
diff --git a/src/Network/BitTorrent/Exchange.hs b/src/Network/BitTorrent/Exchange.hs
deleted file mode 100644
index 143bf090..00000000
--- a/src/Network/BitTorrent/Exchange.hs
+++ /dev/null
@@ -1,35 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8module Network.BitTorrent.Exchange
9 ( -- * Manager
10 Options (..)
11 , Manager
12 , Handler
13 , newManager
14 , closeManager
15
16 -- * Session
17 , Caps
18 , Extension
19 , toCaps
20 , Session
21 , newSession
22 , closeSession
23
24 -- * Query
25 , waitMetadata
26 , takeMetadata
27
28 -- * Connections
29 , connect
30 , connectSink
31 ) where
32
33import Network.BitTorrent.Exchange.Manager
34import Network.BitTorrent.Exchange.Message
35import Network.BitTorrent.Exchange.Session
diff --git a/src/Network/BitTorrent/Exchange/Bitfield.hs b/src/Network/BitTorrent/Exchange/Bitfield.hs
deleted file mode 100644
index 7bae3475..00000000
--- a/src/Network/BitTorrent/Exchange/Bitfield.hs
+++ /dev/null
@@ -1,399 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- This modules provides all necessary machinery to work with
9-- bitfields. Bitfields are used to keep track indices of complete
10-- pieces either this peer have or remote peer have.
11--
12-- There are also commonly used piece seletion algorithms
13-- which used to find out which one next piece to download.
14-- Selectors considered to be used in the following order:
15--
16-- * 'randomFirst' - at the start of download.
17--
18-- * 'rarestFirst' - performed to avoid situation when
19-- rarest piece is unaccessible.
20--
21-- * 'endGame' - performed after a peer has requested all
22-- the subpieces of the content.
23--
24-- Note that BitTorrent protocol recommend (TODO link?) the
25-- 'strictFirst' priority policy for /subpiece/ or /blocks/
26-- selection.
27--
28{-# LANGUAGE CPP #-}
29{-# LANGUAGE BangPatterns #-}
30{-# LANGUAGE RecordWildCards #-}
31module Network.BitTorrent.Exchange.Bitfield
32 ( -- * Bitfield
33 PieceIx
34 , PieceCount
35 , Bitfield
36
37 -- * Construction
38 , haveAll
39 , haveNone
40 , have
41 , singleton
42 , interval
43 , adjustSize
44
45 -- * Query
46 -- ** Cardinality
47 , Network.BitTorrent.Exchange.Bitfield.null
48 , Network.BitTorrent.Exchange.Bitfield.full
49 , haveCount
50 , totalCount
51 , completeness
52
53 -- ** Membership
54 , member
55 , notMember
56 , findMin
57 , findMax
58 , isSubsetOf
59
60 -- ** Availability
61 , complement
62 , Frequency
63 , frequencies
64 , rarest
65
66 -- * Combine
67 , insert
68 , union
69 , intersection
70 , difference
71
72 -- * Conversion
73 , toList
74 , fromList
75
76 -- * Serialization
77 , fromBitmap
78 , toBitmap
79
80 -- * Piece selection
81 , Selector
82 , selector
83 , strategyClass
84
85 , strictFirst
86 , strictLast
87 , rarestFirst
88 , randomFirst
89 , endGame
90 ) where
91
92import Control.Monad
93import Control.Monad.ST
94import Data.ByteString (ByteString)
95import qualified Data.ByteString as B
96import qualified Data.ByteString.Lazy as Lazy
97import Data.Vector.Unboxed (Vector)
98import qualified Data.Vector.Unboxed as V
99import qualified Data.Vector.Unboxed.Mutable as VM
100import Data.IntervalSet (IntSet)
101import qualified Data.IntervalSet as S
102import qualified Data.IntervalSet.ByteString as S
103import Data.List (foldl')
104import Data.Monoid
105import Data.Ratio
106
107import Data.Torrent
108
109-- TODO cache some operations
110
111-- | Bitfields are represented just as integer sets but with
112-- restriction: the each set should be within given interval (or
113-- subset of the specified interval). Size is used to specify
114-- interval, so bitfield of size 10 might contain only indices in
115-- interval [0..9].
116--
117data Bitfield = Bitfield {
118 bfSize :: !PieceCount
119 , bfSet :: !IntSet
120 } deriving (Show, Read, Eq)
121
122-- Invariants: all elements of bfSet lie in [0..bfSize - 1];
123
124instance Monoid Bitfield where
125 {-# SPECIALIZE instance Monoid Bitfield #-}
126 mempty = haveNone 0
127 mappend = union
128 mconcat = unions
129
130{-----------------------------------------------------------------------
131 Construction
132-----------------------------------------------------------------------}
133
134-- | The empty bitfield of the given size.
135haveNone :: PieceCount -> Bitfield
136haveNone s = Bitfield s S.empty
137
138-- | The full bitfield containing all piece indices for the given size.
139haveAll :: PieceCount -> Bitfield
140haveAll s = Bitfield s (S.interval 0 (s - 1))
141
142-- | Insert the index in the set ignoring out of range indices.
143have :: PieceIx -> Bitfield -> Bitfield
144have ix Bitfield {..}
145 | 0 <= ix && ix < bfSize = Bitfield bfSize (S.insert ix bfSet)
146 | otherwise = Bitfield bfSize bfSet
147
148singleton :: PieceIx -> PieceCount -> Bitfield
149singleton ix pc = have ix (haveNone pc)
150
151-- | Assign new size to bitfield. FIXME Normally, size should be only
152-- decreased, otherwise exception raised.
153adjustSize :: PieceCount -> Bitfield -> Bitfield
154adjustSize s Bitfield {..} = Bitfield s bfSet
155
156-- | NOTE: for internal use only
157interval :: PieceCount -> PieceIx -> PieceIx -> Bitfield
158interval pc a b = Bitfield pc (S.interval a b)
159
160{-----------------------------------------------------------------------
161 Query
162-----------------------------------------------------------------------}
163
164-- | Test if bitifield have no one index: peer do not have anything.
165null :: Bitfield -> Bool
166null Bitfield {..} = S.null bfSet
167
168-- | Test if bitfield have all pieces.
169full :: Bitfield -> Bool
170full Bitfield {..} = S.size bfSet == bfSize
171
172-- | Count of peer have pieces.
173haveCount :: Bitfield -> PieceCount
174haveCount = S.size . bfSet
175
176-- | Total count of pieces and its indices.
177totalCount :: Bitfield -> PieceCount
178totalCount = bfSize
179
180-- | Ratio of /have/ piece count to the /total/ piece count.
181--
182-- > forall bf. 0 <= completeness bf <= 1
183--
184completeness :: Bitfield -> Ratio PieceCount
185completeness b = haveCount b % totalCount b
186
187inRange :: PieceIx -> Bitfield -> Bool
188inRange ix Bitfield {..} = 0 <= ix && ix < bfSize
189
190member :: PieceIx -> Bitfield -> Bool
191member ix bf @ Bitfield {..}
192 | ix `inRange` bf = ix `S.member` bfSet
193 | otherwise = False
194
195notMember :: PieceIx -> Bitfield -> Bool
196notMember ix bf @ Bitfield {..}
197 | ix `inRange` bf = ix `S.notMember` bfSet
198 | otherwise = True
199
200-- | Find first available piece index.
201findMin :: Bitfield -> PieceIx
202findMin = S.findMin . bfSet
203{-# INLINE findMin #-}
204
205-- | Find last available piece index.
206findMax :: Bitfield -> PieceIx
207findMax = S.findMax . bfSet
208{-# INLINE findMax #-}
209
210-- | Check if all pieces from first bitfield present if the second bitfield
211isSubsetOf :: Bitfield -> Bitfield -> Bool
212isSubsetOf a b = bfSet a `S.isSubsetOf` bfSet b
213{-# INLINE isSubsetOf #-}
214
215-- | Resulting bitfield includes only missing pieces.
216complement :: Bitfield -> Bitfield
217complement Bitfield {..} = Bitfield
218 { bfSet = uni `S.difference` bfSet
219 , bfSize = bfSize
220 }
221 where
222 Bitfield _ uni = haveAll bfSize
223{-# INLINE complement #-}
224
225{-----------------------------------------------------------------------
226-- Availability
227-----------------------------------------------------------------------}
228
229-- | Frequencies are needed in piece selection startegies which use
230-- availability quantity to find out the optimal next piece index to
231-- download.
232type Frequency = Int
233
234-- TODO rename to availability
235-- | How many times each piece index occur in the given bitfield set.
236frequencies :: [Bitfield] -> Vector Frequency
237frequencies [] = V.fromList []
238frequencies xs = runST $ do
239 v <- VM.new size
240 VM.set v 0
241 forM_ xs $ \ Bitfield {..} -> do
242 forM_ (S.toList bfSet) $ \ x -> do
243 fr <- VM.read v x
244 VM.write v x (succ fr)
245 V.unsafeFreeze v
246 where
247 size = maximum (map bfSize xs)
248
249-- TODO it seems like this operation is veeery slow
250
251-- | Find least available piece index. If no piece available return
252-- 'Nothing'.
253rarest :: [Bitfield] -> Maybe PieceIx
254rarest xs
255 | V.null freqMap = Nothing
256 | otherwise
257 = Just $ fst $ V.ifoldr' minIx (0, freqMap V.! 0) freqMap
258 where
259 freqMap = frequencies xs
260
261 minIx :: PieceIx -> Frequency
262 -> (PieceIx, Frequency)
263 -> (PieceIx, Frequency)
264 minIx ix fr acc@(_, fra)
265 | fr < fra && fr > 0 = (ix, fr)
266 | otherwise = acc
267
268
269{-----------------------------------------------------------------------
270 Combine
271-----------------------------------------------------------------------}
272
273insert :: PieceIx -> Bitfield -> Bitfield
274insert pix bf @ Bitfield {..}
275 | 0 <= pix && pix < bfSize = Bitfield
276 { bfSet = S.insert pix bfSet
277 , bfSize = bfSize
278 }
279 | otherwise = bf
280
281-- | Find indices at least one peer have.
282union :: Bitfield -> Bitfield -> Bitfield
283union a b = {-# SCC union #-} Bitfield {
284 bfSize = bfSize a `max` bfSize b
285 , bfSet = bfSet a `S.union` bfSet b
286 }
287
288-- | Find indices both peers have.
289intersection :: Bitfield -> Bitfield -> Bitfield
290intersection a b = {-# SCC intersection #-} Bitfield {
291 bfSize = bfSize a `min` bfSize b
292 , bfSet = bfSet a `S.intersection` bfSet b
293 }
294
295-- | Find indices which have first peer but do not have the second peer.
296difference :: Bitfield -> Bitfield -> Bitfield
297difference a b = {-# SCC difference #-} Bitfield {
298 bfSize = bfSize a -- FIXME is it reasonable?
299 , bfSet = bfSet a `S.difference` bfSet b
300 }
301
302-- | Find indices the any of the peers have.
303unions :: [Bitfield] -> Bitfield
304unions = {-# SCC unions #-} foldl' union (haveNone 0)
305
306{-----------------------------------------------------------------------
307 Serialization
308-----------------------------------------------------------------------}
309
310-- | List all /have/ indexes.
311toList :: Bitfield -> [PieceIx]
312toList Bitfield {..} = S.toList bfSet
313
314-- | Make bitfield from list of /have/ indexes.
315fromList :: PieceCount -> [PieceIx] -> Bitfield
316fromList s ixs = Bitfield {
317 bfSize = s
318 , bfSet = S.splitGT (-1) $ S.splitLT s $ S.fromList ixs
319 }
320
321-- | Unpack 'Bitfield' from tightly packed bit array. Note resulting
322-- size might be more than real bitfield size, use 'adjustSize'.
323fromBitmap :: ByteString -> Bitfield
324fromBitmap bs = {-# SCC fromBitmap #-} Bitfield {
325 bfSize = B.length bs * 8
326 , bfSet = S.fromByteString bs
327 }
328{-# INLINE fromBitmap #-}
329
330-- | Pack a 'Bitfield' to tightly packed bit array.
331toBitmap :: Bitfield -> Lazy.ByteString
332toBitmap Bitfield {..} = {-# SCC toBitmap #-} Lazy.fromChunks [intsetBM, alignment]
333 where
334 byteSize = bfSize `div` 8 + if bfSize `mod` 8 == 0 then 0 else 1
335 alignment = B.replicate (byteSize - B.length intsetBM) 0
336 intsetBM = S.toByteString bfSet
337
338{-----------------------------------------------------------------------
339-- Piece selection
340-----------------------------------------------------------------------}
341
342type Selector = Bitfield -- ^ Indices of client /have/ pieces.
343 -> Bitfield -- ^ Indices of peer /have/ pieces.
344 -> [Bitfield] -- ^ Indices of other peers /have/ pieces.
345 -> Maybe PieceIx -- ^ Zero-based index of piece to request
346 -- to, if any.
347
348selector :: Selector -- ^ Selector to use at the start.
349 -> Ratio PieceCount
350 -> Selector -- ^ Selector to use after the client have
351 -- the C pieces.
352 -> Selector -- ^ Selector that changes behaviour based
353 -- on completeness.
354selector start pt ready h a xs =
355 case strategyClass pt h of
356 SCBeginning -> start h a xs
357 SCReady -> ready h a xs
358 SCEnd -> endGame h a xs
359
360data StartegyClass
361 = SCBeginning
362 | SCReady
363 | SCEnd
364 deriving (Show, Eq, Ord, Enum, Bounded)
365
366
367strategyClass :: Ratio PieceCount -> Bitfield -> StartegyClass
368strategyClass threshold = classify . completeness
369 where
370 classify c
371 | c < threshold = SCBeginning
372 | c + 1 % numerator c < 1 = SCReady
373 -- FIXME numerator have is not total count
374 | otherwise = SCEnd
375
376
377-- | Select the first available piece.
378strictFirst :: Selector
379strictFirst h a _ = Just $ findMin (difference a h)
380
381-- | Select the last available piece.
382strictLast :: Selector
383strictLast h a _ = Just $ findMax (difference a h)
384
385-- |
386rarestFirst :: Selector
387rarestFirst h a xs = rarest (map (intersection want) xs)
388 where
389 want = difference h a
390
391-- | In average random first is faster than rarest first strategy but
392-- only if all pieces are available.
393randomFirst :: Selector
394randomFirst = do
395-- randomIO
396 error "randomFirst"
397
398endGame :: Selector
399endGame = strictLast
diff --git a/src/Network/BitTorrent/Exchange/Block.hs b/src/Network/BitTorrent/Exchange/Block.hs
deleted file mode 100644
index bc9a3d24..00000000
--- a/src/Network/BitTorrent/Exchange/Block.hs
+++ /dev/null
@@ -1,369 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- Blocks are used to transfer pieces.
9--
10{-# LANGUAGE BangPatterns #-}
11{-# LANGUAGE FlexibleInstances #-}
12{-# LANGUAGE TemplateHaskell #-}
13{-# LANGUAGE DeriveFunctor #-}
14{-# LANGUAGE DeriveDataTypeable #-}
15{-# LANGUAGE GeneralizedNewtypeDeriving #-}
16module Network.BitTorrent.Exchange.Block
17 ( -- * Block attributes
18 BlockOffset
19 , BlockCount
20 , BlockSize
21 , defaultTransferSize
22
23 -- * Block index
24 , BlockIx(..)
25 , blockIxRange
26
27 -- * Block data
28 , Block(..)
29 , blockIx
30 , blockSize
31 , blockRange
32 , isPiece
33 , leadingBlock
34
35 -- * Block bucket
36 , Bucket
37
38 -- ** Query
39 , Network.BitTorrent.Exchange.Block.null
40 , Network.BitTorrent.Exchange.Block.full
41 , Network.BitTorrent.Exchange.Block.size
42 , Network.BitTorrent.Exchange.Block.spans
43
44 -- ** Construction
45 , Network.BitTorrent.Exchange.Block.empty
46 , Network.BitTorrent.Exchange.Block.insert
47 , Network.BitTorrent.Exchange.Block.insertLazy
48 , Network.BitTorrent.Exchange.Block.merge
49 , Network.BitTorrent.Exchange.Block.fromList
50
51 -- ** Rendering
52 , Network.BitTorrent.Exchange.Block.toPiece
53
54 -- ** Debug
55 , Network.BitTorrent.Exchange.Block.valid
56 ) where
57
58import Prelude hiding (span)
59import Control.Applicative
60import Data.ByteString as BS hiding (span)
61import Data.ByteString.Lazy as BL hiding (span)
62import Data.ByteString.Lazy.Builder as BS
63import Data.Default
64import Data.Monoid
65import Data.List as L hiding (span)
66import Data.Serialize as S
67import Data.Typeable
68import Numeric
69import Text.PrettyPrint as PP hiding ((<>))
70import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
71
72import Data.Torrent
73
74{-----------------------------------------------------------------------
75-- Block attributes
76-----------------------------------------------------------------------}
77
78-- | Offset of a block in a piece in bytes. Should be multiple of
79-- the choosen block size.
80type BlockOffset = Int
81
82-- | Size of a block in bytes. Should be power of 2.
83--
84-- Normally block size is equal to 'defaultTransferSize'.
85--
86type BlockSize = Int
87
88-- | Number of block in a piece of a torrent. Used to distinguish
89-- block count from piece count.
90type BlockCount = Int
91
92-- | Widely used semi-official block size. Some clients can ignore if
93-- block size of BlockIx in Request message is not equal to this
94-- value.
95--
96defaultTransferSize :: BlockSize
97defaultTransferSize = 16 * 1024
98
99{-----------------------------------------------------------------------
100 Block Index
101-----------------------------------------------------------------------}
102
103-- | BlockIx correspond.
104data BlockIx = BlockIx {
105 -- | Zero-based piece index.
106 ixPiece :: {-# UNPACK #-} !PieceIx
107
108 -- | Zero-based byte offset within the piece.
109 , ixOffset :: {-# UNPACK #-} !BlockOffset
110
111 -- | Block size starting from offset.
112 , ixLength :: {-# UNPACK #-} !BlockSize
113 } deriving (Show, Eq, Typeable)
114
115-- | First block in torrent. Useful for debugging.
116instance Default BlockIx where
117 def = BlockIx 0 0 defaultTransferSize
118
119getInt :: S.Get Int
120getInt = fromIntegral <$> S.getWord32be
121{-# INLINE getInt #-}
122
123putInt :: S.Putter Int
124putInt = S.putWord32be . fromIntegral
125{-# INLINE putInt #-}
126
127instance Serialize BlockIx where
128 {-# SPECIALIZE instance Serialize BlockIx #-}
129 get = BlockIx <$> getInt
130 <*> getInt
131 <*> getInt
132 {-# INLINE get #-}
133
134 put BlockIx {..} = do
135 putInt ixPiece
136 putInt ixOffset
137 putInt ixLength
138 {-# INLINE put #-}
139
140instance Pretty BlockIx where
141 pPrint BlockIx {..} =
142 ("piece = " <> int ixPiece <> ",") <+>
143 ("offset = " <> int ixOffset <> ",") <+>
144 ("length = " <> int ixLength)
145
146-- | Get location of payload bytes in the torrent content.
147blockIxRange :: (Num a, Integral a) => PieceSize -> BlockIx -> (a, a)
148blockIxRange piSize BlockIx {..} = (offset, offset + len)
149 where
150 offset = fromIntegral piSize * fromIntegral ixPiece
151 + fromIntegral ixOffset
152 len = fromIntegral ixLength
153{-# INLINE blockIxRange #-}
154
155{-----------------------------------------------------------------------
156 Block
157-----------------------------------------------------------------------}
158
159data Block payload = Block {
160 -- | Zero-based piece index.
161 blkPiece :: {-# UNPACK #-} !PieceIx
162
163 -- | Zero-based byte offset within the piece.
164 , blkOffset :: {-# UNPACK #-} !BlockOffset
165
166 -- | Payload bytes.
167 , blkData :: !payload
168 } deriving (Show, Eq, Functor, Typeable)
169
170-- | Payload is ommitted.
171instance Pretty (Block BL.ByteString) where
172 pPrint = pPrint . blockIx
173 {-# INLINE pPrint #-}
174
175-- | Get size of block /payload/ in bytes.
176blockSize :: Block BL.ByteString -> BlockSize
177blockSize = fromIntegral . BL.length . blkData
178{-# INLINE blockSize #-}
179
180-- | Get block index of a block.
181blockIx :: Block BL.ByteString -> BlockIx
182blockIx = BlockIx <$> blkPiece <*> blkOffset <*> blockSize
183
184-- | Get location of payload bytes in the torrent content.
185blockRange :: (Num a, Integral a)
186 => PieceSize -> Block BL.ByteString -> (a, a)
187blockRange piSize = blockIxRange piSize . blockIx
188{-# INLINE blockRange #-}
189
190-- | Test if a block can be safely turned into a piece.
191isPiece :: PieceSize -> Block BL.ByteString -> Bool
192isPiece pieceLen blk @ (Block i offset _) =
193 offset == 0 && blockSize blk == pieceLen && i >= 0
194{-# INLINE isPiece #-}
195
196-- | First block in the piece.
197leadingBlock :: PieceIx -> BlockSize -> BlockIx
198leadingBlock pix blockSize = BlockIx
199 { ixPiece = pix
200 , ixOffset = 0
201 , ixLength = blockSize
202 }
203{-# INLINE leadingBlock #-}
204
205{-----------------------------------------------------------------------
206-- Bucket
207-----------------------------------------------------------------------}
208
209type Pos = Int
210type ChunkSize = Int
211
212-- | A sparse set of blocks used to represent an /in progress/ piece.
213data Bucket
214 = Nil
215 | Span {-# UNPACK #-} !ChunkSize !Bucket
216 | Fill {-# UNPACK #-} !ChunkSize !Builder !Bucket
217
218instance Show Bucket where
219 showsPrec i Nil = showString ""
220 showsPrec i (Span s xs) = showString "Span " <> showInt s
221 <> showString " " <> showsPrec i xs
222 showsPrec i (Fill s _ xs) = showString "Fill " <> showInt s
223 <> showString " " <> showsPrec i xs
224
225-- | INVARIANT: 'Nil' should appear only after 'Span' of 'Fill'.
226nilInvFailed :: a
227nilInvFailed = error "Nil: bucket invariant failed"
228
229valid :: Bucket -> Bool
230valid = check Nothing
231 where
232 check Nothing Nil = False -- see 'nilInvFailed'
233 check (Just _) _ = True
234 check prevIsSpan (Span sz xs) =
235 prevIsSpan /= Just True && -- Span n (NotSpan .. ) invariant
236 sz > 0 && -- Span is always non-empty
237 check (Just True) xs
238 check prevIsSpan (Fill sz b xs) =
239 prevIsSpan /= Just True && -- Fill n (NotFill .. ) invariant
240 sz > 0 && -- Fill is always non-empty
241 check (Just False) xs
242
243instance Pretty Bucket where
244 pPrint Nil = nilInvFailed
245 pPrint bkt = go bkt
246 where
247 go Nil = PP.empty
248 go (Span sz xs) = "Span" <+> PP.int sz <+> go xs
249 go (Fill sz b xs) = "Fill" <+> PP.int sz <+> go xs
250
251-- | Smart constructor: use it when some block is /deleted/ from
252-- bucket.
253span :: ChunkSize -> Bucket -> Bucket
254span sz (Span sz' xs) = Span (sz + sz') xs
255span sz xxs = Span sz xxs
256{-# INLINE span #-}
257
258-- | Smart constructor: use it when some block is /inserted/ to
259-- bucket.
260fill :: ChunkSize -> Builder -> Bucket -> Bucket
261fill sz b (Fill sz' b' xs) = Fill (sz + sz') (b <> b') xs
262fill sz b xxs = Fill sz b xxs
263{-# INLINE fill #-}
264
265{-----------------------------------------------------------------------
266-- Bucket queries
267-----------------------------------------------------------------------}
268
269-- | /O(1)/. Test if this bucket is empty.
270null :: Bucket -> Bool
271null Nil = nilInvFailed
272null (Span _ Nil) = True
273null _ = False
274{-# INLINE null #-}
275
276-- | /O(1)/. Test if this bucket is complete.
277full :: Bucket -> Bool
278full Nil = nilInvFailed
279full (Fill _ _ Nil) = True
280full _ = False
281{-# INLINE full #-}
282
283-- | /O(n)/. Total size of the incompleted piece.
284size :: Bucket -> PieceSize
285size Nil = nilInvFailed
286size bkt = go bkt
287 where
288 go Nil = 0
289 go (Span sz xs) = sz + go xs
290 go (Fill sz _ xs) = sz + go xs
291
292-- | /O(n)/. List incomplete blocks to download. If some block have
293-- size more than the specified 'BlockSize' then block is split into
294-- smaller blocks to satisfy given 'BlockSize'. Small (for
295-- e.g. trailing) blocks is not ignored, but returned in-order.
296spans :: BlockSize -> Bucket -> [(BlockOffset, BlockSize)]
297spans expectedSize = go 0
298 where
299 go _ Nil = []
300 go off (Span sz xs) = listChunks off sz ++ go (off + sz) xs
301 go off (Fill sz _ xs) = go (off + sz) xs
302
303 listChunks off restSize
304 | restSize <= 0 = []
305 | otherwise = (off, blkSize)
306 : listChunks (off + blkSize) (restSize - blkSize)
307 where
308 blkSize = min expectedSize restSize
309
310{-----------------------------------------------------------------------
311-- Bucket contstruction
312-----------------------------------------------------------------------}
313
314-- | /O(1)/. A new empty bucket capable to alloof specified size.
315empty :: PieceSize -> Bucket
316empty sz
317 | sz < 0 = error "empty: Bucket size must be a non-negative value"
318 | otherwise = Span sz Nil
319{-# INLINE empty #-}
320
321insertSpan :: Pos -> BS.ByteString -> ChunkSize -> Bucket -> Bucket
322insertSpan !pos !bs !span_sz !xs =
323 let pref_len = pos
324 fill_len = span_sz - pos `min` BS.length bs
325 suff_len = (span_sz - pos) - fill_len
326 in mkSpan pref_len $
327 fill fill_len (byteString (BS.take fill_len bs)) $
328 mkSpan suff_len $
329 xs
330 where
331 mkSpan 0 xs = xs
332 mkSpan sz xs = Span sz xs
333
334-- | /O(n)/. Insert a strict bytestring at specified position.
335--
336-- Best case: if blocks are inserted in sequential order, then this
337-- operation should take /O(1)/.
338--
339insert :: Pos -> BS.ByteString -> Bucket -> Bucket
340insert _ _ Nil = nilInvFailed
341insert dstPos bs bucket = go 0 bucket
342 where
343 intersects curPos sz = dstPos >= curPos && dstPos <= curPos + sz
344
345 go _ Nil = Nil
346 go curPos (Span sz xs)
347 | intersects curPos sz = insertSpan (dstPos - curPos) bs sz xs
348 | otherwise = span sz (go (curPos + sz) xs)
349 go curPos bkt @ (Fill sz br xs)
350 | intersects curPos sz = bkt
351 | otherwise = fill sz br (go (curPos + sz) xs)
352
353fromList :: PieceSize -> [(Pos, BS.ByteString)] -> Bucket
354fromList s = L.foldr (uncurry Network.BitTorrent.Exchange.Block.insert)
355 (Network.BitTorrent.Exchange.Block.empty s)
356
357-- TODO zero-copy
358insertLazy :: Pos -> BL.ByteString -> Bucket -> Bucket
359insertLazy pos bl = Network.BitTorrent.Exchange.Block.insert pos (BL.toStrict bl)
360
361-- | /O(n)/.
362merge :: Bucket -> Bucket -> Bucket
363merge = error "Bucket.merge: not implemented"
364
365-- | /O(1)/.
366toPiece :: Bucket -> Maybe BL.ByteString
367toPiece Nil = nilInvFailed
368toPiece (Fill _ b Nil) = Just (toLazyByteString b)
369toPiece _ = Nothing
diff --git a/src/Network/BitTorrent/Exchange/Connection.hs b/src/Network/BitTorrent/Exchange/Connection.hs
deleted file mode 100644
index 6804d0a2..00000000
--- a/src/Network/BitTorrent/Exchange/Connection.hs
+++ /dev/null
@@ -1,1012 +0,0 @@
1-- |
2-- Module : Network.BitTorrent.Exchange.Wire
3-- Copyright : (c) Sam Truzjan 2013
4-- (c) Daniel Gröber 2013
5-- License : BSD3
6-- Maintainer : pxqr.sta@gmail.com
7-- Stability : experimental
8-- Portability : portable
9--
10-- Each peer wire connection is identified by triple @(topic,
11-- remote_addr, this_addr)@. This means that connections are the
12-- same if and only if their 'ConnectionId' are the same. Of course,
13-- you /must/ avoid duplicated connections.
14--
15-- This module control /integrity/ of data send and received.
16--
17{-# LANGUAGE DeriveDataTypeable #-}
18{-# LANGUAGE TemplateHaskell #-}
19{-# LANGUAGE MultiParamTypeClasses #-}
20{-# LANGUAGE GeneralizedNewtypeDeriving #-}
21module Network.BitTorrent.Exchange.Connection
22 ( -- * Wire
23 Connected
24 , Wire
25 , ChannelSide (..)
26
27 -- * Connection
28 , Connection
29 , connInitiatedBy
30
31 -- ** Identity
32 , connRemoteAddr
33 , connTopic
34 , connRemotePeerId
35 , connThisPeerId
36
37 -- ** Capabilities
38 , connProtocol
39 , connCaps
40 , connExtCaps
41 , connRemoteEhs
42
43 -- ** State
44 , connStatus
45 , connBitfield
46
47 -- ** Env
48 , connOptions
49 , connSession
50 , connStats
51
52 -- ** Status
53 , PeerStatus (..)
54 , ConnectionStatus (..)
55 , updateStatus
56 , statusUpdates
57 , clientStatus
58 , remoteStatus
59 , canUpload
60 , canDownload
61 , defaultUnchokeSlots
62 , defaultRechokeInterval
63
64
65 -- * Setup
66 , ConnectionPrefs (..)
67 , SessionLink (..)
68 , ConnectionConfig (..)
69
70 -- ** Initiate
71 , connectWire
72
73 -- ** Accept
74 , PendingConnection
75 , newPendingConnection
76 , pendingPeer
77 , pendingCaps
78 , pendingTopic
79 , closePending
80 , acceptWire
81
82 -- ** Post setup actions
83 , resizeBitfield
84
85 -- * Messaging
86 , recvMessage
87 , sendMessage
88 , filterQueue
89 , getMaxQueueLength
90
91 -- * Exceptions
92 , ProtocolError (..)
93 , WireFailure (..)
94 , peerPenalty
95 , isWireFailure
96 , disconnectPeer
97
98 -- * Stats
99 , ByteStats (..)
100 , FlowStats (..)
101 , ConnectionStats (..)
102
103 -- * Flood detection
104 , FloodDetector (..)
105
106 -- * Options
107 , Options (..)
108 ) where
109
110import Control.Applicative
111import Control.Concurrent hiding (yield)
112import Control.Exception
113import Control.Monad.Reader
114import Control.Monad.State
115import Control.Monad.Trans.Resource
116import Control.Lens
117import Data.ByteString as BS
118import Data.ByteString.Lazy as BSL
119import Data.Conduit as C
120import Data.Conduit.Cereal
121import Data.Conduit.List
122import Data.Conduit.Network
123import Data.Default
124import Data.IORef
125import Data.List as L
126import Data.Maybe as M
127import Data.Monoid
128import Data.Serialize as S
129import Data.Typeable
130import Network
131import Network.Socket hiding (Connected)
132import Network.Socket.ByteString as BS
133import Text.PrettyPrint as PP hiding ((<>))
134import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
135import Text.Show.Functions ()
136import System.Log.FastLogger (ToLogStr(..))
137import System.Timeout
138
139import Data.Torrent
140import Network.Address
141import Network.BitTorrent.Exchange.Bitfield as BF
142import Network.BitTorrent.Exchange.Message as Msg
143
144-- TODO handle port message?
145-- TODO handle limits?
146-- TODO filter not requested PIECE messages
147-- TODO metadata piece request flood protection
148-- TODO piece request flood protection
149-- TODO protect against flood attacks
150{-----------------------------------------------------------------------
151-- Exceptions
152-----------------------------------------------------------------------}
153
154-- | Used to specify initiator of 'ProtocolError'.
155data ChannelSide
156 = ThisPeer
157 | RemotePeer
158 deriving (Show, Eq, Enum, Bounded)
159
160instance Default ChannelSide where
161 def = ThisPeer
162
163instance Pretty ChannelSide where
164 pPrint = PP.text . show
165
166-- | A protocol errors occur when a peer violates protocol
167-- specification.
168data ProtocolError
169 -- | Protocol string should be 'BitTorrent Protocol' but remote
170 -- peer have sent a different string.
171 = InvalidProtocol ProtocolName
172
173 -- | Sent and received protocol strings do not match. Can occur
174 -- in 'connectWire' only.
175 | UnexpectedProtocol ProtocolName
176
177 -- | /Remote/ peer replied with invalid 'hsInfoHash' which do not
178 -- match with 'hsInfoHash' /this/ peer have sent. Can occur in
179 -- 'connectWire' or 'acceptWire' only.
180 | UnexpectedTopic InfoHash
181
182 -- | Some trackers or DHT can return 'PeerId' of a peer. If a
183 -- remote peer handshaked with different 'hsPeerId' then this
184 -- exception is raised. Can occur in 'connectWire' only.
185 | UnexpectedPeerId PeerId
186
187 -- | Accepted peer have sent unknown torrent infohash in
188 -- 'hsInfoHash' field. This situation usually happen when /this/
189 -- peer have deleted the requested torrent. The error can occur in
190 -- 'acceptWire' function only.
191 | UnknownTopic InfoHash
192
193 -- | A remote peer have 'ExtExtended' enabled but did not send an
194 -- 'ExtendedHandshake' back.
195 | HandshakeRefused
196
197 -- | 'Network.BitTorrent.Exchange.Message.Bitfield' message MUST
198 -- be send either once or zero times, but either this peer or
199 -- remote peer send a bitfield message the second time.
200 | BitfieldAlreadySent ChannelSide
201
202 -- | Capabilities violation. For example this exception can occur
203 -- when a peer have sent 'Port' message but 'ExtDHT' is not
204 -- allowed in 'connCaps'.
205 | DisallowedMessage
206 { -- | Who sent invalid message.
207 violentSender :: ChannelSide
208
209 -- | If the 'violentSender' reconnect with this extension
210 -- enabled then he can try to send this message.
211 , extensionRequired :: Extension
212 }
213 deriving Show
214
215instance Pretty ProtocolError where
216 pPrint = PP.text . show
217
218errorPenalty :: ProtocolError -> Int
219errorPenalty (InvalidProtocol _) = 1
220errorPenalty (UnexpectedProtocol _) = 1
221errorPenalty (UnexpectedTopic _) = 1
222errorPenalty (UnexpectedPeerId _) = 1
223errorPenalty (UnknownTopic _) = 0
224errorPenalty (HandshakeRefused ) = 1
225errorPenalty (BitfieldAlreadySent _) = 1
226errorPenalty (DisallowedMessage _ _) = 1
227
228-- | Exceptions used to interrupt the current P2P session.
229data WireFailure
230 = ConnectionRefused IOError
231
232 -- | Force termination of wire connection.
233 --
234 -- Normally you should throw only this exception from event loop
235 -- using 'disconnectPeer', other exceptions are thrown
236 -- automatically by functions from this module.
237 --
238 | DisconnectPeer
239
240 -- | A peer not responding and did not send a 'KeepAlive' message
241 -- for a specified period of time.
242 | PeerDisconnected
243
244 -- | A remote peer have sent some unknown message we unable to
245 -- parse.
246 | DecodingError GetException
247
248 -- | See 'ProtocolError' for more details.
249 | ProtocolError ProtocolError
250
251 -- | A possible malicious peer have sent too many control messages
252 -- without making any progress.
253 | FloodDetected ConnectionStats
254 deriving (Show, Typeable)
255
256instance Exception WireFailure
257
258instance Pretty WireFailure where
259 pPrint = PP.text . show
260
261-- TODO
262-- data Penalty = Ban | Penalty Int
263
264peerPenalty :: WireFailure -> Int
265peerPenalty DisconnectPeer = 0
266peerPenalty PeerDisconnected = 0
267peerPenalty (DecodingError _) = 1
268peerPenalty (ProtocolError e) = errorPenalty e
269peerPenalty (FloodDetected _) = 1
270
271-- | Do nothing with exception, used with 'handle' or 'try'.
272isWireFailure :: Monad m => WireFailure -> m ()
273isWireFailure _ = return ()
274
275protocolError :: MonadThrow m => ProtocolError -> m a
276protocolError = monadThrow . ProtocolError
277
278{-----------------------------------------------------------------------
279-- Stats
280-----------------------------------------------------------------------}
281
282-- | Message stats in one direction.
283data FlowStats = FlowStats
284 { -- | Number of the messages sent or received.
285 messageCount :: {-# UNPACK #-} !Int
286 -- | Sum of byte sequences of all messages.
287 , messageBytes :: {-# UNPACK #-} !ByteStats
288 } deriving Show
289
290instance Pretty FlowStats where
291 pPrint FlowStats {..} =
292 PP.int messageCount <+> "messages" $+$
293 pPrint messageBytes
294
295-- | Zeroed stats.
296instance Default FlowStats where
297 def = FlowStats 0 def
298
299-- | Monoid under addition.
300instance Monoid FlowStats where
301 mempty = def
302 mappend a b = FlowStats
303 { messageBytes = messageBytes a <> messageBytes b
304 , messageCount = messageCount a + messageCount b
305 }
306
307-- | Find average length of byte sequences per message.
308avgByteStats :: FlowStats -> ByteStats
309avgByteStats (FlowStats n ByteStats {..}) = ByteStats
310 { overhead = overhead `quot` n
311 , control = control `quot` n
312 , payload = payload `quot` n
313 }
314
315-- | Message stats in both directions. This data can be retrieved
316-- using 'getStats' function.
317--
318-- Note that this stats is completely different from
319-- 'Data.Torrent.Progress.Progress': payload bytes not necessary
320-- equal to downloaded\/uploaded bytes since a peer can send a
321-- broken block.
322--
323data ConnectionStats = ConnectionStats
324 { -- | Received messages stats.
325 incomingFlow :: !FlowStats
326 -- | Sent messages stats.
327 , outcomingFlow :: !FlowStats
328 } deriving Show
329
330instance Pretty ConnectionStats where
331 pPrint ConnectionStats {..} = vcat
332 [ "Recv:" <+> pPrint incomingFlow
333 , "Sent:" <+> pPrint outcomingFlow
334 , "Both:" <+> pPrint (incomingFlow <> outcomingFlow)
335 ]
336
337-- | Zeroed stats.
338instance Default ConnectionStats where
339 def = ConnectionStats def def
340
341-- | Monoid under addition.
342instance Monoid ConnectionStats where
343 mempty = def
344 mappend a b = ConnectionStats
345 { incomingFlow = incomingFlow a <> incomingFlow b
346 , outcomingFlow = outcomingFlow a <> outcomingFlow b
347 }
348
349-- | Aggregate one more message stats in the /specified/ direction.
350addStats :: ChannelSide -> ByteStats -> ConnectionStats -> ConnectionStats
351addStats ThisPeer x s = s { outcomingFlow = (FlowStats 1 x) <> (outcomingFlow s) }
352addStats RemotePeer x s = s { incomingFlow = (FlowStats 1 x) <> (incomingFlow s) }
353
354-- | Sum of overhead and control bytes in both directions.
355wastedBytes :: ConnectionStats -> Int
356wastedBytes ConnectionStats {..} = overhead + control
357 where
358 FlowStats _ ByteStats {..} = incomingFlow <> outcomingFlow
359
360-- | Sum of payload bytes in both directions.
361payloadBytes :: ConnectionStats -> Int
362payloadBytes ConnectionStats {..} =
363 payload (messageBytes (incomingFlow <> outcomingFlow))
364
365-- | Sum of any bytes in both directions.
366transmittedBytes :: ConnectionStats -> Int
367transmittedBytes ConnectionStats {..} =
368 byteLength (messageBytes (incomingFlow <> outcomingFlow))
369
370{-----------------------------------------------------------------------
371-- Flood protection
372-----------------------------------------------------------------------}
373
374defaultFloodFactor :: Int
375defaultFloodFactor = 1
376
377-- | This is a very permissive value, connection setup usually takes
378-- around 10-100KB, including both directions.
379defaultFloodThreshold :: Int
380defaultFloodThreshold = 2 * 1024 * 1024
381
382-- | A flood detection function.
383type Detector stats = Int -- ^ Factor;
384 -> Int -- ^ Threshold;
385 -> stats -- ^ Stats to analyse;
386 -> Bool -- ^ Is this a flooded connection?
387
388defaultDetector :: Detector ConnectionStats
389defaultDetector factor threshold s =
390 transmittedBytes s > threshold &&
391 factor * wastedBytes s > payloadBytes s
392
393-- | Flood detection is used to protect /this/ peer against a /remote/
394-- malicious peer sending meaningless control messages.
395data FloodDetector = FloodDetector
396 { -- | Max ratio of payload bytes to control bytes.
397 floodFactor :: {-# UNPACK #-} !Int
398
399 -- | Max count of bytes connection /setup/ can take including
400 -- 'Handshake', 'ExtendedHandshake', 'Bitfield', 'Have' and 'Port'
401 -- messages. This value is used to avoid false positives at the
402 -- connection initialization.
403 , floodThreshold :: {-# UNPACK #-} !Int
404
405 -- | Flood predicate on the /current/ 'ConnectionStats'.
406 , floodPredicate :: Detector ConnectionStats
407 } deriving Show
408
409instance Eq FloodDetector where
410 a == b = floodFactor a == floodFactor b
411 && floodThreshold a == floodThreshold b
412
413-- | Flood detector with very permissive options.
414instance Default FloodDetector where
415 def = FloodDetector
416 { floodFactor = defaultFloodFactor
417 , floodThreshold = defaultFloodThreshold
418 , floodPredicate = defaultDetector
419 }
420
421-- | This peer might drop connection if the detector gives positive answer.
422runDetector :: FloodDetector -> ConnectionStats -> Bool
423runDetector FloodDetector {..} = floodPredicate floodFactor floodThreshold
424
425{-----------------------------------------------------------------------
426-- Options
427-----------------------------------------------------------------------}
428
429-- | Various connection settings and limits.
430data Options = Options
431 { -- | How often /this/ peer should send 'KeepAlive' messages.
432 keepaliveInterval :: {-# UNPACK #-} !Int
433
434 -- | /This/ peer will drop connection if a /remote/ peer did not
435 -- send any message for this period of time.
436 , keepaliveTimeout :: {-# UNPACK #-} !Int
437
438 , requestQueueLength :: {-# UNPACK #-} !Int
439
440 -- | Used to protect against flood attacks.
441 , floodDetector :: FloodDetector
442
443 -- | Used to protect against flood attacks in /metadata
444 -- exchange/. Normally, a requesting peer should request each
445 -- 'InfoDict' piece only one time, but a malicious peer can
446 -- saturate wire with 'MetadataRequest' messages thus flooding
447 -- responding peer.
448 --
449 -- This value set upper bound for number of 'MetadataRequests'
450 -- for each piece.
451 --
452 , metadataFactor :: {-# UNPACK #-} !Int
453
454 -- | Used to protect against out-of-memory attacks: malicious peer
455 -- can claim that 'totalSize' is, say, 100TB and send some random
456 -- data instead of infodict pieces. Since requesting peer unable
457 -- to check not completed infodict via the infohash, the
458 -- accumulated pieces will allocate the all available memory.
459 --
460 -- This limit set upper bound for 'InfoDict' size. See
461 -- 'ExtendedMetadata' for more info.
462 --
463 , maxInfoDictSize :: {-# UNPACK #-} !Int
464 } deriving (Show, Eq)
465
466-- | Permissive default parameters, most likely you don't need to
467-- change them.
468instance Default Options where
469 def = Options
470 { keepaliveInterval = defaultKeepAliveInterval
471 , keepaliveTimeout = defaultKeepAliveTimeout
472 , requestQueueLength = defaultRequestQueueLength
473 , floodDetector = def
474 , metadataFactor = defaultMetadataFactor
475 , maxInfoDictSize = defaultMaxInfoDictSize
476 }
477
478{-----------------------------------------------------------------------
479-- Peer status
480-----------------------------------------------------------------------}
481
482-- | Connections contain two bits of state on either end: choked or
483-- not, and interested or not.
484data PeerStatus = PeerStatus
485 { -- | Choking is a notification that no data will be sent until
486 -- unchoking happens.
487 _choking :: !Bool
488
489 -- |
490 , _interested :: !Bool
491 } deriving (Show, Eq, Ord)
492
493$(makeLenses ''PeerStatus)
494
495instance Pretty PeerStatus where
496 pPrint PeerStatus {..} =
497 pPrint (Choking _choking) <+> "and" <+> pPrint (Interested _interested)
498
499-- | Connections start out choked and not interested.
500instance Default PeerStatus where
501 def = PeerStatus True False
502
503instance Monoid PeerStatus where
504 mempty = def
505 mappend a b = PeerStatus
506 { _choking = _choking a && _choking b
507 , _interested = _interested a || _interested b
508 }
509
510-- | Can be used to update remote peer status using incoming 'Status'
511-- message.
512updateStatus :: StatusUpdate -> PeerStatus -> PeerStatus
513updateStatus (Choking b) = choking .~ b
514updateStatus (Interested b) = interested .~ b
515
516-- | Can be used to generate outcoming messages.
517statusUpdates :: PeerStatus -> PeerStatus -> [StatusUpdate]
518statusUpdates a b = M.catMaybes $
519 [ if _choking a == _choking b then Nothing
520 else Just $ Choking $ _choking b
521 , if _interested a == _interested b then Nothing
522 else Just $ Interested $ _interested b
523 ]
524
525{-----------------------------------------------------------------------
526-- Connection status
527-----------------------------------------------------------------------}
528
529-- | Status of the both endpoints.
530data ConnectionStatus = ConnectionStatus
531 { _clientStatus :: !PeerStatus
532 , _remoteStatus :: !PeerStatus
533 } deriving (Show, Eq)
534
535$(makeLenses ''ConnectionStatus)
536
537instance Pretty ConnectionStatus where
538 pPrint ConnectionStatus {..} =
539 "this " PP.<+> pPrint _clientStatus PP.$$
540 "remote" PP.<+> pPrint _remoteStatus
541
542-- | Connections start out choked and not interested.
543instance Default ConnectionStatus where
544 def = ConnectionStatus def def
545
546-- | Can the client transfer to the remote peer?
547canUpload :: ConnectionStatus -> Bool
548canUpload ConnectionStatus {..}
549 = _interested _remoteStatus && not (_choking _clientStatus)
550
551-- | Can the client transfer from the remote peer?
552canDownload :: ConnectionStatus -> Bool
553canDownload ConnectionStatus {..}
554 = _interested _clientStatus && not (_choking _remoteStatus)
555
556-- | Indicates how many peers are allowed to download from the client
557-- by default.
558defaultUnchokeSlots :: Int
559defaultUnchokeSlots = 4
560
561-- |
562defaultRechokeInterval :: Int
563defaultRechokeInterval = 10 * 1000 * 1000
564
565{-----------------------------------------------------------------------
566-- Connection
567-----------------------------------------------------------------------}
568
569data ConnectionState = ConnectionState {
570 -- | If @not (allowed ExtExtended connCaps)@ then this set is always
571 -- empty. Otherwise it has the BEP10 extension protocol mandated mapping of
572 -- 'MessageId' to the message type for the remote peer.
573 --
574 -- Note that this value can change in current session if either
575 -- this or remote peer will initiate rehandshaking.
576 --
577 _connExtCaps :: !ExtendedCaps
578
579 -- | Current extended handshake information from the remote peer
580 , _connRemoteEhs :: !ExtendedHandshake
581
582 -- | Various stats about messages sent and received. Stats can be
583 -- used to protect /this/ peer against flood attacks.
584 --
585 -- Note that this value will change with the next sent or received
586 -- message.
587 , _connStats :: !ConnectionStats
588
589 , _connStatus :: !ConnectionStatus
590
591 -- | Bitfield of remote endpoint.
592 , _connBitfield :: !Bitfield
593 }
594
595makeLenses ''ConnectionState
596
597instance Default ConnectionState where
598 def = ConnectionState
599 { _connExtCaps = def
600 , _connRemoteEhs = def
601 , _connStats = def
602 , _connStatus = def
603 , _connBitfield = BF.haveNone 0
604 }
605
606-- | Connection keep various info about both peers.
607data Connection s = Connection
608 { connInitiatedBy :: !ChannelSide
609
610 , connRemoteAddr :: !(PeerAddr IP)
611
612 -- | /Both/ peers handshaked with this protocol string. The only
613 -- value is \"Bittorrent Protocol\" but this can be changed in
614 -- future.
615 , connProtocol :: !ProtocolName
616
617 -- | Set of enabled core extensions, i.e. the pre BEP10 extension
618 -- mechanism. This value is used to check if a message is allowed
619 -- to be sent or received.
620 , connCaps :: !Caps
621
622 -- | /Both/ peers handshaked with this infohash. A connection can
623 -- handle only one topic, use 'reconnect' to change the current
624 -- topic.
625 , connTopic :: !InfoHash
626
627 -- | Typically extracted from handshake.
628 , connRemotePeerId :: !PeerId
629
630 -- | Typically extracted from handshake.
631 , connThisPeerId :: !PeerId
632
633 -- |
634 , connOptions :: !Options
635
636 -- | Mutable connection state, see 'ConnectionState'
637 , connState :: !(IORef ConnectionState)
638
639-- -- | Max request queue length.
640-- , connMaxQueueLen :: !Int
641
642 -- | Environment data.
643 , connSession :: !s
644
645 , connChan :: !(Chan Message)
646 }
647
648instance Pretty (Connection s) where
649 pPrint Connection {..} = "Connection"
650
651instance ToLogStr (Connection s) where
652 toLogStr Connection {..} = mconcat
653 [ toLogStr (show connRemoteAddr)
654 , toLogStr (show connProtocol)
655 , toLogStr (show connCaps)
656 , toLogStr (show connTopic)
657 , toLogStr (show connRemotePeerId)
658 , toLogStr (show connThisPeerId)
659 , toLogStr (show connOptions)
660 ]
661
662-- TODO check extended messages too
663isAllowed :: Connection s -> Message -> Bool
664isAllowed Connection {..} msg
665 | Just ext <- requires msg = ext `allowed` connCaps
666 | otherwise = True
667
668{-----------------------------------------------------------------------
669-- Hanshaking
670-----------------------------------------------------------------------}
671
672sendHandshake :: Socket -> Handshake -> IO ()
673sendHandshake sock hs = sendAll sock (S.encode hs)
674
675recvHandshake :: Socket -> IO Handshake
676recvHandshake sock = do
677 header <- BS.recv sock 1
678 unless (BS.length header == 1) $
679 throw $ userError "Unable to receive handshake header."
680
681 let protocolLen = BS.head header
682 let restLen = handshakeSize protocolLen - 1
683
684 body <- BS.recv sock restLen
685 let resp = BS.cons protocolLen body
686 either (throwIO . userError) return $ S.decode resp
687
688-- | Handshaking with a peer specified by the second argument.
689--
690-- It's important to send handshake first because /accepting/ peer
691-- do not know handshake topic and will wait until /connecting/ peer
692-- will send handshake.
693--
694initiateHandshake :: Socket -> Handshake -> IO Handshake
695initiateHandshake sock hs = do
696 sendHandshake sock hs
697 recvHandshake sock
698
699data HandshakePair = HandshakePair
700 { handshakeSent :: !Handshake
701 , handshakeRecv :: !Handshake
702 } deriving (Show, Eq)
703
704validatePair :: HandshakePair -> PeerAddr IP -> IO ()
705validatePair (HandshakePair hs hs') addr = Prelude.mapM_ checkProp
706 [ (def == hsProtocol hs', InvalidProtocol $ hsProtocol hs')
707 , (hsProtocol hs == hsProtocol hs', UnexpectedProtocol $ hsProtocol hs')
708 , (hsInfoHash hs == hsInfoHash hs', UnexpectedTopic $ hsInfoHash hs')
709 , (hsPeerId hs' == fromMaybe (hsPeerId hs') (peerId addr)
710 , UnexpectedPeerId $ hsPeerId hs')
711 ]
712 where
713 checkProp (t, e) = unless t $ throwIO $ ProtocolError e
714
715-- | Connection state /right/ after handshaking.
716establishedStats :: HandshakePair -> ConnectionStats
717establishedStats HandshakePair {..} = ConnectionStats
718 { outcomingFlow = FlowStats 1 $ handshakeStats handshakeSent
719 , incomingFlow = FlowStats 1 $ handshakeStats handshakeRecv
720 }
721
722{-----------------------------------------------------------------------
723-- Wire
724-----------------------------------------------------------------------}
725
726-- | do not expose this so we can change it without breaking api
727newtype Connected s a = Connected { runConnected :: (ReaderT (Connection s) IO a) }
728 deriving (Functor, Applicative, Monad
729 , MonadIO, MonadReader (Connection s), MonadThrow
730 )
731
732instance MonadState ConnectionState (Connected s) where
733 get = Connected (asks connState) >>= liftIO . readIORef
734 put x = Connected (asks connState) >>= liftIO . flip writeIORef x
735
736-- | A duplex channel connected to a remote peer which keep tracks
737-- connection parameters.
738type Wire s a = ConduitM Message Message (Connected s) a
739
740{-----------------------------------------------------------------------
741-- Wrapper
742-----------------------------------------------------------------------}
743
744putStats :: ChannelSide -> Message -> Connected s ()
745putStats side msg = connStats %= addStats side (stats msg)
746
747validate :: ChannelSide -> Message -> Connected s ()
748validate side msg = do
749 caps <- asks connCaps
750 case requires msg of
751 Nothing -> return ()
752 Just ext
753 | ext `allowed` caps -> return ()
754 | otherwise -> protocolError $ DisallowedMessage side ext
755
756trackFlow :: ChannelSide -> Wire s ()
757trackFlow side = iterM $ do
758 validate side
759 putStats side
760
761{-----------------------------------------------------------------------
762-- Setup
763-----------------------------------------------------------------------}
764
765-- System.Timeout.timeout multiplier
766seconds :: Int
767seconds = 1000000
768
769sinkChan :: MonadIO m => Chan Message -> Sink Message m ()
770sinkChan chan = await >>= maybe (return ()) (liftIO . writeChan chan)
771
772sourceChan :: MonadIO m => Int -> Chan Message -> Source m Message
773sourceChan interval chan = do
774 mmsg <- liftIO $ timeout (interval * seconds) $ readChan chan
775 yield $ fromMaybe Msg.KeepAlive mmsg
776
777-- | Normally you should use 'connectWire' or 'acceptWire'.
778runWire :: Wire s () -> Socket -> Chan Message -> Connection s -> IO ()
779runWire action sock chan conn = flip runReaderT conn $ runConnected $
780 sourceSocket sock $=
781 conduitGet S.get $=
782 trackFlow RemotePeer $=
783 action $=
784 trackFlow ThisPeer C.$$
785 sinkChan chan
786
787-- | This function will block until a peer send new message. You can
788-- also use 'await'.
789recvMessage :: Wire s Message
790recvMessage = await >>= maybe (monadThrow PeerDisconnected) return
791
792-- | You can also use 'yield'.
793sendMessage :: PeerMessage msg => msg -> Wire s ()
794sendMessage msg = do
795 ecaps <- use connExtCaps
796 yield $ envelop ecaps msg
797
798getMaxQueueLength :: Connected s Int
799getMaxQueueLength = do
800 advertisedLen <- ehsQueueLength <$> use connRemoteEhs
801 defaultLen <- asks (requestQueueLength . connOptions)
802 return $ fromMaybe defaultLen advertisedLen
803
804-- | Filter pending messages from send buffer.
805filterQueue :: (Message -> Bool) -> Wire s ()
806filterQueue p = lift $ do
807 chan <- asks connChan
808 liftIO $ getChanContents chan >>= writeList2Chan chan . L.filter p
809
810-- | Forcefully terminate wire session and close socket.
811disconnectPeer :: Wire s a
812disconnectPeer = monadThrow DisconnectPeer
813
814extendedHandshake :: ExtendedCaps -> Wire s ()
815extendedHandshake caps = do
816 -- TODO add other params to the handshake
817 sendMessage $ nullExtendedHandshake caps
818 msg <- recvMessage
819 case msg of
820 Extended (EHandshake remoteEhs@(ExtendedHandshake {..})) -> do
821 connExtCaps .= (ehsCaps <> caps)
822 connRemoteEhs .= remoteEhs
823 _ -> protocolError HandshakeRefused
824
825rehandshake :: ExtendedCaps -> Wire s ()
826rehandshake caps = error "rehandshake"
827
828reconnect :: Wire s ()
829reconnect = error "reconnect"
830
831data ConnectionId = ConnectionId
832 { topic :: !InfoHash
833 , remoteAddr :: !(PeerAddr IP)
834 , thisAddr :: !(PeerAddr (Maybe IP)) -- ^ foreign address of this node.
835 }
836
837-- | /Preffered/ settings of wire. To get the real use 'ask'.
838data ConnectionPrefs = ConnectionPrefs
839 { prefOptions :: !Options
840 , prefProtocol :: !ProtocolName
841 , prefCaps :: !Caps
842 , prefExtCaps :: !ExtendedCaps
843 } deriving (Show, Eq)
844
845instance Default ConnectionPrefs where
846 def = ConnectionPrefs
847 { prefOptions = def
848 , prefProtocol = def
849 , prefCaps = def
850 , prefExtCaps = def
851 }
852
853normalize :: ConnectionPrefs -> ConnectionPrefs
854normalize = error "normalize"
855
856-- | Bridge between 'Connection' and 'Network.BitTorrent.Exchange.Session'.
857data SessionLink s = SessionLink
858 { linkTopic :: !(InfoHash)
859 , linkPeerId :: !(PeerId)
860 , linkMetadataSize :: !(Maybe Int)
861 , linkOutputChan :: !(Maybe (Chan Message))
862 , linkSession :: !(s)
863 }
864
865data ConnectionConfig s = ConnectionConfig
866 { cfgPrefs :: !(ConnectionPrefs)
867 , cfgSession :: !(SessionLink s)
868 , cfgWire :: !(Wire s ())
869 }
870
871configHandshake :: ConnectionConfig s -> Handshake
872configHandshake ConnectionConfig {..} = Handshake
873 { hsProtocol = prefProtocol cfgPrefs
874 , hsReserved = prefCaps cfgPrefs
875 , hsInfoHash = linkTopic cfgSession
876 , hsPeerId = linkPeerId cfgSession
877 }
878
879{-----------------------------------------------------------------------
880-- Pending connections
881-----------------------------------------------------------------------}
882
883-- | Connection in half opened state. A normal usage scenario:
884--
885-- * Opened using 'newPendingConnection', usually in the listener
886-- loop;
887--
888-- * Closed using 'closePending' if 'pendingPeer' is banned,
889-- 'pendingCaps' is prohibited or pendingTopic is unknown;
890--
891-- * Accepted using 'acceptWire' otherwise.
892--
893data PendingConnection = PendingConnection
894 { pendingSock :: Socket
895 , pendingPeer :: PeerAddr IP -- ^ 'peerId' is always non empty;
896 , pendingCaps :: Caps -- ^ advertised by the peer;
897 , pendingTopic :: InfoHash -- ^ possible non-existent topic.
898 }
899
900-- | Reconstruct handshake sent by the remote peer.
901pendingHandshake :: PendingConnection -> Handshake
902pendingHandshake PendingConnection {..} = Handshake
903 { hsProtocol = def
904 , hsReserved = pendingCaps
905 , hsInfoHash = pendingTopic
906 , hsPeerId = fromMaybe (error "pendingHandshake: impossible")
907 (peerId pendingPeer)
908 }
909
910-- |
911--
912-- This function can throw 'WireFailure' exception.
913--
914newPendingConnection :: Socket -> PeerAddr IP -> IO PendingConnection
915newPendingConnection sock addr = do
916 Handshake {..} <- recvHandshake sock
917 unless (hsProtocol == def) $ do
918 throwIO $ ProtocolError $ InvalidProtocol hsProtocol
919 return PendingConnection
920 { pendingSock = sock
921 , pendingPeer = addr { peerId = Just hsPeerId }
922 , pendingCaps = hsReserved
923 , pendingTopic = hsInfoHash
924 }
925
926-- | Release all resources associated with the given connection. Note
927-- that you /must not/ 'closePending' if you 'acceptWire'.
928closePending :: PendingConnection -> IO ()
929closePending PendingConnection {..} = do
930 close pendingSock
931
932{-----------------------------------------------------------------------
933-- Connection setup
934-----------------------------------------------------------------------}
935
936chanToSock :: Int -> Chan Message -> Socket -> IO ()
937chanToSock ka chan sock =
938 sourceChan ka chan $= conduitPut S.put C.$$ sinkSocket sock
939
940afterHandshaking :: ChannelSide -> PeerAddr IP -> Socket -> HandshakePair
941 -> ConnectionConfig s -> IO ()
942afterHandshaking initiator addr sock
943 hpair @ (HandshakePair hs hs')
944 (ConnectionConfig
945 { cfgPrefs = ConnectionPrefs {..}
946 , cfgSession = SessionLink {..}
947 , cfgWire = wire
948 }) = do
949 let caps = hsReserved hs <> hsReserved hs'
950 cstate <- newIORef def { _connStats = establishedStats hpair }
951 chan <- maybe newChan return linkOutputChan
952 let conn = Connection {
953 connInitiatedBy = initiator
954 , connRemoteAddr = addr
955 , connProtocol = hsProtocol hs
956 , connCaps = caps
957 , connTopic = hsInfoHash hs
958 , connRemotePeerId = hsPeerId hs'
959 , connThisPeerId = hsPeerId hs
960 , connOptions = def
961 , connState = cstate
962 , connSession = linkSession
963 , connChan = chan
964 }
965
966 -- TODO make KA interval configurable
967 let kaInterval = defaultKeepAliveInterval
968 wire' = if ExtExtended `allowed` caps
969 then extendedHandshake prefExtCaps >> wire
970 else wire
971
972 bracket (forkIO (chanToSock kaInterval chan sock))
973 (killThread)
974 (\ _ -> runWire wire' sock chan conn)
975
976-- | Initiate 'Wire' connection and handshake with a peer. This function will
977-- also do the BEP10 extension protocol handshake if 'ExtExtended' is enabled on
978-- both sides.
979--
980-- This function can throw 'WireFailure' exception.
981--
982connectWire :: PeerAddr IP -> ConnectionConfig s -> IO ()
983connectWire addr cfg = do
984 let catchRefusal m = try m >>= either (throwIO . ConnectionRefused) return
985 bracket (catchRefusal (peerSocket Stream addr)) close $ \ sock -> do
986 let hs = configHandshake cfg
987 hs' <- initiateHandshake sock hs
988 let hpair = HandshakePair hs hs'
989 validatePair hpair addr
990 afterHandshaking ThisPeer addr sock hpair cfg
991
992-- | Accept 'Wire' connection using already 'Network.Socket.accept'ed
993-- socket. For peer listener loop the 'acceptSafe' should be
994-- prefered against 'accept'. The socket will be closed at exit.
995--
996-- This function can throw 'WireFailure' exception.
997--
998acceptWire :: PendingConnection -> ConnectionConfig s -> IO ()
999acceptWire pc @ PendingConnection {..} cfg = do
1000 bracket (return pendingSock) close $ \ _ -> do
1001 unless (linkTopic (cfgSession cfg) == pendingTopic) $ do
1002 throwIO (ProtocolError (UnexpectedTopic pendingTopic))
1003
1004 let hs = configHandshake cfg
1005 sendHandshake pendingSock hs
1006 let hpair = HandshakePair hs (pendingHandshake pc)
1007
1008 afterHandshaking RemotePeer pendingPeer pendingSock hpair cfg
1009
1010-- | Used when size of bitfield becomes known.
1011resizeBitfield :: Int -> Connected s ()
1012resizeBitfield n = connBitfield %= adjustSize n
diff --git a/src/Network/BitTorrent/Exchange/Download.hs b/src/Network/BitTorrent/Exchange/Download.hs
deleted file mode 100644
index 981db2fb..00000000
--- a/src/Network/BitTorrent/Exchange/Download.hs
+++ /dev/null
@@ -1,296 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8--
9--
10{-# LANGUAGE FlexibleContexts #-}
11{-# LANGUAGE FlexibleInstances #-}
12{-# LANGUAGE MultiParamTypeClasses #-}
13{-# LANGUAGE FunctionalDependencies #-}
14{-# LANGUAGE TemplateHaskell #-}
15module Network.BitTorrent.Exchange.Download
16 ( -- * Downloading
17 Download (..)
18 , Updates
19 , runDownloadUpdates
20
21 -- ** Metadata
22 -- $metadata-download
23 , MetadataDownload
24 , metadataDownload
25
26 -- ** Content
27 -- $content-download
28 , ContentDownload
29 , contentDownload
30 ) where
31
32import Control.Applicative
33import Control.Concurrent
34import Control.Lens
35import Control.Monad.State
36import Data.BEncode as BE
37import Data.ByteString as BS
38import Data.ByteString.Lazy as BL
39import Data.Default
40import Data.List as L
41import Data.Maybe
42import Data.Map as M
43import Data.Tuple
44
45import Data.Torrent as Torrent
46import Network.Address
47import Network.BitTorrent.Exchange.Bitfield as BF
48import Network.BitTorrent.Exchange.Block as Block
49import Network.BitTorrent.Exchange.Message as Msg
50import System.Torrent.Storage (Storage, writePiece)
51
52
53{-----------------------------------------------------------------------
54-- Class
55-----------------------------------------------------------------------}
56
57type Updates s a = StateT s IO a
58
59runDownloadUpdates :: MVar s -> Updates s a -> IO a
60runDownloadUpdates var m = modifyMVar var (fmap swap . runStateT m)
61
62class Download s chunk | s -> chunk where
63 scheduleBlocks :: Int -> PeerAddr IP -> Bitfield -> Updates s [BlockIx]
64
65 -- |
66 scheduleBlock :: PeerAddr IP -> Bitfield -> Updates s (Maybe BlockIx)
67 scheduleBlock addr bf = listToMaybe <$> scheduleBlocks 1 addr bf
68
69 -- | Get number of sent requests to this peer.
70 getRequestQueueLength :: PeerAddr IP -> Updates s Int
71
72 -- | Remove all pending block requests to the remote peer. May be used
73 -- when:
74 --
75 -- * a peer closes connection;
76 --
77 -- * remote peer choked this peer;
78 --
79 -- * timeout expired.
80 --
81 resetPending :: PeerAddr IP -> Updates s ()
82
83 -- | MAY write to storage, if a new piece have been completed.
84 --
85 -- You should check if a returned by peer block is actually have
86 -- been requested and in-flight. This is needed to avoid "I send
87 -- random corrupted block" attacks.
88 pushBlock :: PeerAddr IP -> chunk -> Updates s (Maybe Bool)
89
90{-----------------------------------------------------------------------
91-- Metadata download
92-----------------------------------------------------------------------}
93-- $metadata-download
94-- TODO
95
96data MetadataDownload = MetadataDownload
97 { _pendingPieces :: [(PeerAddr IP, PieceIx)]
98 , _bucket :: Bucket
99 , _topic :: InfoHash
100 }
101
102makeLenses ''MetadataDownload
103
104-- | Create a new scheduler for infodict of the given size.
105metadataDownload :: Int -> InfoHash -> MetadataDownload
106metadataDownload ps = MetadataDownload [] (Block.empty ps)
107
108instance Default MetadataDownload where
109 def = error "instance Default MetadataDownload"
110
111--cancelPending :: PieceIx -> Updates ()
112cancelPending pix = pendingPieces %= L.filter ((pix ==) . snd)
113
114instance Download MetadataDownload (Piece BS.ByteString) where
115 scheduleBlock addr bf = do
116 bkt <- use bucket
117 case spans metadataPieceSize bkt of
118 [] -> return Nothing
119 ((off, _ ) : _) -> do
120 let pix = off `div` metadataPieceSize
121 pendingPieces %= ((addr, pix) :)
122 return (Just (BlockIx pix 0 metadataPieceSize))
123
124 resetPending addr = pendingPieces %= L.filter ((addr ==) . fst)
125
126 pushBlock addr Torrent.Piece {..} = do
127 p <- use pendingPieces
128 when ((addr, pieceIndex) `L.notElem` p) $
129 error "not requested"
130 cancelPending pieceIndex
131
132 bucket %= Block.insert (metadataPieceSize * pieceIndex) pieceData
133 b <- use bucket
134 case toPiece b of
135 Nothing -> return Nothing
136 Just chunks -> do
137 t <- use topic
138 case parseInfoDict (BL.toStrict chunks) t of
139 Right x -> do
140 pendingPieces .= []
141 return undefined -- (Just x)
142 Left e -> do
143 pendingPieces .= []
144 bucket .= Block.empty (Block.size b)
145 return undefined -- Nothing
146 where
147 -- todo use incremental parsing to avoid BS.concat call
148 parseInfoDict :: BS.ByteString -> InfoHash -> Result InfoDict
149 parseInfoDict chunk topic =
150 case BE.decode chunk of
151 Right (infodict @ InfoDict {..})
152 | topic == idInfoHash -> return infodict
153 | otherwise -> Left "broken infodict"
154 Left err -> Left $ "unable to parse infodict " ++ err
155
156{-----------------------------------------------------------------------
157-- Content download
158-----------------------------------------------------------------------}
159-- $content-download
160--
161-- A block can have one of the following status:
162--
163-- 1) /not allowed/: Piece is not in download set.
164--
165-- 2) /waiting/: (allowed?) Block have been allowed to download,
166-- but /this/ peer did not send any 'Request' message for this
167-- block. To allow some piece use
168-- 'Network.BitTorrent.Exchange.Selector' and then 'allowedSet'
169-- and 'allowPiece'.
170--
171-- 3) /inflight/: (pending?) Block have been requested but
172-- /remote/ peer did not send any 'Piece' message for this block.
173-- Related functions 'markInflight'
174--
175-- 4) /pending/: (stalled?) Block have have been downloaded
176-- Related functions 'insertBlock'.
177--
178-- Piece status:
179--
180-- 1) /assembled/: (downloaded?) All blocks in piece have been
181-- downloaded but the piece did not verified yet.
182--
183-- * Valid: go to completed;
184--
185-- * Invalid: go to waiting.
186--
187-- 2) /corrupted/:
188--
189-- 3) /downloaded/: (verified?) A piece have been successfully
190-- verified via the hash. Usually the piece should be stored to
191-- the 'System.Torrent.Storage' and /this/ peer should send 'Have'
192-- messages to the /remote/ peers.
193--
194
195data PieceEntry = PieceEntry
196 { pending :: [(PeerAddr IP, BlockIx)]
197 , stalled :: Bucket
198 }
199
200pieceEntry :: PieceSize -> PieceEntry
201pieceEntry s = PieceEntry [] (Block.empty s)
202
203isEmpty :: PieceEntry -> Bool
204isEmpty PieceEntry {..} = L.null pending && Block.null stalled
205
206_holes :: PieceIx -> PieceEntry -> [BlockIx]
207_holes pix PieceEntry {..} = fmap mkBlockIx (spans defaultTransferSize stalled)
208 where
209 mkBlockIx (off, sz) = BlockIx pix off sz
210
211data ContentDownload = ContentDownload
212 { inprogress :: !(Map PieceIx PieceEntry)
213 , bitfield :: !Bitfield
214 , pieceSize :: !PieceSize
215 , contentStorage :: Storage
216 }
217
218contentDownload :: Bitfield -> PieceSize -> Storage -> ContentDownload
219contentDownload = ContentDownload M.empty
220
221--modifyEntry :: PieceIx -> (PieceEntry -> PieceEntry) -> DownloadUpdates ()
222modifyEntry pix f = modify $ \ s @ ContentDownload {..} -> s
223 { inprogress = alter (g pieceSize) pix inprogress }
224 where
225 g s = h . f . fromMaybe (pieceEntry s)
226 h e
227 | isEmpty e = Nothing
228 | otherwise = Just e
229
230instance Download ContentDownload (Block BL.ByteString) where
231 scheduleBlocks n addr maskBF = do
232 ContentDownload {..} <- get
233 let wantPieces = maskBF `BF.difference` bitfield
234 let wantBlocks = L.concat $ M.elems $ M.mapWithKey _holes $
235 M.filterWithKey (\ pix _ -> pix `BF.member` wantPieces)
236 inprogress
237
238 bixs <- if L.null wantBlocks
239 then do
240 mpix <- choosePiece wantPieces
241 case mpix of -- TODO return 'n' blocks
242 Nothing -> return []
243 Just pix -> return [leadingBlock pix defaultTransferSize]
244 else chooseBlocks wantBlocks n
245
246 forM_ bixs $ \ bix -> do
247 modifyEntry (ixPiece bix) $ \ e @ PieceEntry {..} -> e
248 { pending = (addr, bix) : pending }
249
250 return bixs
251 where
252 -- TODO choose block nearest to pending or stalled sets to reduce disk
253 -- seeks on remote machines
254 --chooseBlocks :: [BlockIx] -> Int -> DownloadUpdates [BlockIx]
255 chooseBlocks xs n = return (L.take n xs)
256
257 -- TODO use selection strategies from Exchange.Selector
258 --choosePiece :: Bitfield -> DownloadUpdates (Maybe PieceIx)
259 choosePiece bf
260 | BF.null bf = return $ Nothing
261 | otherwise = return $ Just $ BF.findMin bf
262
263 getRequestQueueLength addr = do
264 m <- gets (M.map (L.filter ((==) addr . fst) . pending) . inprogress)
265 return $ L.sum $ L.map L.length $ M.elems m
266
267 resetPending addr = modify $ \ s -> s { inprogress = reset (inprogress s) }
268 where
269 reset = fmap $ \ e -> e
270 { pending = L.filter (not . (==) addr . fst) (pending e) }
271
272 pushBlock addr blk @ Block {..} = do
273 mpe <- gets (M.lookup blkPiece . inprogress)
274 case mpe of
275 Nothing -> return Nothing
276 Just (pe @ PieceEntry {..})
277 | blockIx blk `L.notElem` fmap snd pending -> return Nothing
278 | otherwise -> do
279 let bkt' = Block.insertLazy blkOffset blkData stalled
280 case toPiece bkt' of
281 Nothing -> do
282 modifyEntry blkPiece $ \ e @ PieceEntry {..} -> e
283 { pending = L.filter ((==) (blockIx blk) . snd) pending
284 , stalled = bkt'
285 }
286 return (Just False)
287
288 Just pieceData -> do
289 -- TODO verify
290 storage <- gets contentStorage
291 liftIO $ writePiece (Torrent.Piece blkPiece pieceData) storage
292 modify $ \ s @ ContentDownload {..} -> s
293 { inprogress = M.delete blkPiece inprogress
294 , bitfield = BF.insert blkPiece bitfield
295 }
296 return (Just True)
diff --git a/src/Network/BitTorrent/Exchange/Manager.hs b/src/Network/BitTorrent/Exchange/Manager.hs
deleted file mode 100644
index 30a6a607..00000000
--- a/src/Network/BitTorrent/Exchange/Manager.hs
+++ /dev/null
@@ -1,62 +0,0 @@
1module Network.BitTorrent.Exchange.Manager
2 ( Options (..)
3 , Manager
4 , Handler
5 , newManager
6 , closeManager
7 ) where
8
9import Control.Concurrent
10import Control.Exception hiding (Handler)
11import Control.Monad
12import Data.Default
13import Network.Socket
14
15import Data.Torrent
16import Network.Address
17import Network.BitTorrent.Exchange.Connection hiding (Options)
18import Network.BitTorrent.Exchange.Session
19
20
21data Options = Options
22 { optBacklog :: Int
23 , optPeerAddr :: PeerAddr IP
24 } deriving (Show, Eq)
25
26instance Default Options where
27 def = Options
28 { optBacklog = maxListenQueue
29 , optPeerAddr = def
30 }
31
32data Manager = Manager
33 { listener :: !ThreadId
34 }
35
36type Handler = InfoHash -> IO Session
37
38handleNewConn :: Socket -> PeerAddr IP -> Handler -> IO ()
39handleNewConn sock addr handler = do
40 conn <- newPendingConnection sock addr
41 ses <- handler (pendingTopic conn) `onException` closePending conn
42 establish conn ses
43
44listenIncoming :: Options -> Handler -> IO ()
45listenIncoming Options {..} handler = do
46 bracket (socket AF_INET Stream defaultProtocol) close $ \ sock -> do
47 bind sock (toSockAddr optPeerAddr)
48 listen sock optBacklog
49 forever $ do
50 (conn, sockAddr) <- accept sock
51 case fromSockAddr sockAddr of
52 Nothing -> return ()
53 Just addr -> void $ forkIO $ handleNewConn sock addr handler
54
55newManager :: Options -> Handler -> IO Manager
56newManager opts handler = do
57 tid <- forkIO $ listenIncoming opts handler
58 return (Manager tid)
59
60closeManager :: Manager -> IO ()
61closeManager Manager {..} = do
62 killThread listener \ No newline at end of file
diff --git a/src/Network/BitTorrent/Exchange/Message.hs b/src/Network/BitTorrent/Exchange/Message.hs
deleted file mode 100644
index 2c6770f7..00000000
--- a/src/Network/BitTorrent/Exchange/Message.hs
+++ /dev/null
@@ -1,1232 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- Normally peer to peer communication consisting of the following
9-- steps:
10--
11-- * In order to establish the connection between peers we should
12-- send 'Handshake' message. The 'Handshake' is a required message
13-- and must be the first message transmitted by the peer to the
14-- another peer. Another peer should reply with a handshake as well.
15--
16-- * Next peer might sent bitfield message, but might not. In the
17-- former case we should update bitfield peer have. Again, if we
18-- have some pieces we should send bitfield. Normally bitfield
19-- message should sent after the handshake message.
20--
21-- * Regular exchange messages. TODO docs
22--
23-- For more high level API see "Network.BitTorrent.Exchange" module.
24--
25-- For more infomation see:
26-- <https://wiki.theory.org/BitTorrentSpecification#Peer_wire_protocol_.28TCP.29>
27--
28{-# LANGUAGE ViewPatterns #-}
29{-# LANGUAGE FlexibleInstances #-}
30{-# LANGUAGE FlexibleContexts #-}
31{-# LANGUAGE TypeFamilies #-}
32{-# LANGUAGE GeneralizedNewtypeDeriving #-}
33{-# LANGUAGE DeriveDataTypeable #-}
34{-# LANGUAGE TemplateHaskell #-}
35{-# OPTIONS -fno-warn-orphans #-}
36module Network.BitTorrent.Exchange.Message
37 ( -- * Capabilities
38 Capabilities (..)
39 , Extension (..)
40 , Caps
41
42 -- * Handshake
43 , ProtocolName
44 , Handshake(..)
45 , defaultHandshake
46 , handshakeSize
47 , handshakeMaxSize
48 , handshakeStats
49
50 -- * Stats
51 , ByteCount
52 , ByteStats (..)
53 , byteLength
54
55 -- * Messages
56 , Message (..)
57 , defaultKeepAliveTimeout
58 , defaultKeepAliveInterval
59 , PeerMessage (..)
60
61 -- ** Core messages
62 , StatusUpdate (..)
63 , Available (..)
64 , Transfer (..)
65 , defaultRequestQueueLength
66
67 -- ** Fast extension
68 , FastMessage (..)
69
70 -- ** Extension protocol
71 , ExtendedMessage (..)
72
73 -- *** Capabilities
74 , ExtendedExtension (..)
75 , ExtendedCaps (..)
76
77 -- *** Handshake
78 , ExtendedHandshake (..)
79 , defaultQueueLength
80 , nullExtendedHandshake
81
82 -- *** Metadata
83 , ExtendedMetadata (..)
84 , metadataPieceSize
85 , defaultMetadataFactor
86 , defaultMaxInfoDictSize
87 , isLastPiece
88 , isValidPiece
89 ) where
90
91import Control.Applicative
92import Control.Arrow ((&&&), (***))
93import Control.Monad (when)
94import Data.Attoparsec.ByteString.Char8 as BS
95import Data.BEncode as BE
96import Data.BEncode.BDict as BE
97import Data.BEncode.Internal as BE (ppBEncode, parser)
98import Data.BEncode.Types (BDict)
99import Data.Bits
100import Data.ByteString as BS
101import Data.ByteString.Char8 as BC
102import Data.ByteString.Lazy as BL
103import Data.Default
104import Data.List as L
105import Data.Map.Strict as M
106import Data.Maybe
107import Data.Monoid
108import Data.Ord
109import Data.Serialize as S
110import Data.String
111import Data.Text as T
112import Data.Typeable
113import Data.Word
114import Data.IP
115import Network
116import Network.Socket hiding (KeepAlive)
117import Text.PrettyPrint as PP hiding ((<>))
118import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
119
120import Data.Torrent hiding (Piece (..))
121import qualified Data.Torrent as P (Piece (..))
122import Network.Address
123import Network.BitTorrent.Exchange.Bitfield
124import Network.BitTorrent.Exchange.Block
125
126{-----------------------------------------------------------------------
127-- Capabilities
128-----------------------------------------------------------------------}
129
130-- |
131class Capabilities caps where
132 type Ext caps :: *
133
134 -- | Pack extensions to caps.
135 toCaps :: [Ext caps] -> caps
136
137 -- | Unpack extensions from caps.
138 fromCaps :: caps -> [Ext caps]
139
140 -- | Check if an extension is a member of the specified set.
141 allowed :: Ext caps -> caps -> Bool
142
143ppCaps :: Capabilities caps => Pretty (Ext caps) => caps -> Doc
144ppCaps = hcat . punctuate ", " . L.map pPrint . fromCaps
145
146{-----------------------------------------------------------------------
147-- Extensions
148-----------------------------------------------------------------------}
149
150-- | Enumeration of message extension protocols.
151--
152-- For more info see: <http://www.bittorrent.org/beps/bep_0004.html>
153--
154data Extension
155 = ExtDHT -- ^ BEP 5: allow to send PORT messages.
156 | ExtFast -- ^ BEP 6: allow to send FAST messages.
157 | ExtExtended -- ^ BEP 10: allow to send the extension protocol messages.
158 deriving (Show, Eq, Ord, Enum, Bounded)
159
160-- | Full extension names, suitable for logging.
161instance Pretty Extension where
162 pPrint ExtDHT = "Distributed Hash Table Protocol"
163 pPrint ExtFast = "Fast Extension"
164 pPrint ExtExtended = "Extension Protocol"
165
166-- | Extension bitmask as specified by BEP 4.
167extMask :: Extension -> Word64
168extMask ExtDHT = 0x01
169extMask ExtFast = 0x04
170extMask ExtExtended = 0x100000
171
172{-----------------------------------------------------------------------
173-- Capabilities
174-----------------------------------------------------------------------}
175
176-- | Capabilities is a set of 'Extension's usually sent in 'Handshake'
177-- messages.
178newtype Caps = Caps Word64
179 deriving (Show, Eq)
180
181-- | Render set of extensions as comma separated list.
182instance Pretty Caps where
183 pPrint = ppCaps
184 {-# INLINE pPrint #-}
185
186-- | The empty set.
187instance Default Caps where
188 def = Caps 0
189 {-# INLINE def #-}
190
191-- | Monoid under intersection. 'mempty' includes all known extensions.
192instance Monoid Caps where
193 mempty = toCaps [minBound .. maxBound]
194 {-# INLINE mempty #-}
195
196 mappend (Caps a) (Caps b) = Caps (a .&. b)
197 {-# INLINE mappend #-}
198
199-- | 'Handshake' compatible encoding.
200instance Serialize Caps where
201 put (Caps caps) = S.putWord64be caps
202 {-# INLINE put #-}
203
204 get = Caps <$> S.getWord64be
205 {-# INLINE get #-}
206
207instance Capabilities Caps where
208 type Ext Caps = Extension
209
210 allowed e (Caps caps) = (extMask e .&. caps) /= 0
211 {-# INLINE allowed #-}
212
213 toCaps = Caps . L.foldr (.|.) 0 . L.map extMask
214 fromCaps caps = L.filter (`allowed` caps) [minBound..maxBound]
215
216{-----------------------------------------------------------------------
217 Handshake
218-----------------------------------------------------------------------}
219
220maxProtocolNameSize :: Word8
221maxProtocolNameSize = maxBound
222
223-- | The protocol name is used to identify to the local peer which
224-- version of BTP the remote peer uses.
225newtype ProtocolName = ProtocolName BS.ByteString
226 deriving (Eq, Ord, Typeable)
227
228-- | In BTP/1.0 the name is 'BitTorrent protocol'. If this string is
229-- different from the local peers own protocol name, then the
230-- connection is to be dropped.
231instance Default ProtocolName where
232 def = ProtocolName "BitTorrent protocol"
233
234instance Show ProtocolName where
235 show (ProtocolName bs) = show bs
236
237instance Pretty ProtocolName where
238 pPrint (ProtocolName bs) = PP.text $ BC.unpack bs
239
240instance IsString ProtocolName where
241 fromString str
242 | L.length str <= fromIntegral maxProtocolNameSize
243 = ProtocolName (fromString str)
244 | otherwise = error $ "fromString: ProtocolName too long: " ++ str
245
246instance Serialize ProtocolName where
247 put (ProtocolName bs) = do
248 putWord8 $ fromIntegral $ BS.length bs
249 putByteString bs
250
251 get = do
252 len <- getWord8
253 bs <- getByteString $ fromIntegral len
254 return (ProtocolName bs)
255
256-- | Handshake message is used to exchange all information necessary
257-- to establish connection between peers.
258--
259data Handshake = Handshake {
260 -- | Identifier of the protocol. This is usually equal to 'def'.
261 hsProtocol :: ProtocolName
262
263 -- | Reserved bytes used to specify supported BEP's.
264 , hsReserved :: Caps
265
266 -- | Info hash of the info part of the metainfo file. that is
267 -- transmitted in tracker requests. Info hash of the initiator
268 -- handshake and response handshake should match, otherwise
269 -- initiator should break the connection.
270 --
271 , hsInfoHash :: InfoHash
272
273 -- | Peer id of the initiator. This is usually the same peer id
274 -- that is transmitted in tracker requests.
275 --
276 , hsPeerId :: PeerId
277
278 } deriving (Show, Eq)
279
280instance Serialize Handshake where
281 put Handshake {..} = do
282 put hsProtocol
283 put hsReserved
284 put hsInfoHash
285 put hsPeerId
286 get = Handshake <$> get <*> get <*> get <*> get
287
288-- | Show handshake protocol string, caps and fingerprint.
289instance Pretty Handshake where
290 pPrint Handshake {..}
291 = pPrint hsProtocol $$
292 pPrint hsReserved $$
293 pPrint (fingerprint hsPeerId)
294
295-- | Get handshake message size in bytes from the length of protocol
296-- string.
297handshakeSize :: Word8 -> Int
298handshakeSize n = 1 + fromIntegral n + 8 + 20 + 20
299
300-- | Maximum size of handshake message in bytes.
301handshakeMaxSize :: Int
302handshakeMaxSize = handshakeSize maxProtocolNameSize
303
304-- | Handshake with default protocol string and reserved bitmask.
305defaultHandshake :: InfoHash -> PeerId -> Handshake
306defaultHandshake = Handshake def def
307
308handshakeStats :: Handshake -> ByteStats
309handshakeStats (Handshake (ProtocolName bs) _ _ _)
310 = ByteStats 1 (BS.length bs + 8 + 20 + 20) 0
311
312{-----------------------------------------------------------------------
313-- Stats
314-----------------------------------------------------------------------}
315
316-- | Number of bytes.
317type ByteCount = Int
318
319-- | Summary of encoded message byte layout can be used to collect
320-- stats about message flow in both directions. This data can be
321-- retrieved using 'stats' function.
322data ByteStats = ByteStats
323 { -- | Number of bytes used to help encode 'control' and 'payload'
324 -- bytes: message size, message ID's, etc
325 overhead :: {-# UNPACK #-} !ByteCount
326
327 -- | Number of bytes used to exchange peers state\/options: piece
328 -- and block indexes, infohash, port numbers, peer ID\/IP, etc.
329 , control :: {-# UNPACK #-} !ByteCount
330
331 -- | Number of payload bytes: torrent data blocks and infodict
332 -- metadata.
333 , payload :: {-# UNPACK #-} !ByteCount
334 } deriving Show
335
336instance Pretty ByteStats where
337 pPrint s @ ByteStats {..} = fsep
338 [ PP.int overhead, "overhead"
339 , PP.int control, "control"
340 , PP.int payload, "payload"
341 , "bytes"
342 ] $+$ fsep
343 [ PP.int (byteLength s), "total bytes"
344 ]
345
346-- | Empty byte sequences.
347instance Default ByteStats where
348 def = ByteStats 0 0 0
349
350-- | Monoid under addition.
351instance Monoid ByteStats where
352 mempty = def
353 mappend a b = ByteStats
354 { overhead = overhead a + overhead b
355 , control = control a + control b
356 , payload = payload a + payload b
357 }
358
359-- | Sum of the all byte sequences.
360byteLength :: ByteStats -> Int
361byteLength ByteStats {..} = overhead + control + payload
362
363{-----------------------------------------------------------------------
364-- Regular messages
365-----------------------------------------------------------------------}
366
367-- | Messages which can be sent after handshaking. Minimal complete
368-- definition: 'envelop'.
369class PeerMessage a where
370 -- | Construct a message to be /sent/. Note that if 'ExtendedCaps'
371 -- do not contain mapping for this message the default
372 -- 'ExtendedMessageId' is used.
373 envelop :: ExtendedCaps -- ^ The /receiver/ extended capabilities;
374 -> a -- ^ An regular message;
375 -> Message -- ^ Enveloped message to sent.
376
377 -- | Find out the extension this message belong to. Can be used to
378 -- check if this message is allowed to send\/recv in current
379 -- session.
380 requires :: a -> Maybe Extension
381 requires _ = Nothing
382
383 -- | Get sizes of overhead\/control\/payload byte sequences of
384 -- binary message representation without encoding message to binary
385 -- bytestring.
386 --
387 -- This function should obey one law:
388 --
389 -- * 'byteLength' ('stats' msg) == 'BL.length' ('encode' msg)
390 --
391 stats :: a -> ByteStats
392 stats _ = ByteStats 4 0 0
393
394{-----------------------------------------------------------------------
395-- Status messages
396-----------------------------------------------------------------------}
397
398-- | Notification that the sender have updated its
399-- 'Network.BitTorrent.Exchange.Status.PeerStatus'.
400data StatusUpdate
401 -- | Notification that the sender will not upload data to the
402 -- receiver until unchoking happen.
403 = Choking !Bool
404
405 -- | Notification that the sender is interested (or not interested)
406 -- in any of the receiver's data pieces.
407 | Interested !Bool
408 deriving (Show, Eq, Ord, Typeable)
409
410instance Pretty StatusUpdate where
411 pPrint (Choking False) = "not choking"
412 pPrint (Choking True ) = "choking"
413 pPrint (Interested False) = "not interested"
414 pPrint (Interested True ) = "interested"
415
416instance PeerMessage StatusUpdate where
417 envelop _ = Status
418 {-# INLINE envelop #-}
419
420 stats _ = ByteStats 4 1 0
421 {-# INLINE stats #-}
422
423{-----------------------------------------------------------------------
424-- Available messages
425-----------------------------------------------------------------------}
426
427-- | Messages used to inform receiver which pieces of the torrent
428-- sender have.
429data Available =
430 -- | Zero-based index of a piece that has just been successfully
431 -- downloaded and verified via the hash.
432 Have ! PieceIx
433
434 -- | The bitfield message may only be sent immediately after the
435 -- handshaking sequence is complete, and before any other message
436 -- are sent. If client have no pieces then bitfield need not to be
437 -- sent.
438 | Bitfield !Bitfield
439 deriving (Show, Eq)
440
441instance Pretty Available where
442 pPrint (Have ix ) = "Have" <+> int ix
443 pPrint (Bitfield _ ) = "Bitfield"
444
445instance PeerMessage Available where
446 envelop _ = Available
447 {-# INLINE envelop #-}
448
449 stats (Have _) = ByteStats (4 + 1) 4 0
450 stats (Bitfield bf) = ByteStats (4 + 1) (q + trailing) 0
451 where
452 trailing = if r == 0 then 0 else 1
453 (q, r) = quotRem (totalCount bf) 8
454
455{-----------------------------------------------------------------------
456-- Transfer messages
457-----------------------------------------------------------------------}
458
459-- | Messages used to transfer 'Block's.
460data Transfer
461 -- | Request for a particular block. If a client is requested a
462 -- block that another peer do not have the peer might not answer
463 -- at all.
464 = Request ! BlockIx
465
466 -- | Response to a request for a block.
467 | Piece !(Block BL.ByteString)
468
469 -- | Used to cancel block requests. It is typically used during
470 -- "End Game".
471 | Cancel !BlockIx
472 deriving (Show, Eq)
473
474instance Pretty Transfer where
475 pPrint (Request ix ) = "Request" <+> pPrint ix
476 pPrint (Piece blk) = "Piece" <+> pPrint blk
477 pPrint (Cancel i ) = "Cancel" <+> pPrint i
478
479instance PeerMessage Transfer where
480 envelop _ = Transfer
481 {-# INLINE envelop #-}
482
483 stats (Request _ ) = ByteStats (4 + 1) (3 * 4) 0
484 stats (Piece p ) = ByteStats (4 + 1) (4 + 4 + blockSize p) 0
485 stats (Cancel _ ) = ByteStats (4 + 1) (3 * 4) 0
486
487-- TODO increase
488-- | Max number of pending 'Request's inflight.
489defaultRequestQueueLength :: Int
490defaultRequestQueueLength = 1
491
492{-----------------------------------------------------------------------
493-- Fast messages
494-----------------------------------------------------------------------}
495
496-- | BEP6 messages.
497data FastMessage =
498 -- | If a peer have all pieces it might send the 'HaveAll' message
499 -- instead of 'Bitfield' message. Used to save bandwidth.
500 HaveAll
501
502 -- | If a peer have no pieces it might send 'HaveNone' message
503 -- intead of 'Bitfield' message. Used to save bandwidth.
504 | HaveNone
505
506 -- | This is an advisory message meaning "you might like to
507 -- download this piece." Used to avoid excessive disk seeks and
508 -- amount of IO.
509 | SuggestPiece !PieceIx
510
511 -- | Notifies a requesting peer that its request will not be
512 -- satisfied.
513 | RejectRequest !BlockIx
514
515 -- | This is an advisory messsage meaning \"if you ask for this
516 -- piece, I'll give it to you even if you're choked.\" Used to
517 -- shorten starting phase.
518 | AllowedFast !PieceIx
519 deriving (Show, Eq)
520
521instance Pretty FastMessage where
522 pPrint (HaveAll ) = "Have all"
523 pPrint (HaveNone ) = "Have none"
524 pPrint (SuggestPiece pix) = "Suggest" <+> int pix
525 pPrint (RejectRequest bix) = "Reject" <+> pPrint bix
526 pPrint (AllowedFast pix) = "Allowed fast" <+> int pix
527
528instance PeerMessage FastMessage where
529 envelop _ = Fast
530 {-# INLINE envelop #-}
531
532 requires _ = Just ExtFast
533 {-# INLINE requires #-}
534
535 stats HaveAll = ByteStats 4 1 0
536 stats HaveNone = ByteStats 4 1 0
537 stats (SuggestPiece _) = ByteStats 5 4 0
538 stats (RejectRequest _) = ByteStats 5 12 0
539 stats (AllowedFast _) = ByteStats 5 4 0
540
541{-----------------------------------------------------------------------
542-- Extension protocol
543-----------------------------------------------------------------------}
544
545{-----------------------------------------------------------------------
546-- Extended capabilities
547-----------------------------------------------------------------------}
548
549data ExtendedExtension
550 = ExtMetadata -- ^ BEP 9: Extension for Peers to Send Metadata Files
551 deriving (Show, Eq, Ord, Enum, Bounded, Typeable)
552
553instance IsString ExtendedExtension where
554 fromString = fromMaybe (error msg) . fromKey . fromString
555 where
556 msg = "fromString: could not parse ExtendedExtension"
557
558instance Pretty ExtendedExtension where
559 pPrint ExtMetadata = "Extension for Peers to Send Metadata Files"
560
561fromKey :: BKey -> Maybe ExtendedExtension
562fromKey "ut_metadata" = Just ExtMetadata
563fromKey _ = Nothing
564{-# INLINE fromKey #-}
565
566toKey :: ExtendedExtension -> BKey
567toKey ExtMetadata = "ut_metadata"
568{-# INLINE toKey #-}
569
570type ExtendedMessageId = Word8
571
572extId :: ExtendedExtension -> ExtendedMessageId
573extId ExtMetadata = 1
574{-# INLINE extId #-}
575
576type ExtendedMap = Map ExtendedExtension ExtendedMessageId
577
578-- | The extension IDs must be stored for every peer, because every
579-- peer may have different IDs for the same extension.
580--
581newtype ExtendedCaps = ExtendedCaps { extendedCaps :: ExtendedMap }
582 deriving (Show, Eq)
583
584instance Pretty ExtendedCaps where
585 pPrint = ppCaps
586 {-# INLINE pPrint #-}
587
588-- | The empty set.
589instance Default ExtendedCaps where
590 def = ExtendedCaps M.empty
591
592-- | Monoid under intersection:
593--
594-- * The 'mempty' caps includes all known extensions;
595--
596-- * the 'mappend' operation is NOT commutative: it return message
597-- id from the first caps for the extensions existing in both caps.
598--
599instance Monoid ExtendedCaps where
600 mempty = toCaps [minBound..maxBound]
601 mappend (ExtendedCaps a) (ExtendedCaps b) =
602 ExtendedCaps (M.intersection a b)
603
604appendBDict :: BDict -> ExtendedMap -> ExtendedMap
605appendBDict (Cons key val xs) caps
606 | Just ext <- fromKey key
607 , Right eid <- fromBEncode val = M.insert ext eid (appendBDict xs caps)
608 | otherwise = appendBDict xs caps
609appendBDict Nil caps = caps
610
611-- | Handshake compatible encoding.
612instance BEncode ExtendedCaps where
613 toBEncode = BDict . BE.fromAscList . L.sortBy (comparing fst)
614 . L.map (toKey *** toBEncode) . M.toList . extendedCaps
615
616 fromBEncode (BDict bd) = pure $ ExtendedCaps $ appendBDict bd M.empty
617 fromBEncode _ = decodingError "ExtendedCaps"
618
619instance Capabilities ExtendedCaps where
620 type Ext ExtendedCaps = ExtendedExtension
621
622 toCaps = ExtendedCaps . M.fromList . L.map (id &&& extId)
623
624 fromCaps = M.keys . extendedCaps
625 {-# INLINE fromCaps #-}
626
627 allowed e (ExtendedCaps caps) = M.member e caps
628 {-# INLINE allowed #-}
629
630remoteMessageId :: ExtendedExtension -> ExtendedCaps -> ExtendedMessageId
631remoteMessageId ext = fromMaybe (extId ext) . M.lookup ext . extendedCaps
632
633{-----------------------------------------------------------------------
634-- Extended handshake
635-----------------------------------------------------------------------}
636
637-- | This message should be sent immediately after the standard
638-- bittorrent handshake to any peer that supports this extension
639-- protocol. Extended handshakes can be sent more than once, however
640-- an implementation may choose to ignore subsequent handshake
641-- messages.
642--
643data ExtendedHandshake = ExtendedHandshake
644 { -- | If this peer has an IPv4 interface, this is the compact
645 -- representation of that address.
646 ehsIPv4 :: Maybe HostAddress
647
648 -- | If this peer has an IPv6 interface, this is the compact
649 -- representation of that address.
650 , ehsIPv6 :: Maybe HostAddress6
651
652 -- | Dictionary of supported extension messages which maps names
653 -- of extensions to an extended message ID for each extension
654 -- message.
655 , ehsCaps :: ExtendedCaps
656
657 -- | Size of 'Data.Torrent.InfoDict' in bytes. This field should
658 -- be added if 'ExtMetadata' is enabled in current session /and/
659 -- peer have the torrent file.
660 , ehsMetadataSize :: Maybe Int
661
662 -- | Local TCP /listen/ port. Allows each side to learn about the
663 -- TCP port number of the other side.
664 , ehsPort :: Maybe PortNumber
665
666 -- | Request queue the number of outstanding 'Request' messages
667 -- this client supports without dropping any.
668 , ehsQueueLength :: Maybe Int
669
670 -- | Client name and version.
671 , ehsVersion :: Maybe Text
672
673 -- | IP of the remote end
674 , ehsYourIp :: Maybe IP
675 } deriving (Show, Eq, Typeable)
676
677extHandshakeId :: ExtendedMessageId
678extHandshakeId = 0
679
680-- | Default 'Request' queue size.
681defaultQueueLength :: Int
682defaultQueueLength = 1
683
684-- | All fields are empty.
685instance Default ExtendedHandshake where
686 def = ExtendedHandshake def def def def def def def def
687
688instance Monoid ExtendedHandshake where
689 mempty = def { ehsCaps = mempty }
690 mappend old new = ExtendedHandshake {
691 ehsCaps = ehsCaps old <> ehsCaps new,
692 ehsIPv4 = ehsIPv4 old `mergeOld` ehsIPv4 new,
693 ehsIPv6 = ehsIPv6 old `mergeOld` ehsIPv6 new,
694 ehsMetadataSize = ehsMetadataSize old `mergeNew` ehsMetadataSize new,
695 ehsPort = ehsPort old `mergeOld` ehsPort new,
696 ehsQueueLength = ehsQueueLength old `mergeNew` ehsQueueLength new,
697 ehsVersion = ehsVersion old `mergeOld` ehsVersion new,
698 ehsYourIp = ehsYourIp old `mergeOld` ehsYourIp new
699 }
700 where
701 mergeOld mold mnew = mold <|> mnew
702 mergeNew mold mnew = mnew <|> mold
703
704
705instance BEncode ExtendedHandshake where
706 toBEncode ExtendedHandshake {..} = toDict $
707 "ipv4" .=? (S.encode <$> ehsIPv4)
708 .: "ipv6" .=? (S.encode <$> ehsIPv6)
709 .: "m" .=! ehsCaps
710 .: "metadata_size" .=? ehsMetadataSize
711 .: "p" .=? ehsPort
712 .: "reqq" .=? ehsQueueLength
713 .: "v" .=? ehsVersion
714 .: "yourip" .=? (runPut <$> either put put <$> toEither <$> ehsYourIp)
715 .: endDict
716 where
717 toEither (IPv4 v4) = Left v4
718 toEither (IPv6 v6) = Right v6
719
720 fromBEncode = fromDict $ ExtendedHandshake
721 <$>? "ipv4"
722 <*>? "ipv6"
723 <*>! "m"
724 <*>? "metadata_size"
725 <*>? "p"
726 <*>? "reqq"
727 <*>? "v"
728 <*> (opt "yourip" >>= getYourIp)
729
730getYourIp :: Maybe BValue -> BE.Get (Maybe IP)
731getYourIp f =
732 return $ do
733 BString ip <- f
734 either (const Nothing) Just $
735 case BS.length ip of
736 4 -> IPv4 <$> S.decode ip
737 16 -> IPv6 <$> S.decode ip
738 _ -> fail ""
739
740instance Pretty ExtendedHandshake where
741 pPrint = PP.text . show
742
743-- | NOTE: Approximated 'stats'.
744instance PeerMessage ExtendedHandshake where
745 envelop c = envelop c . EHandshake
746 {-# INLINE envelop #-}
747
748 requires _ = Just ExtExtended
749 {-# INLINE requires #-}
750
751 stats _ = ByteStats (4 + 1 + 1) 100 {- is it ok? -} 0 -- FIXME
752 {-# INLINE stats #-}
753
754-- | Set default values and the specified 'ExtendedCaps'.
755nullExtendedHandshake :: ExtendedCaps -> ExtendedHandshake
756nullExtendedHandshake caps = ExtendedHandshake
757 { ehsIPv4 = Nothing
758 , ehsIPv6 = Nothing
759 , ehsCaps = caps
760 , ehsMetadataSize = Nothing
761 , ehsPort = Nothing
762 , ehsQueueLength = Just defaultQueueLength
763 , ehsVersion = Just $ T.pack $ render $ pPrint libFingerprint
764 , ehsYourIp = Nothing
765 }
766
767{-----------------------------------------------------------------------
768-- Metadata exchange extension
769-----------------------------------------------------------------------}
770
771-- | A peer MUST verify that any piece it sends passes the info-hash
772-- verification. i.e. until the peer has the entire metadata, it
773-- cannot run SHA-1 to verify that it yields the same hash as the
774-- info-hash.
775--
776data ExtendedMetadata
777 -- | This message requests the a specified metadata piece. The
778 -- response to this message, from a peer supporting the extension,
779 -- is either a 'MetadataReject' or a 'MetadataData' message.
780 = MetadataRequest PieceIx
781
782 -- | If sender requested a valid 'PieceIx' and receiver have the
783 -- corresponding piece then receiver should respond with this
784 -- message.
785 | MetadataData
786 { -- | A piece of 'Data.Torrent.InfoDict'.
787 piece :: P.Piece BS.ByteString
788
789 -- | This key has the same semantics as the 'ehsMetadataSize' in
790 -- the 'ExtendedHandshake' — it is size of the torrent info
791 -- dict.
792 , totalSize :: Int
793 }
794
795 -- | Peers that do not have the entire metadata MUST respond with
796 -- a reject message to any metadata request.
797 --
798 -- Clients MAY implement flood protection by rejecting request
799 -- messages after a certain number of them have been
800 -- served. Typically the number of pieces of metadata times a
801 -- factor.
802 | MetadataReject PieceIx
803
804 -- | Reserved. By specification we should ignore unknown metadata
805 -- messages.
806 | MetadataUnknown BValue
807 deriving (Show, Eq, Typeable)
808
809-- | Extended metadata message id used in 'msg_type_key'.
810type MetadataId = Int
811
812msg_type_key, piece_key, total_size_key :: BKey
813msg_type_key = "msg_type"
814piece_key = "piece"
815total_size_key = "total_size"
816
817-- | BEP9 compatible encoding.
818instance BEncode ExtendedMetadata where
819 toBEncode (MetadataRequest pix) = toDict $
820 msg_type_key .=! (0 :: MetadataId)
821 .: piece_key .=! pix
822 .: endDict
823 toBEncode (MetadataData (P.Piece pix _) totalSize) = toDict $
824 msg_type_key .=! (1 :: MetadataId)
825 .: piece_key .=! pix
826 .: total_size_key .=! totalSize
827 .: endDict
828 toBEncode (MetadataReject pix) = toDict $
829 msg_type_key .=! (2 :: MetadataId)
830 .: piece_key .=! pix
831 .: endDict
832 toBEncode (MetadataUnknown bval) = bval
833
834 fromBEncode bval = (`fromDict` bval) $ do
835 mid <- field $ req msg_type_key
836 case mid :: MetadataId of
837 0 -> MetadataRequest <$>! piece_key
838 1 -> metadataData <$>! piece_key <*>! total_size_key
839 2 -> MetadataReject <$>! piece_key
840 _ -> pure (MetadataUnknown bval)
841 where
842 metadataData pix s = MetadataData (P.Piece pix BS.empty) s
843
844-- | Piece data bytes are omitted.
845instance Pretty ExtendedMetadata where
846 pPrint (MetadataRequest pix ) = "Request" <+> PP.int pix
847 pPrint (MetadataData p t) = "Data" <+> pPrint p <+> PP.int t
848 pPrint (MetadataReject pix ) = "Reject" <+> PP.int pix
849 pPrint (MetadataUnknown bval ) = "Unknown" <+> ppBEncode bval
850
851-- | NOTE: Approximated 'stats'.
852instance PeerMessage ExtendedMetadata where
853 envelop c = envelop c . EMetadata (remoteMessageId ExtMetadata c)
854 {-# INLINE envelop #-}
855
856 requires _ = Just ExtExtended
857 {-# INLINE requires #-}
858
859 stats (MetadataRequest _) = ByteStats (4 + 1 + 1) {- ~ -} 25 0
860 stats (MetadataData p _) = ByteStats (4 + 1 + 1) {- ~ -} 41 $
861 BS.length (P.pieceData p)
862 stats (MetadataReject _) = ByteStats (4 + 1 + 1) {- ~ -} 25 0
863 stats (MetadataUnknown _) = ByteStats (4 + 1 + 1) {- ? -} 0 0
864
865-- | All 'Piece's in 'MetadataData' messages MUST have size equal to
866-- this value. The last trailing piece can be shorter.
867metadataPieceSize :: PieceSize
868metadataPieceSize = 16 * 1024
869
870isLastPiece :: P.Piece a -> Int -> Bool
871isLastPiece P.Piece {..} total = succ pieceIndex == pcnt
872 where
873 pcnt = q + if r > 0 then 1 else 0
874 (q, r) = quotRem total metadataPieceSize
875
876-- TODO we can check if the piece payload bytestring have appropriate
877-- length; otherwise serialization MUST fail.
878isValidPiece :: P.Piece BL.ByteString -> Int -> Bool
879isValidPiece p @ P.Piece {..} total
880 | isLastPiece p total = pieceSize p <= metadataPieceSize
881 | otherwise = pieceSize p == metadataPieceSize
882
883setMetadataPayload :: BS.ByteString -> ExtendedMetadata -> ExtendedMetadata
884setMetadataPayload bs (MetadataData (P.Piece pix _) t) =
885 MetadataData (P.Piece pix bs) t
886setMetadataPayload _ msg = msg
887
888getMetadataPayload :: ExtendedMetadata -> Maybe BS.ByteString
889getMetadataPayload (MetadataData (P.Piece _ bs) _) = Just bs
890getMetadataPayload _ = Nothing
891
892-- | Metadata BDict usually contain only 'msg_type_key', 'piece_key'
893-- and 'total_size_key' fields so it normally should take less than
894-- 100 bytes. This limit is two order of magnitude larger to be
895-- permissive to 'MetadataUnknown' messages.
896--
897-- See 'maxMessageSize' for further explanation.
898--
899maxMetadataBDictSize :: Int
900maxMetadataBDictSize = 16 * 1024
901
902maxMetadataSize :: Int
903maxMetadataSize = maxMetadataBDictSize + metadataPieceSize
904
905-- to make MetadataData constructor fields a little bit prettier we
906-- cheat here: first we read empty 'pieceData' from bdict, but then we
907-- fill that field with the actual piece data — trailing bytes of
908-- the message
909getMetadata :: Int -> S.Get ExtendedMetadata
910getMetadata len
911 | len > maxMetadataSize = fail $ parseError "size exceeded limit"
912 | otherwise = do
913 bs <- getByteString len
914 parseRes $ BS.parse BE.parser bs
915 where
916 parseError reason = "unable to parse metadata message: " ++ reason
917
918 parseRes (BS.Fail _ _ m) = fail $ parseError $ "bdict: " ++ m
919 parseRes (BS.Partial _) = fail $ parseError "bdict: not enough bytes"
920 parseRes (BS.Done piece bvalueBS)
921 | BS.length piece > metadataPieceSize
922 = fail "infodict piece: size exceeded limit"
923 | otherwise = do
924 metadata <- either (fail . parseError) pure $ fromBEncode bvalueBS
925 return $ setMetadataPayload piece metadata
926
927putMetadata :: ExtendedMetadata -> BL.ByteString
928putMetadata msg
929 | Just bs <- getMetadataPayload msg = BE.encode msg <> BL.fromStrict bs
930 | otherwise = BE.encode msg
931
932-- | Allows a requesting peer to send 2 'MetadataRequest's for the
933-- each piece.
934--
935-- See 'Network.BitTorrent.Wire.Options.metadataFactor' for
936-- explanation why do we need this limit.
937defaultMetadataFactor :: Int
938defaultMetadataFactor = 2
939
940-- | Usually torrent size do not exceed 1MB. This value limit torrent
941-- /content/ size to about 8TB.
942--
943-- See 'Network.BitTorrent.Wire.Options.maxInfoDictSize' for
944-- explanation why do we need this limit.
945defaultMaxInfoDictSize :: Int
946defaultMaxInfoDictSize = 10 * 1024 * 1024
947
948{-----------------------------------------------------------------------
949-- Extension protocol messages
950-----------------------------------------------------------------------}
951
952-- | For more info see <http://www.bittorrent.org/beps/bep_0010.html>
953data ExtendedMessage
954 = EHandshake ExtendedHandshake
955 | EMetadata ExtendedMessageId ExtendedMetadata
956 | EUnknown ExtendedMessageId BS.ByteString
957 deriving (Show, Eq, Typeable)
958
959instance Pretty ExtendedMessage where
960 pPrint (EHandshake ehs) = pPrint ehs
961 pPrint (EMetadata _ msg) = "Metadata" <+> pPrint msg
962 pPrint (EUnknown mid _ ) = "Unknown" <+> PP.text (show mid)
963
964instance PeerMessage ExtendedMessage where
965 envelop _ = Extended
966 {-# INLINE envelop #-}
967
968 requires _ = Just ExtExtended
969 {-# INLINE requires #-}
970
971 stats (EHandshake hs) = stats hs
972 stats (EMetadata _ msg) = stats msg
973 stats (EUnknown _ msg) = ByteStats (4 + 1 + 1) (BS.length msg) 0
974
975{-----------------------------------------------------------------------
976-- The message datatype
977-----------------------------------------------------------------------}
978
979type MessageId = Word8
980
981-- | Messages used in communication between peers.
982--
983-- Note: If some extensions are disabled (not present in extension
984-- mask) and client receive message used by the disabled
985-- extension then the client MUST close the connection.
986--
987data Message
988 -- | Peers may close the TCP connection if they have not received
989 -- any messages for a given period of time, generally 2
990 -- minutes. Thus, the KeepAlive message is sent to keep the
991 -- connection between two peers alive, if no /other/ message has
992 -- been sent in a given period of time.
993 = KeepAlive
994 | Status !StatusUpdate -- ^ Messages used to update peer status.
995 | Available !Available -- ^ Messages used to inform availability.
996 | Transfer !Transfer -- ^ Messages used to transfer 'Block's.
997
998 -- | Peer receiving a handshake indicating the remote peer
999 -- supports the 'ExtDHT' should send a 'Port' message. Peers that
1000 -- receive this message should attempt to ping the node on the
1001 -- received port and IP address of the remote peer.
1002 | Port !PortNumber
1003 | Fast !FastMessage
1004 | Extended !ExtendedMessage
1005 deriving (Show, Eq)
1006
1007instance Default Message where
1008 def = KeepAlive
1009 {-# INLINE def #-}
1010
1011-- | Payload bytes are omitted.
1012instance Pretty Message where
1013 pPrint (KeepAlive ) = "Keep alive"
1014 pPrint (Status m) = "Status" <+> pPrint m
1015 pPrint (Available m) = pPrint m
1016 pPrint (Transfer m) = pPrint m
1017 pPrint (Port p) = "Port" <+> int (fromEnum p)
1018 pPrint (Fast m) = pPrint m
1019 pPrint (Extended m) = pPrint m
1020
1021instance PeerMessage Message where
1022 envelop _ = id
1023 {-# INLINE envelop #-}
1024
1025 requires KeepAlive = Nothing
1026 requires (Status _) = Nothing
1027 requires (Available _) = Nothing
1028 requires (Transfer _) = Nothing
1029 requires (Port _) = Just ExtDHT
1030 requires (Fast _) = Just ExtFast
1031 requires (Extended _) = Just ExtExtended
1032
1033 stats KeepAlive = ByteStats 4 0 0
1034 stats (Status m) = stats m
1035 stats (Available m) = stats m
1036 stats (Transfer m) = stats m
1037 stats (Port _) = ByteStats 5 2 0
1038 stats (Fast m) = stats m
1039 stats (Extended m) = stats m
1040
1041-- | PORT message.
1042instance PeerMessage PortNumber where
1043 envelop _ = Port
1044 {-# INLINE envelop #-}
1045
1046 requires _ = Just ExtDHT
1047 {-# INLINE requires #-}
1048
1049-- | How long /this/ peer should wait before dropping connection, in
1050-- seconds.
1051defaultKeepAliveTimeout :: Int
1052defaultKeepAliveTimeout = 2 * 60
1053
1054-- | How often /this/ peer should send 'KeepAlive' messages, in
1055-- seconds.
1056defaultKeepAliveInterval :: Int
1057defaultKeepAliveInterval = 60
1058
1059getInt :: S.Get Int
1060getInt = fromIntegral <$> S.getWord32be
1061{-# INLINE getInt #-}
1062
1063putInt :: S.Putter Int
1064putInt = S.putWord32be . fromIntegral
1065{-# INLINE putInt #-}
1066
1067-- | This limit should protect against "out-of-memory" attacks: if a
1068-- malicious peer have sent a long varlength message then receiver can
1069-- accumulate too long bytestring in the 'Get'.
1070--
1071-- Normal messages should never exceed this limits.
1072--
1073-- See also 'maxBitfieldSize', 'maxBlockSize' limits.
1074--
1075maxMessageSize :: Int
1076maxMessageSize = 20 + 1024 * 1024
1077
1078-- | This also limit max torrent size to:
1079--
1080-- max_bitfield_size * piece_ix_per_byte * max_piece_size =
1081-- 2 ^ 20 * 8 * 1MB =
1082-- 8TB
1083--
1084maxBitfieldSize :: Int
1085maxBitfieldSize = 1024 * 1024
1086
1087getBitfield :: Int -> S.Get Bitfield
1088getBitfield len
1089 | len > maxBitfieldSize = fail "BITFIELD message size exceeded limit"
1090 | otherwise = fromBitmap <$> getByteString len
1091
1092maxBlockSize :: Int
1093maxBlockSize = 4 * defaultTransferSize
1094
1095getBlock :: Int -> S.Get (Block BL.ByteString)
1096getBlock len
1097 | len > maxBlockSize = fail "BLOCK message size exceeded limit"
1098 | otherwise = Block <$> getInt <*> getInt
1099 <*> getLazyByteString (fromIntegral len)
1100{-# INLINE getBlock #-}
1101
1102instance Serialize Message where
1103 get = do
1104 len <- getInt
1105
1106 when (len > maxMessageSize) $ do
1107 fail "message body size exceeded the limit"
1108
1109 if len == 0 then return KeepAlive
1110 else do
1111 mid <- S.getWord8
1112 case mid of
1113 0x00 -> return $ Status (Choking True)
1114 0x01 -> return $ Status (Choking False)
1115 0x02 -> return $ Status (Interested True)
1116 0x03 -> return $ Status (Interested False)
1117 0x04 -> (Available . Have) <$> getInt
1118 0x05 -> (Available . Bitfield) <$> getBitfield (pred len)
1119 0x06 -> (Transfer . Request) <$> S.get
1120 0x07 -> (Transfer . Piece) <$> getBlock (len - 9)
1121 0x08 -> (Transfer . Cancel) <$> S.get
1122 0x09 -> Port <$> S.get
1123 0x0D -> (Fast . SuggestPiece) <$> getInt
1124 0x0E -> return $ Fast HaveAll
1125 0x0F -> return $ Fast HaveNone
1126 0x10 -> (Fast . RejectRequest) <$> S.get
1127 0x11 -> (Fast . AllowedFast) <$> getInt
1128 0x14 -> Extended <$> getExtendedMessage (pred len)
1129 _ -> do
1130 rm <- S.remaining >>= S.getBytes
1131 fail $ "unknown message ID: " ++ show mid ++ "\n"
1132 ++ "remaining available bytes: " ++ show rm
1133
1134 put KeepAlive = putInt 0
1135 put (Status msg) = putStatus msg
1136 put (Available msg) = putAvailable msg
1137 put (Transfer msg) = putTransfer msg
1138 put (Port p ) = putPort p
1139 put (Fast msg) = putFast msg
1140 put (Extended m ) = putExtendedMessage m
1141
1142statusUpdateId :: StatusUpdate -> MessageId
1143statusUpdateId (Choking choking) = fromIntegral (0 + fromEnum choking)
1144statusUpdateId (Interested choking) = fromIntegral (2 + fromEnum choking)
1145
1146putStatus :: Putter StatusUpdate
1147putStatus su = do
1148 putInt 1
1149 putWord8 (statusUpdateId su)
1150
1151putAvailable :: Putter Available
1152putAvailable (Have i) = do
1153 putInt 5
1154 putWord8 0x04
1155 putInt i
1156putAvailable (Bitfield (toBitmap -> bs)) = do
1157 putInt $ 1 + fromIntegral (BL.length bs)
1158 putWord8 0x05
1159 putLazyByteString bs
1160
1161putBlock :: Putter (Block BL.ByteString)
1162putBlock Block {..} = do
1163 putInt blkPiece
1164 putInt blkOffset
1165 putLazyByteString blkData
1166
1167putTransfer :: Putter Transfer
1168putTransfer (Request blk) = putInt 13 >> S.putWord8 0x06 >> S.put blk
1169putTransfer (Piece blk) = do
1170 putInt (9 + blockSize blk)
1171 putWord8 0x07
1172 putBlock blk
1173putTransfer (Cancel blk) = putInt 13 >> S.putWord8 0x08 >> S.put blk
1174
1175putPort :: Putter PortNumber
1176putPort p = do
1177 putInt 3
1178 putWord8 0x09
1179 put p
1180
1181putFast :: Putter FastMessage
1182putFast HaveAll = putInt 1 >> putWord8 0x0E
1183putFast HaveNone = putInt 1 >> putWord8 0x0F
1184putFast (SuggestPiece pix) = putInt 5 >> putWord8 0x0D >> putInt pix
1185putFast (RejectRequest i ) = putInt 13 >> putWord8 0x10 >> put i
1186putFast (AllowedFast i ) = putInt 5 >> putWord8 0x11 >> putInt i
1187
1188maxEHandshakeSize :: Int
1189maxEHandshakeSize = 16 * 1024
1190
1191getExtendedHandshake :: Int -> S.Get ExtendedHandshake
1192getExtendedHandshake messageSize
1193 | messageSize > maxEHandshakeSize
1194 = fail "extended handshake size exceeded limit"
1195 | otherwise = do
1196 bs <- getByteString messageSize
1197 either fail pure $ BE.decode bs
1198
1199maxEUnknownSize :: Int
1200maxEUnknownSize = 64 * 1024
1201
1202getExtendedUnknown :: Int -> S.Get BS.ByteString
1203getExtendedUnknown len
1204 | len > maxEUnknownSize = fail "unknown extended message size exceeded limit"
1205 | otherwise = getByteString len
1206
1207getExtendedMessage :: Int -> S.Get ExtendedMessage
1208getExtendedMessage messageSize = do
1209 msgId <- getWord8
1210 let msgBodySize = messageSize - 1
1211 case msgId of
1212 0 -> EHandshake <$> getExtendedHandshake msgBodySize
1213 1 -> EMetadata msgId <$> getMetadata msgBodySize
1214 _ -> EUnknown msgId <$> getExtendedUnknown msgBodySize
1215
1216-- | By spec.
1217extendedMessageId :: MessageId
1218extendedMessageId = 20
1219
1220putExt :: ExtendedMessageId -> BL.ByteString -> Put
1221putExt mid lbs = do
1222 putWord32be $ fromIntegral (1 + 1 + BL.length lbs)
1223 putWord8 extendedMessageId
1224 putWord8 mid
1225 putLazyByteString lbs
1226
1227-- NOTE: in contrast to getExtendedMessage this function put length
1228-- and message id too!
1229putExtendedMessage :: Putter ExtendedMessage
1230putExtendedMessage (EHandshake hs) = putExt extHandshakeId $ BE.encode hs
1231putExtendedMessage (EMetadata mid msg) = putExt mid $ putMetadata msg
1232putExtendedMessage (EUnknown mid bs) = putExt mid $ BL.fromStrict bs
diff --git a/src/Network/BitTorrent/Exchange/Session.hs b/src/Network/BitTorrent/Exchange/Session.hs
deleted file mode 100644
index 38a3c3a6..00000000
--- a/src/Network/BitTorrent/Exchange/Session.hs
+++ /dev/null
@@ -1,586 +0,0 @@
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE DeriveDataTypeable #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE StandaloneDeriving #-}
5{-# LANGUAGE TemplateHaskell #-}
6{-# LANGUAGE TypeFamilies #-}
7module Network.BitTorrent.Exchange.Session
8 ( -- * Session
9 Session
10 , Event (..)
11 , LogFun
12 , sessionLogger
13
14 -- * Construction
15 , newSession
16 , closeSession
17 , withSession
18
19 -- * Connection Set
20 , connect
21 , connectSink
22 , establish
23
24 -- * Query
25 , waitMetadata
26 , takeMetadata
27 ) where
28
29import Control.Applicative
30import Control.Concurrent
31import Control.Concurrent.Chan.Split as CS
32import Control.Concurrent.STM
33import Control.Exception hiding (Handler)
34import Control.Lens
35import Control.Monad as M
36import Control.Monad.Logger
37import Control.Monad.Reader
38import Data.ByteString as BS
39import Data.ByteString.Lazy as BL
40import Data.Conduit as C (Sink, awaitForever, (=$=), ($=))
41import qualified Data.Conduit as C
42import Data.Conduit.List as C
43import Data.Map as M
44import Data.Monoid
45import Data.Set as S
46import Data.Text as T
47import Data.Typeable
48import Text.PrettyPrint hiding ((<>))
49import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
50import System.Log.FastLogger (LogStr, ToLogStr (..))
51
52import Data.BEncode as BE
53import Data.Torrent as Torrent
54import Network.BitTorrent.Internal.Types
55import Network.Address
56import Network.BitTorrent.Exchange.Bitfield as BF
57import Network.BitTorrent.Exchange.Block as Block
58import Network.BitTorrent.Exchange.Connection
59import Network.BitTorrent.Exchange.Download as D
60import Network.BitTorrent.Exchange.Message as Message
61import System.Torrent.Storage
62
63#if !MIN_VERSION_iproute(1,2,12)
64deriving instance Ord IP
65#endif
66
67{-----------------------------------------------------------------------
68-- Exceptions
69-----------------------------------------------------------------------}
70
71data ExchangeError
72 = InvalidRequest BlockIx StorageFailure
73 | CorruptedPiece PieceIx
74 deriving (Show, Typeable)
75
76instance Exception ExchangeError
77
78packException :: Exception e => (e -> ExchangeError) -> IO a -> IO a
79packException f m = try m >>= either (throwIO . f) return
80
81{-----------------------------------------------------------------------
82-- Session state
83-----------------------------------------------------------------------}
84-- TODO unmap storage on zero connections
85
86data Cached a = Cached
87 { cachedValue :: !a
88 , cachedData :: BL.ByteString -- keep lazy
89 }
90
91cache :: BEncode a => a -> Cached a
92cache s = Cached s (BE.encode s)
93
94-- | Logger function.
95type LogFun = Loc -> LogSource -> LogLevel -> LogStr -> IO ()
96
97--data SessionStatus = Seeder | Leecher
98
99data SessionState
100 = WaitingMetadata
101 { metadataDownload :: MVar MetadataDownload
102 , metadataCompleted :: MVar InfoDict -- ^ used to unblock waiters
103 , contentRootPath :: FilePath
104 }
105 | HavingMetadata
106 { metadataCache :: Cached InfoDict
107 , contentDownload :: MVar ContentDownload
108 , contentStorage :: Storage
109 }
110
111newSessionState :: FilePath -> Either InfoHash InfoDict -> IO SessionState
112newSessionState rootPath (Left ih ) = do
113 WaitingMetadata <$> newMVar def <*> newEmptyMVar <*> pure rootPath
114newSessionState rootPath (Right dict) = do
115 storage <- openInfoDict ReadWriteEx rootPath dict
116 download <- newMVar $ D.contentDownload (BF.haveNone (totalPieces storage))
117 (piPieceLength (idPieceInfo dict))
118 storage
119 return $ HavingMetadata (cache dict) download storage
120
121closeSessionState :: SessionState -> IO ()
122closeSessionState WaitingMetadata {..} = return ()
123closeSessionState HavingMetadata {..} = close contentStorage
124
125haveMetadata :: InfoDict -> SessionState -> IO SessionState
126haveMetadata dict WaitingMetadata {..} = do
127 storage <- openInfoDict ReadWriteEx contentRootPath dict
128 download <- newMVar $ D.contentDownload (BF.haveNone (totalPieces storage))
129 (piPieceLength (idPieceInfo dict))
130 storage
131 return HavingMetadata
132 { metadataCache = cache dict
133 , contentDownload = download
134 , contentStorage = storage
135 }
136haveMetadata _ s = return s
137
138{-----------------------------------------------------------------------
139-- Session
140-----------------------------------------------------------------------}
141
142data Session = Session
143 { sessionPeerId :: !(PeerId)
144 , sessionTopic :: !(InfoHash)
145 , sessionLogger :: !(LogFun)
146 , sessionEvents :: !(SendPort (Event Session))
147
148 , sessionState :: !(MVar SessionState)
149
150------------------------------------------------------------------------
151 , connectionsPrefs :: !ConnectionPrefs
152
153 -- | Connections either waiting for TCP/uTP 'connect' or waiting
154 -- for BT handshake.
155 , connectionsPending :: !(TVar (Set (PeerAddr IP)))
156
157 -- | Connections successfully handshaked and data transfer can
158 -- take place.
159 , connectionsEstablished :: !(TVar (Map (PeerAddr IP) (Connection Session)))
160
161 -- | TODO implement choking mechanism
162 , connectionsUnchoked :: [PeerAddr IP]
163
164 -- | Messages written to this channel will be sent to the all
165 -- connections, including pending connections (but right after
166 -- handshake).
167 , connectionsBroadcast :: !(Chan Message)
168 }
169
170instance EventSource Session where
171 data Event Session
172 = ConnectingTo (PeerAddr IP)
173 | ConnectionEstablished (PeerAddr IP)
174 | ConnectionAborted
175 | ConnectionClosed (PeerAddr IP)
176 | SessionClosed
177 deriving Show
178
179 listen Session {..} = CS.listen sessionEvents
180
181newSession :: LogFun
182 -> PeerAddr (Maybe IP) -- ^ /external/ address of this peer;
183 -> FilePath -- ^ root directory for content files;
184 -> Either InfoHash InfoDict -- ^ torrent info dictionary;
185 -> IO Session
186newSession logFun addr rootPath source = do
187 let ih = either id idInfoHash source
188 pid <- maybe genPeerId return (peerId addr)
189 eventStream <- newSendPort
190 sState <- newSessionState rootPath source
191 sStateVar <- newMVar sState
192 pSetVar <- newTVarIO S.empty
193 eSetVar <- newTVarIO M.empty
194 chan <- newChan
195 return Session
196 { sessionPeerId = pid
197 , sessionTopic = ih
198 , sessionLogger = logFun
199 , sessionEvents = eventStream
200 , sessionState = sStateVar
201 , connectionsPrefs = def
202 , connectionsPending = pSetVar
203 , connectionsEstablished = eSetVar
204 , connectionsUnchoked = []
205 , connectionsBroadcast = chan
206 }
207
208closeSession :: Session -> IO ()
209closeSession Session {..} = do
210 s <- readMVar sessionState
211 closeSessionState s
212{-
213 hSet <- atomically $ do
214 pSet <- swapTVar connectionsPending S.empty
215 eSet <- swapTVar connectionsEstablished S.empty
216 return pSet
217 mapM_ kill hSet
218-}
219
220withSession :: ()
221withSession = error "withSession"
222
223{-----------------------------------------------------------------------
224-- Logging
225-----------------------------------------------------------------------}
226
227instance MonadLogger (Connected Session) where
228 monadLoggerLog loc src lvl msg = do
229 conn <- ask
230 ses <- asks connSession
231 addr <- asks connRemoteAddr
232 let addrSrc = src <> " @ " <> T.pack (render (pPrint addr))
233 liftIO $ sessionLogger ses loc addrSrc lvl (toLogStr msg)
234
235logMessage :: MonadLogger m => Message -> m ()
236logMessage msg = logDebugN $ T.pack (render (pPrint msg))
237
238logEvent :: MonadLogger m => Text -> m ()
239logEvent = logInfoN
240
241{-----------------------------------------------------------------------
242-- Connection set
243-----------------------------------------------------------------------}
244--- Connection status transition:
245---
246--- pending -> established -> finished -> closed
247--- | \|/ /|\
248--- \-------------------------------------|
249---
250--- Purpose of slots:
251--- 1) to avoid duplicates
252--- 2) connect concurrently
253---
254
255-- | Add connection to the pending set.
256pendingConnection :: PeerAddr IP -> Session -> STM Bool
257pendingConnection addr Session {..} = do
258 pSet <- readTVar connectionsPending
259 eSet <- readTVar connectionsEstablished
260 if (addr `S.member` pSet) || (addr `M.member` eSet)
261 then return False
262 else do
263 modifyTVar' connectionsPending (S.insert addr)
264 return True
265
266-- | Pending connection successfully established, add it to the
267-- established set.
268establishedConnection :: Connected Session ()
269establishedConnection = do
270 conn <- ask
271 addr <- asks connRemoteAddr
272 Session {..} <- asks connSession
273 liftIO $ atomically $ do
274 modifyTVar connectionsPending (S.delete addr)
275 modifyTVar connectionsEstablished (M.insert addr conn)
276
277-- | Either this or remote peer decided to finish conversation
278-- (conversation is alread /established/ connection), remote it from
279-- the established set.
280finishedConnection :: Connected Session ()
281finishedConnection = do
282 Session {..} <- asks connSession
283 addr <- asks connRemoteAddr
284 liftIO $ atomically $ do
285 modifyTVar connectionsEstablished $ M.delete addr
286
287-- | There are no state for this connection, remove it from the all
288-- sets.
289closedConnection :: PeerAddr IP -> Session -> STM ()
290closedConnection addr Session {..} = do
291 modifyTVar connectionsPending $ S.delete addr
292 modifyTVar connectionsEstablished $ M.delete addr
293
294getConnectionConfig :: Session -> IO (ConnectionConfig Session)
295getConnectionConfig s @ Session {..} = do
296 chan <- dupChan connectionsBroadcast
297 let sessionLink = SessionLink {
298 linkTopic = sessionTopic
299 , linkPeerId = sessionPeerId
300 , linkMetadataSize = Nothing
301 , linkOutputChan = Just chan
302 , linkSession = s
303 }
304 return ConnectionConfig
305 { cfgPrefs = connectionsPrefs
306 , cfgSession = sessionLink
307 , cfgWire = mainWire
308 }
309
310type Finalizer = IO ()
311type Runner = (ConnectionConfig Session -> IO ())
312
313runConnection :: Runner -> Finalizer -> PeerAddr IP -> Session -> IO ()
314runConnection runner finalize addr set @ Session {..} = do
315 _ <- forkIO (action `finally` cleanup)
316 return ()
317 where
318 action = do
319 notExist <- atomically $ pendingConnection addr set
320 when notExist $ do
321 cfg <- getConnectionConfig set
322 runner cfg
323
324 cleanup = do
325 finalize
326-- runStatusUpdates status (SS.resetPending addr)
327 -- TODO Metata.resetPending addr
328 atomically $ closedConnection addr set
329
330-- | Establish connection from scratch. If this endpoint is already
331-- connected, no new connections is created. This function do not block.
332connect :: PeerAddr IP -> Session -> IO ()
333connect addr = runConnection (connectWire addr) (return ()) addr
334
335-- | Establish connection with already pre-connected endpoint. If this
336-- endpoint is already connected, no new connections is created. This
337-- function do not block.
338--
339-- 'PendingConnection' will be closed automatically, you do not need
340-- to call 'closePending'.
341establish :: PendingConnection -> Session -> IO ()
342establish conn = runConnection (acceptWire conn) (closePending conn)
343 (pendingPeer conn)
344
345-- | Conduit version of 'connect'.
346connectSink :: MonadIO m => Session -> Sink [PeerAddr IPv4] m ()
347connectSink s = C.mapM_ (liftIO . connectBatch)
348 where
349 connectBatch = M.mapM_ (\ addr -> connect (IPv4 <$> addr) s)
350
351-- | Why do we need this message?
352type BroadcastMessage = ExtendedCaps -> Message
353
354broadcast :: BroadcastMessage -> Session -> IO ()
355broadcast = error "broadcast"
356
357{-----------------------------------------------------------------------
358-- Helpers
359-----------------------------------------------------------------------}
360
361waitMVar :: MVar a -> IO ()
362waitMVar m = withMVar m (const (return ()))
363
364-- This function appear in new GHC "out of box". (moreover it is atomic)
365tryReadMVar :: MVar a -> IO (Maybe a)
366tryReadMVar m = do
367 ma <- tryTakeMVar m
368 maybe (return ()) (putMVar m) ma
369 return ma
370
371readBlock :: BlockIx -> Storage -> IO (Block BL.ByteString)
372readBlock bix @ BlockIx {..} s = do
373 p <- packException (InvalidRequest bix) $ do readPiece ixPiece s
374 let chunk = BL.take (fromIntegral ixLength) $
375 BL.drop (fromIntegral ixOffset) (pieceData p)
376 if BL.length chunk == fromIntegral ixLength
377 then return $ Block ixPiece ixOffset chunk
378 else throwIO $ InvalidRequest bix (InvalidSize ixLength)
379
380-- |
381tryReadMetadataBlock :: PieceIx
382 -> Connected Session (Maybe (Torrent.Piece BS.ByteString, Int))
383tryReadMetadataBlock pix = do
384 Session {..} <- asks connSession
385 s <- liftIO (readMVar sessionState)
386 case s of
387 WaitingMetadata {..} -> error "tryReadMetadataBlock"
388 HavingMetadata {..} -> error "tryReadMetadataBlock"
389
390sendBroadcast :: PeerMessage msg => msg -> Wire Session ()
391sendBroadcast msg = do
392 Session {..} <- asks connSession
393 error "sendBroadcast"
394-- liftIO $ msg `broadcast` sessionConnections
395
396waitMetadata :: Session -> IO InfoDict
397waitMetadata Session {..} = do
398 s <- readMVar sessionState
399 case s of
400 WaitingMetadata {..} -> readMVar metadataCompleted
401 HavingMetadata {..} -> return (cachedValue metadataCache)
402
403takeMetadata :: Session -> IO (Maybe InfoDict)
404takeMetadata Session {..} = do
405 s <- readMVar sessionState
406 case s of
407 WaitingMetadata {..} -> return Nothing
408 HavingMetadata {..} -> return (Just (cachedValue metadataCache))
409
410{-----------------------------------------------------------------------
411-- Triggers
412-----------------------------------------------------------------------}
413
414-- | Trigger is the reaction of a handler at some event.
415type Trigger = Wire Session ()
416
417interesting :: Trigger
418interesting = do
419 addr <- asks connRemoteAddr
420 sendMessage (Interested True)
421 sendMessage (Choking False)
422 tryFillRequestQueue
423
424fillRequestQueue :: Trigger
425fillRequestQueue = do
426 maxN <- lift getMaxQueueLength
427 rbf <- use connBitfield
428 addr <- asks connRemoteAddr
429-- blks <- withStatusUpdates $ do
430-- n <- getRequestQueueLength addr
431-- scheduleBlocks addr rbf (maxN - n)
432-- mapM_ (sendMessage . Request) blks
433 return ()
434
435tryFillRequestQueue :: Trigger
436tryFillRequestQueue = do
437 allowed <- canDownload <$> use connStatus
438 when allowed $ do
439 fillRequestQueue
440
441{-----------------------------------------------------------------------
442-- Incoming message handling
443-----------------------------------------------------------------------}
444
445type Handler msg = msg -> Wire Session ()
446
447handleStatus :: Handler StatusUpdate
448handleStatus s = do
449 connStatus %= over remoteStatus (updateStatus s)
450 case s of
451 Interested _ -> return ()
452 Choking True -> do
453 addr <- asks connRemoteAddr
454-- withStatusUpdates (SS.resetPending addr)
455 return ()
456 Choking False -> tryFillRequestQueue
457
458handleAvailable :: Handler Available
459handleAvailable msg = do
460 connBitfield %= case msg of
461 Have ix -> BF.insert ix
462 Bitfield bf -> const bf
463
464 --thisBf <- getThisBitfield
465 thisBf <- undefined
466 case msg of
467 Have ix
468 | ix `BF.member` thisBf -> return ()
469 | otherwise -> interesting
470 Bitfield bf
471 | bf `BF.isSubsetOf` thisBf -> return ()
472 | otherwise -> interesting
473
474handleTransfer :: Handler Transfer
475handleTransfer (Request bix) = do
476 Session {..} <- asks connSession
477 s <- liftIO $ readMVar sessionState
478 case s of
479 WaitingMetadata {..} -> return ()
480 HavingMetadata {..} -> do
481 bitfield <- undefined -- getThisBitfield
482 upload <- canUpload <$> use connStatus
483 when (upload && ixPiece bix `BF.member` bitfield) $ do
484 blk <- liftIO $ readBlock bix contentStorage
485 sendMessage (Message.Piece blk)
486
487handleTransfer (Message.Piece blk) = do
488 Session {..} <- asks connSession
489 s <- liftIO $ readMVar sessionState
490 case s of
491 WaitingMetadata {..} -> return () -- TODO (?) break connection
492 HavingMetadata {..} -> do
493 isSuccess <- undefined -- withStatusUpdates (SS.pushBlock blk storage)
494 case isSuccess of
495 Nothing -> liftIO $ throwIO $ userError "block is not requested"
496 Just isCompleted -> do
497 when isCompleted $ do
498 sendBroadcast (Have (blkPiece blk))
499-- maybe send not interested
500 tryFillRequestQueue
501
502handleTransfer (Cancel bix) = filterQueue (not . (transferResponse bix))
503 where
504 transferResponse bix (Transfer (Message.Piece blk)) = blockIx blk == bix
505 transferResponse _ _ = False
506
507{-----------------------------------------------------------------------
508-- Metadata exchange
509-----------------------------------------------------------------------}
510-- TODO introduce new metadata exchange specific exceptions
511
512waitForMetadata :: Trigger
513waitForMetadata = do
514 Session {..} <- asks connSession
515 needFetch <- undefined --liftIO (isEmptyMVar infodict)
516 when needFetch $ do
517 canFetch <- allowed ExtMetadata <$> use connExtCaps
518 if canFetch
519 then tryRequestMetadataBlock
520 else undefined -- liftIO (waitMVar infodict)
521
522tryRequestMetadataBlock :: Trigger
523tryRequestMetadataBlock = do
524 mpix <- lift $ undefined --withMetadataUpdates Metadata.scheduleBlock
525 case mpix of
526 Nothing -> error "tryRequestMetadataBlock"
527 Just pix -> sendMessage (MetadataRequest pix)
528
529handleMetadata :: Handler ExtendedMetadata
530handleMetadata (MetadataRequest pix) =
531 lift (tryReadMetadataBlock pix) >>= sendMessage . mkResponse
532 where
533 mkResponse Nothing = MetadataReject pix
534 mkResponse (Just (piece, total)) = MetadataData piece total
535
536handleMetadata (MetadataData {..}) = do
537 ih <- asks connTopic
538 mdict <- lift $ undefined --withMetadataUpdates (Metadata.pushBlock piece ih)
539 case mdict of
540 Nothing -> tryRequestMetadataBlock -- not completed, need all blocks
541 Just dict -> do -- complete, wake up payload fetch
542 Session {..} <- asks connSession
543 liftIO $ modifyMVar_ sessionState (haveMetadata dict)
544
545handleMetadata (MetadataReject pix) = do
546 lift $ undefined -- withMetadataUpdates (Metadata.cancelPending pix)
547
548handleMetadata (MetadataUnknown _ ) = do
549 logInfoN "Unknown metadata message"
550
551{-----------------------------------------------------------------------
552-- Main entry point
553-----------------------------------------------------------------------}
554
555acceptRehandshake :: ExtendedHandshake -> Trigger
556acceptRehandshake ehs = error "acceptRehandshake"
557
558handleExtended :: Handler ExtendedMessage
559handleExtended (EHandshake ehs) = acceptRehandshake ehs
560handleExtended (EMetadata _ msg) = handleMetadata msg
561handleExtended (EUnknown _ _ ) = logWarnN "Unknown extension message"
562
563handleMessage :: Handler Message
564handleMessage KeepAlive = return ()
565handleMessage (Status s) = handleStatus s
566handleMessage (Available msg) = handleAvailable msg
567handleMessage (Transfer msg) = handleTransfer msg
568handleMessage (Port n) = error "handleMessage"
569handleMessage (Fast _) = error "handleMessage"
570handleMessage (Extended msg) = handleExtended msg
571
572exchange :: Wire Session ()
573exchange = do
574 waitForMetadata
575 bf <- undefined --getThisBitfield
576 sendMessage (Bitfield bf)
577 awaitForever handleMessage
578
579mainWire :: Wire Session ()
580mainWire = do
581 lift establishedConnection
582 Session {..} <- asks connSession
583-- lift $ resizeBitfield (totalPieces storage)
584 logEvent "Connection established"
585 iterM logMessage =$= exchange =$= iterM logMessage
586 lift finishedConnection
diff --git a/src/Network/BitTorrent/Internal/Cache.hs b/src/Network/BitTorrent/Internal/Cache.hs
deleted file mode 100644
index 8c74467a..00000000
--- a/src/Network/BitTorrent/Internal/Cache.hs
+++ /dev/null
@@ -1,169 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2014
3-- License : BSD
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- Cached data for tracker responses.
9--
10module Network.BitTorrent.Internal.Cache
11 ( -- * Cache
12 Cached
13 , lastUpdated
14 , updateInterval
15 , minUpdateInterval
16
17 -- * Construction
18 , newCached
19 , newCached_
20
21 -- * Query
22 , isAlive
23 , isStalled
24 , isExpired
25 , canUpdate
26 , shouldUpdate
27
28 -- * Cached data
29 , tryTakeData
30 , unsafeTryTakeData
31 , takeData
32 ) where
33
34import Control.Applicative
35import Data.Monoid
36import Data.Default
37import Data.Time
38import Data.Time.Clock.POSIX
39import System.IO.Unsafe
40
41
42data Cached a = Cached
43 { -- | Time of resource creation.
44 lastUpdated :: !POSIXTime
45
46 -- | Minimum invalidation timeout.
47 , minUpdateInterval :: !NominalDiffTime
48
49 -- | Resource lifetime.
50 , updateInterval :: !NominalDiffTime
51
52 -- | Resource data.
53 , cachedData :: a
54 } deriving (Show, Eq)
55
56-- INVARIANT: minUpdateInterval <= updateInterval
57
58instance Default (Cached a) where
59 def = mempty
60
61instance Functor Cached where
62 fmap f (Cached t i m a) = Cached t i m (f a)
63
64posixEpoch :: NominalDiffTime
65posixEpoch = 1000000000000000000000000000000000000000000000000000000
66
67instance Applicative Cached where
68 pure = Cached 0 posixEpoch posixEpoch
69 f <*> c = Cached
70 { lastUpdated = undefined
71 , minUpdateInterval = undefined
72 , updateInterval = undefined
73 , cachedData = cachedData f (cachedData c)
74 }
75
76instance Alternative Cached where
77 empty = mempty
78 (<|>) = error "cached alternative instance: not implemented"
79
80instance Monad Cached where
81 return = pure
82 Cached {..} >>= f = Cached
83 { lastUpdated = undefined
84 , updateInterval = undefined
85 , minUpdateInterval = undefined
86 , cachedData = undefined
87 }
88
89instance Monoid (Cached a) where
90 mempty = Cached
91 { lastUpdated = 0
92 , minUpdateInterval = 0
93 , updateInterval = 0
94 , cachedData = error "cached mempty: impossible happen"
95 }
96
97 mappend a b
98 | expirationTime a > expirationTime b = a
99 | otherwise = b
100
101normalize :: NominalDiffTime -> NominalDiffTime
102 -> (NominalDiffTime, NominalDiffTime)
103normalize a b
104 | a < b = (a, b)
105 | otherwise = (b, a)
106{-# INLINE normalize #-}
107
108newCached :: NominalDiffTime -> NominalDiffTime -> a -> IO (Cached a)
109newCached minInterval interval x = do
110 t <- getPOSIXTime
111 let (mui, ui) = normalize minInterval interval
112 return Cached
113 { lastUpdated = t
114 , minUpdateInterval = mui
115 , updateInterval = ui
116 , cachedData = x
117 }
118
119newCached_ :: NominalDiffTime -> a -> IO (Cached a)
120newCached_ interval x = newCached interval interval x
121{-# INLINE newCached_ #-}
122
123expirationTime :: Cached a -> POSIXTime
124expirationTime Cached {..} = undefined
125
126isAlive :: Cached a -> IO Bool
127isAlive Cached {..} = do
128 currentTime <- getPOSIXTime
129 return $ lastUpdated + updateInterval > currentTime
130
131isExpired :: Cached a -> IO Bool
132isExpired Cached {..} = undefined
133
134isStalled :: Cached a -> IO Bool
135isStalled Cached {..} = undefined
136
137canUpdate :: Cached a -> IO (Maybe NominalDiffTime)
138canUpdate = undefined --isStaled
139
140shouldUpdate :: Cached a -> IO (Maybe NominalDiffTime)
141shouldUpdate = undefined -- isExpired
142
143tryTakeData :: Cached a -> IO (Maybe a)
144tryTakeData c = do
145 alive <- isAlive c
146 return $ if alive then Just (cachedData c) else Nothing
147
148unsafeTryTakeData :: Cached a -> Maybe a
149unsafeTryTakeData = unsafePerformIO . tryTakeData
150
151invalidateData :: Cached a -> IO a -> IO (Cached a)
152invalidateData Cached {..} action = do
153 t <- getPOSIXTime
154 x <- action
155 return Cached
156 { lastUpdated = t
157 , updateInterval = updateInterval
158 , minUpdateInterval = minUpdateInterval
159 , cachedData = x
160 }
161
162takeData :: Cached a -> IO a -> IO a
163takeData c action = do
164 mdata <- tryTakeData c
165 case mdata of
166 Just a -> return a
167 Nothing -> do
168 c' <- invalidateData c action
169 takeData c' action
diff --git a/src/Network/BitTorrent/Internal/Progress.hs b/src/Network/BitTorrent/Internal/Progress.hs
deleted file mode 100644
index 6ac889e2..00000000
--- a/src/Network/BitTorrent/Internal/Progress.hs
+++ /dev/null
@@ -1,154 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- 'Progress' used to track amount downloaded\/left\/upload bytes
9-- either on per client or per torrent basis. This value is used to
10-- notify the tracker and usually shown to the user. To aggregate
11-- total progress you can use the Monoid instance.
12--
13{-# LANGUAGE TemplateHaskell #-}
14{-# LANGUAGE ViewPatterns #-}
15{-# OPTIONS -fno-warn-orphans #-}
16module Network.BitTorrent.Internal.Progress
17 ( -- * Progress
18 Progress (..)
19
20 -- * Lens
21 , left
22 , uploaded
23 , downloaded
24
25 -- * Construction
26 , startProgress
27 , downloadedProgress
28 , enqueuedProgress
29 , uploadedProgress
30 , dequeuedProgress
31
32 -- * Query
33 , canDownload
34 , canUpload
35 ) where
36
37import Control.Applicative
38import Control.Lens hiding ((%=))
39import Data.ByteString.Lazy.Builder as BS
40import Data.ByteString.Lazy.Builder.ASCII as BS
41import Data.Default
42import Data.Monoid
43import Data.Serialize as S
44import Data.Ratio
45import Data.Word
46import Network.HTTP.Types.QueryLike
47import Text.PrettyPrint as PP
48import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
49
50
51-- | Progress data is considered as dynamic within one client
52-- session. This data also should be shared across client application
53-- sessions (e.g. files), otherwise use 'startProgress' to get initial
54-- 'Progress' value.
55--
56data Progress = Progress
57 { _downloaded :: {-# UNPACK #-} !Word64 -- ^ Total amount of bytes downloaded;
58 , _left :: {-# UNPACK #-} !Word64 -- ^ Total amount of bytes left;
59 , _uploaded :: {-# UNPACK #-} !Word64 -- ^ Total amount of bytes uploaded.
60 } deriving (Show, Read, Eq)
61
62$(makeLenses ''Progress)
63
64-- | UDP tracker compatible encoding.
65instance Serialize Progress where
66 put Progress {..} = do
67 putWord64be $ fromIntegral _downloaded
68 putWord64be $ fromIntegral _left
69 putWord64be $ fromIntegral _uploaded
70
71 get = Progress
72 <$> (fromIntegral <$> getWord64be)
73 <*> (fromIntegral <$> getWord64be)
74 <*> (fromIntegral <$> getWord64be)
75
76instance Default Progress where
77 def = Progress 0 0 0
78 {-# INLINE def #-}
79
80-- | Can be used to aggregate total progress.
81instance Monoid Progress where
82 mempty = def
83 {-# INLINE mempty #-}
84
85 mappend (Progress da la ua) (Progress db lb ub) = Progress
86 { _downloaded = da + db
87 , _left = la + lb
88 , _uploaded = ua + ub
89 }
90 {-# INLINE mappend #-}
91
92instance QueryValueLike Builder where
93 toQueryValue = toQueryValue . BS.toLazyByteString
94
95instance QueryValueLike Word64 where
96 toQueryValue = toQueryValue . BS.word64Dec
97
98-- | HTTP Tracker protocol compatible encoding.
99instance QueryLike Progress where
100 toQuery Progress {..} =
101 [ ("uploaded" , toQueryValue _uploaded)
102 , ("left" , toQueryValue _left)
103 , ("downloaded", toQueryValue _downloaded)
104 ]
105
106instance Pretty Progress where
107 pPrint Progress {..} =
108 "/\\" <+> PP.text (show _uploaded) $$
109 "\\/" <+> PP.text (show _downloaded) $$
110 "left" <+> PP.text (show _left)
111
112-- | Initial progress is used when there are no session before.
113--
114-- Please note that tracker might penalize client some way if the do
115-- not accumulate progress. If possible and save 'Progress' between
116-- client sessions to avoid that.
117--
118startProgress :: Integer -> Progress
119startProgress = Progress 0 0 . fromIntegral
120{-# INLINE startProgress #-}
121
122-- | Used when the client download some data from /any/ peer.
123downloadedProgress :: Int -> Progress -> Progress
124downloadedProgress (fromIntegral -> amount)
125 = (left -~ amount)
126 . (downloaded +~ amount)
127{-# INLINE downloadedProgress #-}
128
129-- | Used when the client upload some data to /any/ peer.
130uploadedProgress :: Int -> Progress -> Progress
131uploadedProgress (fromIntegral -> amount) = uploaded +~ amount
132{-# INLINE uploadedProgress #-}
133
134-- | Used when leecher join client session.
135enqueuedProgress :: Integer -> Progress -> Progress
136enqueuedProgress amount = left +~ fromIntegral amount
137{-# INLINE enqueuedProgress #-}
138
139-- | Used when leecher leave client session.
140-- (e.g. user deletes not completed torrent)
141dequeuedProgress :: Integer -> Progress -> Progress
142dequeuedProgress amount = left -~ fromIntegral amount
143{-# INLINE dequeuedProgress #-}
144
145ri2rw64 :: Ratio Int -> Ratio Word64
146ri2rw64 x = fromIntegral (numerator x) % fromIntegral (denominator x)
147
148-- | Check global /download/ limit by uploaded \/ downloaded ratio.
149canDownload :: Ratio Int -> Progress -> Bool
150canDownload limit Progress {..} = _uploaded % _downloaded > ri2rw64 limit
151
152-- | Check global /upload/ limit by downloaded \/ uploaded ratio.
153canUpload :: Ratio Int -> Progress -> Bool
154canUpload limit Progress {..} = _downloaded % _uploaded > ri2rw64 limit
diff --git a/src/Network/BitTorrent/Internal/Types.hs b/src/Network/BitTorrent/Internal/Types.hs
deleted file mode 100644
index d157db3e..00000000
--- a/src/Network/BitTorrent/Internal/Types.hs
+++ /dev/null
@@ -1,10 +0,0 @@
1{-# LANGUAGE TypeFamilies #-}
2module Network.BitTorrent.Internal.Types
3 ( EventSource (..)
4 ) where
5
6import Control.Concurrent.Chan.Split
7
8class EventSource source where
9 data Event source
10 listen :: source -> IO (ReceivePort (Event source))
diff --git a/src/Network/BitTorrent/Readme.md b/src/Network/BitTorrent/Readme.md
deleted file mode 100644
index ebf9545e..00000000
--- a/src/Network/BitTorrent/Readme.md
+++ /dev/null
@@ -1,10 +0,0 @@
1Layout
2======
3
4| module group | can import | main purpose |
5|:-------------|:------------:|:--------------------------------------:|
6| Core | | common datatypes |
7| DHT | Core | centralized peer discovery |
8| Tracker | Core | decentralized peer discovery |
9| Exchange | Core | torrent content exchange |
10| Client | any other | core of bittorrent client application |
diff --git a/src/Network/BitTorrent/Tracker.hs b/src/Network/BitTorrent/Tracker.hs
deleted file mode 100644
index 6db67559..00000000
--- a/src/Network/BitTorrent/Tracker.hs
+++ /dev/null
@@ -1,50 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : non-portable
7--
8-- This module provides high level API for peer -> tracker
9-- communication. Tracker is used to discover other peers in the
10-- network using torrent info hash.
11--
12{-# LANGUAGE TemplateHaskell #-}
13module Network.BitTorrent.Tracker
14 ( -- * RPC Manager
15 PeerInfo (..)
16 , Options
17 , Manager
18 , newManager
19 , closeManager
20 , withManager
21
22 -- * Multitracker session
23 , trackerList
24 , Session
25 , Event (..)
26 , newSession
27 , closeSession
28 , withSession
29
30 -- ** Events
31 , AnnounceEvent (..)
32 , notify
33 , askPeers
34
35 -- ** Session state
36 , TrackerSession
37 , trackerPeers
38 , trackerScrape
39
40 , tryTakeData
41 , unsafeTryTakeData
42
43 , getSessionState
44 ) where
45
46import Network.BitTorrent.Internal.Cache (tryTakeData, unsafeTryTakeData)
47import Network.BitTorrent.Tracker.Message
48import Network.BitTorrent.Tracker.List
49import Network.BitTorrent.Tracker.RPC
50import Network.BitTorrent.Tracker.Session
diff --git a/src/Network/BitTorrent/Tracker/List.hs b/src/Network/BitTorrent/Tracker/List.hs
deleted file mode 100644
index 0eb11641..00000000
--- a/src/Network/BitTorrent/Tracker/List.hs
+++ /dev/null
@@ -1,193 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2014
3-- License : BSD
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- Multitracker Metadata Extension support.
9--
10-- For more info see: <http://www.bittorrent.org/beps/bep_0012.html>
11--
12{-# LANGUAGE FlexibleInstances #-}
13module Network.BitTorrent.Tracker.List
14 ( -- * Tracker list
15 TierEntry
16 , TrackerList
17
18 -- * Construction
19 , trackerList
20 , shuffleTiers
21 , mapWithURI
22 , Network.BitTorrent.Tracker.List.toList
23
24 -- * Traversals
25 , traverseAll
26 , traverseTiers
27 ) where
28
29import Prelude hiding (mapM, foldr)
30import Control.Arrow
31import Control.Applicative
32import Control.Exception
33import Data.Default
34import Data.List as L (map, elem, any, filter, null)
35import Data.Maybe
36import Data.Foldable
37import Data.Traversable
38import Network.URI
39import System.Random.Shuffle
40
41import Data.Torrent
42import Network.BitTorrent.Tracker.RPC as RPC
43
44{-----------------------------------------------------------------------
45-- Tracker list datatype
46-----------------------------------------------------------------------}
47
48type TierEntry a = (URI, a)
49type Tier a = [TierEntry a]
50
51-- | Tracker list is either a single tracker or list of tiers. All
52-- trackers in each tier must be checked before the client goes on to
53-- the next tier.
54data TrackerList a
55 = Announce (TierEntry a) -- ^ torrent file 'announce' field only
56 | TierList [Tier a] -- ^ torrent file 'announce-list' field only
57 deriving (Show, Eq)
58
59-- | Empty tracker list. Can be used for trackerless torrents.
60instance Default (TrackerList a) where
61 def = TierList []
62
63instance Functor TrackerList where
64 fmap f (Announce (uri, a)) = Announce (uri, f a)
65 fmap f (TierList a) = TierList (fmap (fmap (second f)) a)
66
67instance Foldable TrackerList where
68 foldr f z (Announce e ) = f (snd e) z
69 foldr f z (TierList xs) = foldr (flip (foldr (f . snd))) z xs
70
71_traverseEntry f (uri, a) = (,) uri <$> f a
72
73instance Traversable TrackerList where
74 traverse f (Announce e ) = Announce <$> _traverseEntry f e
75 traverse f (TierList xs) =
76 TierList <$> traverse (traverse (_traverseEntry f)) xs
77
78traverseWithURI :: Applicative f
79 => (TierEntry a -> f b) -> TrackerList a -> f (TrackerList b)
80traverseWithURI f (Announce (uri, a)) = (Announce . (,) uri) <$> f (uri, a)
81traverseWithURI f (TierList xxs ) =
82 TierList <$> traverse (traverse (traverseEntry f)) xxs
83 where
84 traverseEntry f (uri, a) = (,) uri <$> f (uri, a)
85
86{-----------------------------------------------------------------------
87-- List extraction
88-----------------------------------------------------------------------}
89-- BEP12 do not expose any restrictions for the content of
90-- 'announce-list' key - there are some /bad/ cases can happen with
91-- poorly designed or even malicious torrent creation software.
92--
93-- Bad case #1: announce-list is present, but empty.
94--
95-- { tAnnounce = Just "http://a.com"
96-- , tAnnounceList = Just [[]]
97-- }
98--
99-- Bad case #2: announce uri do not present in announce list.
100--
101-- { tAnnounce = Just "http://a.com"
102-- , tAnnounceList = Just [["udp://a.com"]]
103-- }
104--
105-- The addBackup function solves both problems by adding announce uri
106-- as backup tier.
107--
108addBackup :: [[URI]] -> URI -> [[URI]]
109addBackup tiers bkp
110 | L.any (L.elem bkp) tiers = tiers
111 | otherwise = tiers ++ [[bkp]]
112
113fixList :: Maybe [[URI]] -> Maybe URI -> Maybe [[URI]]
114fixList mxss mx = do
115 xss <- mxss
116 let xss' = L.filter (not . L.null) xss
117 return $ maybe xss' (addBackup xss') mx
118
119-- | Extract set of trackers from torrent file. The 'tAnnounce' key is
120-- only ignored if the 'tAnnounceList' key is present.
121trackerList :: Torrent -> TrackerList ()
122trackerList Torrent {..} = fromMaybe (TierList []) $ do
123 (TierList . tierList) <$> (tAnnounceList `fixList` tAnnounce)
124 <|> (Announce . nullEntry) <$> tAnnounce
125 where
126 nullEntry uri = (uri, ())
127 tierList = L.map (L.map nullEntry)
128
129-- | Shuffle /order of trackers/ in each tier, preserving original
130-- /order of tiers/. This can help to balance the load between the
131-- trackers.
132shuffleTiers :: TrackerList a -> IO (TrackerList a)
133shuffleTiers (Announce a ) = return (Announce a)
134shuffleTiers (TierList xs) = TierList <$> mapM shuffleM xs
135
136mapWithURI :: (URI -> a -> b) -> TrackerList a -> TrackerList b
137mapWithURI f (Announce (uri, a)) = Announce (uri, f uri a)
138mapWithURI f (TierList xs ) = TierList (L.map (L.map mapEntry) xs)
139 where
140 mapEntry (uri, a) = (uri, f uri a)
141
142toList :: TrackerList a -> [[TierEntry a]]
143toList (Announce e) = [[e]]
144toList (TierList xxs) = xxs
145
146{-----------------------------------------------------------------------
147-- Special traversals (suppressed RPC exceptions)
148-----------------------------------------------------------------------}
149
150catchRPC :: IO a -> IO a -> IO a
151catchRPC a b = catch a (f b)
152 where
153 f :: a -> RpcException -> a
154 f = const
155
156throwRPC :: String -> IO a
157throwRPC = throwIO . GenericException
158
159-- | Like 'traverse' but ignores 'RpcExceptions'.
160traverseAll :: (TierEntry a -> IO a) -> TrackerList a -> IO (TrackerList a)
161traverseAll action = traverseWithURI (action $?)
162 where
163 f $? x = catchRPC (f x) (return (snd x))
164
165-- | Like 'traverse' but put working trackers to the head of tiers.
166-- This can help to avoid exceessive requests to not available
167-- trackers at each reannounce. If no one action succeed then original
168-- list is returned.
169traverseTiers :: (TierEntry a -> IO a) -> TrackerList a -> IO (TrackerList a)
170traverseTiers action ts = catchRPC (goList ts) (return ts)
171 where
172 goList tl @ (Announce _ ) = traverseWithURI action tl
173 goList (TierList tiers) = TierList <$> goTiers (goTier []) tiers
174
175 goTiers _ [] = throwRPC "traverseTiers: no tiers"
176 goTiers f (x : xs) = catchRPC shortcut failback
177 where
178 shortcut = do
179 x' <- f x
180 return (x' : xs)
181
182 failback = do
183 xs' <- goTiers f xs
184 return (x : xs')
185
186 goTier _ [] = throwRPC "traverseTiers: no trackers in tier"
187 goTier failed ((uri, a) : as) = catchRPC shortcut failback
188 where
189 shortcut = do
190 a' <- action (uri, a)
191 return ((uri, a') : as ++ failed) -- failed trackers at the end
192
193 failback = goTier ((uri, a) : failed) as
diff --git a/src/Network/BitTorrent/Tracker/Message.hs b/src/Network/BitTorrent/Tracker/Message.hs
deleted file mode 100644
index b9b6a9d3..00000000
--- a/src/Network/BitTorrent/Tracker/Message.hs
+++ /dev/null
@@ -1,920 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- (c) Daniel Gröber 2013
4-- License : BSD3
5-- Maintainer : pxqr.sta@gmail.com
6-- Stability : experimental
7-- Portability : portable
8--
9-- Every tracker should support announce query. This query is used
10-- to discover peers within a swarm and have two-fold effect:
11--
12-- * peer doing announce discover other peers using peer list from
13-- the response to the announce query.
14--
15-- * tracker store peer information and use it in the succeeding
16-- requests made by other peers, until the peer info expires.
17--
18-- By convention most trackers support another form of request —
19-- scrape query — which queries the state of a given torrent (or
20-- a list of torrents) that the tracker is managing.
21--
22{-# LANGUAGE FlexibleContexts #-}
23{-# LANGUAGE FlexibleInstances #-}
24{-# LANGUAGE GeneralizedNewtypeDeriving #-}
25{-# LANGUAGE TemplateHaskell #-}
26{-# LANGUAGE DeriveDataTypeable #-}
27{-# LANGUAGE DeriveFunctor #-}
28{-# LANGUAGE ScopedTypeVariables #-}
29{-# LANGUAGE TypeFamilies #-}
30{-# OPTIONS -fno-warn-orphans #-}
31module Network.BitTorrent.Tracker.Message
32 ( -- * Announce
33 -- ** Query
34 AnnounceEvent (..)
35 , AnnounceQuery (..)
36 , renderAnnounceQuery
37 , ParamParseFailure
38 , parseAnnounceQuery
39
40 -- ** Info
41 , PeerList (..)
42 , getPeerList
43 , AnnounceInfo(..)
44 , defaultNumWant
45 , defaultMaxNumWant
46 , defaultReannounceInterval
47
48 -- * Scrape
49 -- ** Query
50 , ScrapeQuery
51 , renderScrapeQuery
52 , parseScrapeQuery
53
54 -- ** Info
55 , ScrapeEntry (..)
56 , ScrapeInfo
57
58 -- * HTTP specific
59 -- ** Routes
60 , PathPiece
61 , defaultAnnouncePath
62 , defaultScrapePath
63
64 -- ** Preferences
65 , AnnouncePrefs (..)
66 , renderAnnouncePrefs
67 , parseAnnouncePrefs
68
69 -- ** Request
70 , AnnounceRequest (..)
71 , parseAnnounceRequest
72 , renderAnnounceRequest
73
74 -- ** Response
75 , announceType
76 , scrapeType
77 , parseFailureStatus
78
79 -- ** Extra
80 , queryToSimpleQuery
81
82 -- * UDP specific
83 -- ** Connection
84 , ConnectionId
85 , initialConnectionId
86
87 -- ** Messages
88 , Request (..)
89 , Response (..)
90 , responseName
91
92 -- ** Transaction
93 , genTransactionId
94 , TransactionId
95 , Transaction (..)
96 )
97 where
98
99import Control.Applicative
100import Control.Monad
101import Data.BEncode as BE hiding (Result)
102import Data.BEncode.BDict as BE
103import Data.ByteString as BS
104import Data.ByteString.Char8 as BC
105import Data.Char as Char
106import Data.Convertible
107import Data.Default
108import Data.Either
109import Data.List as L
110import Data.Maybe
111import Data.Monoid
112import Data.Serialize as S hiding (Result)
113import Data.String
114import Data.Text (Text)
115import Data.Text.Encoding
116import Data.Typeable
117import Data.Word
118import Data.IP
119import Network
120import Network.HTTP.Types.QueryLike
121import Network.HTTP.Types.URI hiding (urlEncode)
122import Network.HTTP.Types.Status
123import Network.Socket hiding (Connected)
124import Numeric
125import System.Entropy
126import Text.Read (readMaybe)
127
128import Data.Torrent
129import Network.Address
130import Network.BitTorrent.Internal.Progress
131
132{-----------------------------------------------------------------------
133-- Events
134-----------------------------------------------------------------------}
135
136-- | Events are used to specify which kind of announce query is performed.
137data AnnounceEvent
138 -- | For the first request: when download first begins.
139 = Started
140
141 -- | This peer stopped downloading /and/ uploading the torrent or
142 -- just shutting down.
143 | Stopped
144
145 -- | This peer completed downloading the torrent. This only happen
146 -- right after last piece have been verified. No 'Completed' is
147 -- sent if the file was completed when 'Started'.
148 | Completed
149 deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable)
150
151-- | HTTP tracker protocol compatible encoding.
152instance QueryValueLike AnnounceEvent where
153 toQueryValue e = toQueryValue (Char.toLower x : xs)
154 where
155 (x : xs) = show e -- INVARIANT: this is always nonempty list
156
157type EventId = Word32
158
159-- | UDP tracker encoding event codes.
160eventId :: AnnounceEvent -> EventId
161eventId Completed = 1
162eventId Started = 2
163eventId Stopped = 3
164
165-- TODO add Regular event
166putEvent :: Putter (Maybe AnnounceEvent)
167putEvent Nothing = putWord32be 0
168putEvent (Just e) = putWord32be (eventId e)
169
170getEvent :: S.Get (Maybe AnnounceEvent)
171getEvent = do
172 eid <- getWord32be
173 case eid of
174 0 -> return Nothing
175 1 -> return $ Just Completed
176 2 -> return $ Just Started
177 3 -> return $ Just Stopped
178 _ -> fail "unknown event id"
179
180{-----------------------------------------------------------------------
181 Announce query
182-----------------------------------------------------------------------}
183-- TODO add &ipv6= and &ipv4= params to AnnounceQuery
184-- http://www.bittorrent.org/beps/bep_0007.html#announce-parameter
185
186-- | A tracker request is HTTP GET request; used to include metrics
187-- from clients that help the tracker keep overall statistics about
188-- the torrent. The most important, requests are used by the tracker
189-- to keep track lists of active peer for a particular torrent.
190--
191data AnnounceQuery = AnnounceQuery
192 {
193 -- | Hash of info part of the torrent usually obtained from
194 -- 'Torrent' or 'Magnet'.
195 reqInfoHash :: !InfoHash
196
197 -- | ID of the peer doing request.
198 , reqPeerId :: !PeerId
199
200 -- | Port to listen to for connections from other
201 -- peers. Tracker should respond with this port when
202 -- some /other/ peer request the tracker with the same info hash.
203 -- Normally, this port is choosed from 'defaultPorts'.
204 , reqPort :: !PortNumber
205
206 -- | Current progress of peer doing request.
207 , reqProgress :: !Progress
208
209 -- | The peer IP. Needed only when client communicated with
210 -- tracker throught a proxy.
211 , reqIP :: Maybe HostAddress
212
213 -- | Number of peers that the peers wants to receive from. It is
214 -- optional for trackers to honor this limit. See note for
215 -- 'defaultNumWant'.
216 , reqNumWant :: Maybe Int
217
218 -- | If not specified, the request is regular periodic
219 -- request. Regular request should be sent
220 , reqEvent :: Maybe AnnounceEvent
221 } deriving (Show, Eq, Typeable)
222
223-- | UDP tracker protocol compatible encoding.
224instance Serialize AnnounceQuery where
225 put AnnounceQuery {..} = do
226 put reqInfoHash
227 put reqPeerId
228 put reqProgress
229 putEvent reqEvent
230 putWord32host $ fromMaybe 0 reqIP
231 putWord32be $ 0 -- TODO what the fuck is "key"?
232 putWord32be $ fromIntegral $ fromMaybe (-1) reqNumWant
233
234 put reqPort
235
236 get = do
237 ih <- get
238 pid <- get
239
240 progress <- get
241
242 ev <- getEvent
243 ip <- getWord32be
244-- key <- getWord32be -- TODO
245 want <- getWord32be
246
247 port <- get
248
249 return $ AnnounceQuery {
250 reqInfoHash = ih
251 , reqPeerId = pid
252 , reqPort = port
253 , reqProgress = progress
254 , reqIP = if ip == 0 then Nothing else Just ip
255 , reqNumWant = if want == -1 then Nothing
256 else Just (fromIntegral want)
257 , reqEvent = ev
258 }
259
260instance QueryValueLike PortNumber where
261 toQueryValue = toQueryValue . show . fromEnum
262
263instance QueryValueLike Word32 where
264 toQueryValue = toQueryValue . show
265
266instance QueryValueLike Int where
267 toQueryValue = toQueryValue . show
268
269-- | HTTP tracker protocol compatible encoding.
270instance QueryLike AnnounceQuery where
271 toQuery AnnounceQuery {..} =
272 toQuery reqProgress ++
273 [ ("info_hash", toQueryValue reqInfoHash) -- TODO use 'paramName'
274 , ("peer_id" , toQueryValue reqPeerId)
275 , ("port" , toQueryValue reqPort)
276 , ("ip" , toQueryValue reqIP)
277 , ("numwant" , toQueryValue reqNumWant)
278 , ("event" , toQueryValue reqEvent)
279 ]
280
281-- | Filter @param=value@ pairs with the unset value.
282queryToSimpleQuery :: Query -> SimpleQuery
283queryToSimpleQuery = catMaybes . L.map f
284 where
285 f (_, Nothing) = Nothing
286 f (a, Just b ) = Just (a, b)
287
288-- | Encode announce query to query string.
289renderAnnounceQuery :: AnnounceQuery -> SimpleQuery
290renderAnnounceQuery = queryToSimpleQuery . toQuery
291
292data QueryParam
293 -- announce query
294 = ParamInfoHash
295 | ParamPeerId
296 | ParamPort
297 | ParamUploaded
298 | ParamLeft
299 | ParamDownloaded
300 | ParamIP
301 | ParamNumWant
302 | ParamEvent
303 -- announce query ext
304 | ParamCompact
305 | ParamNoPeerId
306 deriving (Show, Eq, Ord, Enum)
307
308paramName :: QueryParam -> BS.ByteString
309paramName ParamInfoHash = "info_hash"
310paramName ParamPeerId = "peer_id"
311paramName ParamPort = "port"
312paramName ParamUploaded = "uploaded"
313paramName ParamLeft = "left"
314paramName ParamDownloaded = "downloaded"
315paramName ParamIP = "ip"
316paramName ParamNumWant = "numwant"
317paramName ParamEvent = "event"
318paramName ParamCompact = "compact"
319paramName ParamNoPeerId = "no_peer_id"
320{-# INLINE paramName #-}
321
322class FromParam a where
323 fromParam :: BS.ByteString -> Maybe a
324
325instance FromParam Bool where
326 fromParam "0" = Just False
327 fromParam "1" = Just True
328 fromParam _ = Nothing
329
330instance FromParam InfoHash where
331 fromParam = either (const Nothing) pure . safeConvert
332
333instance FromParam PeerId where
334 fromParam = either (const Nothing) pure . safeConvert
335
336instance FromParam Word32 where
337 fromParam = readMaybe . BC.unpack
338
339instance FromParam Word64 where
340 fromParam = readMaybe . BC.unpack
341
342instance FromParam Int where
343 fromParam = readMaybe . BC.unpack
344
345instance FromParam PortNumber where
346 fromParam bs = fromIntegral <$> (fromParam bs :: Maybe Word32)
347
348instance FromParam AnnounceEvent where
349 fromParam bs = do
350 (x, xs) <- BC.uncons bs
351 readMaybe $ BC.unpack $ BC.cons (Char.toUpper x) xs
352
353-- | 'ParamParseFailure' represent errors can occur while parsing HTTP
354-- tracker requests. In case of failure, this can be used to provide
355-- more informative 'statusCode' and 'statusMessage' in tracker
356-- responses.
357--
358data ParamParseFailure
359 = Missing QueryParam -- ^ param not found in query string;
360 | Invalid QueryParam BS.ByteString -- ^ param present but not valid.
361 deriving (Show, Eq)
362
363type ParseResult = Either ParamParseFailure
364
365withError :: ParamParseFailure -> Maybe a -> ParseResult a
366withError e = maybe (Left e) Right
367
368reqParam :: FromParam a => QueryParam -> SimpleQuery -> ParseResult a
369reqParam param xs = do
370 val <- withError (Missing param) $ L.lookup (paramName param) xs
371 withError (Invalid param val) (fromParam val)
372
373optParam :: FromParam a => QueryParam -> SimpleQuery -> ParseResult (Maybe a)
374optParam param ps
375 | Just x <- L.lookup (paramName param) ps
376 = pure <$> withError (Invalid param x) (fromParam x)
377 | otherwise = pure Nothing
378
379parseProgress :: SimpleQuery -> ParseResult Progress
380parseProgress params = Progress
381 <$> reqParam ParamDownloaded params
382 <*> reqParam ParamLeft params
383 <*> reqParam ParamUploaded params
384
385-- | Parse announce request from a query string.
386parseAnnounceQuery :: SimpleQuery -> ParseResult AnnounceQuery
387parseAnnounceQuery params = AnnounceQuery
388 <$> reqParam ParamInfoHash params
389 <*> reqParam ParamPeerId params
390 <*> reqParam ParamPort params
391 <*> parseProgress params
392 <*> optParam ParamIP params
393 <*> optParam ParamNumWant params
394 <*> optParam ParamEvent params
395
396{-----------------------------------------------------------------------
397-- Announce Info
398-----------------------------------------------------------------------}
399-- TODO check if announceinterval/complete/incomplete is positive ints
400
401-- | Tracker can return peer list in either compact(BEP23) or not
402-- compact form.
403--
404-- For more info see: <http://www.bittorrent.org/beps/bep_0023.html>
405--
406data PeerList ip
407 = PeerList [PeerAddr IP]
408 | CompactPeerList [PeerAddr ip]
409 deriving (Show, Eq, Typeable, Functor)
410
411-- | The empty non-compact peer list.
412instance Default (PeerList IP) where
413 def = PeerList []
414 {-# INLINE def #-}
415
416getPeerList :: PeerList IP -> [PeerAddr IP]
417getPeerList (PeerList xs) = xs
418getPeerList (CompactPeerList xs) = xs
419
420instance Serialize a => BEncode (PeerList a) where
421 toBEncode (PeerList xs) = toBEncode xs
422 toBEncode (CompactPeerList xs) = toBEncode $ runPut (mapM_ put xs)
423
424 fromBEncode (BList l ) = PeerList <$> fromBEncode (BList l)
425 fromBEncode (BString s ) = CompactPeerList <$> runGet (many get) s
426 fromBEncode _ = decodingError "PeerList: should be a BString or BList"
427
428-- | The tracker response includes a peer list that helps the client
429-- participate in the torrent. The most important is 'respPeer' list
430-- used to join the swarm.
431--
432data AnnounceInfo =
433 Failure !Text -- ^ Failure reason in human readable form.
434 | AnnounceInfo {
435 -- | Number of peers completed the torrent. (seeders)
436 respComplete :: !(Maybe Int)
437
438 -- | Number of peers downloading the torrent. (leechers)
439 , respIncomplete :: !(Maybe Int)
440
441 -- | Recommended interval to wait between requests, in seconds.
442 , respInterval :: !Int
443
444 -- | Minimal amount of time between requests, in seconds. A
445 -- peer /should/ make timeout with at least 'respMinInterval'
446 -- value, otherwise tracker might not respond. If not specified
447 -- the same applies to 'respInterval'.
448 , respMinInterval :: !(Maybe Int)
449
450 -- | Peers that must be contacted.
451 , respPeers :: !(PeerList IP)
452
453 -- | Human readable warning.
454 , respWarning :: !(Maybe Text)
455 } deriving (Show, Eq, Typeable)
456
457-- | Empty peer list with default reannounce interval.
458instance Default AnnounceInfo where
459 def = AnnounceInfo
460 { respComplete = Nothing
461 , respIncomplete = Nothing
462 , respInterval = defaultReannounceInterval
463 , respMinInterval = Nothing
464 , respPeers = def
465 , respWarning = Nothing
466 }
467
468-- | HTTP tracker protocol compatible encoding.
469instance BEncode AnnounceInfo where
470 toBEncode (Failure t) = toDict $
471 "failure reason" .=! t
472 .: endDict
473
474 toBEncode AnnounceInfo {..} = toDict $
475 "complete" .=? respComplete
476 .: "incomplete" .=? respIncomplete
477 .: "interval" .=! respInterval
478 .: "min interval" .=? respMinInterval
479 .: "peers" .=! peers
480 .: "peers6" .=? peers6
481 .: "warning message" .=? respWarning
482 .: endDict
483 where
484 (peers, peers6) = prttn respPeers
485
486 prttn :: PeerList IP -> (PeerList IPv4, Maybe (PeerList IPv6))
487 prttn (PeerList xs) = (PeerList xs, Nothing)
488 prttn (CompactPeerList xs) = mk $ partitionEithers $ toEither <$> xs
489 where
490 mk (v4s, v6s)
491 | L.null v6s = (CompactPeerList v4s, Nothing)
492 | otherwise = (CompactPeerList v4s, Just (CompactPeerList v6s))
493
494 toEither :: PeerAddr IP -> Either (PeerAddr IPv4) (PeerAddr IPv6)
495 toEither PeerAddr {..} = case peerHost of
496 IPv4 ipv4 -> Left $ PeerAddr peerId ipv4 peerPort
497 IPv6 ipv6 -> Right $ PeerAddr peerId ipv6 peerPort
498
499 fromBEncode (BDict d)
500 | Just t <- BE.lookup "failure reason" d = Failure <$> fromBEncode t
501 | otherwise = (`fromDict` (BDict d)) $
502 AnnounceInfo
503 <$>? "complete"
504 <*>? "incomplete"
505 <*>! "interval"
506 <*>? "min interval"
507 <*> (uncurry merge =<< (,) <$>! "peers" <*>? "peers6")
508 <*>? "warning message"
509 where
510 merge :: PeerList IPv4 -> Maybe (PeerList IPv6) -> BE.Get (PeerList IP)
511 merge (PeerList ips) Nothing = pure (PeerList ips)
512 merge (PeerList _ ) (Just _)
513 = fail "PeerList: non-compact peer list provided, \
514 \but the `peers6' field present"
515
516 merge (CompactPeerList ipv4s) Nothing
517 = pure $ CompactPeerList (fmap IPv4 <$> ipv4s)
518
519 merge (CompactPeerList _ ) (Just (PeerList _))
520 = fail "PeerList: the `peers6' field value \
521 \should contain *compact* peer list"
522
523 merge (CompactPeerList ipv4s) (Just (CompactPeerList ipv6s))
524 = pure $ CompactPeerList $
525 (fmap IPv4 <$> ipv4s) <> (fmap IPv6 <$> ipv6s)
526
527 fromBEncode _ = decodingError "Announce info"
528
529-- | UDP tracker protocol compatible encoding.
530instance Serialize AnnounceInfo where
531 put (Failure msg) = put $ encodeUtf8 msg
532 put AnnounceInfo {..} = do
533 putWord32be $ fromIntegral respInterval
534 putWord32be $ fromIntegral $ fromMaybe 0 respIncomplete
535 putWord32be $ fromIntegral $ fromMaybe 0 respComplete
536 forM_ (fmap ipv4 <$> getPeerList respPeers) put
537
538 get = do
539 interval <- getWord32be
540 leechers <- getWord32be
541 seeders <- getWord32be
542 peers <- many $ fmap IPv4 <$> get
543
544 return $ AnnounceInfo {
545 respWarning = Nothing
546 , respInterval = fromIntegral interval
547 , respMinInterval = Nothing
548 , respIncomplete = Just $ fromIntegral leechers
549 , respComplete = Just $ fromIntegral seeders
550 , respPeers = PeerList peers
551 }
552
553-- | Decodes announce response from bencoded string, for debugging only.
554instance IsString AnnounceInfo where
555 fromString str = either (error . format) id $ BE.decode (fromString str)
556 where
557 format msg = "fromString: unable to decode AnnounceInfo: " ++ msg
558
559-- | Above 25, new peers are highly unlikely to increase download
560-- speed. Even 30 peers is /plenty/, the official client version 3
561-- in fact only actively forms new connections if it has less than
562-- 30 peers and will refuse connections if it has 55.
563--
564-- <https://wiki.theory.org/BitTorrent_Tracker_Protocol#Basic_Tracker_Announce_Request>
565--
566defaultNumWant :: Int
567defaultNumWant = 50
568
569-- | Reasonable upper bound of numwant parameter.
570defaultMaxNumWant :: Int
571defaultMaxNumWant = 200
572
573-- | Widely used reannounce interval. Note: tracker clients should not
574-- use this value!
575defaultReannounceInterval :: Int
576defaultReannounceInterval = 30 * 60
577
578{-----------------------------------------------------------------------
579 Scrape message
580-----------------------------------------------------------------------}
581
582-- | Scrape query used to specify a set of torrent to scrape.
583-- If list is empty then tracker should return scrape info about each
584-- torrent.
585type ScrapeQuery = [InfoHash]
586
587-- TODO
588-- data ScrapeQuery
589-- = ScrapeAll
590-- | ScrapeSingle InfoHash
591-- | ScrapeMulti (HashSet InfoHash)
592-- deriving (Show)
593--
594-- data ScrapeInfo
595-- = ScrapeAll (HashMap InfoHash ScrapeEntry)
596-- | ScrapeSingle InfoHash ScrapeEntry
597-- | ScrapeMulti (HashMap InfoHash ScrapeEntry)
598--
599
600scrapeParam :: BS.ByteString
601scrapeParam = "info_hash"
602
603isScrapeParam :: BS.ByteString -> Bool
604isScrapeParam = (==) scrapeParam
605
606-- | Parse scrape query to query string.
607parseScrapeQuery :: SimpleQuery -> ScrapeQuery
608parseScrapeQuery
609 = catMaybes . L.map (fromParam . snd) . L.filter (isScrapeParam . fst)
610
611-- | Render scrape query to query string.
612renderScrapeQuery :: ScrapeQuery -> SimpleQuery
613renderScrapeQuery = queryToSimpleQuery . L.map mkPair
614 where
615 mkPair ih = (scrapeParam, toQueryValue ih)
616
617-- | Overall information about particular torrent.
618data ScrapeEntry = ScrapeEntry {
619 -- | Number of seeders - peers with the entire file.
620 siComplete :: {-# UNPACK #-} !Int
621
622 -- | Total number of times the tracker has registered a completion.
623 , siDownloaded :: {-# UNPACK #-} !Int
624
625 -- | Number of leechers.
626 , siIncomplete :: {-# UNPACK #-} !Int
627
628 -- | Name of the torrent file, as specified by the "name"
629 -- file in the info section of the .torrent file.
630 , siName :: !(Maybe Text)
631 } deriving (Show, Eq, Typeable)
632
633-- | HTTP tracker protocol compatible encoding.
634instance BEncode ScrapeEntry where
635 toBEncode ScrapeEntry {..} = toDict $
636 "complete" .=! siComplete
637 .: "downloaded" .=! siDownloaded
638 .: "incomplete" .=! siIncomplete
639 .: "name" .=? siName
640 .: endDict
641
642 fromBEncode = fromDict $ ScrapeEntry
643 <$>! "complete"
644 <*>! "downloaded"
645 <*>! "incomplete"
646 <*>? "name"
647
648-- | UDP tracker protocol compatible encoding.
649instance Serialize ScrapeEntry where
650 put ScrapeEntry {..} = do
651 putWord32be $ fromIntegral siComplete
652 putWord32be $ fromIntegral siDownloaded
653 putWord32be $ fromIntegral siIncomplete
654
655 get = ScrapeEntry
656 <$> (fromIntegral <$> getWord32be)
657 <*> (fromIntegral <$> getWord32be)
658 <*> (fromIntegral <$> getWord32be)
659 <*> pure Nothing
660
661-- | Scrape info about a set of torrents.
662type ScrapeInfo = [(InfoHash, ScrapeEntry)]
663
664{-----------------------------------------------------------------------
665-- HTTP specific
666-----------------------------------------------------------------------}
667
668-- | Some HTTP trackers allow to choose prefered representation of the
669-- 'AnnounceInfo'. It's optional for trackers to honor any of this
670-- options.
671data AnnouncePrefs = AnnouncePrefs
672 { -- | If specified, "compact" parameter is used to advise the
673 -- tracker to send peer id list as:
674 --
675 -- * bencoded list (extCompact = Just False);
676 -- * or more compact binary string (extCompact = Just True).
677 --
678 -- The later is prefered since compact peer list will reduce the
679 -- size of tracker responses. Hovewer, if tracker do not support
680 -- this extension then it can return peer list in either form.
681 --
682 -- For more info see: <http://www.bittorrent.org/beps/bep_0023.html>
683 --
684 extCompact :: !(Maybe Bool)
685
686 -- | If specified, "no_peer_id" parameter is used advise tracker
687 -- to either send or not to send peer id in tracker response.
688 -- Tracker may not support this extension as well.
689 --
690 -- For more info see:
691 -- <http://permalink.gmane.org/gmane.network.bit-torrent.general/4030>
692 --
693 , extNoPeerId :: !(Maybe Bool)
694 } deriving (Show, Eq, Typeable)
695
696instance Default AnnouncePrefs where
697 def = AnnouncePrefs Nothing Nothing
698
699instance QueryLike AnnouncePrefs where
700 toQuery AnnouncePrefs {..} =
701 [ ("compact", toQueryFlag <$> extCompact) -- TODO use 'paramName'
702 , ("no_peer_id", toQueryFlag <$> extNoPeerId)
703 ]
704 where
705 toQueryFlag False = "0"
706 toQueryFlag True = "1"
707
708-- | Parse announce query extended part from query string.
709parseAnnouncePrefs :: SimpleQuery -> AnnouncePrefs
710parseAnnouncePrefs params = either (const def) id $
711 AnnouncePrefs
712 <$> optParam ParamCompact params
713 <*> optParam ParamNoPeerId params
714
715-- | Render announce preferences to query string.
716renderAnnouncePrefs :: AnnouncePrefs -> SimpleQuery
717renderAnnouncePrefs = queryToSimpleQuery . toQuery
718
719-- | HTTP tracker request with preferences.
720data AnnounceRequest = AnnounceRequest
721 { announceQuery :: AnnounceQuery -- ^ Request query params.
722 , announcePrefs :: AnnouncePrefs -- ^ Optional advises to the tracker.
723 } deriving (Show, Eq, Typeable)
724
725instance QueryLike AnnounceRequest where
726 toQuery AnnounceRequest{..} =
727 toQuery announcePrefs <>
728 toQuery announceQuery
729
730-- | Parse announce request from query string.
731parseAnnounceRequest :: SimpleQuery -> ParseResult AnnounceRequest
732parseAnnounceRequest params = AnnounceRequest
733 <$> parseAnnounceQuery params
734 <*> pure (parseAnnouncePrefs params)
735
736-- | Render announce request to query string.
737renderAnnounceRequest :: AnnounceRequest -> SimpleQuery
738renderAnnounceRequest = queryToSimpleQuery . toQuery
739
740type PathPiece = BS.ByteString
741
742defaultAnnouncePath :: PathPiece
743defaultAnnouncePath = "announce"
744
745defaultScrapePath :: PathPiece
746defaultScrapePath = "scrape"
747
748missingOffset :: Int
749missingOffset = 101
750
751invalidOffset :: Int
752invalidOffset = 150
753
754parseFailureCode :: ParamParseFailure -> Int
755parseFailureCode (Missing param ) = missingOffset + fromEnum param
756parseFailureCode (Invalid param _) = invalidOffset + fromEnum param
757
758parseFailureMessage :: ParamParseFailure -> BS.ByteString
759parseFailureMessage e = BS.concat $ case e of
760 Missing p -> ["Missing parameter: ", paramName p]
761 Invalid p v -> ["Invalid parameter: ", paramName p, " = ", v]
762
763-- | HTTP response /content type/ for announce info.
764announceType :: ByteString
765announceType = "text/plain"
766
767-- | HTTP response /content type/ for scrape info.
768scrapeType :: ByteString
769scrapeType = "text/plain"
770
771-- | Get HTTP response status from a announce params parse failure.
772--
773-- For more info see:
774-- <https://wiki.theory.org/BitTorrent_Tracker_Protocol#Response_Codes>
775--
776parseFailureStatus :: ParamParseFailure -> Status
777parseFailureStatus = mkStatus <$> parseFailureCode <*> parseFailureMessage
778
779{-----------------------------------------------------------------------
780-- UDP specific message types
781-----------------------------------------------------------------------}
782
783genToken :: IO Word64
784genToken = do
785 bs <- getEntropy 8
786 either err return $ runGet getWord64be bs
787 where
788 err = error "genToken: impossible happen"
789
790-- | Connection Id is used for entire tracker session.
791newtype ConnectionId = ConnectionId Word64
792 deriving (Eq, Serialize)
793
794instance Show ConnectionId where
795 showsPrec _ (ConnectionId cid) = showString "0x" <> showHex cid
796
797initialConnectionId :: ConnectionId
798initialConnectionId = ConnectionId 0x41727101980
799
800-- | Transaction Id is used within a UDP RPC.
801newtype TransactionId = TransactionId Word32
802 deriving (Eq, Ord, Enum, Bounded, Serialize)
803
804instance Show TransactionId where
805 showsPrec _ (TransactionId tid) = showString "0x" <> showHex tid
806
807genTransactionId :: IO TransactionId
808genTransactionId = (TransactionId . fromIntegral) <$> genToken
809
810data Request
811 = Connect
812 | Announce AnnounceQuery
813 | Scrape ScrapeQuery
814 deriving Show
815
816data Response
817 = Connected ConnectionId
818 | Announced AnnounceInfo
819 | Scraped [ScrapeEntry]
820 | Failed Text
821 deriving Show
822
823responseName :: Response -> String
824responseName (Connected _) = "connected"
825responseName (Announced _) = "announced"
826responseName (Scraped _) = "scraped"
827responseName (Failed _) = "failed"
828
829data family Transaction a
830data instance Transaction Request = TransactionQ
831 { connIdQ :: {-# UNPACK #-} !ConnectionId
832 , transIdQ :: {-# UNPACK #-} !TransactionId
833 , request :: !Request
834 } deriving Show
835data instance Transaction Response = TransactionR
836 { transIdR :: {-# UNPACK #-} !TransactionId
837 , response :: !Response
838 } deriving Show
839
840-- TODO newtype
841newtype MessageId = MessageId Word32
842 deriving (Show, Eq, Num, Serialize)
843
844connectId, announceId, scrapeId, errorId :: MessageId
845connectId = 0
846announceId = 1
847scrapeId = 2
848errorId = 3
849
850instance Serialize (Transaction Request) where
851 put TransactionQ {..} = do
852 case request of
853 Connect -> do
854 put initialConnectionId
855 put connectId
856 put transIdQ
857
858 Announce ann -> do
859 put connIdQ
860 put announceId
861 put transIdQ
862 put ann
863
864 Scrape hashes -> do
865 put connIdQ
866 put scrapeId
867 put transIdQ
868 forM_ hashes put
869
870 get = do
871 cid <- get
872 mid <- get
873 TransactionQ cid <$> S.get <*> getBody mid
874 where
875 getBody :: MessageId -> S.Get Request
876 getBody msgId
877 | msgId == connectId = pure Connect
878 | msgId == announceId = Announce <$> get
879 | msgId == scrapeId = Scrape <$> many get
880 | otherwise = fail errMsg
881 where
882 errMsg = "unknown request: " ++ show msgId
883
884instance Serialize (Transaction Response) where
885 put TransactionR {..} = do
886 case response of
887 Connected conn -> do
888 put connectId
889 put transIdR
890 put conn
891
892 Announced info -> do
893 put announceId
894 put transIdR
895 put info
896
897 Scraped infos -> do
898 put scrapeId
899 put transIdR
900 forM_ infos put
901
902 Failed info -> do
903 put errorId
904 put transIdR
905 put (encodeUtf8 info)
906
907
908 get = do
909 mid <- get
910 TransactionR <$> get <*> getBody mid
911 where
912 getBody :: MessageId -> S.Get Response
913 getBody msgId
914 | msgId == connectId = Connected <$> get
915 | msgId == announceId = Announced <$> get
916 | msgId == scrapeId = Scraped <$> many get
917 | msgId == errorId = (Failed . decodeUtf8) <$> get
918 | otherwise = fail msg
919 where
920 msg = "unknown response: " ++ show msgId
diff --git a/src/Network/BitTorrent/Tracker/RPC.hs b/src/Network/BitTorrent/Tracker/RPC.hs
deleted file mode 100644
index 45fef05e..00000000
--- a/src/Network/BitTorrent/Tracker/RPC.hs
+++ /dev/null
@@ -1,175 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- This module provides unified RPC interface to BitTorrent
9-- trackers. The tracker is an UDP/HTTP/HTTPS service used to
10-- discovery peers for a particular existing torrent and keep
11-- statistics about the swarm. This module also provides a way to
12-- request scrape info for a particular torrent list.
13--
14{-# LANGUAGE DeriveDataTypeable #-}
15module Network.BitTorrent.Tracker.RPC
16 ( PeerInfo (..)
17
18 -- * Manager
19 , Options (..)
20 , Manager
21 , newManager
22 , closeManager
23 , withManager
24
25 -- * RPC
26 , SAnnounceQuery (..)
27 , RpcException (..)
28 , Network.BitTorrent.Tracker.RPC.announce
29 , scrape
30 ) where
31
32import Control.Exception
33import Data.Default
34import Data.Typeable
35import Network
36import Network.URI
37import Network.Socket (HostAddress)
38
39import Data.Torrent
40import Network.Address
41import Network.BitTorrent.Internal.Progress
42import Network.BitTorrent.Tracker.Message
43import qualified Network.BitTorrent.Tracker.RPC.HTTP as HTTP
44import qualified Network.BitTorrent.Tracker.RPC.UDP as UDP
45
46
47{-----------------------------------------------------------------------
48-- Simplified announce
49-----------------------------------------------------------------------}
50
51-- | Info to advertise to trackers.
52data PeerInfo = PeerInfo
53 { peerId :: !PeerId
54 , peerIP :: !(Maybe HostAddress)
55 , peerPort :: !PortNumber
56 } deriving (Show, Eq)
57
58instance Default PeerInfo where
59 def = PeerInfo def Nothing 6881
60
61-- | Simplified announce query.
62data SAnnounceQuery = SAnnounceQuery
63 { sInfoHash :: InfoHash
64 , sProgress :: Progress
65 , sNumWant :: Maybe Int
66 , sEvent :: Maybe AnnounceEvent
67 }
68
69fillAnnounceQuery :: PeerInfo -> SAnnounceQuery -> AnnounceQuery
70fillAnnounceQuery PeerInfo{..} SAnnounceQuery {..} = AnnounceQuery
71 { reqInfoHash = sInfoHash
72 , reqPeerId = peerId
73 , reqPort = peerPort
74 , reqProgress = sProgress
75 , reqIP = peerIP
76 , reqNumWant = sNumWant
77 , reqEvent = sEvent
78 }
79
80{-----------------------------------------------------------------------
81-- RPC manager
82-----------------------------------------------------------------------}
83
84-- | Tracker manager settings.
85data Options = Options
86 { -- | HTTP tracker protocol specific options.
87 optHttpRPC :: !HTTP.Options
88
89 -- | UDP tracker protocol specific options.
90 , optUdpRPC :: !UDP.Options
91
92 -- | Whether to use multitracker extension.
93 , optMultitracker :: !Bool
94 }
95
96instance Default Options where
97 def = Options
98 { optHttpRPC = def
99 , optUdpRPC = def
100 , optMultitracker = True
101 }
102
103-- | Tracker RPC Manager.
104data Manager = Manager
105 { options :: !Options
106 , peerInfo :: !PeerInfo
107 , httpMgr :: !HTTP.Manager
108 , udpMgr :: !UDP.Manager
109 }
110
111-- | Create a new 'Manager'. You /must/ manually 'closeManager'
112-- otherwise resource leakage is possible. Normally, a bittorrent
113-- client need a single RPC manager only.
114--
115-- This function can throw 'IOException' on invalid 'Options'.
116--
117newManager :: Options -> PeerInfo -> IO Manager
118newManager opts info = do
119 h <- HTTP.newManager (optHttpRPC opts)
120 u <- UDP.newManager (optUdpRPC opts) `onException` HTTP.closeManager h
121 return $ Manager opts info h u
122
123-- | Close all pending RPCs. Behaviour of currently in-flight RPCs can
124-- differ depending on underlying protocol used. No rpc calls should
125-- be performed after manager becomes closed.
126closeManager :: Manager -> IO ()
127closeManager Manager {..} = do
128 UDP.closeManager udpMgr `finally` HTTP.closeManager httpMgr
129
130-- | Normally you need to use 'Control.Monad.Trans.Resource.allocate'.
131withManager :: Options -> PeerInfo -> (Manager -> IO a) -> IO a
132withManager opts info = bracket (newManager opts info) closeManager
133
134{-----------------------------------------------------------------------
135-- Exceptions
136-----------------------------------------------------------------------}
137-- TODO Catch IO exceptions on rpc calls (?)
138
139data RpcException
140 = UdpException UDP.RpcException -- ^ UDP RPC driver failure;
141 | HttpException HTTP.RpcException -- ^ HTTP RPC driver failure;
142 | UnrecognizedScheme String -- ^ unsupported scheme in announce URI;
143 | GenericException String -- ^ for furter extensibility.
144 deriving (Show, Typeable)
145
146instance Exception RpcException
147
148packException :: Exception e => (e -> RpcException) -> IO a -> IO a
149packException f m = try m >>= either (throwIO . f) return
150{-# INLINE packException #-}
151
152{-----------------------------------------------------------------------
153-- RPC calls
154-----------------------------------------------------------------------}
155
156dispatch :: URI -> IO a -> IO a -> IO a
157dispatch URI {..} http udp
158 | uriScheme == "http:" ||
159 uriScheme == "https:" = packException HttpException http
160 | uriScheme == "udp:" = packException UdpException udp
161 | otherwise = throwIO $ UnrecognizedScheme uriScheme
162
163announce :: Manager -> URI -> SAnnounceQuery -> IO AnnounceInfo
164announce Manager {..} uri simpleQuery
165 = dispatch uri
166 (HTTP.announce httpMgr uri annQ)
167 ( UDP.announce udpMgr uri annQ)
168 where
169 annQ = fillAnnounceQuery peerInfo simpleQuery
170
171scrape :: Manager -> URI -> ScrapeQuery -> IO ScrapeInfo
172scrape Manager {..} uri q
173 = dispatch uri
174 (HTTP.scrape httpMgr uri q)
175 ( UDP.scrape udpMgr uri q)
diff --git a/src/Network/BitTorrent/Tracker/RPC/HTTP.hs b/src/Network/BitTorrent/Tracker/RPC/HTTP.hs
deleted file mode 100644
index 9b6e056a..00000000
--- a/src/Network/BitTorrent/Tracker/RPC/HTTP.hs
+++ /dev/null
@@ -1,191 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013
3-- License : BSD
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : provisional
6-- Portability : portable
7--
8-- This module implement HTTP tracker protocol.
9--
10-- For more information see:
11-- <https://wiki.theory.org/BitTorrentSpecification#Tracker_HTTP.2FHTTPS_Protocol>
12--
13{-# LANGUAGE DeriveDataTypeable #-}
14module Network.BitTorrent.Tracker.RPC.HTTP
15 ( -- * Manager
16 Options (..)
17 , Manager
18 , newManager
19 , closeManager
20 , withManager
21
22 -- * RPC
23 , RpcException (..)
24 , announce
25 , scrape
26 , scrapeOne
27 ) where
28
29import Control.Applicative
30import Control.Exception
31import Control.Monad
32import Control.Monad.Trans.Resource
33import Data.BEncode as BE
34import Data.ByteString as BS
35import Data.ByteString.Char8 as BC
36import Data.ByteString.Lazy as BL
37import Data.Default
38import Data.List as L
39import Data.Monoid
40import Data.Typeable hiding (Proxy)
41import Network.URI
42import Network.HTTP.Conduit hiding
43 (Manager, newManager, closeManager, withManager)
44import Network.HTTP.Client (defaultManagerSettings)
45import Network.HTTP.Client.Internal (setUri)
46import qualified Network.HTTP.Conduit as HTTP
47import Network.HTTP.Types.Header (hUserAgent)
48import Network.HTTP.Types.URI (SimpleQuery, renderSimpleQuery)
49
50import Data.Torrent (InfoHash)
51import Network.Address (libUserAgent)
52import Network.BitTorrent.Tracker.Message hiding (Request, Response)
53
54{-----------------------------------------------------------------------
55-- Exceptions
56-----------------------------------------------------------------------}
57
58data RpcException
59 = RequestFailed HttpException -- ^ failed HTTP request.
60 | ParserFailure String -- ^ unable to decode tracker response;
61 | ScrapelessTracker -- ^ tracker do not support scraping;
62 | BadScrape -- ^ unable to find info hash in response dict;
63 deriving (Show, Typeable)
64
65instance Exception RpcException
66
67packHttpException :: IO a -> IO a
68packHttpException m = try m >>= either (throwIO . RequestFailed) return
69
70{-----------------------------------------------------------------------
71-- Manager
72-----------------------------------------------------------------------}
73
74-- | HTTP tracker specific RPC options.
75data Options = Options
76 { -- | Global HTTP announce query preferences.
77 optAnnouncePrefs :: !AnnouncePrefs
78
79 -- | Whether to use HTTP proxy for HTTP tracker requests.
80 , optHttpProxy :: !(Maybe Proxy)
81
82 -- | Value to put in HTTP user agent header.
83 , optUserAgent :: !BS.ByteString
84
85 -- | HTTP manager options.
86 , optHttpOptions :: !ManagerSettings
87 }
88
89instance Default Options where
90 def = Options
91 { optAnnouncePrefs = def
92 , optHttpProxy = Nothing
93 , optUserAgent = BC.pack libUserAgent
94 , optHttpOptions = defaultManagerSettings
95 }
96
97-- | HTTP tracker manager.
98data Manager = Manager
99 { options :: !Options
100 , httpMgr :: !HTTP.Manager
101 }
102
103-- |
104newManager :: Options -> IO Manager
105newManager opts = Manager opts <$> HTTP.newManager (optHttpOptions opts)
106
107-- |
108closeManager :: Manager -> IO ()
109closeManager Manager {..} = HTTP.closeManager httpMgr
110
111-- | Normally you need to use 'Control.Monad.Trans.Resource.allocate'.
112withManager :: Options -> (Manager -> IO a) -> IO a
113withManager opts = bracket (newManager opts) closeManager
114
115{-----------------------------------------------------------------------
116-- Queries
117-----------------------------------------------------------------------}
118
119fillRequest :: Options -> SimpleQuery -> Request -> Request
120fillRequest Options {..} q r = r
121 { queryString = joinQuery (queryString r) (renderSimpleQuery False q)
122 , requestHeaders = (hUserAgent, optUserAgent) : requestHeaders r
123 , proxy = optHttpProxy
124 }
125 where
126 joinQuery a b
127 | BS.null a = b
128 | otherwise = a <> "&" <> b
129
130httpTracker :: BEncode a => Manager -> URI -> SimpleQuery -> IO a
131httpTracker Manager {..} uri q = packHttpException $ do
132 request <- fillRequest options q <$> setUri def {- http-client instance for Request -} uri
133 response <- runResourceT $ httpLbs request httpMgr
134 case BE.decode $ BL.toStrict $ responseBody response of
135 Left msg -> throwIO (ParserFailure msg)
136 Right info -> return info
137
138{-----------------------------------------------------------------------
139-- RPC
140-----------------------------------------------------------------------}
141
142-- | Send request and receive response from the tracker specified in
143-- announce list.
144--
145-- This function can throw 'RpcException'.
146--
147announce :: Manager -> URI -> AnnounceQuery -> IO AnnounceInfo
148announce mgr uri q = httpTracker mgr uri (renderAnnounceRequest uriQ)
149 where
150 uriQ = AnnounceRequest
151 { announceQuery = q
152 , announcePrefs = optAnnouncePrefs (options mgr)
153 }
154
155-- | Trying to convert /announce/ URL to /scrape/ URL. If 'scrapeURL'
156-- gives 'Nothing' then tracker do not support scraping.
157--
158scrapeURL :: URI -> Maybe URI
159scrapeURL uri = do
160 newPath <- replace (BC.pack (uriPath uri))
161 return uri { uriPath = BC.unpack newPath }
162 where
163 replace p = do
164 let ps = BC.splitWith (== '/') p
165 guard (not (L.null ps))
166 guard ("announce" `BS.isPrefixOf` L.last ps)
167 let newSuff = "scrape" <> BS.drop (BS.length "announce") (L.last ps)
168 return (BS.intercalate "/" (L.init ps ++ [newSuff]))
169
170-- | For each 'InfoHash' of torrents request scrape info from the tracker.
171-- However if the info hash list is 'null', the tracker should list
172-- all available torrents.
173--
174-- This function can throw 'RpcException'.
175--
176scrape :: Manager -> URI -> ScrapeQuery -> IO ScrapeInfo
177scrape m u q = do
178 case scrapeURL u of
179 Nothing -> throwIO ScrapelessTracker
180 Just uri -> httpTracker m uri (renderScrapeQuery q)
181
182-- | More particular version of 'scrape', just for one torrent.
183--
184-- This function can throw 'RpcException'.
185--
186scrapeOne :: Manager -> URI -> InfoHash -> IO ScrapeEntry
187scrapeOne m uri ih = do
188 xs <- scrape m uri [ih]
189 case L.lookup ih xs of
190 Nothing -> throwIO BadScrape
191 Just a -> return a
diff --git a/src/Network/BitTorrent/Tracker/RPC/UDP.hs b/src/Network/BitTorrent/Tracker/RPC/UDP.hs
deleted file mode 100644
index 31b6b870..00000000
--- a/src/Network/BitTorrent/Tracker/RPC/UDP.hs
+++ /dev/null
@@ -1,454 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2013-2014
3-- License : BSD3
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : provisional
6-- Portability : portable
7--
8-- This module implement UDP tracker protocol.
9--
10-- For protocol details and uri scheme see:
11-- <http://www.bittorrent.org/beps/bep_0015.html>,
12-- <https://www.iana.org/assignments/uri-schemes/prov/udp>
13--
14{-# LANGUAGE RecordWildCards #-}
15{-# LANGUAGE FlexibleInstances #-}
16{-# LANGUAGE GeneralizedNewtypeDeriving #-}
17{-# LANGUAGE DeriveDataTypeable #-}
18module Network.BitTorrent.Tracker.RPC.UDP
19 ( -- * Manager
20 Options (..)
21 , Manager
22 , newManager
23 , closeManager
24 , withManager
25
26 -- * RPC
27 , RpcException (..)
28 , announce
29 , scrape
30 ) where
31
32import Control.Applicative
33import Control.Concurrent
34import Control.Exception
35import Control.Monad
36import Data.Default
37import Data.IORef
38import Data.List as L
39import Data.Map as M
40import Data.Maybe
41import Data.Serialize
42import Data.Text as T
43import Data.Time
44import Data.Time.Clock.POSIX
45import Data.Traversable
46import Data.Typeable
47import Text.Read (readMaybe)
48import Network.Socket hiding (Connected, connect, listen)
49import Network.Socket.ByteString as BS
50import Network.URI
51import System.Timeout
52
53import Network.BitTorrent.Tracker.Message
54
55{-----------------------------------------------------------------------
56-- Options
57-----------------------------------------------------------------------}
58
59-- | 'System.Timeout.timeout' specific.
60sec :: Int
61sec = 1000000
62
63-- | See <http://www.bittorrent.org/beps/bep_0015.html#time-outs>
64defMinTimeout :: Int
65defMinTimeout = 15
66
67-- | See <http://www.bittorrent.org/beps/bep_0015.html#time-outs>
68defMaxTimeout :: Int
69defMaxTimeout = 15 * 2 ^ (8 :: Int)
70
71-- | See: <http://www.bittorrent.org/beps/bep_0015.html#time-outs>
72defMultiplier :: Int
73defMultiplier = 2
74
75-- TODO why 98?
76defMaxPacketSize :: Int
77defMaxPacketSize = 98
78
79-- | Manager configuration.
80data Options = Options
81 { -- | Max size of a /response/ packet.
82 --
83 -- 'optMaxPacketSize' /must/ be a positive value.
84 --
85 optMaxPacketSize :: {-# UNPACK #-} !Int
86
87 -- | Starting timeout interval in seconds. If a response is not
88 -- received after 'optMinTimeout' then 'Manager' repeat RPC with
89 -- timeout interval multiplied by 'optMultiplier' and so on until
90 -- timeout interval reach 'optMaxTimeout'.
91 --
92 -- 'optMinTimeout' /must/ be a positive value.
93 --
94 , optMinTimeout :: {-# UNPACK #-} !Int
95
96 -- | Final timeout interval in seconds. After 'optMaxTimeout'
97 -- reached and tracker still not responding both 'announce' and
98 -- 'scrape' functions will throw 'TimeoutExpired' exception.
99 --
100 -- 'optMaxTimeout' /must/ be greater than 'optMinTimeout'.
101 --
102 , optMaxTimeout :: {-# UNPACK #-} !Int
103
104 -- | 'optMultiplier' /must/ be a positive value.
105 , optMultiplier :: {-# UNPACK #-} !Int
106 } deriving (Show, Eq)
107
108-- | Options suitable for bittorrent client.
109instance Default Options where
110 def = Options
111 { optMaxPacketSize = defMaxPacketSize
112 , optMinTimeout = defMinTimeout
113 , optMaxTimeout = defMaxTimeout
114 , optMultiplier = defMultiplier
115 }
116
117checkOptions :: Options -> IO ()
118checkOptions Options {..} = do
119 unless (optMaxPacketSize > 0) $ do
120 throwIO $ userError "optMaxPacketSize must be positive"
121
122 unless (optMinTimeout > 0) $ do
123 throwIO $ userError "optMinTimeout must be positive"
124
125 unless (optMaxTimeout > 0) $ do
126 throwIO $ userError "optMaxTimeout must be positive"
127
128 unless (optMultiplier > 0) $ do
129 throwIO $ userError "optMultiplier must be positive"
130
131 unless (optMaxTimeout > optMinTimeout) $ do
132 throwIO $ userError "optMaxTimeout must be greater than optMinTimeout"
133
134
135{-----------------------------------------------------------------------
136-- Manager state
137-----------------------------------------------------------------------}
138
139type ConnectionCache = Map SockAddr Connection
140
141type PendingResponse = MVar (Either RpcException Response)
142type PendingTransactions = Map TransactionId PendingResponse
143type PendingQueries = Map SockAddr PendingTransactions
144
145-- | UDP tracker manager.
146data Manager = Manager
147 { options :: !Options
148 , sock :: !Socket
149-- , dnsCache :: !(IORef (Map URI SockAddr))
150 , connectionCache :: !(IORef ConnectionCache)
151 , pendingResps :: !(MVar PendingQueries)
152 , listenerThread :: !(MVar ThreadId)
153 }
154
155initManager :: Options -> IO Manager
156initManager opts = Manager opts
157 <$> socket AF_INET Datagram defaultProtocol
158 <*> newIORef M.empty
159 <*> newMVar M.empty
160 <*> newEmptyMVar
161
162unblockAll :: PendingQueries -> IO ()
163unblockAll m = traverse (traverse unblockCall) m >> return ()
164 where
165 unblockCall ares = putMVar ares (Left ManagerClosed)
166
167resetState :: Manager -> IO ()
168resetState Manager {..} = do
169 writeIORef connectionCache err
170 m <- swapMVar pendingResps err
171 unblockAll m
172 mtid <- tryTakeMVar listenerThread
173 case mtid of
174 Nothing -> return () -- thread killed by 'closeManager'
175 Just _ -> return () -- thread killed by exception from 'listen'
176 return ()
177 where
178 err = error "UDP tracker manager closed"
179
180-- | This function will throw 'IOException' on invalid 'Options'.
181newManager :: Options -> IO Manager
182newManager opts = do
183 checkOptions opts
184 mgr <- initManager opts
185 tid <- forkIO (listen mgr `finally` resetState mgr)
186 putMVar (listenerThread mgr) tid
187 return mgr
188
189-- | Unblock all RPCs by throwing 'ManagerClosed' exception. No rpc
190-- calls should be performed after manager becomes closed.
191closeManager :: Manager -> IO ()
192closeManager Manager {..} = do
193 close sock
194 mtid <- tryTakeMVar listenerThread
195 case mtid of
196 Nothing -> return ()
197 Just tid -> killThread tid
198
199-- | Normally you need to use 'Control.Monad.Trans.Resource.allocate'.
200withManager :: Options -> (Manager -> IO a) -> IO a
201withManager opts = bracket (newManager opts) closeManager
202
203{-----------------------------------------------------------------------
204-- Exceptions
205-----------------------------------------------------------------------}
206
207data RpcException
208 -- | Unable to lookup hostname;
209 = HostUnknown
210
211 -- | Unable to lookup hostname;
212 | HostLookupFailed
213
214 -- | Expecting 'udp:', but some other scheme provided.
215 | UnrecognizedScheme String
216
217 -- | Tracker exists but not responding for specific number of seconds.
218 | TimeoutExpired Int
219
220 -- | Tracker responded with unexpected message type.
221 | UnexpectedResponse
222 { expectedMsg :: String
223 , actualMsg :: String
224 }
225
226 -- | RPC succeed, but tracker responded with error code.
227 | QueryFailed Text
228
229 -- | RPC manager closed while waiting for response.
230 | ManagerClosed
231 deriving (Eq, Show, Typeable)
232
233instance Exception RpcException
234
235{-----------------------------------------------------------------------
236-- Host Addr resolution
237-----------------------------------------------------------------------}
238
239setPort :: PortNumber -> SockAddr -> SockAddr
240setPort p (SockAddrInet _ h) = SockAddrInet p h
241setPort p (SockAddrInet6 _ f h s) = SockAddrInet6 p f h s
242setPort _ addr = addr
243
244resolveURI :: URI -> IO SockAddr
245resolveURI URI { uriAuthority = Just (URIAuth {..}) } = do
246 infos <- getAddrInfo Nothing (Just uriRegName) Nothing
247 let port = fromMaybe 0 (readMaybe (L.drop 1 uriPort) :: Maybe Int)
248 case infos of
249 AddrInfo {..} : _ -> return $ setPort (fromIntegral port) addrAddress
250 _ -> throwIO HostLookupFailed
251resolveURI _ = throwIO HostUnknown
252
253-- TODO caching?
254getTrackerAddr :: Manager -> URI -> IO SockAddr
255getTrackerAddr _ uri
256 | uriScheme uri == "udp:" = resolveURI uri
257 | otherwise = throwIO (UnrecognizedScheme (uriScheme uri))
258
259{-----------------------------------------------------------------------
260 Connection
261-----------------------------------------------------------------------}
262
263connectionLifetime :: NominalDiffTime
264connectionLifetime = 60
265
266data Connection = Connection
267 { connectionId :: ConnectionId
268 , connectionTimestamp :: UTCTime
269 } deriving Show
270
271-- placeholder for the first 'connect'
272initialConnection :: Connection
273initialConnection = Connection initialConnectionId (posixSecondsToUTCTime 0)
274
275establishedConnection :: ConnectionId -> IO Connection
276establishedConnection cid = Connection cid <$> getCurrentTime
277
278isExpired :: Connection -> IO Bool
279isExpired Connection {..} = do
280 currentTime <- getCurrentTime
281 let timeDiff = diffUTCTime currentTime connectionTimestamp
282 return $ timeDiff > connectionLifetime
283
284{-----------------------------------------------------------------------
285-- Transactions
286-----------------------------------------------------------------------}
287
288-- | Sometimes 'genTransactionId' may return already used transaction
289-- id. We use a good entropy source but the issue /still/ (with very
290-- small probabality) may happen. If the collision happen then this
291-- function tries to find nearest unused slot, otherwise pending
292-- transactions table is full.
293firstUnused :: SockAddr -> TransactionId -> PendingQueries -> TransactionId
294firstUnused addr rid m = do
295 case M.splitLookup rid <$> M.lookup addr m of
296 Nothing -> rid
297 Just (_ , Nothing, _ ) -> rid
298 Just (lt, Just _ , gt) ->
299 case backwardHole (keys lt) rid <|> forwardHole rid (keys gt) of
300 Nothing -> error "firstUnused: table is full" -- impossible
301 Just tid -> tid
302 where
303 forwardHole a []
304 | a == maxBound = Nothing
305 | otherwise = Just (succ a)
306 forwardHole a (b : xs)
307 | succ a == b = forwardHole b xs
308 | otherwise = Just (succ a)
309
310 backwardHole [] a
311 | a == minBound = Nothing
312 | otherwise = Just (pred a)
313 backwardHole (b : xs) a
314 | b == pred a = backwardHole xs b
315 | otherwise = Just (pred a)
316
317register :: SockAddr -> TransactionId -> PendingResponse
318 -> PendingQueries -> PendingQueries
319register addr tid ares = M.alter insertId addr
320 where
321 insertId Nothing = Just (M.singleton tid ares)
322 insertId (Just m) = Just (M.insert tid ares m)
323
324unregister :: SockAddr -> TransactionId
325 -> PendingQueries -> PendingQueries
326unregister addr tid = M.update deleteId addr
327 where
328 deleteId m
329 | M.null m' = Nothing
330 | otherwise = Just m'
331 where
332 m' = M.delete tid m
333
334-- | Generate a new unused transaction id and register as pending.
335allocTransaction :: Manager -> SockAddr -> PendingResponse -> IO TransactionId
336allocTransaction Manager {..} addr ares =
337 modifyMVar pendingResps $ \ m -> do
338 rndId <- genTransactionId
339 let tid = firstUnused addr rndId m
340 return (register addr tid ares m, tid)
341
342-- | Wake up blocked thread and return response back.
343commitTransaction :: Manager -> SockAddr -> TransactionId -> Response -> IO ()
344commitTransaction Manager {..} addr tid resp =
345 modifyMVarMasked_ pendingResps $ \ m -> do
346 case M.lookup tid =<< M.lookup addr m of
347 Nothing -> return m -- tracker responded after 'cancelTransaction' fired
348 Just ares -> do
349 putMVar ares (Right resp)
350 return $ unregister addr tid m
351
352-- | Abort transaction forcefully.
353cancelTransaction :: Manager -> SockAddr -> TransactionId -> IO ()
354cancelTransaction Manager {..} addr tid =
355 modifyMVarMasked_ pendingResps $ \m ->
356 return $ unregister addr tid m
357
358-- | Handle responses from trackers.
359listen :: Manager -> IO ()
360listen mgr @ Manager {..} = do
361 forever $ do
362 (bs, addr) <- BS.recvFrom sock (optMaxPacketSize options)
363 case decode bs of
364 Left _ -> return () -- parser failed, ignoring
365 Right (TransactionR {..}) -> commitTransaction mgr addr transIdR response
366
367-- | Perform RPC transaction. If the action interrupted transaction
368-- will be aborted.
369transaction :: Manager -> SockAddr -> Connection -> Request -> IO Response
370transaction mgr @ Manager {..} addr conn request = do
371 ares <- newEmptyMVar
372 tid <- allocTransaction mgr addr ares
373 performTransaction tid ares
374 `onException` cancelTransaction mgr addr tid
375 where
376 performTransaction tid ares = do
377 let trans = TransactionQ (connectionId conn) tid request
378 BS.sendAllTo sock (encode trans) addr
379 takeMVar ares >>= either throwIO return
380
381{-----------------------------------------------------------------------
382-- Connection cache
383-----------------------------------------------------------------------}
384
385connect :: Manager -> SockAddr -> Connection -> IO ConnectionId
386connect m addr conn = do
387 resp <- transaction m addr conn Connect
388 case resp of
389 Connected cid -> return cid
390 Failed msg -> throwIO $ QueryFailed msg
391 _ -> throwIO $ UnexpectedResponse "connected" (responseName resp)
392
393newConnection :: Manager -> SockAddr -> IO Connection
394newConnection m addr = do
395 connId <- connect m addr initialConnection
396 establishedConnection connId
397
398refreshConnection :: Manager -> SockAddr -> Connection -> IO Connection
399refreshConnection mgr addr conn = do
400 expired <- isExpired conn
401 if expired
402 then do
403 connId <- connect mgr addr conn
404 establishedConnection connId
405 else do
406 return conn
407
408withCache :: Manager -> SockAddr
409 -> (Maybe Connection -> IO Connection) -> IO Connection
410withCache mgr addr action = do
411 cache <- readIORef (connectionCache mgr)
412 conn <- action (M.lookup addr cache)
413 writeIORef (connectionCache mgr) (M.insert addr conn cache)
414 return conn
415
416getConnection :: Manager -> SockAddr -> IO Connection
417getConnection mgr addr = withCache mgr addr $
418 maybe (newConnection mgr addr) (refreshConnection mgr addr)
419
420{-----------------------------------------------------------------------
421-- RPC
422-----------------------------------------------------------------------}
423
424retransmission :: Options -> IO a -> IO a
425retransmission Options {..} action = go optMinTimeout
426 where
427 go curTimeout
428 | curTimeout > optMaxTimeout = throwIO $ TimeoutExpired curTimeout
429 | otherwise = do
430 r <- timeout (curTimeout * sec) action
431 maybe (go (optMultiplier * curTimeout)) return r
432
433queryTracker :: Manager -> URI -> Request -> IO Response
434queryTracker mgr uri req = do
435 addr <- getTrackerAddr mgr uri
436 retransmission (options mgr) $ do
437 conn <- getConnection mgr addr
438 transaction mgr addr conn req
439
440-- | This function can throw 'RpcException'.
441announce :: Manager -> URI -> AnnounceQuery -> IO AnnounceInfo
442announce mgr uri q = do
443 resp <- queryTracker mgr uri (Announce q)
444 case resp of
445 Announced info -> return info
446 _ -> throwIO $ UnexpectedResponse "announce" (responseName resp)
447
448-- | This function can throw 'RpcException'.
449scrape :: Manager -> URI -> ScrapeQuery -> IO ScrapeInfo
450scrape mgr uri ihs = do
451 resp <- queryTracker mgr uri (Scrape ihs)
452 case resp of
453 Scraped info -> return $ L.zip ihs info
454 _ -> throwIO $ UnexpectedResponse "scrape" (responseName resp)
diff --git a/src/Network/BitTorrent/Tracker/Session.hs b/src/Network/BitTorrent/Tracker/Session.hs
deleted file mode 100644
index aa4a832f..00000000
--- a/src/Network/BitTorrent/Tracker/Session.hs
+++ /dev/null
@@ -1,306 +0,0 @@
1-- |
2-- Copyright : (c) Sam Truzjan 2014
3-- License : BSD
4-- Maintainer : pxqr.sta@gmail.com
5-- Stability : experimental
6-- Portability : portable
7--
8-- Multitracker sessions.
9--
10{-# LANGUAGE FlexibleInstances #-}
11{-# LANGUAGE TypeFamilies #-}
12{-# LANGUAGE TypeSynonymInstances #-}
13{-# LANGUAGE TemplateHaskell #-}
14module Network.BitTorrent.Tracker.Session
15 ( -- * Session
16 Session
17 , Event (..)
18 , newSession
19 , closeSession
20 , withSession
21
22 -- * Client send notifications
23 , notify
24 , askPeers
25
26 -- * Session state
27 -- ** Status
28 , Status (..)
29 , getStatus
30
31 -- ** Single tracker sessions
32 , LastScrape (..)
33 , TrackerSession
34 , trackerPeers
35 , trackerScrape
36 , getSessionState
37
38 -- * Tracker Exchange
39 -- | BEP28: <http://www.bittorrent.org/beps/bep_0028.html>
40 , addTracker
41 , removeTracker
42 , getTrustedTrackers
43 ) where
44
45import Control.Applicative
46import Control.Exception
47import Control.Concurrent
48import Control.Concurrent.Chan.Split as CS
49import Control.Monad
50import Data.Default
51import Data.Fixed
52import Data.Foldable as F
53import Data.IORef
54import Data.List as L
55import Data.Maybe
56import Data.Time
57import Data.Traversable
58import Network.URI
59
60import Data.Torrent
61import Network.Address
62import Network.BitTorrent.Internal.Cache
63import Network.BitTorrent.Internal.Types
64import Network.BitTorrent.Tracker.List as TL
65import Network.BitTorrent.Tracker.Message
66import Network.BitTorrent.Tracker.RPC as RPC
67
68{-----------------------------------------------------------------------
69-- Single tracker session
70-----------------------------------------------------------------------}
71
72-- | Status of this client.
73data Status
74 = Running -- ^ This client is announced and listenning for incoming
75 -- connections.
76 | Paused -- ^ This client does not expecting incoming connections.
77 deriving (Show, Eq, Bounded, Enum)
78
79-- | Client starting in the paused state.
80instance Default Status where
81 def = Paused
82
83-- | Tracker session starts with scrape unknown.
84instance Default LastScrape where
85 def = LastScrape Nothing Nothing
86
87data LastScrape = LastScrape
88 { -- | Count of leechers the tracker aware of.
89 scrapeLeechers :: Maybe Int
90
91 -- | Count of seeders the tracker aware of.
92 , scrapeSeeders :: Maybe Int
93 } deriving (Show, Eq)
94
95-- | Single tracker session.
96data TrackerSession = TrackerSession
97 { -- | Used to notify 'Stopped' and 'Completed' events.
98 statusSent :: !(Maybe Status)
99
100 -- | Can be used to retrieve peer set.
101 , trackerPeers :: Cached [PeerAddr IP]
102
103 -- | Can be used to show brief swarm stats in client GUI.
104 , trackerScrape :: Cached LastScrape
105 }
106
107-- | Not contacted.
108instance Default TrackerSession where
109 def = TrackerSession Nothing def def
110
111-- | Do we need to notify this /specific/ tracker?
112needNotify :: AnnounceEvent -> Maybe Status -> Maybe Bool
113needNotify Started Nothing = Just True
114needNotify Stopped Nothing = Just False
115needNotify Completed Nothing = Just False
116needNotify Started (Just Running) = Nothing
117needNotify Stopped (Just Running) = Just True
118needNotify Completed (Just Running) = Just True
119needNotify Started (Just Paused ) = Just True
120needNotify Stopped (Just Paused ) = Just False
121needNotify Completed (Just Paused ) = Just True
122
123-- | Client status after event announce succeed.
124nextStatus :: AnnounceEvent -> Maybe Status
125nextStatus Started = Just Running
126nextStatus Stopped = Just Paused
127nextStatus Completed = Nothing -- must keep previous status
128
129seconds :: Int -> NominalDiffTime
130seconds n = realToFrac (toEnum n :: Uni)
131
132cachePeers :: AnnounceInfo -> IO (Cached [PeerAddr IP])
133cachePeers AnnounceInfo {..} =
134 newCached (seconds respInterval)
135 (seconds (fromMaybe respInterval respMinInterval))
136 (getPeerList respPeers)
137
138cacheScrape :: AnnounceInfo -> IO (Cached LastScrape)
139cacheScrape AnnounceInfo {..} =
140 newCached (seconds respInterval)
141 (seconds (fromMaybe respInterval respMinInterval))
142 LastScrape
143 { scrapeSeeders = respComplete
144 , scrapeLeechers = respIncomplete
145 }
146
147-- | Make announce request to specific tracker returning new state.
148notifyTo :: Manager -> Session -> AnnounceEvent
149 -> TierEntry TrackerSession -> IO TrackerSession
150notifyTo mgr s @ Session {..} event (uri, entry @ TrackerSession {..}) = do
151 let shouldNotify = needNotify event statusSent
152 mustNotify <- maybe (isExpired trackerPeers) return shouldNotify
153 if not mustNotify
154 then return entry
155 else do
156 let q = SAnnounceQuery sessionTopic def Nothing (Just event)
157 res <- RPC.announce mgr uri q
158 when (statusSent == Nothing) $ do
159 send sessionEvents (TrackerConfirmed uri)
160 send sessionEvents (AnnouncedTo uri)
161 let status' = nextStatus event <|> statusSent
162 TrackerSession status' <$> cachePeers res <*> cacheScrape res
163
164{-----------------------------------------------------------------------
165-- Multitracker Session
166-----------------------------------------------------------------------}
167
168-- | Multitracker session.
169data Session = Session
170 { -- | Infohash to announce at each 'announce' request.
171 sessionTopic :: !InfoHash
172
173 -- | Current status of this client is used to filter duplicated
174 -- notifications, for e.g. we don't want to notify a tracker with
175 -- ['Stopped', 'Stopped'], the last should be ignored.
176 , sessionStatus :: !(IORef Status)
177
178 -- | A set of single-tracker sessions. Any request to a tracker
179 -- must take a lock.
180 , sessionTrackers :: !(MVar (TrackerList TrackerSession))
181
182 , sessionEvents :: !(SendPort (Event Session))
183 }
184
185instance EventSource Session where
186 data Event Session
187 = TrackerAdded URI
188 | TrackerConfirmed URI
189 | TrackerRemoved URI
190 | AnnouncedTo URI
191 | SessionClosed
192
193 listen Session {..} = CS.listen sessionEvents
194
195
196-- | Create a new multitracker session in paused state. Tracker list
197-- must contant only /trusted/ tracker uris. To start announcing
198-- client presence use 'notify'.
199newSession :: InfoHash -> TrackerList () -> IO Session
200newSession ih origUris = do
201 urisList <- shuffleTiers origUris
202 statusRef <- newIORef def
203 entriesVar <- newMVar (fmap (const def) urisList)
204 eventStream <- newSendPort
205 return Session
206 { sessionTopic = ih
207 , sessionStatus = statusRef
208 , sessionTrackers = entriesVar
209 , sessionEvents = eventStream
210 }
211
212-- | Release scarce resources associated with the given session. This
213-- function block until all trackers tied with this peer notified with
214-- 'Stopped' event.
215closeSession :: Manager -> Session -> IO ()
216closeSession m s @ Session {..} = do
217 notify m s Stopped
218 send sessionEvents SessionClosed
219
220{-----------------------------------------------------------------------
221-- Operations
222-----------------------------------------------------------------------}
223
224-- | Normally you need to use 'Control.Monad.Trans.Resource.alloc'.
225withSession :: Manager -> InfoHash -> TrackerList ()
226 -> (Session -> IO ()) -> IO ()
227withSession m ih uris = bracket (newSession ih uris) (closeSession m)
228
229-- | Get last announced status. The only action can alter this status
230-- is 'notify'.
231getStatus :: Session -> IO Status
232getStatus Session {..} = readIORef sessionStatus
233
234getSessionState :: Session -> IO [[TierEntry TrackerSession]]
235getSessionState Session {..} = TL.toList <$> readMVar sessionTrackers
236
237-- | Do we need to sent this event to a first working tracker or to
238-- the all known good trackers?
239allNotify :: AnnounceEvent -> Bool
240allNotify Started = False
241allNotify Stopped = True
242allNotify Completed = True
243
244notifyAll :: Manager -> Session -> AnnounceEvent -> IO ()
245notifyAll mgr s @ Session {..} event = do
246 modifyMVar_ sessionTrackers $
247 (traversal (notifyTo mgr s event))
248 where
249 traversal
250 | allNotify event = traverseAll
251 | otherwise = traverseTiers
252
253-- TODO send notifications to tracker periodically.
254-- |
255--
256-- This function /may/ block until tracker query proceed.
257notify :: Manager -> Session -> AnnounceEvent -> IO ()
258notify mgr ses event = do
259 prevStatus <- atomicModifyIORef (sessionStatus ses) $ \ s ->
260 (fromMaybe s (nextStatus event), s)
261 when (needNotify event (Just prevStatus) == Just True) $ do
262 notifyAll mgr ses event
263
264-- TODO run announce if sesion have no peers
265-- | The returned list of peers can have duplicates.
266-- This function /may/ block. Use async if needed.
267askPeers :: Manager -> Session -> IO [PeerAddr IP]
268askPeers _mgr ses = do
269 list <- readMVar (sessionTrackers ses)
270 L.concat <$> collect (tryTakeData . trackerPeers) list
271
272collect :: (a -> IO (Maybe b)) -> TrackerList a -> IO [b]
273collect f lst = (catMaybes . F.toList) <$> traverse f lst
274
275--sourcePeers :: Session -> Source (PeerAddr IP)
276--sourcePeers
277
278{-----------------------------------------------------------------------
279-- Tracker exchange
280-----------------------------------------------------------------------}
281
282-- Trackers discovered through this protocol SHOULD be treated with a
283-- certain amount of suspicion. Since the source of a tracker exchange
284-- message cannot be trusted, an implementation SHOULD have a lower
285-- number of retries before giving up entirely.
286
287addTracker :: Session -> URI -> IO ()
288addTracker Session {..} uri = do
289 undefined
290 send sessionEvents (TrackerAdded uri)
291
292removeTracker :: Manager -> Session -> URI -> IO ()
293removeTracker m Session {..} uri = do
294 send sessionEvents (TrackerRemoved uri)
295
296-- Also, as specified under the definitions section, a tracker that
297-- has not worked should never be propagated to other peers over the
298-- tracker exchange protocol.
299
300-- | Return all known trackers.
301getTrackers :: Session -> IO [URI]
302getTrackers = undefined
303
304-- | Return trackers from torrent file and
305getTrustedTrackers :: Session -> IO [URI]
306getTrustedTrackers = undefined
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-11 17:16:19 +0000