summaryrefslogtreecommitdiff
path: root/bittorrent/src/Network/BitTorrent
diff options
context:
space:
mode:
authorJames Crayne <jim.crayne@gmail.com>2019-09-28 13:43:29 -0400
committerJoe Crayne <joe@jerkface.net>2020-01-01 19:27:53 -0500
commit11987749fc6e6d3e53ea737d46d5ab13a16faeb8 (patch)
tree5716463275c2d3e902889db619908ded2a73971c /bittorrent/src/Network/BitTorrent
parentadd2c76bced51fde5e9917e7449ef52be70faf87 (diff)
Factor out some new libraries
word64-map: Data.Word64Map network-addr: Network.Address tox-crypto: Crypto.Tox lifted-concurrent: Control.Concurrent.Lifted.Instrument Control.Concurrent.Async.Lifted.Instrument psq-wrap: Data.Wrapper.PSQInt Data.Wrapper.PSQ minmax-psq: Data.MinMaxPSQ tasks: Control.Concurrent.Tasks kad: Network.Kademlia Network.Kademlia.Bootstrap Network.Kademlia.Routing Network.Kademlia.CommonAPI Network.Kademlia.Persistence Network.Kademlia.Search
Diffstat (limited to 'bittorrent/src/Network/BitTorrent')
-rw-r--r--bittorrent/src/Network/BitTorrent/Client.hs195
-rw-r--r--bittorrent/src/Network/BitTorrent/Client/Handle.hs188
-rw-r--r--bittorrent/src/Network/BitTorrent/Client/Types.hs163
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange.hs35
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Bitfield.hs405
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Block.hs369
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Connection.hs1012
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Download.hs296
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Manager.hs62
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Message.hs1237
-rw-r--r--bittorrent/src/Network/BitTorrent/Exchange/Session.hs586
-rw-r--r--bittorrent/src/Network/BitTorrent/Internal/Cache.hs169
-rw-r--r--bittorrent/src/Network/BitTorrent/Internal/Progress.hs154
-rw-r--r--bittorrent/src/Network/BitTorrent/Internal/Types.hs10
-rw-r--r--bittorrent/src/Network/BitTorrent/Readme.md10
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker.hs51
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/List.hs197
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/Message.hs925
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/RPC.hs175
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/RPC/HTTP.hs191
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/RPC/UDP.hs454
-rw-r--r--bittorrent/src/Network/BitTorrent/Tracker/Session.hs306
22 files changed, 0 insertions, 7190 deletions
diff --git a/bittorrent/src/Network/BitTorrent/Client.hs b/bittorrent/src/Network/BitTorrent/Client.hs
deleted file mode 100644
index c84290dd..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Client/Handle.hs b/bittorrent/src/Network/BitTorrent/Client/Handle.hs
deleted file mode 100644
index 66baac48..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Client/Types.hs b/bittorrent/src/Network/BitTorrent/Client/Types.hs
deleted file mode 100644
index e2ad858f..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange.hs b/bittorrent/src/Network/BitTorrent/Exchange.hs
deleted file mode 100644
index 143bf090..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange/Bitfield.hs b/bittorrent/src/Network/BitTorrent/Exchange/Bitfield.hs
deleted file mode 100644
index 1be9f970..00000000
--- a/bittorrent/src/Network/BitTorrent/Exchange/Bitfield.hs
+++ /dev/null
@@ -1,405 +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 selection 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 a restriction:
112-- each integer in the set should be within the given interval. The greatest
113-- lower bound of the interval must be zero, so intervals may be specified by
114-- providing a maximum set size. For example, a bitfield of size 10 might
115-- contain only indices in interval [0..9].
116--
117-- By convention, we use the following aliases for Int:
118--
119-- [ PieceIx ] an Int member of the Bitfield.
120--
121-- [ PieceCount ] maximum set size for a Bitfield.
122data Bitfield = Bitfield {
123 bfSize :: !PieceCount
124 , bfSet :: !IntSet
125 } deriving (Show, Read, Eq)
126
127-- Invariants: all elements of bfSet lie in [0..bfSize - 1];
128
129instance Monoid Bitfield where
130 {-# SPECIALIZE instance Monoid Bitfield #-}
131 mempty = haveNone 0
132 mappend = union
133 mconcat = unions
134
135{-----------------------------------------------------------------------
136 Construction
137-----------------------------------------------------------------------}
138
139-- | The empty bitfield of the given size.
140haveNone :: PieceCount -> Bitfield
141haveNone s = Bitfield s S.empty
142
143-- | The full bitfield containing all piece indices for the given size.
144haveAll :: PieceCount -> Bitfield
145haveAll s = Bitfield s (S.interval 0 (s - 1))
146
147-- | Insert the index in the set ignoring out of range indices.
148have :: PieceIx -> Bitfield -> Bitfield
149have ix Bitfield {..}
150 | 0 <= ix && ix < bfSize = Bitfield bfSize (S.insert ix bfSet)
151 | otherwise = Bitfield bfSize bfSet
152
153singleton :: PieceIx -> PieceCount -> Bitfield
154singleton ix pc = have ix (haveNone pc)
155
156-- | Assign new size to bitfield. FIXME Normally, size should be only
157-- decreased, otherwise exception raised.
158adjustSize :: PieceCount -> Bitfield -> Bitfield
159adjustSize s Bitfield {..} = Bitfield s bfSet
160
161-- | NOTE: for internal use only
162interval :: PieceCount -> PieceIx -> PieceIx -> Bitfield
163interval pc a b = Bitfield pc (S.interval a b)
164
165{-----------------------------------------------------------------------
166 Query
167-----------------------------------------------------------------------}
168
169-- | Test if bitifield have no one index: peer do not have anything.
170null :: Bitfield -> Bool
171null Bitfield {..} = S.null bfSet
172
173-- | Test if bitfield have all pieces.
174full :: Bitfield -> Bool
175full Bitfield {..} = S.size bfSet == bfSize
176
177-- | Count of peer have pieces.
178haveCount :: Bitfield -> PieceCount
179haveCount = S.size . bfSet
180
181-- | Total count of pieces and its indices.
182totalCount :: Bitfield -> PieceCount
183totalCount = bfSize
184
185-- | Ratio of /have/ piece count to the /total/ piece count.
186--
187-- > forall bf. 0 <= completeness bf <= 1
188--
189completeness :: Bitfield -> Ratio PieceCount
190completeness b = haveCount b % totalCount b
191
192inRange :: PieceIx -> Bitfield -> Bool
193inRange ix Bitfield {..} = 0 <= ix && ix < bfSize
194
195member :: PieceIx -> Bitfield -> Bool
196member ix bf @ Bitfield {..}
197 | ix `inRange` bf = ix `S.member` bfSet
198 | otherwise = False
199
200notMember :: PieceIx -> Bitfield -> Bool
201notMember ix bf @ Bitfield {..}
202 | ix `inRange` bf = ix `S.notMember` bfSet
203 | otherwise = True
204
205-- | Find first available piece index.
206findMin :: Bitfield -> PieceIx
207findMin = S.findMin . bfSet
208{-# INLINE findMin #-}
209
210-- | Find last available piece index.
211findMax :: Bitfield -> PieceIx
212findMax = S.findMax . bfSet
213{-# INLINE findMax #-}
214
215-- | Check if all pieces from first bitfield present if the second bitfield
216isSubsetOf :: Bitfield -> Bitfield -> Bool
217isSubsetOf a b = bfSet a `S.isSubsetOf` bfSet b
218{-# INLINE isSubsetOf #-}
219
220-- | Resulting bitfield includes only missing pieces.
221complement :: Bitfield -> Bitfield
222complement Bitfield {..} = Bitfield
223 { bfSet = uni `S.difference` bfSet
224 , bfSize = bfSize
225 }
226 where
227 Bitfield _ uni = haveAll bfSize
228{-# INLINE complement #-}
229
230{-----------------------------------------------------------------------
231-- Availability
232-----------------------------------------------------------------------}
233
234-- | Frequencies are needed in piece selection startegies which use
235-- availability quantity to find out the optimal next piece index to
236-- download.
237type Frequency = Int
238
239-- TODO rename to availability
240-- | How many times each piece index occur in the given bitfield set.
241frequencies :: [Bitfield] -> Vector Frequency
242frequencies [] = V.fromList []
243frequencies xs = runST $ do
244 v <- VM.new size
245 VM.set v 0
246 forM_ xs $ \ Bitfield {..} -> do
247 forM_ (S.toList bfSet) $ \ x -> do
248 fr <- VM.read v x
249 VM.write v x (succ fr)
250 V.unsafeFreeze v
251 where
252 size = maximum (map bfSize xs)
253
254-- TODO it seems like this operation is veeery slow
255
256-- | Find least available piece index. If no piece available return
257-- 'Nothing'.
258rarest :: [Bitfield] -> Maybe PieceIx
259rarest xs
260 | V.null freqMap = Nothing
261 | otherwise
262 = Just $ fst $ V.ifoldr' minIx (0, freqMap V.! 0) freqMap
263 where
264 freqMap = frequencies xs
265 {-# NOINLINE freqMap #-}
266
267 minIx :: PieceIx -> Frequency
268 -> (PieceIx, Frequency)
269 -> (PieceIx, Frequency)
270 minIx ix fr acc@(_, fra)
271 | fr < fra && fr > 0 = (ix, fr)
272 | otherwise = acc
273
274
275{-----------------------------------------------------------------------
276 Combine
277-----------------------------------------------------------------------}
278
279insert :: PieceIx -> Bitfield -> Bitfield
280insert pix bf @ Bitfield {..}
281 | 0 <= pix && pix < bfSize = Bitfield
282 { bfSet = S.insert pix bfSet
283 , bfSize = bfSize
284 }
285 | otherwise = bf
286
287-- | Find indices at least one peer have.
288union :: Bitfield -> Bitfield -> Bitfield
289union a b = {-# SCC union #-} Bitfield {
290 bfSize = bfSize a `max` bfSize b
291 , bfSet = bfSet a `S.union` bfSet b
292 }
293
294-- | Find indices both peers have.
295intersection :: Bitfield -> Bitfield -> Bitfield
296intersection a b = {-# SCC intersection #-} Bitfield {
297 bfSize = bfSize a `min` bfSize b
298 , bfSet = bfSet a `S.intersection` bfSet b
299 }
300
301-- | Find indices which have first peer but do not have the second peer.
302difference :: Bitfield -> Bitfield -> Bitfield
303difference a b = {-# SCC difference #-} Bitfield {
304 bfSize = bfSize a -- FIXME is it reasonable?
305 , bfSet = bfSet a `S.difference` bfSet b
306 }
307
308-- | Find indices the any of the peers have.
309unions :: [Bitfield] -> Bitfield
310unions = {-# SCC unions #-} foldl' union (haveNone 0)
311
312{-----------------------------------------------------------------------
313 Serialization
314-----------------------------------------------------------------------}
315
316-- | List all /have/ indexes.
317toList :: Bitfield -> [PieceIx]
318toList Bitfield {..} = S.toList bfSet
319
320-- | Make bitfield from list of /have/ indexes.
321fromList :: PieceCount -> [PieceIx] -> Bitfield
322fromList s ixs = Bitfield {
323 bfSize = s
324 , bfSet = S.splitGT (-1) $ S.splitLT s $ S.fromList ixs
325 }
326
327-- | Unpack 'Bitfield' from tightly packed bit array. Note resulting
328-- size might be more than real bitfield size, use 'adjustSize'.
329fromBitmap :: ByteString -> Bitfield
330fromBitmap bs = {-# SCC fromBitmap #-} Bitfield {
331 bfSize = B.length bs * 8
332 , bfSet = S.fromByteString bs
333 }
334{-# INLINE fromBitmap #-}
335
336-- | Pack a 'Bitfield' to tightly packed bit array.
337toBitmap :: Bitfield -> Lazy.ByteString
338toBitmap Bitfield {..} = {-# SCC toBitmap #-} Lazy.fromChunks [intsetBM, alignment]
339 where
340 byteSize = bfSize `div` 8 + if bfSize `mod` 8 == 0 then 0 else 1
341 alignment = B.replicate (byteSize - B.length intsetBM) 0
342 intsetBM = S.toByteString bfSet
343
344{-----------------------------------------------------------------------
345-- Piece selection
346-----------------------------------------------------------------------}
347
348type Selector = Bitfield -- ^ Indices of client /have/ pieces.
349 -> Bitfield -- ^ Indices of peer /have/ pieces.
350 -> [Bitfield] -- ^ Indices of other peers /have/ pieces.
351 -> Maybe PieceIx -- ^ Zero-based index of piece to request
352 -- to, if any.
353
354selector :: Selector -- ^ Selector to use at the start.
355 -> Ratio PieceCount
356 -> Selector -- ^ Selector to use after the client have
357 -- the C pieces.
358 -> Selector -- ^ Selector that changes behaviour based
359 -- on completeness.
360selector start pt ready h a xs =
361 case strategyClass pt h of
362 SCBeginning -> start h a xs
363 SCReady -> ready h a xs
364 SCEnd -> endGame h a xs
365
366data StartegyClass
367 = SCBeginning
368 | SCReady
369 | SCEnd
370 deriving (Show, Eq, Ord, Enum, Bounded)
371
372
373strategyClass :: Ratio PieceCount -> Bitfield -> StartegyClass
374strategyClass threshold = classify . completeness
375 where
376 classify c
377 | c < threshold = SCBeginning
378 | c + 1 % numerator c < 1 = SCReady
379 -- FIXME numerator have is not total count
380 | otherwise = SCEnd
381
382
383-- | Select the first available piece.
384strictFirst :: Selector
385strictFirst h a _ = Just $ findMin (difference a h)
386
387-- | Select the last available piece.
388strictLast :: Selector
389strictLast h a _ = Just $ findMax (difference a h)
390
391-- |
392rarestFirst :: Selector
393rarestFirst h a xs = rarest (map (intersection want) xs)
394 where
395 want = difference h a
396
397-- | In average random first is faster than rarest first strategy but
398-- only if all pieces are available.
399randomFirst :: Selector
400randomFirst = do
401-- randomIO
402 error "TODO: randomFirst"
403
404endGame :: Selector
405endGame = strictLast
diff --git a/bittorrent/src/Network/BitTorrent/Exchange/Block.hs b/bittorrent/src/Network/BitTorrent/Exchange/Block.hs
deleted file mode 100644
index bc9a3d24..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange/Connection.hs b/bittorrent/src/Network/BitTorrent/Exchange/Connection.hs
deleted file mode 100644
index 6804d0a2..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange/Download.hs b/bittorrent/src/Network/BitTorrent/Exchange/Download.hs
deleted file mode 100644
index 981db2fb..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange/Manager.hs b/bittorrent/src/Network/BitTorrent/Exchange/Manager.hs
deleted file mode 100644
index 30a6a607..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Exchange/Message.hs b/bittorrent/src/Network/BitTorrent/Exchange/Message.hs
deleted file mode 100644
index 5c096523..00000000
--- a/bittorrent/src/Network/BitTorrent/Exchange/Message.hs
+++ /dev/null
@@ -1,1237 +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{-# LANGUAGE CPP #-}
36{-# OPTIONS -fno-warn-orphans #-}
37module Network.BitTorrent.Exchange.Message
38 ( -- * Capabilities
39 Capabilities (..)
40 , Extension (..)
41 , Caps
42
43 -- * Handshake
44 , ProtocolName
45 , Handshake(..)
46 , defaultHandshake
47 , handshakeSize
48 , handshakeMaxSize
49 , handshakeStats
50
51 -- * Stats
52 , ByteCount
53 , ByteStats (..)
54 , byteLength
55
56 -- * Messages
57 , Message (..)
58 , defaultKeepAliveTimeout
59 , defaultKeepAliveInterval
60 , PeerMessage (..)
61
62 -- ** Core messages
63 , StatusUpdate (..)
64 , Available (..)
65 , Transfer (..)
66 , defaultRequestQueueLength
67
68 -- ** Fast extension
69 , FastMessage (..)
70
71 -- ** Extension protocol
72 , ExtendedMessage (..)
73
74 -- *** Capabilities
75 , ExtendedExtension (..)
76 , ExtendedCaps (..)
77
78 -- *** Handshake
79 , ExtendedHandshake (..)
80 , defaultQueueLength
81 , nullExtendedHandshake
82
83 -- *** Metadata
84 , ExtendedMetadata (..)
85 , metadataPieceSize
86 , defaultMetadataFactor
87 , defaultMaxInfoDictSize
88 , isLastPiece
89 , isValidPiece
90 ) where
91
92import Control.Applicative
93import Control.Arrow ((&&&), (***))
94import Control.Monad (when)
95import Data.Attoparsec.ByteString.Char8 as BS
96import Data.BEncode as BE
97import Data.BEncode.BDict as BE
98import Data.BEncode.Internal as BE (ppBEncode, parser)
99import Data.BEncode.Types (BDict)
100import Data.Bits
101import Data.ByteString as BS
102import Data.ByteString.Char8 as BC
103import Data.ByteString.Lazy as BL
104import Data.Default
105import Data.List as L
106import Data.Map.Strict as M
107import Data.Maybe
108import Data.Monoid
109import Data.Ord
110import Data.Serialize as S
111import Data.String
112import Data.Text as T
113import Data.Typeable
114import Data.Word
115#if MIN_VERSION_iproute(1,7,4)
116import Data.IP hiding (fromSockAddr)
117#else
118import Data.IP
119#endif
120import Network
121import Network.Socket hiding (KeepAlive)
122import Text.PrettyPrint as PP hiding ((<>))
123import Text.PrettyPrint.HughesPJClass hiding ((<>),($$))
124
125import Data.Torrent hiding (Piece (..))
126import qualified Data.Torrent as P (Piece (..))
127import Network.Address
128import Network.BitTorrent.Exchange.Bitfield
129import Network.BitTorrent.Exchange.Block
130
131{-----------------------------------------------------------------------
132-- Capabilities
133-----------------------------------------------------------------------}
134
135-- |
136class Capabilities caps where
137 type Ext caps :: *
138
139 -- | Pack extensions to caps.
140 toCaps :: [Ext caps] -> caps
141
142 -- | Unpack extensions from caps.
143 fromCaps :: caps -> [Ext caps]
144
145 -- | Check if an extension is a member of the specified set.
146 allowed :: Ext caps -> caps -> Bool
147
148ppCaps :: Capabilities caps => Pretty (Ext caps) => caps -> Doc
149ppCaps = hcat . punctuate ", " . L.map pPrint . fromCaps
150
151{-----------------------------------------------------------------------
152-- Extensions
153-----------------------------------------------------------------------}
154
155-- | Enumeration of message extension protocols.
156--
157-- For more info see: <http://www.bittorrent.org/beps/bep_0004.html>
158--
159data Extension
160 = ExtDHT -- ^ BEP 5: allow to send PORT messages.
161 | ExtFast -- ^ BEP 6: allow to send FAST messages.
162 | ExtExtended -- ^ BEP 10: allow to send the extension protocol messages.
163 deriving (Show, Eq, Ord, Enum, Bounded)
164
165-- | Full extension names, suitable for logging.
166instance Pretty Extension where
167 pPrint ExtDHT = "Distributed Hash Table Protocol"
168 pPrint ExtFast = "Fast Extension"
169 pPrint ExtExtended = "Extension Protocol"
170
171-- | Extension bitmask as specified by BEP 4.
172extMask :: Extension -> Word64
173extMask ExtDHT = 0x01
174extMask ExtFast = 0x04
175extMask ExtExtended = 0x100000
176
177{-----------------------------------------------------------------------
178-- Capabilities
179-----------------------------------------------------------------------}
180
181-- | Capabilities is a set of 'Extension's usually sent in 'Handshake'
182-- messages.
183newtype Caps = Caps Word64
184 deriving (Show, Eq)
185
186-- | Render set of extensions as comma separated list.
187instance Pretty Caps where
188 pPrint = ppCaps
189 {-# INLINE pPrint #-}
190
191-- | The empty set.
192instance Default Caps where
193 def = Caps 0
194 {-# INLINE def #-}
195
196-- | Monoid under intersection. 'mempty' includes all known extensions.
197instance Monoid Caps where
198 mempty = toCaps [minBound .. maxBound]
199 {-# INLINE mempty #-}
200
201 mappend (Caps a) (Caps b) = Caps (a .&. b)
202 {-# INLINE mappend #-}
203
204-- | 'Handshake' compatible encoding.
205instance Serialize Caps where
206 put (Caps caps) = S.putWord64be caps
207 {-# INLINE put #-}
208
209 get = Caps <$> S.getWord64be
210 {-# INLINE get #-}
211
212instance Capabilities Caps where
213 type Ext Caps = Extension
214
215 allowed e (Caps caps) = (extMask e .&. caps) /= 0
216 {-# INLINE allowed #-}
217
218 toCaps = Caps . L.foldr (.|.) 0 . L.map extMask
219 fromCaps caps = L.filter (`allowed` caps) [minBound..maxBound]
220
221{-----------------------------------------------------------------------
222 Handshake
223-----------------------------------------------------------------------}
224
225maxProtocolNameSize :: Word8
226maxProtocolNameSize = maxBound
227
228-- | The protocol name is used to identify to the local peer which
229-- version of BTP the remote peer uses.
230newtype ProtocolName = ProtocolName BS.ByteString
231 deriving (Eq, Ord, Typeable)
232
233-- | In BTP/1.0 the name is 'BitTorrent protocol'. If this string is
234-- different from the local peers own protocol name, then the
235-- connection is to be dropped.
236instance Default ProtocolName where
237 def = ProtocolName "BitTorrent protocol"
238
239instance Show ProtocolName where
240 show (ProtocolName bs) = show bs
241
242instance Pretty ProtocolName where
243 pPrint (ProtocolName bs) = PP.text $ BC.unpack bs
244
245instance IsString ProtocolName where
246 fromString str
247 | L.length str <= fromIntegral maxProtocolNameSize
248 = ProtocolName (fromString str)
249 | otherwise = error $ "fromString: ProtocolName too long: " ++ str
250
251instance Serialize ProtocolName where
252 put (ProtocolName bs) = do
253 putWord8 $ fromIntegral $ BS.length bs
254 putByteString bs
255
256 get = do
257 len <- getWord8
258 bs <- getByteString $ fromIntegral len
259 return (ProtocolName bs)
260
261-- | Handshake message is used to exchange all information necessary
262-- to establish connection between peers.
263--
264data Handshake = Handshake {
265 -- | Identifier of the protocol. This is usually equal to 'def'.
266 hsProtocol :: ProtocolName
267
268 -- | Reserved bytes used to specify supported BEP's.
269 , hsReserved :: Caps
270
271 -- | Info hash of the info part of the metainfo file. that is
272 -- transmitted in tracker requests. Info hash of the initiator
273 -- handshake and response handshake should match, otherwise
274 -- initiator should break the connection.
275 --
276 , hsInfoHash :: InfoHash
277
278 -- | Peer id of the initiator. This is usually the same peer id
279 -- that is transmitted in tracker requests.
280 --
281 , hsPeerId :: PeerId
282
283 } deriving (Show, Eq)
284
285instance Serialize Handshake where
286 put Handshake {..} = do
287 put hsProtocol
288 put hsReserved
289 put hsInfoHash
290 put hsPeerId
291 get = Handshake <$> get <*> get <*> get <*> get
292
293-- | Show handshake protocol string, caps and fingerprint.
294instance Pretty Handshake where
295 pPrint Handshake {..}
296 = pPrint hsProtocol $$
297 pPrint hsReserved $$
298 pPrint (fingerprint hsPeerId)
299
300-- | Get handshake message size in bytes from the length of protocol
301-- string.
302handshakeSize :: Word8 -> Int
303handshakeSize n = 1 + fromIntegral n + 8 + 20 + 20
304
305-- | Maximum size of handshake message in bytes.
306handshakeMaxSize :: Int
307handshakeMaxSize = handshakeSize maxProtocolNameSize
308
309-- | Handshake with default protocol string and reserved bitmask.
310defaultHandshake :: InfoHash -> PeerId -> Handshake
311defaultHandshake = Handshake def def
312
313handshakeStats :: Handshake -> ByteStats
314handshakeStats (Handshake (ProtocolName bs) _ _ _)
315 = ByteStats 1 (BS.length bs + 8 + 20 + 20) 0
316
317{-----------------------------------------------------------------------
318-- Stats
319-----------------------------------------------------------------------}
320
321-- | Number of bytes.
322type ByteCount = Int
323
324-- | Summary of encoded message byte layout can be used to collect
325-- stats about message flow in both directions. This data can be
326-- retrieved using 'stats' function.
327data ByteStats = ByteStats
328 { -- | Number of bytes used to help encode 'control' and 'payload'
329 -- bytes: message size, message ID's, etc
330 overhead :: {-# UNPACK #-} !ByteCount
331
332 -- | Number of bytes used to exchange peers state\/options: piece
333 -- and block indexes, infohash, port numbers, peer ID\/IP, etc.
334 , control :: {-# UNPACK #-} !ByteCount
335
336 -- | Number of payload bytes: torrent data blocks and infodict
337 -- metadata.
338 , payload :: {-# UNPACK #-} !ByteCount
339 } deriving Show
340
341instance Pretty ByteStats where
342 pPrint s @ ByteStats {..} = fsep
343 [ PP.int overhead, "overhead"
344 , PP.int control, "control"
345 , PP.int payload, "payload"
346 , "bytes"
347 ] $+$ fsep
348 [ PP.int (byteLength s), "total bytes"
349 ]
350
351-- | Empty byte sequences.
352instance Default ByteStats where
353 def = ByteStats 0 0 0
354
355-- | Monoid under addition.
356instance Monoid ByteStats where
357 mempty = def
358 mappend a b = ByteStats
359 { overhead = overhead a + overhead b
360 , control = control a + control b
361 , payload = payload a + payload b
362 }
363
364-- | Sum of the all byte sequences.
365byteLength :: ByteStats -> Int
366byteLength ByteStats {..} = overhead + control + payload
367
368{-----------------------------------------------------------------------
369-- Regular messages
370-----------------------------------------------------------------------}
371
372-- | Messages which can be sent after handshaking. Minimal complete
373-- definition: 'envelop'.
374class PeerMessage a where
375 -- | Construct a message to be /sent/. Note that if 'ExtendedCaps'
376 -- do not contain mapping for this message the default
377 -- 'ExtendedMessageId' is used.
378 envelop :: ExtendedCaps -- ^ The /receiver/ extended capabilities;
379 -> a -- ^ An regular message;
380 -> Message -- ^ Enveloped message to sent.
381
382 -- | Find out the extension this message belong to. Can be used to
383 -- check if this message is allowed to send\/recv in current
384 -- session.
385 requires :: a -> Maybe Extension
386 requires _ = Nothing
387
388 -- | Get sizes of overhead\/control\/payload byte sequences of
389 -- binary message representation without encoding message to binary
390 -- bytestring.
391 --
392 -- This function should obey one law:
393 --
394 -- * 'byteLength' ('stats' msg) == 'BL.length' ('encode' msg)
395 --
396 stats :: a -> ByteStats
397 stats _ = ByteStats 4 0 0
398
399{-----------------------------------------------------------------------
400-- Status messages
401-----------------------------------------------------------------------}
402
403-- | Notification that the sender have updated its
404-- 'Network.BitTorrent.Exchange.Status.PeerStatus'.
405data StatusUpdate
406 -- | Notification that the sender will not upload data to the
407 -- receiver until unchoking happen.
408 = Choking !Bool
409
410 -- | Notification that the sender is interested (or not interested)
411 -- in any of the receiver's data pieces.
412 | Interested !Bool
413 deriving (Show, Eq, Ord, Typeable)
414
415instance Pretty StatusUpdate where
416 pPrint (Choking False) = "not choking"
417 pPrint (Choking True ) = "choking"
418 pPrint (Interested False) = "not interested"
419 pPrint (Interested True ) = "interested"
420
421instance PeerMessage StatusUpdate where
422 envelop _ = Status
423 {-# INLINE envelop #-}
424
425 stats _ = ByteStats 4 1 0
426 {-# INLINE stats #-}
427
428{-----------------------------------------------------------------------
429-- Available messages
430-----------------------------------------------------------------------}
431
432-- | Messages used to inform receiver which pieces of the torrent
433-- sender have.
434data Available =
435 -- | Zero-based index of a piece that has just been successfully
436 -- downloaded and verified via the hash.
437 Have ! PieceIx
438
439 -- | The bitfield message may only be sent immediately after the
440 -- handshaking sequence is complete, and before any other message
441 -- are sent. If client have no pieces then bitfield need not to be
442 -- sent.
443 | Bitfield !Bitfield
444 deriving (Show, Eq)
445
446instance Pretty Available where
447 pPrint (Have ix ) = "Have" <+> int ix
448 pPrint (Bitfield _ ) = "Bitfield"
449
450instance PeerMessage Available where
451 envelop _ = Available
452 {-# INLINE envelop #-}
453
454 stats (Have _) = ByteStats (4 + 1) 4 0
455 stats (Bitfield bf) = ByteStats (4 + 1) (q + trailing) 0
456 where
457 trailing = if r == 0 then 0 else 1
458 (q, r) = quotRem (totalCount bf) 8
459
460{-----------------------------------------------------------------------
461-- Transfer messages
462-----------------------------------------------------------------------}
463
464-- | Messages used to transfer 'Block's.
465data Transfer
466 -- | Request for a particular block. If a client is requested a
467 -- block that another peer do not have the peer might not answer
468 -- at all.
469 = Request ! BlockIx
470
471 -- | Response to a request for a block.
472 | Piece !(Block BL.ByteString)
473
474 -- | Used to cancel block requests. It is typically used during
475 -- "End Game".
476 | Cancel !BlockIx
477 deriving (Show, Eq)
478
479instance Pretty Transfer where
480 pPrint (Request ix ) = "Request" <+> pPrint ix
481 pPrint (Piece blk) = "Piece" <+> pPrint blk
482 pPrint (Cancel i ) = "Cancel" <+> pPrint i
483
484instance PeerMessage Transfer where
485 envelop _ = Transfer
486 {-# INLINE envelop #-}
487
488 stats (Request _ ) = ByteStats (4 + 1) (3 * 4) 0
489 stats (Piece p ) = ByteStats (4 + 1) (4 + 4 + blockSize p) 0
490 stats (Cancel _ ) = ByteStats (4 + 1) (3 * 4) 0
491
492-- TODO increase
493-- | Max number of pending 'Request's inflight.
494defaultRequestQueueLength :: Int
495defaultRequestQueueLength = 1
496
497{-----------------------------------------------------------------------
498-- Fast messages
499-----------------------------------------------------------------------}
500
501-- | BEP6 messages.
502data FastMessage =
503 -- | If a peer have all pieces it might send the 'HaveAll' message
504 -- instead of 'Bitfield' message. Used to save bandwidth.
505 HaveAll
506
507 -- | If a peer have no pieces it might send 'HaveNone' message
508 -- intead of 'Bitfield' message. Used to save bandwidth.
509 | HaveNone
510
511 -- | This is an advisory message meaning "you might like to
512 -- download this piece." Used to avoid excessive disk seeks and
513 -- amount of IO.
514 | SuggestPiece !PieceIx
515
516 -- | Notifies a requesting peer that its request will not be
517 -- satisfied.
518 | RejectRequest !BlockIx
519
520 -- | This is an advisory messsage meaning \"if you ask for this
521 -- piece, I'll give it to you even if you're choked.\" Used to
522 -- shorten starting phase.
523 | AllowedFast !PieceIx
524 deriving (Show, Eq)
525
526instance Pretty FastMessage where
527 pPrint (HaveAll ) = "Have all"
528 pPrint (HaveNone ) = "Have none"
529 pPrint (SuggestPiece pix) = "Suggest" <+> int pix
530 pPrint (RejectRequest bix) = "Reject" <+> pPrint bix
531 pPrint (AllowedFast pix) = "Allowed fast" <+> int pix
532
533instance PeerMessage FastMessage where
534 envelop _ = Fast
535 {-# INLINE envelop #-}
536
537 requires _ = Just ExtFast
538 {-# INLINE requires #-}
539
540 stats HaveAll = ByteStats 4 1 0
541 stats HaveNone = ByteStats 4 1 0
542 stats (SuggestPiece _) = ByteStats 5 4 0
543 stats (RejectRequest _) = ByteStats 5 12 0
544 stats (AllowedFast _) = ByteStats 5 4 0
545
546{-----------------------------------------------------------------------
547-- Extension protocol
548-----------------------------------------------------------------------}
549
550{-----------------------------------------------------------------------
551-- Extended capabilities
552-----------------------------------------------------------------------}
553
554data ExtendedExtension
555 = ExtMetadata -- ^ BEP 9: Extension for Peers to Send Metadata Files
556 deriving (Show, Eq, Ord, Enum, Bounded, Typeable)
557
558instance IsString ExtendedExtension where
559 fromString = fromMaybe (error msg) . fromKey . fromString
560 where
561 msg = "fromString: could not parse ExtendedExtension"
562
563instance Pretty ExtendedExtension where
564 pPrint ExtMetadata = "Extension for Peers to Send Metadata Files"
565
566fromKey :: BKey -> Maybe ExtendedExtension
567fromKey "ut_metadata" = Just ExtMetadata
568fromKey _ = Nothing
569{-# INLINE fromKey #-}
570
571toKey :: ExtendedExtension -> BKey
572toKey ExtMetadata = "ut_metadata"
573{-# INLINE toKey #-}
574
575type ExtendedMessageId = Word8
576
577extId :: ExtendedExtension -> ExtendedMessageId
578extId ExtMetadata = 1
579{-# INLINE extId #-}
580
581type ExtendedMap = Map ExtendedExtension ExtendedMessageId
582
583-- | The extension IDs must be stored for every peer, because every
584-- peer may have different IDs for the same extension.
585--
586newtype ExtendedCaps = ExtendedCaps { extendedCaps :: ExtendedMap }
587 deriving (Show, Eq)
588
589instance Pretty ExtendedCaps where
590 pPrint = ppCaps
591 {-# INLINE pPrint #-}
592
593-- | The empty set.
594instance Default ExtendedCaps where
595 def = ExtendedCaps M.empty
596
597-- | Monoid under intersection:
598--
599-- * The 'mempty' caps includes all known extensions;
600--
601-- * the 'mappend' operation is NOT commutative: it return message
602-- id from the first caps for the extensions existing in both caps.
603--
604instance Monoid ExtendedCaps where
605 mempty = toCaps [minBound..maxBound]
606 mappend (ExtendedCaps a) (ExtendedCaps b) =
607 ExtendedCaps (M.intersection a b)
608
609appendBDict :: BDict -> ExtendedMap -> ExtendedMap
610appendBDict (Cons key val xs) caps
611 | Just ext <- fromKey key
612 , Right eid <- fromBEncode val = M.insert ext eid (appendBDict xs caps)
613 | otherwise = appendBDict xs caps
614appendBDict Nil caps = caps
615
616-- | Handshake compatible encoding.
617instance BEncode ExtendedCaps where
618 toBEncode = BDict . BE.fromAscList . L.sortBy (comparing fst)
619 . L.map (toKey *** toBEncode) . M.toList . extendedCaps
620
621 fromBEncode (BDict bd) = pure $ ExtendedCaps $ appendBDict bd M.empty
622 fromBEncode _ = decodingError "ExtendedCaps"
623
624instance Capabilities ExtendedCaps where
625 type Ext ExtendedCaps = ExtendedExtension
626
627 toCaps = ExtendedCaps . M.fromList . L.map (id &&& extId)
628
629 fromCaps = M.keys . extendedCaps
630 {-# INLINE fromCaps #-}
631
632 allowed e (ExtendedCaps caps) = M.member e caps
633 {-# INLINE allowed #-}
634
635remoteMessageId :: ExtendedExtension -> ExtendedCaps -> ExtendedMessageId
636remoteMessageId ext = fromMaybe (extId ext) . M.lookup ext . extendedCaps
637
638{-----------------------------------------------------------------------
639-- Extended handshake
640-----------------------------------------------------------------------}
641
642-- | This message should be sent immediately after the standard
643-- bittorrent handshake to any peer that supports this extension
644-- protocol. Extended handshakes can be sent more than once, however
645-- an implementation may choose to ignore subsequent handshake
646-- messages.
647--
648data ExtendedHandshake = ExtendedHandshake
649 { -- | If this peer has an IPv4 interface, this is the compact
650 -- representation of that address.
651 ehsIPv4 :: Maybe HostAddress
652
653 -- | If this peer has an IPv6 interface, this is the compact
654 -- representation of that address.
655 , ehsIPv6 :: Maybe HostAddress6
656
657 -- | Dictionary of supported extension messages which maps names
658 -- of extensions to an extended message ID for each extension
659 -- message.
660 , ehsCaps :: ExtendedCaps
661
662 -- | Size of 'Data.Torrent.InfoDict' in bytes. This field should
663 -- be added if 'ExtMetadata' is enabled in current session /and/
664 -- peer have the torrent file.
665 , ehsMetadataSize :: Maybe Int
666
667 -- | Local TCP /listen/ port. Allows each side to learn about the
668 -- TCP port number of the other side.
669 , ehsPort :: Maybe PortNumber
670
671 -- | Request queue the number of outstanding 'Request' messages
672 -- this client supports without dropping any.
673 , ehsQueueLength :: Maybe Int
674
675 -- | Client name and version.
676 , ehsVersion :: Maybe Text
677
678 -- | IP of the remote end
679 , ehsYourIp :: Maybe IP
680 } deriving (Show, Eq, Typeable)
681
682extHandshakeId :: ExtendedMessageId
683extHandshakeId = 0
684
685-- | Default 'Request' queue size.
686defaultQueueLength :: Int
687defaultQueueLength = 1
688
689-- | All fields are empty.
690instance Default ExtendedHandshake where
691 def = ExtendedHandshake def def def def def def def def
692
693instance Monoid ExtendedHandshake where
694 mempty = def { ehsCaps = mempty }
695 mappend old new = ExtendedHandshake {
696 ehsCaps = ehsCaps old <> ehsCaps new,
697 ehsIPv4 = ehsIPv4 old `mergeOld` ehsIPv4 new,
698 ehsIPv6 = ehsIPv6 old `mergeOld` ehsIPv6 new,
699 ehsMetadataSize = ehsMetadataSize old `mergeNew` ehsMetadataSize new,
700 ehsPort = ehsPort old `mergeOld` ehsPort new,
701 ehsQueueLength = ehsQueueLength old `mergeNew` ehsQueueLength new,
702 ehsVersion = ehsVersion old `mergeOld` ehsVersion new,
703 ehsYourIp = ehsYourIp old `mergeOld` ehsYourIp new
704 }
705 where
706 mergeOld mold mnew = mold <|> mnew
707 mergeNew mold mnew = mnew <|> mold
708
709
710instance BEncode ExtendedHandshake where
711 toBEncode ExtendedHandshake {..} = toDict $
712 "ipv4" .=? (S.encode <$> ehsIPv4)
713 .: "ipv6" .=? (S.encode <$> ehsIPv6)
714 .: "m" .=! ehsCaps
715 .: "metadata_size" .=? ehsMetadataSize
716 .: "p" .=? ehsPort
717 .: "reqq" .=? ehsQueueLength
718 .: "v" .=? ehsVersion
719 .: "yourip" .=? (runPut <$> either put put <$> toEither <$> ehsYourIp)
720 .: endDict
721 where
722 toEither (IPv4 v4) = Left v4
723 toEither (IPv6 v6) = Right v6
724
725 fromBEncode = fromDict $ ExtendedHandshake
726 <$>? "ipv4"
727 <*>? "ipv6"
728 <*>! "m"
729 <*>? "metadata_size"
730 <*>? "p"
731 <*>? "reqq"
732 <*>? "v"
733 <*> (opt "yourip" >>= getYourIp)
734
735getYourIp :: Maybe BValue -> BE.Get (Maybe IP)
736getYourIp f =
737 return $ do
738 BString ip <- f
739 either (const Nothing) Just $
740 case BS.length ip of
741 4 -> IPv4 <$> S.decode ip
742 16 -> IPv6 <$> S.decode ip
743 _ -> fail ""
744
745instance Pretty ExtendedHandshake where
746 pPrint = PP.text . show
747
748-- | NOTE: Approximated 'stats'.
749instance PeerMessage ExtendedHandshake where
750 envelop c = envelop c . EHandshake
751 {-# INLINE envelop #-}
752
753 requires _ = Just ExtExtended
754 {-# INLINE requires #-}
755
756 stats _ = ByteStats (4 + 1 + 1) 100 {- is it ok? -} 0 -- FIXME
757 {-# INLINE stats #-}
758
759-- | Set default values and the specified 'ExtendedCaps'.
760nullExtendedHandshake :: ExtendedCaps -> ExtendedHandshake
761nullExtendedHandshake caps = ExtendedHandshake
762 { ehsIPv4 = Nothing
763 , ehsIPv6 = Nothing
764 , ehsCaps = caps
765 , ehsMetadataSize = Nothing
766 , ehsPort = Nothing
767 , ehsQueueLength = Just defaultQueueLength
768 , ehsVersion = Just $ T.pack $ render $ pPrint libFingerprint
769 , ehsYourIp = Nothing
770 }
771
772{-----------------------------------------------------------------------
773-- Metadata exchange extension
774-----------------------------------------------------------------------}
775
776-- | A peer MUST verify that any piece it sends passes the info-hash
777-- verification. i.e. until the peer has the entire metadata, it
778-- cannot run SHA-1 to verify that it yields the same hash as the
779-- info-hash.
780--
781data ExtendedMetadata
782 -- | This message requests the a specified metadata piece. The
783 -- response to this message, from a peer supporting the extension,
784 -- is either a 'MetadataReject' or a 'MetadataData' message.
785 = MetadataRequest PieceIx
786
787 -- | If sender requested a valid 'PieceIx' and receiver have the
788 -- corresponding piece then receiver should respond with this
789 -- message.
790 | MetadataData
791 { -- | A piece of 'Data.Torrent.InfoDict'.
792 piece :: P.Piece BS.ByteString
793
794 -- | This key has the same semantics as the 'ehsMetadataSize' in
795 -- the 'ExtendedHandshake' — it is size of the torrent info
796 -- dict.
797 , totalSize :: Int
798 }
799
800 -- | Peers that do not have the entire metadata MUST respond with
801 -- a reject message to any metadata request.
802 --
803 -- Clients MAY implement flood protection by rejecting request
804 -- messages after a certain number of them have been
805 -- served. Typically the number of pieces of metadata times a
806 -- factor.
807 | MetadataReject PieceIx
808
809 -- | Reserved. By specification we should ignore unknown metadata
810 -- messages.
811 | MetadataUnknown BValue
812 deriving (Show, Eq, Typeable)
813
814-- | Extended metadata message id used in 'msg_type_key'.
815type MetadataId = Int
816
817msg_type_key, piece_key, total_size_key :: BKey
818msg_type_key = "msg_type"
819piece_key = "piece"
820total_size_key = "total_size"
821
822-- | BEP9 compatible encoding.
823instance BEncode ExtendedMetadata where
824 toBEncode (MetadataRequest pix) = toDict $
825 msg_type_key .=! (0 :: MetadataId)
826 .: piece_key .=! pix
827 .: endDict
828 toBEncode (MetadataData (P.Piece pix _) totalSize) = toDict $
829 msg_type_key .=! (1 :: MetadataId)
830 .: piece_key .=! pix
831 .: total_size_key .=! totalSize
832 .: endDict
833 toBEncode (MetadataReject pix) = toDict $
834 msg_type_key .=! (2 :: MetadataId)
835 .: piece_key .=! pix
836 .: endDict
837 toBEncode (MetadataUnknown bval) = bval
838
839 fromBEncode bval = (`fromDict` bval) $ do
840 mid <- field $ req msg_type_key
841 case mid :: MetadataId of
842 0 -> MetadataRequest <$>! piece_key
843 1 -> metadataData <$>! piece_key <*>! total_size_key
844 2 -> MetadataReject <$>! piece_key
845 _ -> pure (MetadataUnknown bval)
846 where
847 metadataData pix s = MetadataData (P.Piece pix BS.empty) s
848
849-- | Piece data bytes are omitted.
850instance Pretty ExtendedMetadata where
851 pPrint (MetadataRequest pix ) = "Request" <+> PP.int pix
852 pPrint (MetadataData p t) = "Data" <+> pPrint p <+> PP.int t
853 pPrint (MetadataReject pix ) = "Reject" <+> PP.int pix
854 pPrint (MetadataUnknown bval ) = "Unknown" <+> ppBEncode bval
855
856-- | NOTE: Approximated 'stats'.
857instance PeerMessage ExtendedMetadata where
858 envelop c = envelop c . EMetadata (remoteMessageId ExtMetadata c)
859 {-# INLINE envelop #-}
860
861 requires _ = Just ExtExtended
862 {-# INLINE requires #-}
863
864 stats (MetadataRequest _) = ByteStats (4 + 1 + 1) {- ~ -} 25 0
865 stats (MetadataData p _) = ByteStats (4 + 1 + 1) {- ~ -} 41 $
866 BS.length (P.pieceData p)
867 stats (MetadataReject _) = ByteStats (4 + 1 + 1) {- ~ -} 25 0
868 stats (MetadataUnknown _) = ByteStats (4 + 1 + 1) {- ? -} 0 0
869
870-- | All 'Piece's in 'MetadataData' messages MUST have size equal to
871-- this value. The last trailing piece can be shorter.
872metadataPieceSize :: PieceSize
873metadataPieceSize = 16 * 1024
874
875isLastPiece :: P.Piece a -> Int -> Bool
876isLastPiece P.Piece {..} total = succ pieceIndex == pcnt
877 where
878 pcnt = q + if r > 0 then 1 else 0
879 (q, r) = quotRem total metadataPieceSize
880
881-- TODO we can check if the piece payload bytestring have appropriate
882-- length; otherwise serialization MUST fail.
883isValidPiece :: P.Piece BL.ByteString -> Int -> Bool
884isValidPiece p @ P.Piece {..} total
885 | isLastPiece p total = pieceSize p <= metadataPieceSize
886 | otherwise = pieceSize p == metadataPieceSize
887
888setMetadataPayload :: BS.ByteString -> ExtendedMetadata -> ExtendedMetadata
889setMetadataPayload bs (MetadataData (P.Piece pix _) t) =
890 MetadataData (P.Piece pix bs) t
891setMetadataPayload _ msg = msg
892
893getMetadataPayload :: ExtendedMetadata -> Maybe BS.ByteString
894getMetadataPayload (MetadataData (P.Piece _ bs) _) = Just bs
895getMetadataPayload _ = Nothing
896
897-- | Metadata BDict usually contain only 'msg_type_key', 'piece_key'
898-- and 'total_size_key' fields so it normally should take less than
899-- 100 bytes. This limit is two order of magnitude larger to be
900-- permissive to 'MetadataUnknown' messages.
901--
902-- See 'maxMessageSize' for further explanation.
903--
904maxMetadataBDictSize :: Int
905maxMetadataBDictSize = 16 * 1024
906
907maxMetadataSize :: Int
908maxMetadataSize = maxMetadataBDictSize + metadataPieceSize
909
910-- to make MetadataData constructor fields a little bit prettier we
911-- cheat here: first we read empty 'pieceData' from bdict, but then we
912-- fill that field with the actual piece data — trailing bytes of
913-- the message
914getMetadata :: Int -> S.Get ExtendedMetadata
915getMetadata len
916 | len > maxMetadataSize = fail $ parseError "size exceeded limit"
917 | otherwise = do
918 bs <- getByteString len
919 parseRes $ BS.parse BE.parser bs
920 where
921 parseError reason = "unable to parse metadata message: " ++ reason
922
923 parseRes (BS.Fail _ _ m) = fail $ parseError $ "bdict: " ++ m
924 parseRes (BS.Partial _) = fail $ parseError "bdict: not enough bytes"
925 parseRes (BS.Done piece bvalueBS)
926 | BS.length piece > metadataPieceSize
927 = fail "infodict piece: size exceeded limit"
928 | otherwise = do
929 metadata <- either (fail . parseError) pure $ fromBEncode bvalueBS
930 return $ setMetadataPayload piece metadata
931
932putMetadata :: ExtendedMetadata -> BL.ByteString
933putMetadata msg
934 | Just bs <- getMetadataPayload msg = BE.encode msg <> BL.fromStrict bs
935 | otherwise = BE.encode msg
936
937-- | Allows a requesting peer to send 2 'MetadataRequest's for the
938-- each piece.
939--
940-- See 'Network.BitTorrent.Wire.Options.metadataFactor' for
941-- explanation why do we need this limit.
942defaultMetadataFactor :: Int
943defaultMetadataFactor = 2
944
945-- | Usually torrent size do not exceed 1MB. This value limit torrent
946-- /content/ size to about 8TB.
947--
948-- See 'Network.BitTorrent.Wire.Options.maxInfoDictSize' for
949-- explanation why do we need this limit.
950defaultMaxInfoDictSize :: Int
951defaultMaxInfoDictSize = 10 * 1024 * 1024
952
953{-----------------------------------------------------------------------
954-- Extension protocol messages
955-----------------------------------------------------------------------}
956
957-- | For more info see <http://www.bittorrent.org/beps/bep_0010.html>
958data ExtendedMessage
959 = EHandshake ExtendedHandshake
960 | EMetadata ExtendedMessageId ExtendedMetadata
961 | EUnknown ExtendedMessageId BS.ByteString
962 deriving (Show, Eq, Typeable)
963
964instance Pretty ExtendedMessage where
965 pPrint (EHandshake ehs) = pPrint ehs
966 pPrint (EMetadata _ msg) = "Metadata" <+> pPrint msg
967 pPrint (EUnknown mid _ ) = "Unknown" <+> PP.text (show mid)
968
969instance PeerMessage ExtendedMessage where
970 envelop _ = Extended
971 {-# INLINE envelop #-}
972
973 requires _ = Just ExtExtended
974 {-# INLINE requires #-}
975
976 stats (EHandshake hs) = stats hs
977 stats (EMetadata _ msg) = stats msg
978 stats (EUnknown _ msg) = ByteStats (4 + 1 + 1) (BS.length msg) 0
979
980{-----------------------------------------------------------------------
981-- The message datatype
982-----------------------------------------------------------------------}
983
984type MessageId = Word8
985
986-- | Messages used in communication between peers.
987--
988-- Note: If some extensions are disabled (not present in extension
989-- mask) and client receive message used by the disabled
990-- extension then the client MUST close the connection.
991--
992data Message
993 -- | Peers may close the TCP connection if they have not received
994 -- any messages for a given period of time, generally 2
995 -- minutes. Thus, the KeepAlive message is sent to keep the
996 -- connection between two peers alive, if no /other/ message has
997 -- been sent in a given period of time.
998 = KeepAlive
999 | Status !StatusUpdate -- ^ Messages used to update peer status.
1000 | Available !Available -- ^ Messages used to inform availability.
1001 | Transfer !Transfer -- ^ Messages used to transfer 'Block's.
1002
1003 -- | Peer receiving a handshake indicating the remote peer
1004 -- supports the 'ExtDHT' should send a 'Port' message. Peers that
1005 -- receive this message should attempt to ping the node on the
1006 -- received port and IP address of the remote peer.
1007 | Port !PortNumber
1008 | Fast !FastMessage
1009 | Extended !ExtendedMessage
1010 deriving (Show, Eq)
1011
1012instance Default Message where
1013 def = KeepAlive
1014 {-# INLINE def #-}
1015
1016-- | Payload bytes are omitted.
1017instance Pretty Message where
1018 pPrint (KeepAlive ) = "Keep alive"
1019 pPrint (Status m) = "Status" <+> pPrint m
1020 pPrint (Available m) = pPrint m
1021 pPrint (Transfer m) = pPrint m
1022 pPrint (Port p) = "Port" <+> int (fromEnum p)
1023 pPrint (Fast m) = pPrint m
1024 pPrint (Extended m) = pPrint m
1025
1026instance PeerMessage Message where
1027 envelop _ = id
1028 {-# INLINE envelop #-}
1029
1030 requires KeepAlive = Nothing
1031 requires (Status _) = Nothing
1032 requires (Available _) = Nothing
1033 requires (Transfer _) = Nothing
1034 requires (Port _) = Just ExtDHT
1035 requires (Fast _) = Just ExtFast
1036 requires (Extended _) = Just ExtExtended
1037
1038 stats KeepAlive = ByteStats 4 0 0
1039 stats (Status m) = stats m
1040 stats (Available m) = stats m
1041 stats (Transfer m) = stats m
1042 stats (Port _) = ByteStats 5 2 0
1043 stats (Fast m) = stats m
1044 stats (Extended m) = stats m
1045
1046-- | PORT message.
1047instance PeerMessage PortNumber where
1048 envelop _ = Port
1049 {-# INLINE envelop #-}
1050
1051 requires _ = Just ExtDHT
1052 {-# INLINE requires #-}
1053
1054-- | How long /this/ peer should wait before dropping connection, in
1055-- seconds.
1056defaultKeepAliveTimeout :: Int
1057defaultKeepAliveTimeout = 2 * 60
1058
1059-- | How often /this/ peer should send 'KeepAlive' messages, in
1060-- seconds.
1061defaultKeepAliveInterval :: Int
1062defaultKeepAliveInterval = 60
1063
1064getInt :: S.Get Int
1065getInt = fromIntegral <$> S.getWord32be
1066{-# INLINE getInt #-}
1067
1068putInt :: S.Putter Int
1069putInt = S.putWord32be . fromIntegral
1070{-# INLINE putInt #-}
1071
1072-- | This limit should protect against "out-of-memory" attacks: if a
1073-- malicious peer have sent a long varlength message then receiver can
1074-- accumulate too long bytestring in the 'Get'.
1075--
1076-- Normal messages should never exceed this limits.
1077--
1078-- See also 'maxBitfieldSize', 'maxBlockSize' limits.
1079--
1080maxMessageSize :: Int
1081maxMessageSize = 20 + 1024 * 1024
1082
1083-- | This also limit max torrent size to:
1084--
1085-- max_bitfield_size * piece_ix_per_byte * max_piece_size =
1086-- 2 ^ 20 * 8 * 1MB =
1087-- 8TB
1088--
1089maxBitfieldSize :: Int
1090maxBitfieldSize = 1024 * 1024
1091
1092getBitfield :: Int -> S.Get Bitfield
1093getBitfield len
1094 | len > maxBitfieldSize = fail "BITFIELD message size exceeded limit"
1095 | otherwise = fromBitmap <$> getByteString len
1096
1097maxBlockSize :: Int
1098maxBlockSize = 4 * defaultTransferSize
1099
1100getBlock :: Int -> S.Get (Block BL.ByteString)
1101getBlock len
1102 | len > maxBlockSize = fail "BLOCK message size exceeded limit"
1103 | otherwise = Block <$> getInt <*> getInt
1104 <*> getLazyByteString (fromIntegral len)
1105{-# INLINE getBlock #-}
1106
1107instance Serialize Message where
1108 get = do
1109 len <- getInt
1110
1111 when (len > maxMessageSize) $ do
1112 fail "message body size exceeded the limit"
1113
1114 if len == 0 then return KeepAlive
1115 else do
1116 mid <- S.getWord8
1117 case mid of
1118 0x00 -> return $ Status (Choking True)
1119 0x01 -> return $ Status (Choking False)
1120 0x02 -> return $ Status (Interested True)
1121 0x03 -> return $ Status (Interested False)
1122 0x04 -> (Available . Have) <$> getInt
1123 0x05 -> (Available . Bitfield) <$> getBitfield (pred len)
1124 0x06 -> (Transfer . Request) <$> S.get
1125 0x07 -> (Transfer . Piece) <$> getBlock (len - 9)
1126 0x08 -> (Transfer . Cancel) <$> S.get
1127 0x09 -> Port <$> S.get
1128 0x0D -> (Fast . SuggestPiece) <$> getInt
1129 0x0E -> return $ Fast HaveAll
1130 0x0F -> return $ Fast HaveNone
1131 0x10 -> (Fast . RejectRequest) <$> S.get
1132 0x11 -> (Fast . AllowedFast) <$> getInt
1133 0x14 -> Extended <$> getExtendedMessage (pred len)
1134 _ -> do
1135 rm <- S.remaining >>= S.getBytes
1136 fail $ "unknown message ID: " ++ show mid ++ "\n"
1137 ++ "remaining available bytes: " ++ show rm
1138
1139 put KeepAlive = putInt 0
1140 put (Status msg) = putStatus msg
1141 put (Available msg) = putAvailable msg
1142 put (Transfer msg) = putTransfer msg
1143 put (Port p ) = putPort p
1144 put (Fast msg) = putFast msg
1145 put (Extended m ) = putExtendedMessage m
1146
1147statusUpdateId :: StatusUpdate -> MessageId
1148statusUpdateId (Choking choking) = fromIntegral (0 + fromEnum choking)
1149statusUpdateId (Interested choking) = fromIntegral (2 + fromEnum choking)
1150
1151putStatus :: Putter StatusUpdate
1152putStatus su = do
1153 putInt 1
1154 putWord8 (statusUpdateId su)
1155
1156putAvailable :: Putter Available
1157putAvailable (Have i) = do
1158 putInt 5
1159 putWord8 0x04
1160 putInt i
1161putAvailable (Bitfield (toBitmap -> bs)) = do
1162 putInt $ 1 + fromIntegral (BL.length bs)
1163 putWord8 0x05
1164 putLazyByteString bs
1165
1166putBlock :: Putter (Block BL.ByteString)
1167putBlock Block {..} = do
1168 putInt blkPiece
1169 putInt blkOffset
1170 putLazyByteString blkData
1171
1172putTransfer :: Putter Transfer
1173putTransfer (Request blk) = putInt 13 >> S.putWord8 0x06 >> S.put blk
1174putTransfer (Piece blk) = do
1175 putInt (9 + blockSize blk)
1176 putWord8 0x07
1177 putBlock blk
1178putTransfer (Cancel blk) = putInt 13 >> S.putWord8 0x08 >> S.put blk
1179
1180putPort :: Putter PortNumber
1181putPort p = do
1182 putInt 3
1183 putWord8 0x09
1184 put p
1185
1186putFast :: Putter FastMessage
1187putFast HaveAll = putInt 1 >> putWord8 0x0E
1188putFast HaveNone = putInt 1 >> putWord8 0x0F
1189putFast (SuggestPiece pix) = putInt 5 >> putWord8 0x0D >> putInt pix
1190putFast (RejectRequest i ) = putInt 13 >> putWord8 0x10 >> put i
1191putFast (AllowedFast i ) = putInt 5 >> putWord8 0x11 >> putInt i
1192
1193maxEHandshakeSize :: Int
1194maxEHandshakeSize = 16 * 1024
1195
1196getExtendedHandshake :: Int -> S.Get ExtendedHandshake
1197getExtendedHandshake messageSize
1198 | messageSize > maxEHandshakeSize
1199 = fail "extended handshake size exceeded limit"
1200 | otherwise = do
1201 bs <- getByteString messageSize
1202 either fail pure $ BE.decode bs
1203
1204maxEUnknownSize :: Int
1205maxEUnknownSize = 64 * 1024
1206
1207getExtendedUnknown :: Int -> S.Get BS.ByteString
1208getExtendedUnknown len
1209 | len > maxEUnknownSize = fail "unknown extended message size exceeded limit"
1210 | otherwise = getByteString len
1211
1212getExtendedMessage :: Int -> S.Get ExtendedMessage
1213getExtendedMessage messageSize = do
1214 msgId <- getWord8
1215 let msgBodySize = messageSize - 1
1216 case msgId of
1217 0 -> EHandshake <$> getExtendedHandshake msgBodySize
1218 1 -> EMetadata msgId <$> getMetadata msgBodySize
1219 _ -> EUnknown msgId <$> getExtendedUnknown msgBodySize
1220
1221-- | By spec.
1222extendedMessageId :: MessageId
1223extendedMessageId = 20
1224
1225putExt :: ExtendedMessageId -> BL.ByteString -> Put
1226putExt mid lbs = do
1227 putWord32be $ fromIntegral (1 + 1 + BL.length lbs)
1228 putWord8 extendedMessageId
1229 putWord8 mid
1230 putLazyByteString lbs
1231
1232-- NOTE: in contrast to getExtendedMessage this function put length
1233-- and message id too!
1234putExtendedMessage :: Putter ExtendedMessage
1235putExtendedMessage (EHandshake hs) = putExt extHandshakeId $ BE.encode hs
1236putExtendedMessage (EMetadata mid msg) = putExt mid $ putMetadata msg
1237putExtendedMessage (EUnknown mid bs) = putExt mid $ BL.fromStrict bs
diff --git a/bittorrent/src/Network/BitTorrent/Exchange/Session.hs b/bittorrent/src/Network/BitTorrent/Exchange/Session.hs
deleted file mode 100644
index 38a3c3a6..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Internal/Cache.hs b/bittorrent/src/Network/BitTorrent/Internal/Cache.hs
deleted file mode 100644
index 8c74467a..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Internal/Progress.hs b/bittorrent/src/Network/BitTorrent/Internal/Progress.hs
deleted file mode 100644
index 6ac889e2..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Internal/Types.hs b/bittorrent/src/Network/BitTorrent/Internal/Types.hs
deleted file mode 100644
index d157db3e..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Readme.md b/bittorrent/src/Network/BitTorrent/Readme.md
deleted file mode 100644
index ebf9545e..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Tracker.hs b/bittorrent/src/Network/BitTorrent/Tracker.hs
deleted file mode 100644
index 1191f921..00000000
--- a/bittorrent/src/Network/BitTorrent/Tracker.hs
+++ /dev/null
@@ -1,51 +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 , trackers
27 , newSession
28 , closeSession
29 , withSession
30
31 -- ** Events
32 , AnnounceEvent (..)
33 , notify
34 , askPeers
35
36 -- ** Session state
37 , TrackerSession
38 , trackerPeers
39 , trackerScrape
40
41 , tryTakeData
42 , unsafeTryTakeData
43
44 , getSessionState
45 ) where
46
47import Network.BitTorrent.Internal.Cache (tryTakeData, unsafeTryTakeData)
48import Network.BitTorrent.Tracker.Message
49import Network.BitTorrent.Tracker.List
50import Network.BitTorrent.Tracker.RPC
51import Network.BitTorrent.Tracker.Session
diff --git a/bittorrent/src/Network/BitTorrent/Tracker/List.hs b/bittorrent/src/Network/BitTorrent/Tracker/List.hs
deleted file mode 100644
index 1507b4be..00000000
--- a/bittorrent/src/Network/BitTorrent/Tracker/List.hs
+++ /dev/null
@@ -1,197 +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 , trackers
20 , trackerList
21 , shuffleTiers
22 , mapWithURI
23 , Network.BitTorrent.Tracker.List.toList
24
25 -- * Traversals
26 , traverseAll
27 , traverseTiers
28 ) where
29
30import Prelude hiding (mapM, foldr)
31import Control.Arrow
32import Control.Applicative
33import Control.Exception
34import Data.Default
35import Data.List as L (map, elem, any, filter, null)
36import Data.Maybe
37import Data.Foldable
38import Data.Traversable
39import Network.URI
40import System.Random.Shuffle
41
42import Data.Torrent
43import Network.BitTorrent.Tracker.RPC as RPC
44
45{-----------------------------------------------------------------------
46-- Tracker list datatype
47-----------------------------------------------------------------------}
48
49type TierEntry a = (URI, a)
50type Tier a = [TierEntry a]
51
52-- | Tracker list is either a single tracker or list of tiers. All
53-- trackers in each tier must be checked before the client goes on to
54-- the next tier.
55data TrackerList a
56 = Announce (TierEntry a) -- ^ torrent file 'announce' field only
57 | TierList [Tier a] -- ^ torrent file 'announce-list' field only
58 deriving (Show, Eq)
59
60-- | Empty tracker list. Can be used for trackerless torrents.
61instance Default (TrackerList a) where
62 def = TierList []
63
64instance Functor TrackerList where
65 fmap f (Announce (uri, a)) = Announce (uri, f a)
66 fmap f (TierList a) = TierList (fmap (fmap (second f)) a)
67
68instance Foldable TrackerList where
69 foldr f z (Announce e ) = f (snd e) z
70 foldr f z (TierList xs) = foldr (flip (foldr (f . snd))) z xs
71
72_traverseEntry f (uri, a) = (,) uri <$> f a
73
74instance Traversable TrackerList where
75 traverse f (Announce e ) = Announce <$> _traverseEntry f e
76 traverse f (TierList xs) =
77 TierList <$> traverse (traverse (_traverseEntry f)) xs
78
79traverseWithURI :: Applicative f
80 => (TierEntry a -> f b) -> TrackerList a -> f (TrackerList b)
81traverseWithURI f (Announce (uri, a)) = (Announce . (,) uri) <$> f (uri, a)
82traverseWithURI f (TierList xxs ) =
83 TierList <$> traverse (traverse (traverseEntry f)) xxs
84 where
85 traverseEntry f (uri, a) = (,) uri <$> f (uri, a)
86
87{-----------------------------------------------------------------------
88-- List extraction
89-----------------------------------------------------------------------}
90-- BEP12 do not expose any restrictions for the content of
91-- 'announce-list' key - there are some /bad/ cases can happen with
92-- poorly designed or even malicious torrent creation software.
93--
94-- Bad case #1: announce-list is present, but empty.
95--
96-- { tAnnounce = Just "http://a.com"
97-- , tAnnounceList = Just [[]]
98-- }
99--
100-- Bad case #2: announce uri do not present in announce list.
101--
102-- { tAnnounce = Just "http://a.com"
103-- , tAnnounceList = Just [["udp://a.com"]]
104-- }
105--
106-- The addBackup function solves both problems by adding announce uri
107-- as backup tier.
108--
109addBackup :: [[URI]] -> URI -> [[URI]]
110addBackup tiers bkp
111 | L.any (L.elem bkp) tiers = tiers
112 | otherwise = tiers ++ [[bkp]]
113
114fixList :: Maybe [[URI]] -> Maybe URI -> Maybe [[URI]]
115fixList mxss mx = do
116 xss <- mxss
117 let xss' = L.filter (not . L.null) xss
118 return $ maybe xss' (addBackup xss') mx
119
120trackers :: [URI] -> TrackerList ()
121trackers uris = TierList $ map (\uri -> [(uri,())]) uris
122
123-- | Extract set of trackers from torrent file. The 'tAnnounce' key is
124-- only ignored if the 'tAnnounceList' key is present.
125trackerList :: Torrent -> TrackerList ()
126trackerList Torrent {..} = fromMaybe (TierList []) $ do
127 (TierList . tierList) <$> (tAnnounceList `fixList` tAnnounce)
128 <|> (Announce . nullEntry) <$> tAnnounce
129 where
130 nullEntry uri = (uri, ())
131 tierList = L.map (L.map nullEntry)
132
133-- | Shuffle /order of trackers/ in each tier, preserving original
134-- /order of tiers/. This can help to balance the load between the
135-- trackers.
136shuffleTiers :: TrackerList a -> IO (TrackerList a)
137shuffleTiers (Announce a ) = return (Announce a)
138shuffleTiers (TierList xs) = TierList <$> mapM shuffleM xs
139
140mapWithURI :: (URI -> a -> b) -> TrackerList a -> TrackerList b
141mapWithURI f (Announce (uri, a)) = Announce (uri, f uri a)
142mapWithURI f (TierList xs ) = TierList (L.map (L.map mapEntry) xs)
143 where
144 mapEntry (uri, a) = (uri, f uri a)
145
146toList :: TrackerList a -> [[TierEntry a]]
147toList (Announce e) = [[e]]
148toList (TierList xxs) = xxs
149
150{-----------------------------------------------------------------------
151-- Special traversals (suppressed RPC exceptions)
152-----------------------------------------------------------------------}
153
154catchRPC :: IO a -> IO a -> IO a
155catchRPC a b = catch a (f b)
156 where
157 f :: a -> RpcException -> a
158 f = const
159
160throwRPC :: String -> IO a
161throwRPC = throwIO . GenericException
162
163-- | Like 'traverse' but ignores 'RpcExceptions'.
164traverseAll :: (TierEntry a -> IO a) -> TrackerList a -> IO (TrackerList a)
165traverseAll action = traverseWithURI (action $?)
166 where
167 f $? x = catchRPC (f x) (return (snd x))
168
169-- | Like 'traverse' but put working trackers to the head of tiers.
170-- This can help to avoid exceessive requests to not available
171-- trackers at each reannounce. If no one action succeed then original
172-- list is returned.
173traverseTiers :: (TierEntry a -> IO a) -> TrackerList a -> IO (TrackerList a)
174traverseTiers action ts = catchRPC (goList ts) (return ts)
175 where
176 goList tl @ (Announce _ ) = traverseWithURI action tl
177 goList (TierList tiers) = TierList <$> goTiers (goTier []) tiers
178
179 goTiers _ [] = throwRPC "traverseTiers: no tiers"
180 goTiers f (x : xs) = catchRPC shortcut failback
181 where
182 shortcut = do
183 x' <- f x
184 return (x' : xs)
185
186 failback = do
187 xs' <- goTiers f xs
188 return (x : xs')
189
190 goTier _ [] = throwRPC "traverseTiers: no trackers in tier"
191 goTier failed ((uri, a) : as) = catchRPC shortcut failback
192 where
193 shortcut = do
194 a' <- action (uri, a)
195 return ((uri, a') : as ++ failed) -- failed trackers at the end
196
197 failback = goTier ((uri, a) : failed) as
diff --git a/bittorrent/src/Network/BitTorrent/Tracker/Message.hs b/bittorrent/src/Network/BitTorrent/Tracker/Message.hs
deleted file mode 100644
index ab492275..00000000
--- a/bittorrent/src/Network/BitTorrent/Tracker/Message.hs
+++ /dev/null
@@ -1,925 +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{-# LANGUAGE CPP #-}
31{-# OPTIONS -fno-warn-orphans #-}
32module Network.BitTorrent.Tracker.Message
33 ( -- * Announce
34 -- ** Query
35 AnnounceEvent (..)
36 , AnnounceQuery (..)
37 , renderAnnounceQuery
38 , ParamParseFailure
39 , parseAnnounceQuery
40
41 -- ** Info
42 , PeerList (..)
43 , getPeerList
44 , AnnounceInfo(..)
45 , defaultNumWant
46 , defaultMaxNumWant
47 , defaultReannounceInterval
48
49 -- * Scrape
50 -- ** Query
51 , ScrapeQuery
52 , renderScrapeQuery
53 , parseScrapeQuery
54
55 -- ** Info
56 , ScrapeEntry (..)
57 , ScrapeInfo
58
59 -- * HTTP specific
60 -- ** Routes
61 , PathPiece
62 , defaultAnnouncePath
63 , defaultScrapePath
64
65 -- ** Preferences
66 , AnnouncePrefs (..)
67 , renderAnnouncePrefs
68 , parseAnnouncePrefs
69
70 -- ** Request
71 , AnnounceRequest (..)
72 , parseAnnounceRequest
73 , renderAnnounceRequest
74
75 -- ** Response
76 , announceType
77 , scrapeType
78 , parseFailureStatus
79
80 -- ** Extra
81 , queryToSimpleQuery
82
83 -- * UDP specific
84 -- ** Connection
85 , ConnectionId
86 , initialConnectionId
87
88 -- ** Messages
89 , Request (..)
90 , Response (..)
91 , responseName
92
93 -- ** Transaction
94 , genTransactionId
95 , TransactionId
96 , Transaction (..)
97 )
98 where
99
100import Control.Applicative
101import Control.Monad
102import Data.BEncode as BE hiding (Result)
103import Data.BEncode.BDict as BE
104import Data.ByteString as BS
105import Data.ByteString.Char8 as BC
106import Data.Char as Char
107import Data.Convertible
108import Data.Default
109import Data.Either
110import Data.List as L
111import Data.Maybe
112import Data.Monoid
113import Data.Serialize as S hiding (Result)
114import Data.String
115import Data.Text (Text)
116import Data.Text.Encoding
117import Data.Typeable
118import Data.Word
119#if MIN_VERSION_iproute(1,7,4)
120import Data.IP hiding (fromSockAddr)
121#else
122import Data.IP
123#endif
124import Network
125import Network.HTTP.Types.QueryLike
126import Network.HTTP.Types.URI hiding (urlEncode)
127import Network.HTTP.Types.Status
128import Network.Socket hiding (Connected)
129import Numeric
130import System.Entropy
131import Text.Read (readMaybe)
132
133import Data.Torrent
134import Network.Address
135import Network.BitTorrent.Internal.Progress
136
137{-----------------------------------------------------------------------
138-- Events
139-----------------------------------------------------------------------}
140
141-- | Events are used to specify which kind of announce query is performed.
142data AnnounceEvent
143 -- | For the first request: when download first begins.
144 = Started
145
146 -- | This peer stopped downloading /and/ uploading the torrent or
147 -- just shutting down.
148 | Stopped
149
150 -- | This peer completed downloading the torrent. This only happen
151 -- right after last piece have been verified. No 'Completed' is
152 -- sent if the file was completed when 'Started'.
153 | Completed
154 deriving (Show, Read, Eq, Ord, Enum, Bounded, Typeable)
155
156-- | HTTP tracker protocol compatible encoding.
157instance QueryValueLike AnnounceEvent where
158 toQueryValue e = toQueryValue (Char.toLower x : xs)
159 where
160 (x : xs) = show e -- INVARIANT: this is always nonempty list
161
162type EventId = Word32
163
164-- | UDP tracker encoding event codes.
165eventId :: AnnounceEvent -> EventId
166eventId Completed = 1
167eventId Started = 2
168eventId Stopped = 3
169
170-- TODO add Regular event
171putEvent :: Putter (Maybe AnnounceEvent)
172putEvent Nothing = putWord32be 0
173putEvent (Just e) = putWord32be (eventId e)
174
175getEvent :: S.Get (Maybe AnnounceEvent)
176getEvent = do
177 eid <- getWord32be
178 case eid of
179 0 -> return Nothing
180 1 -> return $ Just Completed
181 2 -> return $ Just Started
182 3 -> return $ Just Stopped
183 _ -> fail "unknown event id"
184
185{-----------------------------------------------------------------------
186 Announce query
187-----------------------------------------------------------------------}
188-- TODO add &ipv6= and &ipv4= params to AnnounceQuery
189-- http://www.bittorrent.org/beps/bep_0007.html#announce-parameter
190
191-- | A tracker request is HTTP GET request; used to include metrics
192-- from clients that help the tracker keep overall statistics about
193-- the torrent. The most important, requests are used by the tracker
194-- to keep track lists of active peer for a particular torrent.
195--
196data AnnounceQuery = AnnounceQuery
197 {
198 -- | Hash of info part of the torrent usually obtained from
199 -- 'Torrent' or 'Magnet'.
200 reqInfoHash :: !InfoHash
201
202 -- | ID of the peer doing request.
203 , reqPeerId :: !PeerId
204
205 -- | Port to listen to for connections from other
206 -- peers. Tracker should respond with this port when
207 -- some /other/ peer request the tracker with the same info hash.
208 -- Normally, this port is choosed from 'defaultPorts'.
209 , reqPort :: !PortNumber
210
211 -- | Current progress of peer doing request.
212 , reqProgress :: !Progress
213
214 -- | The peer IP. Needed only when client communicated with
215 -- tracker throught a proxy.
216 , reqIP :: Maybe HostAddress
217
218 -- | Number of peers that the peers wants to receive from. It is
219 -- optional for trackers to honor this limit. See note for
220 -- 'defaultNumWant'.
221 , reqNumWant :: Maybe Int
222
223 -- | If not specified, the request is regular periodic
224 -- request. Regular request should be sent
225 , reqEvent :: Maybe AnnounceEvent
226 } deriving (Show, Eq, Typeable)
227
228-- | UDP tracker protocol compatible encoding.
229instance Serialize AnnounceQuery where
230 put AnnounceQuery {..} = do
231 put reqInfoHash
232 put reqPeerId
233 put reqProgress
234 putEvent reqEvent
235 putWord32host $ fromMaybe 0 reqIP
236 putWord32be $ 0 -- TODO what the fuck is "key"?
237 putWord32be $ fromIntegral $ fromMaybe (-1) reqNumWant
238
239 put reqPort
240
241 get = do
242 ih <- get
243 pid <- get
244
245 progress <- get
246
247 ev <- getEvent
248 ip <- getWord32be
249-- key <- getWord32be -- TODO
250 want <- getWord32be
251
252 port <- get
253
254 return $ AnnounceQuery {
255 reqInfoHash = ih
256 , reqPeerId = pid
257 , reqPort = port
258 , reqProgress = progress
259 , reqIP = if ip == 0 then Nothing else Just ip
260 , reqNumWant = if want == -1 then Nothing
261 else Just (fromIntegral want)
262 , reqEvent = ev
263 }
264
265instance QueryValueLike PortNumber where
266 toQueryValue = toQueryValue . show . fromEnum
267
268instance QueryValueLike Word32 where
269 toQueryValue = toQueryValue . show
270
271instance QueryValueLike Int where
272 toQueryValue = toQueryValue . show
273
274-- | HTTP tracker protocol compatible encoding.
275instance QueryLike AnnounceQuery where
276 toQuery AnnounceQuery {..} =
277 toQuery reqProgress ++
278 [ ("info_hash", toQueryValue reqInfoHash) -- TODO use 'paramName'
279 , ("peer_id" , toQueryValue reqPeerId)
280 , ("port" , toQueryValue reqPort)
281 , ("ip" , toQueryValue reqIP)
282 , ("numwant" , toQueryValue reqNumWant)
283 , ("event" , toQueryValue reqEvent)
284 ]
285
286-- | Filter @param=value@ pairs with the unset value.
287queryToSimpleQuery :: Query -> SimpleQuery
288queryToSimpleQuery = catMaybes . L.map f
289 where
290 f (_, Nothing) = Nothing
291 f (a, Just b ) = Just (a, b)
292
293-- | Encode announce query to query string.
294renderAnnounceQuery :: AnnounceQuery -> SimpleQuery
295renderAnnounceQuery = queryToSimpleQuery . toQuery
296
297data QueryParam
298 -- announce query
299 = ParamInfoHash
300 | ParamPeerId
301 | ParamPort
302 | ParamUploaded
303 | ParamLeft
304 | ParamDownloaded
305 | ParamIP
306 | ParamNumWant
307 | ParamEvent
308 -- announce query ext
309 | ParamCompact
310 | ParamNoPeerId
311 deriving (Show, Eq, Ord, Enum)
312
313paramName :: QueryParam -> BS.ByteString
314paramName ParamInfoHash = "info_hash"
315paramName ParamPeerId = "peer_id"
316paramName ParamPort = "port"
317paramName ParamUploaded = "uploaded"
318paramName ParamLeft = "left"
319paramName ParamDownloaded = "downloaded"
320paramName ParamIP = "ip"
321paramName ParamNumWant = "numwant"
322paramName ParamEvent = "event"
323paramName ParamCompact = "compact"
324paramName ParamNoPeerId = "no_peer_id"
325{-# INLINE paramName #-}
326
327class FromParam a where
328 fromParam :: BS.ByteString -> Maybe a
329
330instance FromParam Bool where
331 fromParam "0" = Just False
332 fromParam "1" = Just True
333 fromParam _ = Nothing
334
335instance FromParam InfoHash where
336 fromParam = either (const Nothing) pure . safeConvert
337
338instance FromParam PeerId where
339 fromParam = either (const Nothing) pure . safeConvert
340
341instance FromParam Word32 where
342 fromParam = readMaybe . BC.unpack
343
344instance FromParam Word64 where
345 fromParam = readMaybe . BC.unpack
346
347instance FromParam Int where
348 fromParam = readMaybe . BC.unpack
349
350instance FromParam PortNumber where
351 fromParam bs = fromIntegral <$> (fromParam bs :: Maybe Word32)
352
353instance FromParam AnnounceEvent where
354 fromParam bs = do
355 (x, xs) <- BC.uncons bs
356 readMaybe $ BC.unpack $ BC.cons (Char.toUpper x) xs
357
358-- | 'ParamParseFailure' represent errors can occur while parsing HTTP
359-- tracker requests. In case of failure, this can be used to provide
360-- more informative 'statusCode' and 'statusMessage' in tracker
361-- responses.
362--
363data ParamParseFailure
364 = Missing QueryParam -- ^ param not found in query string;
365 | Invalid QueryParam BS.ByteString -- ^ param present but not valid.
366 deriving (Show, Eq)
367
368type ParseResult = Either ParamParseFailure
369
370withError :: ParamParseFailure -> Maybe a -> ParseResult a
371withError e = maybe (Left e) Right
372
373reqParam :: FromParam a => QueryParam -> SimpleQuery -> ParseResult a
374reqParam param xs = do
375 val <- withError (Missing param) $ L.lookup (paramName param) xs
376 withError (Invalid param val) (fromParam val)
377
378optParam :: FromParam a => QueryParam -> SimpleQuery -> ParseResult (Maybe a)
379optParam param ps
380 | Just x <- L.lookup (paramName param) ps
381 = pure <$> withError (Invalid param x) (fromParam x)
382 | otherwise = pure Nothing
383
384parseProgress :: SimpleQuery -> ParseResult Progress
385parseProgress params = Progress
386 <$> reqParam ParamDownloaded params
387 <*> reqParam ParamLeft params
388 <*> reqParam ParamUploaded params
389
390-- | Parse announce request from a query string.
391parseAnnounceQuery :: SimpleQuery -> ParseResult AnnounceQuery
392parseAnnounceQuery params = AnnounceQuery
393 <$> reqParam ParamInfoHash params
394 <*> reqParam ParamPeerId params
395 <*> reqParam ParamPort params
396 <*> parseProgress params
397 <*> optParam ParamIP params
398 <*> optParam ParamNumWant params
399 <*> optParam ParamEvent params
400
401{-----------------------------------------------------------------------
402-- Announce Info
403-----------------------------------------------------------------------}
404-- TODO check if announceinterval/complete/incomplete is positive ints
405
406-- | Tracker can return peer list in either compact(BEP23) or not
407-- compact form.
408--
409-- For more info see: <http://www.bittorrent.org/beps/bep_0023.html>
410--
411data PeerList ip
412 = PeerList [PeerAddr]
413 | CompactPeerList [PeerAddr]
414 deriving (Show, Eq, Typeable, Functor)
415
416-- | The empty non-compact peer list.
417instance Default (PeerList IP) where
418 def = PeerList []
419 {-# INLINE def #-}
420
421getPeerList :: PeerList IP -> [PeerAddr]
422getPeerList (PeerList xs) = xs
423getPeerList (CompactPeerList xs) = xs
424
425instance BEncode (PeerList a) where
426 toBEncode (PeerList xs) = toBEncode xs
427 toBEncode (CompactPeerList xs) = toBEncode $ runPut (mapM_ put xs)
428
429 fromBEncode (BList l ) = PeerList <$> fromBEncode (BList l)
430 fromBEncode (BString s ) = CompactPeerList <$> runGet (many get) s
431 fromBEncode _ = decodingError "PeerList: should be a BString or BList"
432
433-- | The tracker response includes a peer list that helps the client
434-- participate in the torrent. The most important is 'respPeer' list
435-- used to join the swarm.
436--
437data AnnounceInfo =
438 Failure !Text -- ^ Failure reason in human readable form.
439 | AnnounceInfo {
440 -- | Number of peers completed the torrent. (seeders)
441 respComplete :: !(Maybe Int)
442
443 -- | Number of peers downloading the torrent. (leechers)
444 , respIncomplete :: !(Maybe Int)
445
446 -- | Recommended interval to wait between requests, in seconds.
447 , respInterval :: !Int
448
449 -- | Minimal amount of time between requests, in seconds. A
450 -- peer /should/ make timeout with at least 'respMinInterval'
451 -- value, otherwise tracker might not respond. If not specified
452 -- the same applies to 'respInterval'.
453 , respMinInterval :: !(Maybe Int)
454
455 -- | Peers that must be contacted.
456 , respPeers :: !(PeerList IP)
457
458 -- | Human readable warning.
459 , respWarning :: !(Maybe Text)
460 } deriving (Show, Eq, Typeable)
461
462-- | Empty peer list with default reannounce interval.
463instance Default AnnounceInfo where
464 def = AnnounceInfo
465 { respComplete = Nothing
466 , respIncomplete = Nothing
467 , respInterval = defaultReannounceInterval
468 , respMinInterval = Nothing
469 , respPeers = def
470 , respWarning = Nothing
471 }
472
473-- | HTTP tracker protocol compatible encoding.
474instance BEncode AnnounceInfo where
475 toBEncode (Failure t) = toDict $
476 "failure reason" .=! t
477 .: endDict
478
479 toBEncode AnnounceInfo {..} = toDict $
480 "complete" .=? respComplete
481 .: "incomplete" .=? respIncomplete
482 .: "interval" .=! respInterval
483 .: "min interval" .=? respMinInterval
484 .: "peers" .=! peers
485 .: "peers6" .=? peers6
486 .: "warning message" .=? respWarning
487 .: endDict
488 where
489 (peers, peers6) = prttn respPeers
490
491 prttn :: PeerList IP -> (PeerList IPv4, Maybe (PeerList IPv6))
492 prttn (PeerList xs) = (PeerList xs, Nothing)
493 prttn (CompactPeerList xs) = mk $ partitionEithers $ toEither <$> xs
494 where
495 mk (v4s, v6s)
496 | L.null v6s = (CompactPeerList v4s, Nothing)
497 | otherwise = (CompactPeerList v4s, Just (CompactPeerList v6s))
498
499 toEither :: PeerAddr -> Either PeerAddr PeerAddr
500 toEither PeerAddr {..} = case peerHost of
501 ipv4@IPv4{} -> Left $ PeerAddr peerId ipv4 peerPort
502 ipv6@IPv6{} -> Right $ PeerAddr peerId ipv6 peerPort
503
504 fromBEncode (BDict d)
505 | Just t <- BE.lookup "failure reason" d = Failure <$> fromBEncode t
506 | otherwise = (`fromDict` (BDict d)) $
507 AnnounceInfo
508 <$>? "complete"
509 <*>? "incomplete"
510 <*>! "interval"
511 <*>? "min interval"
512 <*> (uncurry merge =<< (,) <$>! "peers" <*>? "peers6")
513 <*>? "warning message"
514 where
515 merge :: PeerList IPv4 -> Maybe (PeerList IPv6) -> BE.Get (PeerList IP)
516 merge (PeerList ips) Nothing = pure (PeerList ips)
517 merge (PeerList _ ) (Just _)
518 = fail "PeerList: non-compact peer list provided, \
519 \but the `peers6' field present"
520
521 merge (CompactPeerList ipv4s) Nothing
522 = pure $ CompactPeerList ipv4s
523
524 merge (CompactPeerList _ ) (Just (PeerList _))
525 = fail "PeerList: the `peers6' field value \
526 \should contain *compact* peer list"
527
528 merge (CompactPeerList ipv4s) (Just (CompactPeerList ipv6s))
529 = pure $ CompactPeerList $
530 ipv4s <> ipv6s
531
532 fromBEncode _ = decodingError "Announce info"
533
534-- | UDP tracker protocol compatible encoding.
535instance Serialize AnnounceInfo where
536 put (Failure msg) = put $ encodeUtf8 msg
537 put AnnounceInfo {..} = do
538 putWord32be $ fromIntegral respInterval
539 putWord32be $ fromIntegral $ fromMaybe 0 respIncomplete
540 putWord32be $ fromIntegral $ fromMaybe 0 respComplete
541 forM_ (getPeerList respPeers) put
542
543 get = do
544 interval <- getWord32be
545 leechers <- getWord32be
546 seeders <- getWord32be
547 peers <- many $ isolate 6 get -- isolated to specify IPv4.
548
549 return $ AnnounceInfo {
550 respWarning = Nothing
551 , respInterval = fromIntegral interval
552 , respMinInterval = Nothing
553 , respIncomplete = Just $ fromIntegral leechers
554 , respComplete = Just $ fromIntegral seeders
555 , respPeers = PeerList peers
556 }
557
558-- | Decodes announce response from bencoded string, for debugging only.
559instance IsString AnnounceInfo where
560 fromString str = either (error . format) id $ BE.decode (fromString str)
561 where
562 format msg = "fromString: unable to decode AnnounceInfo: " ++ msg
563
564-- | Above 25, new peers are highly unlikely to increase download
565-- speed. Even 30 peers is /plenty/, the official client version 3
566-- in fact only actively forms new connections if it has less than
567-- 30 peers and will refuse connections if it has 55.
568--
569-- <https://wiki.theory.org/BitTorrent_Tracker_Protocol#Basic_Tracker_Announce_Request>
570--
571defaultNumWant :: Int
572defaultNumWant = 50
573
574-- | Reasonable upper bound of numwant parameter.
575defaultMaxNumWant :: Int
576defaultMaxNumWant = 200
577
578-- | Widely used reannounce interval. Note: tracker clients should not
579-- use this value!
580defaultReannounceInterval :: Int
581defaultReannounceInterval = 30 * 60
582
583{-----------------------------------------------------------------------
584 Scrape message
585-----------------------------------------------------------------------}
586
587-- | Scrape query used to specify a set of torrent to scrape.
588-- If list is empty then tracker should return scrape info about each
589-- torrent.
590type ScrapeQuery = [InfoHash]
591
592-- TODO
593-- data ScrapeQuery
594-- = ScrapeAll
595-- | ScrapeSingle InfoHash
596-- | ScrapeMulti (HashSet InfoHash)
597-- deriving (Show)
598--
599-- data ScrapeInfo
600-- = ScrapeAll (HashMap InfoHash ScrapeEntry)
601-- | ScrapeSingle InfoHash ScrapeEntry
602-- | ScrapeMulti (HashMap InfoHash ScrapeEntry)
603--
604
605scrapeParam :: BS.ByteString
606scrapeParam = "info_hash"
607
608isScrapeParam :: BS.ByteString -> Bool
609isScrapeParam = (==) scrapeParam
610
611-- | Parse scrape query to query string.
612parseScrapeQuery :: SimpleQuery -> ScrapeQuery
613parseScrapeQuery
614 = catMaybes . L.map (fromParam . snd) . L.filter (isScrapeParam . fst)
615
616-- | Render scrape query to query string.
617renderScrapeQuery :: ScrapeQuery -> SimpleQuery
618renderScrapeQuery = queryToSimpleQuery . L.map mkPair
619 where
620 mkPair ih = (scrapeParam, toQueryValue ih)
621
622-- | Overall information about particular torrent.
623data ScrapeEntry = ScrapeEntry {
624 -- | Number of seeders - peers with the entire file.
625 siComplete :: {-# UNPACK #-} !Int
626
627 -- | Total number of times the tracker has registered a completion.
628 , siDownloaded :: {-# UNPACK #-} !Int
629
630 -- | Number of leechers.
631 , siIncomplete :: {-# UNPACK #-} !Int
632
633 -- | Name of the torrent file, as specified by the "name"
634 -- file in the info section of the .torrent file.
635 , siName :: !(Maybe Text)
636 } deriving (Show, Eq, Typeable)
637
638-- | HTTP tracker protocol compatible encoding.
639instance BEncode ScrapeEntry where
640 toBEncode ScrapeEntry {..} = toDict $
641 "complete" .=! siComplete
642 .: "downloaded" .=! siDownloaded
643 .: "incomplete" .=! siIncomplete
644 .: "name" .=? siName
645 .: endDict
646
647 fromBEncode = fromDict $ ScrapeEntry
648 <$>! "complete"
649 <*>! "downloaded"
650 <*>! "incomplete"
651 <*>? "name"
652
653-- | UDP tracker protocol compatible encoding.
654instance Serialize ScrapeEntry where
655 put ScrapeEntry {..} = do
656 putWord32be $ fromIntegral siComplete
657 putWord32be $ fromIntegral siDownloaded
658 putWord32be $ fromIntegral siIncomplete
659
660 get = ScrapeEntry
661 <$> (fromIntegral <$> getWord32be)
662 <*> (fromIntegral <$> getWord32be)
663 <*> (fromIntegral <$> getWord32be)
664 <*> pure Nothing
665
666-- | Scrape info about a set of torrents.
667type ScrapeInfo = [(InfoHash, ScrapeEntry)]
668
669{-----------------------------------------------------------------------
670-- HTTP specific
671-----------------------------------------------------------------------}
672
673-- | Some HTTP trackers allow to choose prefered representation of the
674-- 'AnnounceInfo'. It's optional for trackers to honor any of this
675-- options.
676data AnnouncePrefs = AnnouncePrefs
677 { -- | If specified, "compact" parameter is used to advise the
678 -- tracker to send peer id list as:
679 --
680 -- * bencoded list (extCompact = Just False);
681 -- * or more compact binary string (extCompact = Just True).
682 --
683 -- The later is prefered since compact peer list will reduce the
684 -- size of tracker responses. Hovewer, if tracker do not support
685 -- this extension then it can return peer list in either form.
686 --
687 -- For more info see: <http://www.bittorrent.org/beps/bep_0023.html>
688 --
689 extCompact :: !(Maybe Bool)
690
691 -- | If specified, "no_peer_id" parameter is used advise tracker
692 -- to either send or not to send peer id in tracker response.
693 -- Tracker may not support this extension as well.
694 --
695 -- For more info see:
696 -- <http://permalink.gmane.org/gmane.network.bit-torrent.general/4030>
697 --
698 , extNoPeerId :: !(Maybe Bool)
699 } deriving (Show, Eq, Typeable)
700
701instance Default AnnouncePrefs where
702 def = AnnouncePrefs Nothing Nothing
703
704instance QueryLike AnnouncePrefs where
705 toQuery AnnouncePrefs {..} =
706 [ ("compact", toQueryFlag <$> extCompact) -- TODO use 'paramName'
707 , ("no_peer_id", toQueryFlag <$> extNoPeerId)
708 ]
709 where
710 toQueryFlag False = "0"
711 toQueryFlag True = "1"
712
713-- | Parse announce query extended part from query string.
714parseAnnouncePrefs :: SimpleQuery -> AnnouncePrefs
715parseAnnouncePrefs params = either (const def) id $
716 AnnouncePrefs
717 <$> optParam ParamCompact params
718 <*> optParam ParamNoPeerId params
719
720-- | Render announce preferences to query string.
721renderAnnouncePrefs :: AnnouncePrefs -> SimpleQuery
722renderAnnouncePrefs = queryToSimpleQuery . toQuery
723
724-- | HTTP tracker request with preferences.
725data AnnounceRequest = AnnounceRequest
726 { announceQuery :: AnnounceQuery -- ^ Request query params.
727 , announcePrefs :: AnnouncePrefs -- ^ Optional advises to the tracker.
728 } deriving (Show, Eq, Typeable)
729
730instance QueryLike AnnounceRequest where
731 toQuery AnnounceRequest{..} =
732 toQuery announcePrefs <>
733 toQuery announceQuery
734
735-- | Parse announce request from query string.
736parseAnnounceRequest :: SimpleQuery -> ParseResult AnnounceRequest
737parseAnnounceRequest params = AnnounceRequest
738 <$> parseAnnounceQuery params
739 <*> pure (parseAnnouncePrefs params)
740
741-- | Render announce request to query string.
742renderAnnounceRequest :: AnnounceRequest -> SimpleQuery
743renderAnnounceRequest = queryToSimpleQuery . toQuery
744
745type PathPiece = BS.ByteString
746
747defaultAnnouncePath :: PathPiece
748defaultAnnouncePath = "announce"
749
750defaultScrapePath :: PathPiece
751defaultScrapePath = "scrape"
752
753missingOffset :: Int
754missingOffset = 101
755
756invalidOffset :: Int
757invalidOffset = 150
758
759parseFailureCode :: ParamParseFailure -> Int
760parseFailureCode (Missing param ) = missingOffset + fromEnum param
761parseFailureCode (Invalid param _) = invalidOffset + fromEnum param
762
763parseFailureMessage :: ParamParseFailure -> BS.ByteString
764parseFailureMessage e = BS.concat $ case e of
765 Missing p -> ["Missing parameter: ", paramName p]
766 Invalid p v -> ["Invalid parameter: ", paramName p, " = ", v]
767
768-- | HTTP response /content type/ for announce info.
769announceType :: ByteString
770announceType = "text/plain"
771
772-- | HTTP response /content type/ for scrape info.
773scrapeType :: ByteString
774scrapeType = "text/plain"
775
776-- | Get HTTP response status from a announce params parse failure.
777--
778-- For more info see:
779-- <https://wiki.theory.org/BitTorrent_Tracker_Protocol#Response_Codes>
780--
781parseFailureStatus :: ParamParseFailure -> Status
782parseFailureStatus = mkStatus <$> parseFailureCode <*> parseFailureMessage
783
784{-----------------------------------------------------------------------
785-- UDP specific message types
786-----------------------------------------------------------------------}
787
788genToken :: IO Word64
789genToken = do
790 bs <- getEntropy 8
791 either err return $ runGet getWord64be bs
792 where
793 err = error "genToken: impossible happen"
794
795-- | Connection Id is used for entire tracker session.
796newtype ConnectionId = ConnectionId Word64
797 deriving (Eq, Serialize)
798
799instance Show ConnectionId where
800 showsPrec _ (ConnectionId cid) = showString "0x" <> showHex cid
801
802initialConnectionId :: ConnectionId
803initialConnectionId = ConnectionId 0x41727101980
804
805-- | Transaction Id is used within a UDP RPC.
806newtype TransactionId = TransactionId Word32
807 deriving (Eq, Ord, Enum, Bounded, Serialize)
808
809instance Show TransactionId where
810 showsPrec _ (TransactionId tid) = showString "0x" <> showHex tid
811
812genTransactionId :: IO TransactionId
813genTransactionId = (TransactionId . fromIntegral) <$> genToken
814
815data Request
816 = Connect
817 | Announce AnnounceQuery
818 | Scrape ScrapeQuery
819 deriving Show
820
821data Response
822 = Connected ConnectionId
823 | Announced AnnounceInfo
824 | Scraped [ScrapeEntry]
825 | Failed Text
826 deriving Show
827
828responseName :: Response -> String
829responseName (Connected _) = "connected"
830responseName (Announced _) = "announced"
831responseName (Scraped _) = "scraped"
832responseName (Failed _) = "failed"
833
834data family Transaction a
835data instance Transaction Request = TransactionQ
836 { connIdQ :: {-# UNPACK #-} !ConnectionId
837 , transIdQ :: {-# UNPACK #-} !TransactionId
838 , request :: !Request
839 } deriving Show
840data instance Transaction Response = TransactionR
841 { transIdR :: {-# UNPACK #-} !TransactionId
842 , response :: !Response
843 } deriving Show
844
845-- TODO newtype
846newtype MessageId = MessageId Word32
847 deriving (Show, Eq, Num, Serialize)
848
849connectId, announceId, scrapeId, errorId :: MessageId
850connectId = 0
851announceId = 1
852scrapeId = 2
853errorId = 3
854
855instance Serialize (Transaction Request) where
856 put TransactionQ {..} = do
857 case request of
858 Connect -> do
859 put initialConnectionId
860 put connectId
861 put transIdQ
862
863 Announce ann -> do
864 put connIdQ
865 put announceId
866 put transIdQ
867 put ann
868
869 Scrape hashes -> do
870 put connIdQ
871 put scrapeId
872 put transIdQ
873 forM_ hashes put
874
875 get = do
876 cid <- get
877 mid <- get
878 TransactionQ cid <$> S.get <*> getBody mid
879 where
880 getBody :: MessageId -> S.Get Request
881 getBody msgId
882 | msgId == connectId = pure Connect
883 | msgId == announceId = Announce <$> get
884 | msgId == scrapeId = Scrape <$> many get
885 | otherwise = fail errMsg
886 where
887 errMsg = "unknown request: " ++ show msgId
888
889instance Serialize (Transaction Response) where
890 put TransactionR {..} = do
891 case response of
892 Connected conn -> do
893 put connectId
894 put transIdR
895 put conn
896
897 Announced info -> do
898 put announceId
899 put transIdR
900 put info
901
902 Scraped infos -> do
903 put scrapeId
904 put transIdR
905 forM_ infos put
906
907 Failed info -> do
908 put errorId
909 put transIdR
910 put (encodeUtf8 info)
911
912
913 get = do
914 mid <- get
915 TransactionR <$> get <*> getBody mid
916 where
917 getBody :: MessageId -> S.Get Response
918 getBody msgId
919 | msgId == connectId = Connected <$> get
920 | msgId == announceId = Announced <$> get
921 | msgId == scrapeId = Scraped <$> many get
922 | msgId == errorId = (Failed . decodeUtf8) <$> get
923 | otherwise = fail msg
924 where
925 msg = "unknown response: " ++ show msgId
diff --git a/bittorrent/src/Network/BitTorrent/Tracker/RPC.hs b/bittorrent/src/Network/BitTorrent/Tracker/RPC.hs
deleted file mode 100644
index 45fef05e..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Tracker/RPC/HTTP.hs b/bittorrent/src/Network/BitTorrent/Tracker/RPC/HTTP.hs
deleted file mode 100644
index 6f7a53bf..00000000
--- a/bittorrent/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 defaultRequest 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/bittorrent/src/Network/BitTorrent/Tracker/RPC/UDP.hs b/bittorrent/src/Network/BitTorrent/Tracker/RPC/UDP.hs
deleted file mode 100644
index 31b6b870..00000000
--- a/bittorrent/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/bittorrent/src/Network/BitTorrent/Tracker/Session.hs b/bittorrent/src/Network/BitTorrent/Tracker/Session.hs
deleted file mode 100644
index db6ebaff..00000000
--- a/bittorrent/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]
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])
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]
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 15:26:54 +0000