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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
|
{-# LANGUAGE CPP #-}
{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE NondecreasingIndentation #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PartialTypeSignatures #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeFamilies #-}
import Control.Arrow
import Control.Concurrent.STM
import Control.DeepSeq
import Control.Exception
import Control.Monad
import Data.Char
import Data.List
import qualified Data.Map as Map
import Data.Time.Clock
import GHC.Stats
import Network.Socket
import System.Environment
import System.IO
import System.Mem
import System.Posix.Process
import Text.Printf
import Text.Read
#ifdef THREAD_DEBUG
import Control.Concurrent.Lifted.Instrument
#else
import Control.Concurrent.Lifted
import GHC.Conc (labelThread)
#endif
import Network.Address hiding (NodeId, NodeInfo(..))
import Network.QueryResponse
import Network.StreamServer
import Kademlia
import qualified Mainline
import Network.DHT.Routing as R
import Data.Aeson as J (ToJSON, FromJSON)
import qualified Data.Aeson as J
import qualified Data.ByteString.Lazy as L
import Control.Concurrent.Async.Pool
import System.IO.Error
showReport :: [(String,String)] -> String
showReport kvs = do
let colwidth = maximum $ map (length . fst) kvs
(k,v) <- kvs
concat [ printf " %-*s" (colwidth+1) k, v, "\n" ]
marshalForClient :: String -> String
marshalForClient s = show (length s) ++ ":" ++ s
hPutClient :: Handle -> String -> IO ()
hPutClient h s = hPutStr h ('.' : marshalForClient s)
data DHT = forall ni. ( Show ni
, Read ni
, ToJSON ni
, FromJSON ni
) =>
DHT
{ dhtBuckets :: TVar (BucketList ni)
, dhtPing :: ni -> IO Bool
-- dhtClient :: Client err meth tid addr x
}
nodesFileName :: String -> String
nodesFileName netname = netname ++ "-nodes.json"
saveNodes :: String -> DHT -> IO ()
saveNodes netname (DHT var _) = do
bkts <- atomically $ readTVar var
let ns = map fst $ concat $ R.toList bkts
bs = J.encode ns
fname = nodesFileName netname
L.writeFile fname bs
loadNodes :: FromJSON ni => String -> IO [ni]
loadNodes netname = do
let fname = nodesFileName netname
attempt <- tryIOError $ do
J.decode <$> L.readFile fname
>>= maybe (ioError $ userError "Nothing") return
either (const $ return []) return attempt
pingNodes :: String -> DHT -> IO Bool
pingNodes netname (DHT _ ping) = do
let fname = nodesFileName netname
attempt <- tryIOError $ do
J.decode <$> L.readFile fname
>>= maybe (ioError $ userError "Nothing") return
case attempt of
Left _ -> return False
Right ns -> do fork $ do
myThreadId >>= flip labelThread ("pinging."++fname)
putStrLn $ "Forked "++show fname
withTaskGroup 10 $ \g -> do
mapTasks_ g (map ping ns)
putStrLn $ "Load finished "++show fname
return True
reportTable :: Show ni => BucketList ni -> [(String,String)]
reportTable bkts = map (show *** show . fst)
$ concat
$ zipWith map (map (,) [0::Int ..])
$ R.toList
$ bkts
clientSession netname dhts signalQuit sock n h = do
line <- map toLower . dropWhile isSpace <$> hGetLine h
let (c,args) = second (dropWhile isSpace) $ break isSpace line
cmd0 :: IO () -> IO ()
cmd0 action = action >> clientSession netname dhts signalQuit sock n h
switchNetwork dest = do hPutClient h ("Network: "++dest)
clientSession dest dhts signalQuit sock n h
case (c,args) of
("stop", _) -> do hPutClient h "Terminating DHT Daemon."
hClose h
putMVar signalQuit ()
("quit", _) -> hPutClient h "" >> hClose h
("pid", _) -> cmd0 $ do
pid <- getProcessID
hPutClient h (show pid)
#ifdef THREAD_DEBUG
("threads", _) -> cmd0 $ do
ts <- threadsInformation
tm <- getCurrentTime
let r = map (\PerThread{..} -> (show lbl,show (diffUTCTime tm startTime))) ts
hPutClient h $ showReport r
("mem", s) -> cmd0 $ do
case s of
"gc" -> do hPutClient h "Performing garbage collection..."
performMajorGC
"" -> do
is_enabled <- getGCStatsEnabled
if is_enabled
then do
GCStats{..} <- getGCStats
let r = [ ("bytesAllocated", show bytesAllocated)
, ("numGcs", show numGcs)
, ("maxBytesUsed", show maxBytesUsed)
, ("numByteUsageSamples", show numByteUsageSamples)
, ("cumulativeBytesUsed", show cumulativeBytesUsed)
, ("bytesCopied", show bytesCopied)
, ("currentBytesUsed", show currentBytesUsed)
, ("currentBytesSlop", show currentBytesSlop)
, ("maxBytesSlop", show maxBytesSlop)
, ("peakMegabytesAllocated", show peakMegabytesAllocated)
, ("mutatorCpuSeconds", show mutatorCpuSeconds)
, ("mutatorWallSeconds", show mutatorWallSeconds)
, ("gcCpuSeconds", show gcCpuSeconds)
, ("gcWallSeconds", show gcWallSeconds)
, ("cpuSeconds", show cpuSeconds)
, ("wallSeconds", show wallSeconds)
, ("parTotBytesCopied", show parTotBytesCopied)
, ("parMaxBytesCopied", show parMaxBytesCopied)
]
hPutClient h $ showReport r
else hPutClient h "Run with +RTS -T to obtain live memory-usage information."
_ -> hPutClient h "error."
#endif
("ls", _) | Just (DHT var _) <- Map.lookup netname dhts
-> cmd0 $ do
bkts <- atomically $ readTVar var
let r = reportTable bkts
hPutClient h $
showReport $
r ++ [ ("buckets", show $ R.shape bkts)
, ("node-id", show $ thisNode bkts)
, ("network", netname) ]
("ping", s) | Just (DHT _ ping) <- Map.lookup netname dhts
-> cmd0 $ do
case readEither s of
Right addr -> do result <- ping addr
let rs = [" ", show result]
hPutClient h $ unlines rs
Left er -> hPutClient h er
("save", _) | Just dht <- Map.lookup netname dhts
-> cmd0 $ do
saveNodes netname dht
hPutClient h $ "Saved " ++ nodesFileName netname ++ "."
("load", _) | Just dht <- Map.lookup netname dhts
-> cmd0 $ do
b <- pingNodes netname dht
if b then hPutClient h $ "Pinging " ++ nodesFileName netname ++ "."
else hPutClient h $ "Failed: " ++ nodesFileName netname ++ "."
(n, _) | n `elem` Map.keys dhts -> switchNetwork n
_ -> cmd0 $ hPutClient h "error."
defaultPort = "6881"
main = do
args <- getArgs
p <- case take 2 (dropWhile (/="-p") args) of
["-p",port] | not ("-" `isPrefixOf` port) -> return port
("-p":_) -> error "Port not specified! (-p PORT)"
_ -> return defaultPort
addr <- getBindAddress p True{- ipv6 -}
(bt,btR) <- Mainline.newClient addr
quitBt <- forkListener bt
tox <- return $ error "TODO: Tox.newClient"
quitTox <- return $ return () -- TODO: forkListener tox
let dhts = Map.fromList
[ ("bt4", DHT (Mainline.routing4 btR) (Mainline.ping bt))
, ("bt6", DHT (Mainline.routing6 btR) (Mainline.ping bt))
]
waitForSignal <- do
signalQuit <- newEmptyMVar
srv <- streamServer (withSession $ clientSession "bt4" dhts signalQuit) (SockAddrUnix "dht.sock")
return $ do
() <- takeMVar signalQuit
quitListening srv
let bkts4 = Mainline.routing4 btR
btSaved4 <- loadNodes "bt4" :: IO [Mainline.NodeInfo]
putStrLn $ "Loaded "++show (length btSaved4)++" nodes for bt4."
fallbackNodes4 <- Mainline.bootstrapNodes Mainline.Want_IP4
fork $ do
myThreadId >>= flip labelThread "bootstrap.Mainline4"
bootstrap (Mainline.nodeSearch bt) bkts4 (Mainline.ping bt) btSaved4 fallbackNodes4
saveNodes "bt4" (dhts Map.! "bt4")
btSaved6 <- loadNodes "bt6"
putStrLn $ "Loaded "++show (length btSaved6)++" nodes for bt6."
let bkts6 = Mainline.routing6 btR
fallbackNodes6 <- Mainline.bootstrapNodes Mainline.Want_IP6
fork $ do
myThreadId >>= flip labelThread "bootstrap.Mainline6"
bootstrap (Mainline.nodeSearch bt) bkts6 (Mainline.ping bt) btSaved6 fallbackNodes6
saveNodes "bt6" (dhts Map.! "bt6")
hPutStr stderr $ showReport $ map (("bootstrap (IPv4)",) . show) fallbackNodes4
++ map (("bootstrap (IPv6)",) . show) fallbackNodes6
waitForSignal
quitBt
quitTox
|
