summaryrefslogtreecommitdiff
path: root/src/Erebos/Network.hs
diff options
context:
space:
mode:
Diffstat (limited to 'src/Erebos/Network.hs')
-rw-r--r--src/Erebos/Network.hs188
1 files changed, 89 insertions, 99 deletions
diff --git a/src/Erebos/Network.hs b/src/Erebos/Network.hs
index b341974..6265bbf 100644
--- a/src/Erebos/Network.hs
+++ b/src/Erebos/Network.hs
@@ -1,5 +1,3 @@
-{-# LANGUAGE CPP #-}
-
module Erebos.Network (
Server,
startServer,
@@ -10,8 +8,8 @@ module Erebos.Network (
ServerOptions(..), serverIdentity, defaultServerOptions,
Peer, peerServer, peerStorage,
- PeerAddress(..), peerAddress,
- PeerIdentity(..), peerIdentity,
+ PeerAddress(..), getPeerAddress, getPeerAddresses,
+ PeerIdentity(..), getPeerIdentity,
WaitingRef, wrDigest,
Service(..),
@@ -20,9 +18,7 @@ module Erebos.Network (
serverPeer,
serverPeerCustom,
-#ifdef ENABLE_ICE_SUPPORT
- serverPeerIce,
-#endif
+ findPeer,
dropPeer,
isPeerDropped,
sendToPeer, sendManyToPeer,
@@ -66,10 +62,8 @@ import Network.Socket hiding (ControlMessage)
import Network.Socket.ByteString qualified as S
import Erebos.Error
-#ifdef ENABLE_ICE_SUPPORT
-import Erebos.ICE
-#endif
import Erebos.Identity
+import Erebos.Network.Address
import Erebos.Network.Channel
import Erebos.Network.Protocol
import Erebos.Object.Internal
@@ -121,12 +115,16 @@ getNextPeerChange = atomically . readTChan . serverChanPeer
data ServerOptions = ServerOptions
{ serverPort :: PortNumber
, serverLocalDiscovery :: Bool
+ , serverErrorPrefix :: String
+ , serverTestLog :: Bool
}
defaultServerOptions :: ServerOptions
defaultServerOptions = ServerOptions
{ serverPort = discoveryPort
, serverLocalDiscovery = True
+ , serverErrorPrefix = ""
+ , serverTestLog = False
}
@@ -141,6 +139,14 @@ data Peer = Peer
, peerWaitingRefs :: TMVar [WaitingRef]
}
+-- | Get current main address of the peer (used to send new packets).
+getPeerAddress :: MonadIO m => Peer -> m PeerAddress
+getPeerAddress = liftIO . return . peerAddress
+
+-- | Get all known addresses of given peer.
+getPeerAddresses :: MonadIO m => Peer -> m [ PeerAddress ]
+getPeerAddresses = fmap (: []) . getPeerAddress
+
peerServer :: Peer -> Server
peerServer = peerServer_
@@ -166,36 +172,24 @@ instance Eq Peer where
class (Eq addr, Ord addr, Show addr, Typeable addr) => PeerAddressType addr where
sendBytesToAddress :: addr -> ByteString -> IO ()
+ connectionToAddressClosed :: addr -> IO ()
data PeerAddress
= forall addr. PeerAddressType addr => CustomPeerAddress addr
| DatagramAddress SockAddr
-#ifdef ENABLE_ICE_SUPPORT
- | PeerIceSession IceSession
-#endif
instance Show PeerAddress where
show (CustomPeerAddress addr) = show addr
- show (DatagramAddress saddr) = unwords $ case IP.fromSockAddr saddr of
- Just (IP.IPv6 ipv6, port)
- | (0, 0, 0xffff, ipv4) <- IP.fromIPv6w ipv6
- -> [show (IP.toIPv4w ipv4), show port]
- Just (addr, port)
- -> [show addr, show port]
- _ -> [show saddr]
-
-#ifdef ENABLE_ICE_SUPPORT
- show (PeerIceSession ice) = show ice
-#endif
+ show (DatagramAddress saddr) =
+ case inetFromSockAddr saddr of
+ Just ( addr, port ) -> unwords [ show addr, show port ]
+ _ -> show saddr
instance Eq PeerAddress where
CustomPeerAddress addr == CustomPeerAddress addr'
| Just addr'' <- cast addr' = addr == addr''
DatagramAddress addr == DatagramAddress addr' = addr == addr'
-#ifdef ENABLE_ICE_SUPPORT
- PeerIceSession ice == PeerIceSession ice' = ice == ice'
-#endif
_ == _ = False
instance Ord PeerAddress where
@@ -206,20 +200,16 @@ instance Ord PeerAddress where
compare _ (CustomPeerAddress _ ) = GT
compare (DatagramAddress addr) (DatagramAddress addr') = compare addr addr'
-#ifdef ENABLE_ICE_SUPPORT
- compare (DatagramAddress _ ) _ = LT
- compare _ (DatagramAddress _ ) = GT
- compare (PeerIceSession ice ) (PeerIceSession ice') = compare ice ice'
-#endif
+data PeerIdentity
+ = PeerIdentityUnknown (TVar [ UnifiedIdentity -> ExceptT ErebosError IO () ])
+ | PeerIdentityRef WaitingRef (TVar [ UnifiedIdentity -> ExceptT ErebosError IO () ])
+ | PeerIdentityFull UnifiedIdentity
-data PeerIdentity = PeerIdentityUnknown (TVar [UnifiedIdentity -> ExceptT ErebosError IO ()])
- | PeerIdentityRef WaitingRef (TVar [UnifiedIdentity -> ExceptT ErebosError IO ()])
- | PeerIdentityFull UnifiedIdentity
-
-peerIdentity :: MonadIO m => Peer -> m PeerIdentity
-peerIdentity = liftIO . atomically . readTVar . peerIdentityVar
+-- | Get currently known identity of the given peer
+getPeerIdentity :: MonadIO m => Peer -> m PeerIdentity
+getPeerIdentity = liftIO . atomically . readTVar . peerIdentityVar
data PeerState
@@ -277,7 +267,16 @@ startServer serverOptions serverOrigHead logd' serverServices = do
let logd = writeTQueue serverErrorLog
forkServerThread server $ forever $ do
- logd' =<< atomically (readTQueue serverErrorLog)
+ logd' . (serverErrorPrefix serverOptions <>) =<< atomically (readTQueue serverErrorLog)
+
+ logt <- if
+ | serverTestLog serverOptions -> do
+ serverTestLog <- newTQueueIO
+ forkServerThread server $ forever $ do
+ logd' =<< atomically (readTQueue serverTestLog)
+ return $ writeTQueue serverTestLog
+ | otherwise -> do
+ return $ \_ -> return ()
forkServerThread server $ dataResponseWorker server
forkServerThread server $ forever $ do
@@ -327,13 +326,18 @@ startServer serverOptions serverOrigHead logd' serverServices = do
announceUpdate idt
forM_ serverServices $ \(SomeService service _) -> do
- forM_ (serviceStorageWatchers service) $ \(SomeStorageWatcher sel act) -> do
- watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
- withMVar serverPeers $ mapM_ $ \peer -> atomically $ do
- readTVar (peerIdentityVar peer) >>= \case
- PeerIdentityFull _ -> writeTQueue serverIOActions $ do
- runPeerService peer $ act x
- _ -> return ()
+ forM_ (serviceStorageWatchers service) $ \case
+ SomeStorageWatcher sel act -> do
+ watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
+ withMVar serverPeers $ mapM_ $ \peer -> atomically $ do
+ readTVar (peerIdentityVar peer) >>= \case
+ PeerIdentityFull _ -> writeTQueue serverIOActions $ do
+ runPeerService peer $ act x
+ _ -> return ()
+ GlobalStorageWatcher sel act -> do
+ watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
+ atomically $ writeTQueue serverIOActions $ do
+ act server x
forkServerThread server $ forever $ do
(msg, saddr) <- S.recvFrom sock 4096
@@ -345,9 +349,6 @@ startServer serverOptions serverOrigHead logd' serverServices = do
case paddr of
CustomPeerAddress addr -> sendBytesToAddress addr msg
DatagramAddress addr -> void $ S.sendTo sock msg addr
-#ifdef ENABLE_ICE_SUPPORT
- PeerIceSession ice -> iceSend ice msg
-#endif
forkServerThread server $ forever $ do
readFlowIO serverControlFlow >>= \case
@@ -389,9 +390,13 @@ startServer serverOptions serverOrigHead logd' serverServices = do
prefs <- forM objs $ storeObject $ peerInStorage peer
identity <- readMVar serverIdentity_
let svcs = map someServiceID serverServices
- handlePacket identity secure peer chanSvc svcs header prefs
+ handlePacket paddr identity secure peer chanSvc svcs header prefs
peerLoop
Nothing -> do
+ case paddr of
+ DatagramAddress _ -> return ()
+ CustomPeerAddress caddr -> connectionToAddressClosed caddr
+
dropPeer peer
atomically $ writeTChan serverChanPeer peer
peerLoop
@@ -399,7 +404,7 @@ startServer serverOptions serverOrigHead logd' serverServices = do
ReceivedAnnounce addr _ -> do
void $ serverPeer' server addr
- erebosNetworkProtocol (headLocalIdentity serverOrigHead) logd protocolRawPath protocolControlFlow
+ erebosNetworkProtocol (headLocalIdentity serverOrigHead) logd logt protocolRawPath protocolControlFlow
forkServerThread server $ withSocketsDo $ do
let hints = defaultHints
@@ -411,9 +416,9 @@ startServer serverOptions serverOrigHead logd' serverServices = do
bracket (open addr) close loop
forkServerThread server $ forever $ do
- ( peer, svc, ref, streams ) <- atomically $ readTQueue chanSvc
+ ( peer, paddr, svc, ref, streams ) <- atomically $ readTQueue chanSvc
case find ((svc ==) . someServiceID) serverServices of
- Just service@(SomeService (_ :: Proxy s) attr) -> runPeerServiceOn (Just ( service, attr )) streams peer (serviceHandler $ wrappedLoad @s ref)
+ Just service@(SomeService (_ :: Proxy s) attr) -> runPeerServiceOn (Just ( service, attr )) streams paddr peer (serviceHandler $ wrappedLoad @s ref)
_ -> atomically $ logd $ "unhandled service '" ++ show (toUUID svc) ++ "'"
return server
@@ -560,10 +565,10 @@ appendDistinct x (y:ys) | x == y = y : ys
| otherwise = y : appendDistinct x ys
appendDistinct x [] = [x]
-handlePacket :: UnifiedIdentity -> Bool
- -> Peer -> TQueue ( Peer, ServiceID, Ref, [ RawStreamReader ]) -> [ ServiceID ]
+handlePacket :: PeerAddress -> UnifiedIdentity -> Bool
+ -> Peer -> TQueue ( Peer, PeerAddress, ServiceID, Ref, [ RawStreamReader ] ) -> [ ServiceID ]
-> TransportHeader -> [ PartialRef ] -> IO ()
-handlePacket identity secure peer chanSvc svcs (TransportHeader headers) prefs = atomically $ do
+handlePacket paddr identity secure peer chanSvc svcs (TransportHeader headers) prefs = atomically $ do
let server = peerServer peer
ochannel <- getPeerChannel peer
let sidentity = idData identity
@@ -699,7 +704,7 @@ handlePacket identity secure peer chanSvc svcs (TransportHeader headers) prefs =
then do
streamReaders <- mapM acceptStream $ lookupNewStreams headers
void $ newWaitingRef dgst $ \ref ->
- liftIO $ atomically $ writeTQueue chanSvc ( peer, svc, ref, streamReaders )
+ liftIO $ atomically $ writeTQueue chanSvc ( peer, paddr, svc, ref, streamReaders )
else throwError $ "missing service object " ++ show dgst
| otherwise -> addHeader $ Rejected dgst
| otherwise -> throwError $ "service ref without type"
@@ -779,7 +784,7 @@ finalizedChannel peer@Peer {..} ch self = do
-- Notify services about new peer
readTVar peerIdentityVar >>= \case
- PeerIdentityFull _ -> notifyServicesOfPeer peer
+ PeerIdentityFull _ -> notifyServicesOfPeer True peer
_ -> return ()
@@ -805,7 +810,7 @@ handleIdentityAnnounce self peer ref = liftIO $ atomically $ do
PeerIdentityFull pid
| idData pid `precedes` wrappedLoad ref
-> validateAndUpdate (idUpdates pid) $ \_ -> do
- notifyServicesOfPeer peer
+ notifyServicesOfPeer False peer
_ -> return ()
@@ -817,15 +822,18 @@ handleIdentityUpdate peer ref = liftIO $ atomically $ do
-> do
writeTVar (peerIdentityVar peer) $ PeerIdentityFull pid'
writeTChan (serverChanPeer $ peerServer peer) peer
- when (idData pid /= idData pid') $ notifyServicesOfPeer peer
+ when (pid /= pid') $ do
+ notifyServicesOfPeer False peer
| otherwise -> return ()
-notifyServicesOfPeer :: Peer -> STM ()
-notifyServicesOfPeer peer@Peer { peerServer_ = Server {..} } = do
+notifyServicesOfPeer :: Bool -> Peer -> STM ()
+notifyServicesOfPeer new peer@Peer { peerServer_ = Server {..} } = do
writeTQueue serverIOActions $ do
+ paddr <- getPeerAddress peer
forM_ serverServices $ \service@(SomeService _ attrs) ->
- runPeerServiceOn (Just ( service, attrs )) [] peer serviceNewPeer
+ runPeerServiceOn (Just ( service, attrs )) [] paddr peer $
+ if new then serviceNewPeer else serviceUpdatedPeer
receivedFromCustomAddress :: PeerAddressType addr => Server -> addr -> ByteString -> IO ()
@@ -853,15 +861,6 @@ serverPeer server paddr = do
serverPeerCustom :: PeerAddressType addr => Server -> addr -> IO Peer
serverPeerCustom server addr = serverPeer' server (CustomPeerAddress addr)
-#ifdef ENABLE_ICE_SUPPORT
-serverPeerIce :: Server -> IceSession -> IO Peer
-serverPeerIce server@Server {..} ice = do
- let paddr = PeerIceSession ice
- peer <- serverPeer' server paddr
- iceSetChan ice $ mapFlow undefined (paddr,) serverRawPath
- return peer
-#endif
-
serverPeer' :: Server -> PeerAddress -> IO Peer
serverPeer' server paddr = do
(peer, hello) <- modifyMVar (serverPeers server) $ \pvalue -> do
@@ -874,6 +873,13 @@ serverPeer' server paddr = do
writeFlow (serverControlFlow server) (RequestConnection paddr)
return peer
+findPeer :: Server -> (Peer -> IO Bool) -> IO (Maybe Peer)
+findPeer server test = withMVar (serverPeers server) (helper . M.elems)
+ where
+ helper (p : ps) = test p >>= \case True -> return (Just p)
+ False -> helper ps
+ helper [] = return Nothing
+
dropPeer :: MonadIO m => Peer -> m ()
dropPeer peer = liftIO $ do
modifyMVar_ (serverPeers $ peerServer peer) $ \pvalue -> do
@@ -983,10 +989,12 @@ lookupService proxy (service@(SomeService (_ :: Proxy t) attr) : rest)
lookupService _ [] = Nothing
runPeerService :: forall s m. (Service s, MonadIO m) => Peer -> ServiceHandler s () -> m ()
-runPeerService = runPeerServiceOn Nothing []
+runPeerService peer handler = do
+ paddr <- getPeerAddress peer
+ runPeerServiceOn Nothing [] paddr peer handler
-runPeerServiceOn :: forall s m. (Service s, MonadIO m) => Maybe ( SomeService, ServiceAttributes s ) -> [ RawStreamReader ] -> Peer -> ServiceHandler s () -> m ()
-runPeerServiceOn mbservice newStreams peer handler = liftIO $ do
+runPeerServiceOn :: forall s m. (Service s, MonadIO m) => Maybe ( SomeService, ServiceAttributes s ) -> [ RawStreamReader ] -> PeerAddress -> Peer -> ServiceHandler s () -> m ()
+runPeerServiceOn mbservice newStreams paddr peer handler = liftIO $ do
let server = peerServer peer
proxy = Proxy @s
svc = serviceID proxy
@@ -1008,6 +1016,7 @@ runPeerServiceOn mbservice newStreams peer handler = liftIO $ do
let inp = ServiceInput
{ svcAttributes = attr
, svcPeer = peer
+ , svcPeerAddress = paddr
, svcPeerIdentity = peerId
, svcServer = server
, svcPrintOp = atomically . logd
@@ -1027,7 +1036,7 @@ runPeerServiceOn mbservice newStreams peer handler = liftIO $ do
putTMVar (peerServiceState peer) $ M.insert svc (SomeServiceState proxy s') svcs
putTMVar (serverServiceStates server) $ M.insert svc (SomeServiceGlobalState proxy gs') global
_ -> do
- atomically $ logd $ "can't run service handler on peer with incomplete identity " ++ show (peerAddress peer)
+ atomically $ logd $ "can't run service handler on peer with incomplete identity " ++ show paddr
_ -> atomically $ do
logd $ "unhandled service '" ++ show (toUUID svc) ++ "'"
@@ -1054,30 +1063,11 @@ modifyServiceGlobalState server proxy f = do
throwOtherError $ "unhandled service '" ++ show (toUUID svc) ++ "'"
-foreign import ccall unsafe "Network/ifaddrs.h join_multicast" cJoinMulticast :: CInt -> Ptr CSize -> IO (Ptr Word32)
-foreign import ccall unsafe "Network/ifaddrs.h local_addresses" cLocalAddresses :: Ptr CSize -> IO (Ptr InetAddress)
-foreign import ccall unsafe "Network/ifaddrs.h broadcast_addresses" cBroadcastAddresses :: IO (Ptr Word32)
+foreign import ccall unsafe "Network/ifaddrs.h erebos_join_multicast" cJoinMulticast :: CInt -> Ptr CSize -> IO (Ptr Word32)
+foreign import ccall unsafe "Network/ifaddrs.h erebos_local_addresses" cLocalAddresses :: Ptr CSize -> IO (Ptr InetAddress)
+foreign import ccall unsafe "Network/ifaddrs.h erebos_broadcast_addresses" cBroadcastAddresses :: IO (Ptr Word32)
foreign import ccall unsafe "stdlib.h free" cFree :: Ptr a -> IO ()
-data InetAddress = InetAddress { fromInetAddress :: IP.IP }
-
-instance F.Storable InetAddress where
- sizeOf _ = sizeOf (undefined :: CInt) + 16
- alignment _ = 8
-
- peek ptr = (unpackFamily <$> peekByteOff ptr 0) >>= \case
- AF_INET -> InetAddress . IP.IPv4 . IP.fromHostAddress <$> peekByteOff ptr (sizeOf (undefined :: CInt))
- AF_INET6 -> InetAddress . IP.IPv6 . IP.toIPv6b . map fromIntegral <$> peekArray 16 (ptr `plusPtr` sizeOf (undefined :: CInt) :: Ptr Word8)
- _ -> fail "InetAddress: unknown family"
-
- poke ptr (InetAddress addr) = case addr of
- IP.IPv4 ip -> do
- pokeByteOff ptr 0 (packFamily AF_INET)
- pokeByteOff ptr (sizeOf (undefined :: CInt)) (IP.toHostAddress ip)
- IP.IPv6 ip -> do
- pokeByteOff ptr 0 (packFamily AF_INET6)
- pokeArray (ptr `plusPtr` sizeOf (undefined :: CInt) :: Ptr Word8) (map fromIntegral $ IP.fromIPv6b ip)
-
joinMulticast :: Socket -> IO [ Word32 ]
joinMulticast sock =
withFdSocket sock $ \fd ->
@@ -1104,7 +1094,7 @@ getServerAddresses Server {..} = do
count <- fromIntegral <$> peek pcount
res <- peekArray count ptr
cFree ptr
- return $ map (IP.toSockAddr . (, serverPort serverOptions ) . fromInetAddress) res
+ return $ map (inetToSockAddr . (, serverPort serverOptions )) res
getBroadcastAddresses :: PortNumber -> IO [SockAddr]
getBroadcastAddresses port = do
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-16 13:31:45 +0000