5 files changed, 95 insertions, 33 deletions

diff --git a/bittorrent.cabal b/bittorrent.cabal
index 73e8bef2..c8f3d4ed 100644
--- a/bittorrent.cabal
+++ b/bittorrent.cabal
@@ -58,11 +58,19 @@ library
 
                        -- Control packages
                      , mtl
+                     , resourcet
                      , lens
 
                        -- Concurrency packages
+                     , monad-fork
                      , SafeSemaphore
-                     , stm        >= 2.4
+                     , BoundedChan >= 1.0.1.0
+                     , stm         >= 2.4
+
+                       -- Conduits
+                     , conduit          == 1.*
+                     , network-conduit  == 1.*
+                     , cereal-conduit   >= 0.5
 
                        -- Data packages
                      , array      >= 0.4
@@ -87,12 +95,6 @@ library
                      , HTTP       >= 4000.2
                      , krpc
 
-                       -- Conduits
-                     , conduit              == 1.*
-                     , network-conduit      == 1.*
-                     , cereal-conduit       >= 0.5
-                     , resourcet
-
                        -- Misc
                      , data-default
                      , cryptohash
diff --git a/exsamples/Main.hs b/exsamples/Main.hs
index 9ed311ae..ebc81b02 100644
--- a/exsamples/Main.hs
+++ b/exsamples/Main.hs
@@ -15,7 +15,7 @@ main = do
   [path]  <- getArgs
   torrent <- fromFile path
 
-  client  <- defaultClient
+  client  <- newClient 2 []
   swarm   <- newLeacher  client torrent
 
   discover swarm $ do
diff --git a/src/Network/BitTorrent/Exchange.hs b/src/Network/BitTorrent/Exchange.hs
index 978e86db..be9a455b 100644
--- a/src/Network/BitTorrent/Exchange.hs
+++ b/src/Network/BitTorrent/Exchange.hs
@@ -26,6 +26,7 @@ import Control.Applicative
 import Control.Exception
 import Control.Concurrent
 import Control.Lens
+import Control.Monad.Fork.Class
 import Control.Monad.Reader
 import Control.Monad.Trans.Resource
 
@@ -283,6 +284,7 @@ newtype P2P a = P2P {
              , MonadIO, MonadThrow, MonadActive
              , MonadReader PeerSession
              )
+-- TODO instance for MonadFork
 
 runSession :: SwarmSession -> PeerAddr -> P2P () -> IO ()
 runSession  se addr p2p =
diff --git a/src/Network/BitTorrent/Internal.hs b/src/Network/BitTorrent/Internal.hs
index afe1fff1..918bfed7 100644
--- a/src/Network/BitTorrent/Internal.hs
+++ b/src/Network/BitTorrent/Internal.hs
@@ -89,6 +89,9 @@ import Network.BitTorrent.Peer
 import Network.BitTorrent.Exchange.Protocol as BT
 import Network.BitTorrent.Tracker.Protocol as BT
 
+{-----------------------------------------------------------------------
+    Progress
+-----------------------------------------------------------------------}
 
 -- | 'Progress' contains upload/download/left stats about
 --   current client state.
@@ -109,6 +112,7 @@ startProgress = Progress 0 0
     Client session
 -----------------------------------------------------------------------}
 
+-- TODO comment thread count bounding
 type ThreadCount = Int
 
 defaultThreadCount :: ThreadCount
@@ -169,6 +173,7 @@ newClient n exts = do
     Swarm session
 -----------------------------------------------------------------------}
 
+-- TODO document P2P sessions bounding
 type SessionCount = Int
 
 defSeederConns :: SessionCount
diff --git a/src/Network/BitTorrent/Tracker.hs b/src/Network/BitTorrent/Tracker.hs
index c3bce63a..ea45b75d 100644
--- a/src/Network/BitTorrent/Tracker.hs
+++ b/src/Network/BitTorrent/Tracker.hs
@@ -24,7 +24,7 @@ module Network.BitTorrent.Tracker
 
          -- * Session
        , TSession
-       , getPeerAddr, getPeerList
+       , getPeerAddr
        , getProgress, waitInterval
 
          -- * Re-export
@@ -38,6 +38,7 @@ module Network.BitTorrent.Tracker
 
 import Control.Applicative
 import Control.Concurrent
+import Control.Concurrent.BoundedChan as BC
 import Control.Concurrent.STM
 import Control.Exception
 import Control.Monad
@@ -45,6 +46,7 @@ import Data.BEncode
 import           Data.ByteString (ByteString)
 import qualified Data.ByteString as B
 import qualified Data.ByteString.Char8 as BC
+import Data.List as L
 import           Data.Map (Map)
 import qualified Data.Map as M
 import Data.Monoid
@@ -59,9 +61,13 @@ import Network.BitTorrent.Internal
 import Network.BitTorrent.Peer
 import Network.BitTorrent.Tracker.Protocol
 
+{-----------------------------------------------------------------------
+    Tracker connection
+-----------------------------------------------------------------------}
 
--- | 'TConnection' (shorthand for Tracker session) combines tracker request
---   fields neccessary for tracker, torrent and client identification.
+-- | 'TConnection' (shorthand for Tracker session) combines tracker
+-- request fields neccessary for tracker, torrent and client
+-- identification.
 --
 --   This data is considered as static within one session.
 --
@@ -136,49 +142,84 @@ completedReq ses pr = (genericReq ses pr) {
   , reqEvent      = Just Completed
   }
 
+{-----------------------------------------------------------------------
+    Tracker session
+-----------------------------------------------------------------------}
 
+{-  Why use BoundedChan?
 
+Because most times we need just a list of peer at the start and all
+the rest time we will take little by little. On the other hand tracker
+will give us some constant count of peers and channel will grow with
+time. To avoid space leaks and long lists of peers (which we don't
+need) we use bounded chaan.
+
+   Chan size.
+
+Should be at least (count_of_workers * 2) to accumulate long enough
+peer list.
+
+  Order of peers in chan.
+
+Old peers in head, new ones in tail. Old peers should be used in the
+first place because by statistics they are most likely will present in
+network a long time than a new.
+
+-}
+
+type TimeInterval = Int
 
 data TSession = TSession {
+  -- TODO synchonize progress with client session
     seProgress   :: TVar Progress
-  , seInterval   :: IORef Int
-  , sePeers      :: Chan PeerAddr
-    -- TODO use something like 'TVar (Set PeerAddr)'
-    -- otherwise we might get space leak
-    -- TODO or maybe BoundedChan?
+  , seInterval   :: IORef TimeInterval
+  , sePeers      :: BoundedChan PeerAddr
   }
 
-newSession :: Progress -> Int -> [PeerAddr] -> IO TSession
-newSession pr i ps = do
-  chan <- newChan
-  writeList2Chan chan ps
-  TSession <$> newTVarIO pr
-           <*> newIORef i
-           <*> pure chan
+type PeerCount = Int
 
-getPeerAddr :: TSession -> IO PeerAddr
-getPeerAddr = readChan . sePeers
+defaultChanSize :: PeerCount
+defaultChanSize = defaultNumWant * 2
 
-getPeerList :: TSession -> IO [PeerAddr]
-getPeerList = getChanContents . sePeers
+getPeerAddr :: TSession -> IO PeerAddr
+getPeerAddr = BC.readChan . sePeers
 
 getProgress :: TSession -> IO Progress
 getProgress = readTVarIO . seProgress
 
-sec :: Int
-sec = 1000 * 1000
+newSession :: PeerCount -> Progress -> TimeInterval -> [PeerAddr]
+           -> IO TSession
+newSession chanSize pr i ps
+  | chanSize < 1
+  = throwIO $ userError "size of chan should be more that 1"
+
+  | otherwise = do
+    chan <- newBoundedChan chanSize
+
+    -- if length of the "ps" is more than the "chanSize" we will block
+    -- forever; to avoid this we remove excessive peers
+    let ps' = take chanSize ps
+    BC.writeList2Chan chan ps'
+
+    TSession <$> newTVarIO pr
+             <*> newIORef i
+             <*> pure chan
 
 waitInterval :: TSession -> IO ()
 waitInterval se @ TSession {..} = do
-  delay <- readIORef seInterval
-  threadDelay (delay * sec)
+    delay <- readIORef seInterval
+    threadDelay (delay * sec)
+  where
+    sec = 1000 * 1000 :: Int
 
 withTracker :: Progress -> TConnection -> (TSession -> IO a) -> IO a
 withTracker initProgress conn action = bracket start end (action . fst)
   where
     start = do
       resp <- askTracker (startedReq conn initProgress)
-      se   <- newSession initProgress (respInterval resp) (respPeers resp)
+      se   <- newSession defaultChanSize initProgress
+                         (respInterval resp) (respPeers resp)
+
       tid  <- forkIO (syncSession se)
       return (se, tid)
 
@@ -190,7 +231,16 @@ withTracker initProgress conn action = bracket start end (action . fst)
         case resp of
           Right (OK {..}) -> do
             writeIORef seInterval respInterval
-            writeList2Chan sePeers respPeers
+
+            -- we rely on the fact that union on lists is not
+            -- commutative: this implements the heuristic "old peers
+            -- in head"
+            old <- BC.getChanContents sePeers
+            let new = respPeers
+            let combined = L.union old new
+
+            BC.writeList2Chan sePeers combined
+
           _ -> return ()
       where
         isIOException :: IOException -> Maybe IOException
@@ -201,6 +251,9 @@ withTracker initProgress conn action = bracket start end (action . fst)
       pr <- getProgress se
       leaveTracker $ stoppedReq conn pr
 
+{-----------------------------------------------------------------------
+    Scrape
+-----------------------------------------------------------------------}
 
 
 -- | Information about particular torrent.