1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
-- |
-- Copyright : (c) Sam Truzjan 2013
-- License : BSD3
-- Maintainer : pxqr.sta@gmail.com
-- Stability : experimental
-- Portability : portable
--
-- BitTorrent uses a \"distributed sloppy hash table\" (DHT) for
-- storing peer contact information for \"trackerless\" torrents. In
-- effect, each peer becomes a tracker.
--
-- Normally you don't need to import other DHT modules.
--
-- For more info see:
-- <http://www.bittorrent.org/beps/bep_0005.html>
--
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TypeOperators #-}
module Network.BitTorrent.DHT
( -- * Distributed Hash Table
DHT
, Options (..)
, dht
-- * Bootstrapping
, tNodes
, defaultBootstrapNodes
, resolveHostName
, bootstrap
, isBootstrapped
-- * Initialization
, snapshot
, restore
-- * Operations
, Network.BitTorrent.DHT.lookup
, Network.BitTorrent.DHT.insert
, Network.BitTorrent.DHT.delete
-- * Embedding
-- ** Session
, LogFun
, Node
, defaultHandlers
, newNode
, closeNode
-- ** Monad
, MonadDHT (..)
, runDHT
) where
import Control.Applicative
import Control.Monad.Logger
import Control.Monad.Reader
import Control.Exception
import Data.ByteString as BS
import Data.Conduit as C
import Data.Conduit.List as C
import Network.Socket
import Data.Torrent (tNodes)
import Data.Torrent.InfoHash
import Network.BitTorrent.Core
import Network.BitTorrent.DHT.Query
import Network.BitTorrent.DHT.Session
import Network.BitTorrent.DHT.Routing as T
{-----------------------------------------------------------------------
-- DHT types
-----------------------------------------------------------------------}
class MonadDHT m where
liftDHT :: DHT IPv4 a -> m a
instance MonadDHT (DHT IPv4) where
liftDHT = id
-- | Run DHT on specified port. <add note about resources>
dht :: Address ip
=> Options -- ^ normally you need to use 'Data.Default.def';
-> NodeAddr ip -- ^ address to bind this node;
-> DHT ip a -- ^ actions to run: 'bootstrap', 'lookup', etc;
-> IO a -- ^ result.
dht opts addr action = do
runStderrLoggingT $ LoggingT $ \ logger -> do
bracket (newNode defaultHandlers opts addr logger) closeNode $
\ node -> runDHT node action
{-# INLINE dht #-}
{-----------------------------------------------------------------------
-- Bootstrapping
-----------------------------------------------------------------------}
-- Do not include the following hosts in the default bootstrap nodes list:
--
-- * "dht.aelitis.com" and "dht6.azureusplatform.com" - since
-- Azureus client have a different (and probably incompatible) DHT
-- protocol implementation.
--
-- * "router.utorrent.com" since it is just an alias to
-- "router.bittorrent.com".
-- | List of bootstrap nodes maintained by different bittorrent
-- software authors.
defaultBootstrapNodes :: [NodeAddr HostName]
defaultBootstrapNodes =
[ NodeAddr "router.bittorrent.com" 6881 -- by BitTorrent Inc.
-- doesn't work at the moment (use git blame) of commit
, NodeAddr "dht.transmissionbt.com" 6881 -- by Transmission project
]
-- TODO Multihomed hosts
-- | Resolve either a numeric network address or a hostname to a
-- numeric IP address of the node. Usually used to resolve
-- 'defaultBootstrapNodes' or 'Data.Torrent.tNodes' lists.
resolveHostName :: NodeAddr HostName -> IO (NodeAddr IPv4)
resolveHostName NodeAddr {..} = do
let hints = defaultHints { addrFamily = AF_INET, addrSocketType = Datagram }
-- getAddrInfo throws exception on empty list, so the pattern matching never fail
info : _ <- getAddrInfo (Just hints) (Just nodeHost) (Just (show nodePort))
case fromSockAddr (addrAddress info) of
Nothing -> error "resolveNodeAddr: impossible"
Just addr -> return addr
-- | One good node may be sufficient.
--
-- This operation do block, use
-- 'Control.Concurrent.Async.Lifted.async' if needed.
--
bootstrap :: Address ip => [NodeAddr ip] -> DHT ip ()
bootstrap startNodes = do
$(logInfoS) "bootstrap" "Start node bootstrapping"
nid <- asks thisNodeId
-- TODO filter duplicated in startNodes list
-- TODO retransmissions for startNodes
aliveNodes <- queryParallel (pingQ <$> startNodes)
_ <- sourceList [aliveNodes] $= search nid (findNodeQ nid) $$ C.consume
$(logInfoS) "bootstrap" "Node bootstrapping finished"
-- | Check if this node is already bootstrapped.
-- @bootstrap [good_node] >> isBootstrapped@@ should always return 'True'.
--
-- This operation do not block.
--
isBootstrapped :: DHT ip Bool
isBootstrapped = T.full <$> getTable
{-----------------------------------------------------------------------
-- Initialization
-----------------------------------------------------------------------}
-- | Load previous session. (corrupted - exception/ignore ?)
--
-- This is blocking operation, use
-- 'Control.Concurrent.Async.Lifted.async' if needed.
restore :: ByteString -> IO (Node ip)
restore = error "DHT.restore: not implemented"
-- | Serialize current DHT session to byte string.
--
-- This is blocking operation, use
-- 'Control.Concurrent.Async.Lifted.async' if needed.
snapshot :: DHT ip ByteString
snapshot = error "DHT.snapshot: not implemented"
{-----------------------------------------------------------------------
-- Operations
-----------------------------------------------------------------------}
-- | Get list of peers which downloading this torrent.
--
-- This operation is incremental and do block.
--
lookup :: Address ip => InfoHash -> DHT ip `Source` [PeerAddr ip]
lookup topic = do -- TODO retry getClosest if bucket is empty
closest <- lift $ getClosest topic
sourceList [closest] $= search topic (getPeersQ topic)
-- TODO do not republish if the topic is already in announceSet
-- | Announce that /this/ peer may have some pieces of the specified
-- torrent. DHT will reannounce this data periodically using
-- 'optReannounce' interval.
--
-- This operation is synchronous and do block, use
-- 'Control.Concurrent.Async.Lifted.async' if needed.
--
insert :: Address ip => InfoHash -> PortNumber -> DHT ip ()
insert ih p = do
publish ih p
insertTopic ih p
-- | Stop announcing /this/ peer for the specified torrent.
--
-- This operation is atomic and may block for a while.
--
delete :: Address ip => InfoHash -> PortNumber -> DHT ip ()
delete = deleteTopic
{-# INLINE delete #-}
|
