mirror of https://github.com/voidlizard/hbs2
parent
29fdd23602
commit
a1bbc634f4
|
@ -524,7 +524,7 @@ runMe conf = withLogging $ flip runReaderT conf do
|
||||||
when here do
|
when here do
|
||||||
liftIO $ removeFile sa
|
liftIO $ removeFile sa
|
||||||
|
|
||||||
server <- newMessagingUnix True 1.0 sa
|
server <- newMessagingUnixOpts [MUNoFork] True 1.0 sa
|
||||||
|
|
||||||
abus <- async $ runMessagingUnix server
|
abus <- async $ runMessagingUnix server
|
||||||
|
|
||||||
|
|
|
@ -60,7 +60,7 @@ instance Hashable (Peer UNIX) where
|
||||||
{- HLINT ignore "Use newtype instead of data" -}
|
{- HLINT ignore "Use newtype instead of data" -}
|
||||||
data MessagingUnixOpts =
|
data MessagingUnixOpts =
|
||||||
MUWatchdog Int
|
MUWatchdog Int
|
||||||
| MUFork
|
| MUNoFork
|
||||||
deriving (Eq,Ord,Show,Generic,Data)
|
deriving (Eq,Ord,Show,Generic,Data)
|
||||||
|
|
||||||
-- FIXME: use-bounded-queues
|
-- FIXME: use-bounded-queues
|
||||||
|
@ -137,7 +137,12 @@ runMessagingUnix env = do
|
||||||
liftIO $ bind sock $ SockAddrUnix (msgUnixSockPath env)
|
liftIO $ bind sock $ SockAddrUnix (msgUnixSockPath env)
|
||||||
liftIO $ listen sock 5
|
liftIO $ listen sock 5
|
||||||
|
|
||||||
let withSession = void . async . runResourceT
|
-- let withSession = void . async . runResourceT
|
||||||
|
|
||||||
|
let doFork = not $ Set.member MUNoFork (msgUnixOpts env)
|
||||||
|
|
||||||
|
let withSession | doFork = void . async . runResourceT
|
||||||
|
| otherwise = void . runResourceT
|
||||||
|
|
||||||
watchdog <- async $ do
|
watchdog <- async $ do
|
||||||
|
|
||||||
|
@ -174,7 +179,10 @@ runMessagingUnix env = do
|
||||||
|
|
||||||
withSession do
|
withSession do
|
||||||
|
|
||||||
let that = msgUnixSelf env & over fromPeerUnix (<> "#" <> show peerNum)
|
let that = if doFork then
|
||||||
|
msgUnixSelf env & over fromPeerUnix (<> "#" <> show peerNum)
|
||||||
|
else
|
||||||
|
msgUnixSelf env
|
||||||
|
|
||||||
void $ allocate ( createQueues env that ) dropQueuesFor
|
void $ allocate ( createQueues env that ) dropQueuesFor
|
||||||
|
|
||||||
|
|
|
@ -7,6 +7,8 @@ import HBS2.Hash
|
||||||
import HBS2.Merkle
|
import HBS2.Merkle
|
||||||
import HBS2.Storage
|
import HBS2.Storage
|
||||||
|
|
||||||
|
import HBS2.System.Logger.Simple
|
||||||
|
|
||||||
import Streaming.Prelude qualified as S
|
import Streaming.Prelude qualified as S
|
||||||
import Control.Monad.Trans.Maybe
|
import Control.Monad.Trans.Maybe
|
||||||
import Control.Monad
|
import Control.Monad
|
||||||
|
@ -16,12 +18,14 @@ import Data.Maybe
|
||||||
findMissedBlocks :: (MonadIO m) => AnyStorage -> HashRef -> m [HashRef]
|
findMissedBlocks :: (MonadIO m) => AnyStorage -> HashRef -> m [HashRef]
|
||||||
findMissedBlocks sto href = do
|
findMissedBlocks sto href = do
|
||||||
|
|
||||||
|
trace $ "findMissedBlocks" <+> pretty href
|
||||||
|
|
||||||
S.toList_ $
|
S.toList_ $
|
||||||
|
|
||||||
walkMerkle (fromHashRef href) (lift . getBlock sto) $ \(hr :: Either (Hash HbSync) [HashRef]) -> do
|
walkMerkle (fromHashRef href) (lift . getBlock sto) $ \(hr :: Either (Hash HbSync) [HashRef]) -> do
|
||||||
case hr of
|
case hr of
|
||||||
-- FIXME: investigate-this-wtf
|
-- FIXME: investigate-this-wtf
|
||||||
Left{} -> pure ()
|
Left hx -> S.yield (HashRef hx)
|
||||||
Right (hrr :: [HashRef]) -> do
|
Right (hrr :: [HashRef]) -> do
|
||||||
forM_ hrr $ \hx -> runMaybeT do
|
forM_ hrr $ \hx -> runMaybeT do
|
||||||
blk <- lift $ getBlock sto (fromHashRef hx)
|
blk <- lift $ getBlock sto (fromHashRef hx)
|
||||||
|
|
|
@ -1037,7 +1037,7 @@ runPeer opts = U.handle (\e -> myException e
|
||||||
]
|
]
|
||||||
|
|
||||||
let rpcSa = getRpcSocketName conf
|
let rpcSa = getRpcSocketName conf
|
||||||
rpcmsg <- newMessagingUnixOpts [MUFork] True 1.0 rpcSa
|
rpcmsg <- newMessagingUnix True 1.0 rpcSa
|
||||||
|
|
||||||
let rpcctx = RPC2Context { rpcConfig = fromPeerConfig conf
|
let rpcctx = RPC2Context { rpcConfig = fromPeerConfig conf
|
||||||
, rpcMessaging = rpcmsg
|
, rpcMessaging = rpcmsg
|
||||||
|
|
|
@ -34,6 +34,7 @@ import HBS2.Net.Proto.Peer
|
||||||
import HBS2.Net.Proto.RefChan
|
import HBS2.Net.Proto.RefChan
|
||||||
import HBS2.Net.Proto.Sessions
|
import HBS2.Net.Proto.Sessions
|
||||||
import HBS2.Storage
|
import HBS2.Storage
|
||||||
|
import HBS2.Storage.Operations.Missed
|
||||||
|
|
||||||
import HBS2.System.Logger.Simple
|
import HBS2.System.Logger.Simple
|
||||||
|
|
||||||
|
|
|
@ -151,7 +151,7 @@ main = do
|
||||||
|
|
||||||
let soname = tmp </> "unix.socket"
|
let soname = tmp </> "unix.socket"
|
||||||
|
|
||||||
server <- newMessagingUnixOpts [MUFork] True 1.0 soname
|
server <- newMessagingUnix True 1.0 soname
|
||||||
|
|
||||||
client1 <- newMessagingUnix False 1.0 soname
|
client1 <- newMessagingUnix False 1.0 soname
|
||||||
client2 <- newMessagingUnix False 1.0 soname
|
client2 <- newMessagingUnix False 1.0 soname
|
||||||
|
|
Loading…
Reference in New Issue