DDD works

hbs2-peer/app/BlockDownloadNew.hs

@@ -47,6 +47,7 @@ import Data.Vector ((!?))
 import Data.ByteString qualified as BS
 import Data.List qualified as L
 import Data.Coerce
+import Data.Kind
 import Numeric
 import UnliftIO
 import Control.Concurrent.STM.TSem (TSem)
@@ -154,6 +155,12 @@ queryBlockSizeFromPeer cache e h peer = do
     Right x -> pure (Right x)
 
 
+class Monad m => IsBurstMachine a m  where
+  getCurrentBurst :: a -> m Int
+  burstMachineAddErrors ::  a -> Int -> m ()
+  burstMachineReset :: a -> m ()
+  runBurstMachine ::  a -> m ()
+
 data BurstMachine =
   BurstMachine
   { _buTimeout  :: Double
@@ -165,15 +172,15 @@ data BurstMachine =
   , _buErrors   :: TVar Int
   }
 
-burstMachineAddErrors :: MonadUnliftIO m => BurstMachine -> Int -> m ()
+data ConstBurstMachine = ConstBurstMachine Int
-burstMachineAddErrors BurstMachine{..} n =
-  atomically $ modifyTVar _buErrors (+n)
 
-burstMachineReset :: MonadUnliftIO m => BurstMachine -> m ()
+data AnyBurstMachine (m :: Type -> Type) = forall a . IsBurstMachine a m => AnyBurstMachine a
-burstMachineReset BurstMachine{..} = do
+
-  atomically do
+instance MonadIO m => IsBurstMachine (AnyBurstMachine m) m where
-    writeTVar _buCurrent _buStart
+  getCurrentBurst (AnyBurstMachine a) = getCurrentBurst a
-    writeTVar _buErrors 0
+  burstMachineAddErrors (AnyBurstMachine a) = burstMachineAddErrors a
+  burstMachineReset (AnyBurstMachine a) = burstMachineReset a
+  runBurstMachine (AnyBurstMachine a) = runBurstMachine a
 
 newBurstMachine :: MonadUnliftIO m
                 => Double      -- ^ timeout
@@ -193,13 +200,16 @@ newBurstMachine t0 buMax buStart up' down' = do
     down = min 0.85 down'
     up   = min 0.5 up'
 
-getCurrentBurst :: MonadUnliftIO m => BurstMachine -> m Int
+instance MonadUnliftIO m => IsBurstMachine BurstMachine m where
   getCurrentBurst BurstMachine{..} = readTVarIO _buCurrent <&> round
 
+  burstMachineAddErrors BurstMachine{..} n =
+    atomically $ modifyTVar _buErrors (+n)
 
-runBurstMachine :: MonadUnliftIO m
+  burstMachineReset BurstMachine{..} = do
-                => BurstMachine
+    atomically do
-                -> m ()
+        writeTVar _buCurrent _buStart
+        writeTVar _buErrors 0
 
   runBurstMachine BurstMachine{..} = do
 
@@ -280,6 +290,12 @@ runBurstMachine BurstMachine{..} = do
         pause @'Seconds dt
         next eNew
 
+instance MonadIO m => IsBurstMachine ConstBurstMachine m where
+  getCurrentBurst (ConstBurstMachine n) = pure n
+  burstMachineAddErrors _ _ = none
+  burstMachineReset _ = none
+  runBurstMachine _ = forever $ pause @'Seconds 300
+
 data S =
     SInit
   | SFetchQ
@@ -689,8 +705,8 @@ downloadDispatcher brains env = flip runContT pure do
 
       bm <- liftIO do
               case _sockType p of
-                TCP -> newBurstMachine 20 256 (Just 256) 0.25 0.45
+                TCP -> AnyBurstMachine @IO <$> newBurstMachine 30 256 (Just 256) 0.20 0.10
-                UDP -> newBurstMachine 3 256 (Just 50)  0.10 0.25
+                UDP -> AnyBurstMachine @IO <$> newBurstMachine  5 256 (Just 100) 0.15 0.35
 
       void $ ContT $ bracket none $ const do
         debug $ "Cancelling thread for" <+> pretty p
@@ -706,7 +722,7 @@ downloadDispatcher brains env = flip runContT pure do
 
         when (b0 == b1) do
           debug $ blue "Reset burst machine" <+> pretty p
-          burstMachineReset bm
+          liftIO $ burstMachineReset bm
 
         s <- readTVarIO wip <&> HM.size
 
@@ -721,7 +737,7 @@ downloadDispatcher brains env = flip runContT pure do
             unless here do
               modifyTVar _sizeCache (HM.delete h)
 
-      bmt <- ContT $ withAsync $ runBurstMachine bm
+      bmt <- ContT $ withAsync $ liftIO $ runBurstMachine bm
 
       tstat <- ContT $ withAsync $ forever $ (>> pause @'Seconds 5) do
                  tss <- readTVarIO btimes
@@ -734,7 +750,7 @@ downloadDispatcher brains env = flip runContT pure do
         pinfo' <- lift $ find @e (PeerInfoKey p) id
         PeerInfo{..} <- ContT $ maybe1 pinfo' none
 
-        bu <- lift $ getCurrentBurst bm
+        bu <- liftIO $ getCurrentBurst bm
         atomically do
           erno <- readTVar _errors
           down <- readTVar _blknum
@@ -827,10 +843,10 @@ downloadDispatcher brains env = flip runContT pure do
 
           PFetchBlock hx dcb size -> do
 
-            bu <- lift $ getCurrentBurst bm
+            bu <- liftIO $ getCurrentBurst bm
 
             t0 <- getTimeCoarse
-            r <- lift $ downloadFromPeer bu (KnownSize size) env (coerce hx) p
+            r <- downloadFromPeer bu (KnownSize size) env (coerce hx) p
             t1 <- getTimeCoarse
 
             case r of
@@ -841,7 +857,7 @@ downloadDispatcher brains env = flip runContT pure do
                 avg <- readTVarIO _avg
 
                 when (dtsec > avg * 1.15) do
-                  burstMachineAddErrors bm 1
+                  liftIO $ burstMachineAddErrors bm 1
 
                 atomically do
                   modifyTVar  btimes ( take 100 . (dtsec :) )
@@ -852,7 +868,7 @@ downloadDispatcher brains env = flip runContT pure do
                 go (PReleaseBlock hx dcb True)
 
               Left e -> do
-                burstMachineAddErrors bm 1
+                liftIO $ burstMachineAddErrors bm 1
                 atomically $ modifyTVar _errors succ
                 debug $ red "BLOCK DOWNLOAD FUCKED" <+> pretty p <+> pretty hx <+> viaShow e