erebos-0.2.0: src/Erebos/Network.hs
module Erebos.Network (
Server,
startServer,
stopServer,
getCurrentPeerList,
getNextPeerChange,
getServerAddresses,
ServerOptions(..), serverIdentity, defaultServerOptions,
Peer, peerServer, peerStorage,
PeerAddress(..), getPeerAddress, getPeerAddresses,
PeerIdentity(..), getPeerIdentity,
WaitingRef, wrDigest,
Service(..),
PeerAddressType(..),
receivedFromCustomAddress,
serverPeer,
serverPeerCustom,
findPeer,
dropPeer,
isPeerDropped,
sendToPeer, sendManyToPeer,
sendToPeerStored, sendManyToPeerStored,
sendToPeerWith,
runPeerService,
modifyServiceGlobalState,
discoveryPort,
) where
import Control.Concurrent
import Control.Concurrent.STM
import Control.Exception
import Control.Monad
import Control.Monad.Except
import Control.Monad.Reader
import Control.Monad.State
import Data.ByteString (ByteString)
import Data.ByteString.Char8 qualified as BC
import Data.ByteString.Lazy qualified as BL
import Data.Function
import Data.IP qualified as IP
import Data.List
import Data.Map (Map)
import Data.Map qualified as M
import Data.Maybe
import Data.Typeable
import Data.Word
import Foreign.C.Types
import Foreign.Marshal.Alloc
import Foreign.Marshal.Array
import Foreign.Ptr
import Foreign.Storable as F
import GHC.Conc.Sync (unsafeIOToSTM)
import Network.Socket hiding (ControlMessage)
import Network.Socket.ByteString qualified as S
import Erebos.Error
import Erebos.Identity
import Erebos.Network.Address
import Erebos.Network.Channel
import Erebos.Network.Protocol
import Erebos.Object.Internal
import Erebos.PubKey
import Erebos.Service
import Erebos.State
import Erebos.Storage
import Erebos.Storage.Key
import Erebos.Storage.Merge
discoveryPort :: PortNumber
discoveryPort = 29665
discoveryMulticastGroup :: HostAddress6
discoveryMulticastGroup = tupleToHostAddress6 (0xff12, 0xb6a4, 0x6b1f, 0x0969, 0xcaee, 0xacc2, 0x5c93, 0x73e1) -- ff12:b6a4:6b1f:969:caee:acc2:5c93:73e1
announceIntervalSeconds :: Int
announceIntervalSeconds = 60
data Server = Server
{ serverStorage :: Storage
, serverOptions :: ServerOptions
, serverOrigHead :: Head LocalState
, serverIdentity_ :: MVar UnifiedIdentity
, serverThreads :: MVar [ThreadId]
, serverSocket :: MVar Socket
, serverRawPath :: SymFlow (PeerAddress, BC.ByteString)
, serverControlFlow :: Flow (ControlMessage PeerAddress) (ControlRequest PeerAddress)
, serverDataResponse :: TQueue (Peer, Maybe PartialRef)
, serverIOActions :: TQueue (ExceptT ErebosError IO ())
, serverServices :: [SomeService]
, serverServiceStates :: TMVar (M.Map ServiceID SomeServiceGlobalState)
, serverPeers :: MVar (Map PeerAddress Peer)
, serverChanPeer :: TChan Peer
, serverErrorLog :: TQueue String
}
serverIdentity :: Server -> IO UnifiedIdentity
serverIdentity = readMVar . serverIdentity_
getCurrentPeerList :: Server -> IO [Peer]
getCurrentPeerList = fmap M.elems . readMVar . serverPeers
getNextPeerChange :: Server -> IO Peer
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
}
data Peer = Peer
{ peerAddress :: PeerAddress
, peerServer_ :: Server
, peerState :: TVar PeerState
, peerIdentityVar :: TVar PeerIdentity
, peerStorage_ :: Storage
, peerInStorage :: PartialStorage
, peerServiceState :: TMVar (M.Map ServiceID SomeServiceState)
, 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_
peerStorage :: Peer -> Storage
peerStorage = peerStorage_
getPeerChannel :: Peer -> STM ChannelState
getPeerChannel Peer {..} =
readTVar peerState >>= \case
PeerInit _ -> return ChannelNone
PeerConnected conn -> connGetChannel conn
PeerDropped -> return ChannelClosed
setPeerChannel :: Peer -> ChannelState -> STM ()
setPeerChannel Peer {..} ch = do
readTVar peerState >>= \case
PeerInit _ -> retry
PeerConnected conn -> connSetChannel conn ch
PeerDropped -> return ()
instance Eq Peer where
(==) = (==) `on` peerIdentityVar
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
instance Show PeerAddress where
show (CustomPeerAddress addr) = show addr
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'
_ == _ = False
instance Ord PeerAddress where
compare (CustomPeerAddress addr) (CustomPeerAddress addr')
| Just addr'' <- cast addr' = compare addr addr''
| otherwise = compare (typeOf addr) (typeOf addr')
compare (CustomPeerAddress _ ) _ = LT
compare _ (CustomPeerAddress _ ) = GT
compare (DatagramAddress addr) (DatagramAddress addr') = compare addr addr'
data PeerIdentity
= PeerIdentityUnknown (TVar [ UnifiedIdentity -> ExceptT ErebosError IO () ])
| PeerIdentityRef WaitingRef (TVar [ UnifiedIdentity -> ExceptT ErebosError IO () ])
| PeerIdentityFull UnifiedIdentity
-- | Get currently known identity of the given peer
getPeerIdentity :: MonadIO m => Peer -> m PeerIdentity
getPeerIdentity = liftIO . atomically . readTVar . peerIdentityVar
data PeerState
= PeerInit [ ( SecurityRequirement, TransportPacket Ref, [ TransportHeaderItem ] ) ]
| PeerConnected (Connection PeerAddress)
| PeerDropped
lookupServiceType :: [TransportHeaderItem] -> Maybe ServiceID
lookupServiceType (ServiceType stype : _) = Just stype
lookupServiceType (_ : hs) = lookupServiceType hs
lookupServiceType [] = Nothing
lookupNewStreams :: [TransportHeaderItem] -> [Word8]
lookupNewStreams (StreamOpen num : rest) = num : lookupNewStreams rest
lookupNewStreams (_ : rest) = lookupNewStreams rest
lookupNewStreams [] = []
newWaitingRef :: RefDigest -> (Ref -> WaitingRefCallback) -> PacketHandler WaitingRef
newWaitingRef dgst act = do
peer@Peer {..} <- gets phPeer
wref <- WaitingRef peerStorage_ (partialRefFromDigest peerInStorage dgst) act <$> liftSTM (newTVar (Left []))
modifyTMVarP peerWaitingRefs (wref:)
liftSTM $ writeTQueue (serverDataResponse $ peerServer peer) (peer, Nothing)
return wref
forkServerThread :: Server -> IO () -> IO ()
forkServerThread server act = do
modifyMVar_ (serverThreads server) $ \ts -> do
t <- forkIO $ do
t <- myThreadId
act
modifyMVar_ (serverThreads server) $ return . filter (/=t)
return (t:ts)
startServer :: ServerOptions -> Head LocalState -> (String -> IO ()) -> [SomeService] -> IO Server
startServer serverOptions serverOrigHead logd' serverServices = do
let serverStorage = headStorage serverOrigHead
serverIdentity_ <- newMVar $ headLocalIdentity serverOrigHead
serverThreads <- newMVar []
serverSocket <- newEmptyMVar
(serverRawPath, protocolRawPath) <- newFlowIO
(serverControlFlow, protocolControlFlow) <- newFlowIO
serverDataResponse <- newTQueueIO
serverIOActions <- newTQueueIO
serverServiceStates <- newTMVarIO M.empty
serverPeers <- newMVar M.empty
serverChanPeer <- newTChanIO
serverErrorLog <- newTQueueIO
let server = Server {..}
chanSvc <- newTQueueIO
let logd = writeTQueue serverErrorLog
forkServerThread server $ forever $ do
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
either (atomically . logd . showErebosError) return =<< runExceptT =<<
atomically (readTQueue serverIOActions)
let open addr = do
sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
putMVar serverSocket sock
setSocketOption sock ReuseAddr 1
setSocketOption sock Broadcast 1
withFdSocket sock setCloseOnExecIfNeeded
bind sock (addrAddress addr)
return sock
loop sock = do
when (serverLocalDiscovery serverOptions) $ forkServerThread server $ do
announceAddreses <- fmap concat $ sequence $
[ map (SockAddrInet6 discoveryPort 0 discoveryMulticastGroup) <$> joinMulticast sock
, getBroadcastAddresses discoveryPort
]
forever $ do
atomically $ writeFlowBulk serverControlFlow $ map (SendAnnounce . DatagramAddress) announceAddreses
threadDelay $ announceIntervalSeconds * 1000 * 1000
let announceUpdate identity = do
st <- derivePartialStorage serverStorage
let selfRef = partialRef st $ storedRef $ idExtData identity
updateRefs = map refDigest $ selfRef : map (partialRef st . storedRef) (idUpdates identity)
ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- updateRefs ]
hitems = map AnnounceUpdate updateRefs
packet = TransportPacket (TransportHeader $ hitems) []
ps <- readMVar serverPeers
forM_ ps $ \peer -> atomically $ do
((,) <$> readTVar (peerIdentityVar peer) <*> getPeerChannel peer) >>= \case
(PeerIdentityFull _, ChannelEstablished _) ->
sendToPeerS peer ackedBy packet
_ -> return ()
void $ watchHead serverOrigHead $ \h -> do
let idt = headLocalIdentity h
changedId <- modifyMVar serverIdentity_ $ \cur ->
return (idt, cur /= idt)
when changedId $ do
writeFlowIO serverControlFlow $ UpdateSelfIdentity idt
announceUpdate idt
forM_ serverServices $ \(SomeService service _) -> do
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
writeFlowIO serverRawPath (DatagramAddress saddr, msg)
forkServerThread server $ forever $ do
(paddr, msg) <- readFlowIO serverRawPath
handle (\(e :: SomeException) -> atomically . logd $ "failed to send packet to " ++ show paddr ++ ": " ++ show e) $ do
case paddr of
CustomPeerAddress addr -> sendBytesToAddress addr msg
DatagramAddress addr -> void $ S.sendTo sock msg addr
forkServerThread server $ forever $ do
readFlowIO serverControlFlow >>= \case
NewConnection conn mbpid -> do
let paddr = connAddress conn
peer <- modifyMVar serverPeers $ \pvalue -> do
case M.lookup paddr pvalue of
Just peer -> return (pvalue, peer)
Nothing -> do
peer <- mkPeer server paddr
return (M.insert paddr peer pvalue, peer)
forkServerThread server $ do
atomically $ do
readTVar (peerState peer) >>= \case
PeerInit packets -> do
writeFlowBulk (connData conn) $ reverse packets
writeTVar (peerState peer) (PeerConnected conn)
PeerConnected _ -> do
writeTVar (peerState peer) (PeerConnected conn)
PeerDropped -> do
connClose conn
case mbpid of
Just dgst -> do
identity <- readMVar serverIdentity_
atomically $ runPacketHandler False peer $ do
wref <- newWaitingRef dgst $ handleIdentityAnnounce identity peer
readTVarP (peerIdentityVar peer) >>= \case
PeerIdentityUnknown idwait -> do
addHeader $ AnnounceSelf $ refDigest $ storedRef $ idData identity
writeTVarP (peerIdentityVar peer) $ PeerIdentityRef wref idwait
liftSTM $ writeTChan serverChanPeer peer
_ -> return ()
Nothing -> return ()
let peerLoop = readFlowIO (connData conn) >>= \case
Just (secure, TransportPacket header objs) -> do
prefs <- forM objs $ storeObject $ peerInStorage peer
identity <- readMVar serverIdentity_
let svcs = map someServiceID serverServices
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
ReceivedAnnounce addr _ -> do
void $ serverPeer' server addr
erebosNetworkProtocol (headLocalIdentity serverOrigHead) logd logt protocolRawPath protocolControlFlow
forkServerThread server $ withSocketsDo $ do
let hints = defaultHints
{ addrFlags = [AI_PASSIVE]
, addrFamily = AF_INET6
, addrSocketType = Datagram
}
addr:_ <- getAddrInfo (Just hints) Nothing (Just $ show $ serverPort serverOptions)
bracket (open addr) close loop
forkServerThread server $ forever $ do
( 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 paddr peer (serviceHandler $ wrappedLoad @s ref)
_ -> atomically $ logd $ "unhandled service '" ++ show (toUUID svc) ++ "'"
return server
stopServer :: Server -> IO ()
stopServer server@Server {..} = do
withMVar serverPeers $ \peers -> do
( global, peerStates ) <- atomically $ (,)
<$> takeTMVar serverServiceStates
<*> (forM (M.elems peers) $ \p@Peer {..} -> ( p, ) <$> takeTMVar peerServiceState)
forM_ global $ \(SomeServiceGlobalState (proxy :: Proxy s) gs) -> do
ps <- forM peerStates $ \( peer, states ) ->
return $ ( peer, ) $ case M.lookup (serviceID proxy) states of
Just (SomeServiceState (_ :: Proxy ps) pstate)
| Just (Refl :: s :~: ps) <- eqT
-> pstate
_ -> emptyServiceState proxy
serviceStopServer proxy server gs ps
mapM_ killThread =<< takeMVar serverThreads
dataResponseWorker :: Server -> IO ()
dataResponseWorker server = forever $ do
(peer, npref) <- atomically (readTQueue $ serverDataResponse server)
wait <- atomically $ takeTMVar (peerWaitingRefs peer)
list <- forM wait $ \wr@WaitingRef { wrefStatus = tvar } ->
atomically (readTVar tvar) >>= \case
Left ds -> case maybe id (filter . (/=) . refDigest) npref $ ds of
[] -> copyRef (wrefStorage wr) (wrefPartial wr) >>= \case
Right ref -> do
atomically (writeTVar tvar $ Right ref)
forkServerThread server $ runExceptT (wrefAction wr ref) >>= \case
Left err -> atomically $ writeTQueue (serverErrorLog server) (showErebosError err)
Right () -> return ()
return (Nothing, [])
Left dgst -> do
atomically (writeTVar tvar $ Left [dgst])
return (Just wr, [dgst])
ds' -> do
atomically (writeTVar tvar $ Left ds')
return (Just wr, [])
Right _ -> return (Nothing, [])
atomically $ putTMVar (peerWaitingRefs peer) $ catMaybes $ map fst list
let reqs = concat $ map snd list
when (not $ null reqs) $ do
let packet = TransportPacket (TransportHeader $ map DataRequest reqs) []
ackedBy = concat [[ Rejected r, DataResponse r ] | r <- reqs ]
atomically $ sendToPeerPlain peer ackedBy packet
newtype PacketHandler a = PacketHandler { unPacketHandler :: StateT PacketHandlerState (ExceptT String STM) a }
deriving (Functor, Applicative, Monad, MonadState PacketHandlerState, MonadError String)
instance MonadFail PacketHandler where
fail = throwError
runPacketHandler :: Bool -> Peer -> PacketHandler () -> STM ()
runPacketHandler secure peer@Peer {..} act = do
let logd = writeTQueue $ serverErrorLog peerServer_
runExceptT (flip execStateT (PacketHandlerState peer [] [] [] Nothing False) $ unPacketHandler act) >>= \case
Left err -> do
logd $ "Error in handling packet from " ++ show peerAddress ++ ": " ++ err
Right ph -> do
when (not $ null $ phHead ph) $ do
body <- case phBodyStream ph of
Nothing -> return $ phBody ph
Just stream -> do
writeTQueue (serverIOActions peerServer_) $ void $ liftIO $ forkIO $ do
writeByteStringToStream stream $ BL.concat $ map lazyLoadBytes $ phBody ph
return []
let packet = TransportPacket (TransportHeader $ phHead ph) body
secreq = case (secure, phPlaintextReply ph) of
(True, _) -> EncryptedOnly
(False, False) -> PlaintextAllowed
(False, True) -> PlaintextOnly
sendToPeerS' secreq peer (phAckedBy ph) packet
liftSTM :: STM a -> PacketHandler a
liftSTM = PacketHandler . lift . lift
readTVarP :: TVar a -> PacketHandler a
readTVarP = liftSTM . readTVar
writeTVarP :: TVar a -> a -> PacketHandler ()
writeTVarP v = liftSTM . writeTVar v
modifyTMVarP :: TMVar a -> (a -> a) -> PacketHandler ()
modifyTMVarP v f = liftSTM $ putTMVar v . f =<< takeTMVar v
data PacketHandlerState = PacketHandlerState
{ phPeer :: Peer
, phHead :: [TransportHeaderItem]
, phAckedBy :: [TransportHeaderItem]
, phBody :: [Ref]
, phBodyStream :: Maybe RawStreamWriter
, phPlaintextReply :: Bool
}
addHeader :: TransportHeaderItem -> PacketHandler ()
addHeader h = modify $ \ph -> ph { phHead = h `appendDistinct` phHead ph }
addAckedBy :: [TransportHeaderItem] -> PacketHandler ()
addAckedBy hs = modify $ \ph -> ph { phAckedBy = foldr appendDistinct (phAckedBy ph) hs }
addBody :: Ref -> PacketHandler ()
addBody r = modify $ \ph -> ph { phBody = r `appendDistinct` phBody ph }
sendBodyAsStream :: PacketHandler ()
sendBodyAsStream = do
gets phBodyStream >>= \case
Nothing -> do
stream <- openStream
modify $ \ph -> ph { phBodyStream = Just stream }
Just _ -> return ()
keepPlaintextReply :: PacketHandler ()
keepPlaintextReply = modify $ \ph -> ph { phPlaintextReply = True }
openStream :: PacketHandler RawStreamWriter
openStream = do
Peer {..} <- gets phPeer
conn <- readTVarP peerState >>= \case
PeerConnected conn -> return conn
_ -> throwError "can't open stream without established connection"
(hdr, writer, handler) <- liftEither =<< liftSTM (connAddWriteStream conn)
liftSTM $ writeTQueue (serverIOActions peerServer_) (liftIO $ forkServerThread peerServer_ handler)
addHeader hdr
return writer
acceptStream :: Word8 -> PacketHandler RawStreamReader
acceptStream streamNumber = do
Peer {..} <- gets phPeer
conn <- readTVarP peerState >>= \case
PeerConnected conn -> return conn
_ -> throwError "can't accept stream without established connection"
liftSTM $ connAddReadStream conn streamNumber
appendDistinct :: Eq a => a -> [a] -> [a]
appendDistinct x (y:ys) | x == y = y : ys
| otherwise = y : appendDistinct x ys
appendDistinct x [] = [x]
handlePacket :: PeerAddress -> UnifiedIdentity -> Bool
-> Peer -> TQueue ( Peer, PeerAddress, ServiceID, Ref, [ RawStreamReader ] ) -> [ ServiceID ]
-> TransportHeader -> [ PartialRef ] -> IO ()
handlePacket paddr identity secure peer chanSvc svcs (TransportHeader headers) prefs = atomically $ do
let server = peerServer peer
ochannel <- getPeerChannel peer
let sidentity = idData identity
plaintextRefs = map (refDigest . storedRef) $ concatMap (collectStoredObjects . wrappedLoad) $ concat
[ [ storedRef sidentity ]
, map storedRef $ idUpdates identity
, case ochannel of
ChannelOurRequest req -> [ storedRef req ]
ChannelOurAccept acc _ -> [ storedRef acc ]
_ -> []
]
runPacketHandler secure peer $ do
let logd = liftSTM . writeTQueue (serverErrorLog server)
forM_ headers $ \case
Acknowledged dgst -> do
liftSTM (getPeerChannel peer) >>= \case
ChannelOurAccept acc ch | refDigest (storedRef acc) == dgst -> do
liftSTM $ finalizedChannel peer ch identity
_ -> return ()
Rejected dgst
| peerRequest : _ <- mapMaybe (\case TrChannelRequest d -> Just d; _ -> Nothing) headers
, peerRequest < dgst
-> return () -- Our request was rejected due to lower priority
| otherwise -> logd $ "rejected by peer: " ++ show dgst
DataRequest dgst
| secure || dgst `elem` plaintextRefs -> do
Right mref <- liftSTM $ unsafeIOToSTM $
copyRef (peerStorage peer) $
partialRefFromDigest (peerInStorage peer) dgst
addHeader $ DataResponse dgst
addAckedBy [ Acknowledged dgst, Rejected dgst ]
-- Plaintext request may indicate the peer has restarted/changed or
-- otherwise lost the channel, so keep the reply plaintext as well.
when (not secure) keepPlaintextReply
-- TODO: MTU
when (secure && BL.length (lazyLoadBytes mref) > 500)
sendBodyAsStream
addBody $ mref
| otherwise -> do
logd $ "unauthorized data request for " ++ show dgst
addHeader $ Rejected dgst
DataResponse dgst -> if
| Just pref <- find ((==dgst) . refDigest) prefs -> do
when (not secure) $ do
addHeader $ Acknowledged dgst
liftSTM $ writeTQueue (serverDataResponse server) (peer, Just pref)
| streamNumber : _ <- lookupNewStreams headers -> do
streamReader <- acceptStream streamNumber
liftSTM $ writeTQueue (serverIOActions server) $ void $ liftIO $ forkIO $ do
(runExcept <$> readObjectsFromStream (peerInStorage peer) streamReader) >>= \case
Left err -> atomically $ writeTQueue (serverErrorLog server) $
"failed to receive object from stream: " <> showErebosError err
Right objs -> do
forM_ objs $ \obj -> do
pref <- storeObject (peerInStorage peer) obj
atomically $ writeTQueue (serverDataResponse server) (peer, Just pref)
| otherwise -> throwError $ "mismatched data response " ++ show dgst
AnnounceSelf dgst
| dgst == refDigest (storedRef sidentity) -> return ()
| otherwise -> do
wref <- newWaitingRef dgst $ handleIdentityAnnounce identity peer
readTVarP (peerIdentityVar peer) >>= \case
PeerIdentityUnknown idwait -> do
addHeader $ AnnounceSelf $ refDigest $ storedRef $ idData identity
writeTVarP (peerIdentityVar peer) $ PeerIdentityRef wref idwait
liftSTM $ writeTChan (serverChanPeer $ peerServer peer) peer
_ -> return ()
AnnounceUpdate dgst -> do
readTVarP (peerIdentityVar peer) >>= \case
PeerIdentityFull _ -> do
void $ newWaitingRef dgst $ handleIdentityUpdate peer
_ -> return ()
TrChannelRequest dgst -> do
let process = do
addHeader $ Acknowledged dgst
wref <- newWaitingRef dgst $ handleChannelRequest peer identity
liftSTM $ setPeerChannel peer $ ChannelPeerRequest wref
reject = addHeader $ Rejected dgst
liftSTM (getPeerChannel peer) >>= \case
ChannelNone {} -> return ()
ChannelCookieWait {} -> return ()
ChannelCookieReceived {} -> process
ChannelCookieConfirmed {} -> process
ChannelOurRequest our
| dgst < refDigest (storedRef our) -> process
| otherwise -> do
-- Reject peer channel request with lower priority
addHeader $ TrChannelRequest $ refDigest $ storedRef our
reject
ChannelPeerRequest prev
| dgst == wrDigest prev -> addHeader $ Acknowledged dgst
| otherwise -> process
ChannelOurAccept {} -> reject
ChannelEstablished {} -> process
ChannelClosed {} -> return ()
TrChannelAccept dgst -> do
let process = do
handleChannelAccept identity $ partialRefFromDigest (peerInStorage peer) dgst
reject = addHeader $ Rejected dgst
liftSTM (getPeerChannel peer) >>= \case
ChannelNone {} -> reject
ChannelCookieWait {} -> reject
ChannelCookieReceived {} -> reject
ChannelCookieConfirmed {} -> reject
ChannelOurRequest {} -> process
ChannelPeerRequest {} -> process
ChannelOurAccept our _ | dgst < refDigest (storedRef our) -> process
| otherwise -> addHeader $ Rejected dgst
ChannelEstablished {} -> process
ChannelClosed {} -> return ()
ServiceType _ -> return ()
ServiceRef dgst
| not secure -> throwError $ "service packet without secure channel"
| Just svc <- lookupServiceType headers -> if
| svc `elem` svcs -> do
if dgst `elem` map refDigest prefs || True {- TODO: used by Message service to confirm receive -}
then do
streamReaders <- mapM acceptStream $ lookupNewStreams headers
void $ newWaitingRef dgst $ \ref ->
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"
_ -> return ()
withPeerIdentity :: MonadIO m => Peer -> (UnifiedIdentity -> ExceptT ErebosError IO ()) -> m ()
withPeerIdentity peer act = liftIO $ atomically $ readTVar (peerIdentityVar peer) >>= \case
PeerIdentityUnknown tvar -> modifyTVar' tvar (act:)
PeerIdentityRef _ tvar -> modifyTVar' tvar (act:)
PeerIdentityFull idt -> writeTQueue (serverIOActions $ peerServer peer) (act idt)
setupChannel :: UnifiedIdentity -> Peer -> UnifiedIdentity -> WaitingRefCallback
setupChannel identity peer upid = do
req <- flip runReaderT (peerStorage peer) $ createChannelRequest identity upid
let reqref = refDigest $ storedRef req
let hitems =
[ TrChannelRequest reqref
, AnnounceSelf $ refDigest $ storedRef $ idData identity
]
let sendChannelRequest = do
sendToPeerPlain peer [ Acknowledged reqref, Rejected reqref ] $
TransportPacket (TransportHeader hitems) [storedRef req]
setPeerChannel peer $ ChannelOurRequest req
liftIO $ atomically $ do
getPeerChannel peer >>= \case
ChannelCookieReceived -> sendChannelRequest
ChannelCookieConfirmed -> sendChannelRequest
_ -> return ()
handleChannelRequest :: Peer -> UnifiedIdentity -> Ref -> WaitingRefCallback
handleChannelRequest peer identity req = do
withPeerIdentity peer $ \upid -> do
(acc, ch) <- flip runReaderT (peerStorage peer) $ acceptChannelRequest identity upid (wrappedLoad req)
liftIO $ atomically $ do
getPeerChannel peer >>= \case
ChannelPeerRequest wr | wrDigest wr == refDigest req -> do
setPeerChannel peer $ ChannelOurAccept acc ch
let accref = refDigest $ storedRef acc
header = TrChannelAccept accref
ackedBy = [ Acknowledged accref, Rejected accref ]
sendToPeerPlain peer ackedBy $ TransportPacket (TransportHeader [header]) $ concat
[ [ storedRef $ acc ]
, [ storedRef $ signedData $ fromStored acc ]
, [ storedRef $ caKey $ fromStored $ signedData $ fromStored acc ]
, map storedRef $ signedSignature $ fromStored acc
]
_ -> writeTQueue (serverErrorLog $ peerServer peer) $ "unexpected channel request"
handleChannelAccept :: UnifiedIdentity -> PartialRef -> PacketHandler ()
handleChannelAccept identity accref = do
peer <- gets phPeer
liftSTM $ writeTQueue (serverIOActions $ peerServer peer) $ do
withPeerIdentity peer $ \upid -> do
copyRef (peerStorage peer) accref >>= \case
Right acc -> do
ch <- acceptedChannel identity upid (wrappedLoad acc)
liftIO $ atomically $ do
sendToPeerS peer [] $ TransportPacket (TransportHeader [Acknowledged $ refDigest accref]) []
finalizedChannel peer ch identity
Left dgst -> throwOtherError $ "missing accept data " ++ BC.unpack (showRefDigest dgst)
finalizedChannel :: Peer -> Channel -> UnifiedIdentity -> STM ()
finalizedChannel peer@Peer {..} ch self = do
setPeerChannel peer $ ChannelEstablished ch
-- Identity update
writeTQueue (serverIOActions peerServer_) $ liftIO $ atomically $ do
let selfRef = refDigest $ storedRef $ idExtData $ self
updateRefs = selfRef : map (refDigest . storedRef) (idUpdates self)
ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- updateRefs ]
sendToPeerS peer ackedBy $ flip TransportPacket [] $ TransportHeader $ map AnnounceUpdate updateRefs
-- Notify services about new peer
readTVar peerIdentityVar >>= \case
PeerIdentityFull _ -> notifyServicesOfPeer True peer
_ -> return ()
handleIdentityAnnounce :: UnifiedIdentity -> Peer -> Ref -> WaitingRefCallback
handleIdentityAnnounce self peer ref = liftIO $ atomically $ do
let validateAndUpdate upds act = case validateIdentity $ wrappedLoad ref of
Just pid' -> do
let pid = fromMaybe pid' $ toUnifiedIdentity (updateIdentity upds pid')
writeTVar (peerIdentityVar peer) $ PeerIdentityFull pid
writeTChan (serverChanPeer $ peerServer peer) peer
act pid
writeTQueue (serverIOActions $ peerServer peer) $ do
setupChannel self peer pid
Nothing -> return ()
readTVar (peerIdentityVar peer) >>= \case
PeerIdentityRef wref wact
| wrDigest wref == refDigest ref
-> validateAndUpdate [] $ \pid -> do
mapM_ (writeTQueue (serverIOActions $ peerServer peer) . ($ pid)) .
reverse =<< readTVar wact
PeerIdentityFull pid
| idData pid `precedes` wrappedLoad ref
-> validateAndUpdate (idUpdates pid) $ \_ -> do
notifyServicesOfPeer False peer
_ -> return ()
handleIdentityUpdate :: Peer -> Ref -> WaitingRefCallback
handleIdentityUpdate peer ref = liftIO $ atomically $ do
pidentity <- readTVar (peerIdentityVar peer)
if | PeerIdentityFull pid <- pidentity
, Just pid' <- toUnifiedIdentity $ updateIdentity [wrappedLoad ref] pid
-> do
writeTVar (peerIdentityVar peer) $ PeerIdentityFull pid'
writeTChan (serverChanPeer $ peerServer peer) peer
when (pid /= pid') $ do
notifyServicesOfPeer False peer
| otherwise -> return ()
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 )) [] paddr peer $
if new then serviceNewPeer else serviceUpdatedPeer
receivedFromCustomAddress :: PeerAddressType addr => Server -> addr -> ByteString -> IO ()
receivedFromCustomAddress Server {..} addr msg = do
writeFlowIO serverRawPath ( CustomPeerAddress addr, msg )
mkPeer :: Server -> PeerAddress -> IO Peer
mkPeer peerServer_ peerAddress = do
peerState <- newTVarIO (PeerInit [])
peerIdentityVar <- newTVarIO . PeerIdentityUnknown =<< newTVarIO []
peerStorage_ <- deriveEphemeralStorage $ serverStorage peerServer_
peerInStorage <- derivePartialStorage peerStorage_
peerServiceState <- newTMVarIO M.empty
peerWaitingRefs <- newTMVarIO []
return Peer {..}
serverPeer :: Server -> SockAddr -> IO Peer
serverPeer server paddr = do
let paddr' = case IP.fromSockAddr paddr of
Just (IP.IPv4 ipv4, port)
-> IP.toSockAddr (IP.IPv6 $ IP.toIPv6w (0, 0, 0xffff, IP.fromIPv4w ipv4), port)
_ -> paddr
serverPeer' server (DatagramAddress paddr')
serverPeerCustom :: PeerAddressType addr => Server -> addr -> IO Peer
serverPeerCustom server addr = serverPeer' server (CustomPeerAddress addr)
serverPeer' :: Server -> PeerAddress -> IO Peer
serverPeer' server paddr = do
(peer, hello) <- modifyMVar (serverPeers server) $ \pvalue -> do
case M.lookup paddr pvalue of
Just peer -> return (pvalue, (peer, False))
Nothing -> do
peer <- mkPeer server paddr
return (M.insert paddr peer pvalue, (peer, True))
when hello $ atomically $ 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
atomically $ do
readTVar (peerState peer) >>= \case
PeerConnected conn -> connClose conn
_ -> return()
writeTVar (peerState peer) PeerDropped
return $ M.delete (peerAddress peer) pvalue
isPeerDropped :: MonadIO m => Peer -> m Bool
isPeerDropped peer = liftIO $ atomically $ readTVar (peerState peer) >>= \case
PeerDropped -> return True
_ -> return False
sendToPeer :: (Service s, MonadIO m) => Peer -> s -> m ()
sendToPeer peer = sendManyToPeer peer . (: [])
sendManyToPeer :: (Service s, MonadIO m) => Peer -> [ s ] -> m ()
sendManyToPeer peer = sendToPeerList peer . map (\part -> ServiceReply (Left part) True)
sendToPeerStored :: (Service s, MonadIO m) => Peer -> Stored s -> m ()
sendToPeerStored peer = sendManyToPeerStored peer . (: [])
sendManyToPeerStored :: (Service s, MonadIO m) => Peer -> [ Stored s ] -> m ()
sendManyToPeerStored peer = sendToPeerList peer . map (\part -> ServiceReply (Right part) True)
sendToPeerList :: (Service s, MonadIO m) => Peer -> [ ServiceReply s ] -> m ()
sendToPeerList peer parts = do
let st = peerStorage peer
res <- runExceptT $ do
srefs <- liftIO $ fmap catMaybes $ forM parts $ \case
ServiceReply (Left x) use -> Just . (,use) <$> store st x
ServiceReply (Right sx) use -> return $ Just (storedRef sx, use)
_ -> return Nothing
streamHeaders <- concat <$> do
(liftEither =<<) $ liftIO $ atomically $ runExceptT $ do
forM parts $ \case
ServiceOpenStream cb -> do
conn <- lift (readTVar (peerState peer)) >>= \case
PeerConnected conn -> return conn
_ -> throwError "can't open stream without established connection"
(hdr, writer, handler) <- liftEither =<< lift (connAddWriteStream conn)
lift $ writeTQueue (serverIOActions (peerServer peer)) $ do
liftIO $ forkServerThread (peerServer peer) handler
return [ ( hdr, cb writer ) ]
_ -> return []
liftIO $ sequence_ $ map snd streamHeaders
liftIO $ forM_ parts $ \case
ServiceFinally act -> act
_ -> return ()
let dgsts = map (refDigest . fst) srefs
let content = map fst $ filter (\(ref, use) -> use && BL.length (lazyLoadBytes ref) < 500) srefs -- TODO: MTU
header = TransportHeader $ concat
[ [ ServiceType (serviceID $ head parts) ]
, map ServiceRef dgsts
, map fst streamHeaders
]
packet = TransportPacket header content
ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- dgsts ]
liftIO $ atomically $ sendToPeerS peer ackedBy packet
case res of
Right () -> return ()
Left err -> liftIO $ atomically $ writeTQueue (serverErrorLog $ peerServer peer) $
"failed to send packet to " <> show (peerAddress peer) <> ": " <> err
sendToPeerS' :: SecurityRequirement -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()
sendToPeerS' secure Peer {..} ackedBy packet = do
readTVar peerState >>= \case
PeerInit xs -> writeTVar peerState $ PeerInit $ (secure, packet, ackedBy) : xs
PeerConnected conn -> writeFlow (connData conn) (secure, packet, ackedBy)
PeerDropped -> return ()
sendToPeerS :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()
sendToPeerS = sendToPeerS' EncryptedOnly
sendToPeerPlain :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()
sendToPeerPlain = sendToPeerS' PlaintextAllowed
sendToPeerWith :: forall s m e. (Service s, MonadIO m, MonadError e m, FromErebosError e) => Peer -> (ServiceState s -> ExceptT ErebosError IO (Maybe s, ServiceState s)) -> m ()
sendToPeerWith peer fobj = do
let sproxy = Proxy @s
sid = serviceID sproxy
res <- liftIO $ do
svcs <- atomically $ takeTMVar (peerServiceState peer)
(svcs', res) <- runExceptT (fobj $ fromMaybe (emptyServiceState sproxy) $ fromServiceState sproxy =<< M.lookup sid svcs) >>= \case
Right (obj, s') -> return $ (M.insert sid (SomeServiceState sproxy s') svcs, Right obj)
Left err -> return $ (svcs, Left err)
atomically $ putTMVar (peerServiceState peer) svcs'
return res
case res of
Right (Just obj) -> sendToPeer peer obj
Right Nothing -> return ()
Left err -> throwError $ fromErebosError err
lookupService :: forall s proxy. Service s => proxy s -> [SomeService] -> Maybe (SomeService, ServiceAttributes s)
lookupService proxy (service@(SomeService (_ :: Proxy t) attr) : rest)
| Just (Refl :: s :~: t) <- eqT = Just (service, attr)
| otherwise = lookupService proxy rest
lookupService _ [] = Nothing
runPeerService :: forall s m. (Service s, MonadIO m) => Peer -> ServiceHandler s () -> m ()
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 ] -> PeerAddress -> Peer -> ServiceHandler s () -> m ()
runPeerServiceOn mbservice newStreams paddr peer handler = liftIO $ do
let server = peerServer peer
proxy = Proxy @s
svc = serviceID proxy
logd = writeTQueue (serverErrorLog server)
case mbservice `mplus` lookupService proxy (serverServices server) of
Just (service, attr) ->
atomically (readTVar (peerIdentityVar peer)) >>= \case
PeerIdentityFull peerId -> do
(global, svcs) <- atomically $ (,)
<$> takeTMVar (serverServiceStates server)
<*> takeTMVar (peerServiceState peer)
case (fromMaybe (someServiceEmptyState service) $ M.lookup svc svcs,
fromMaybe (someServiceEmptyGlobalState service) $ M.lookup svc global) of
((SomeServiceState (_ :: Proxy ps) ps),
(SomeServiceGlobalState (_ :: Proxy gs) gs)) -> do
Just (Refl :: s :~: ps) <- return $ eqT
Just (Refl :: s :~: gs) <- return $ eqT
let inp = ServiceInput
{ svcAttributes = attr
, svcPeer = peer
, svcPeerAddress = paddr
, svcPeerIdentity = peerId
, svcServer = server
, svcPrintOp = atomically . logd
, svcNewStreams = newStreams
}
reloadHead (serverOrigHead server) >>= \case
Nothing -> atomically $ do
logd $ "current head deleted"
putTMVar (peerServiceState peer) svcs
putTMVar (serverServiceStates server) global
Just h -> do
(rsp, (s', gs')) <- runServiceHandler h inp ps gs handler
moveKeys (peerStorage peer) (serverStorage server)
when (not (null rsp)) $ do
sendToPeerList peer rsp
atomically $ 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 paddr
_ -> atomically $ do
logd $ "unhandled service '" ++ show (toUUID svc) ++ "'"
modifyServiceGlobalState
:: forall s a m e proxy. (Service s, MonadIO m, MonadError e m, FromErebosError e)
=> Server -> proxy s
-> (ServiceGlobalState s -> ( ServiceGlobalState s, a ))
-> m a
modifyServiceGlobalState server proxy f = do
let svc = serviceID proxy
case lookupService proxy (serverServices server) of
Just ( service, _ ) -> do
liftIO $ atomically $ do
global <- takeTMVar (serverServiceStates server)
( global', res ) <- case fromMaybe (someServiceEmptyGlobalState service) $ M.lookup svc global of
SomeServiceGlobalState (_ :: Proxy gs) gs -> do
(Refl :: s :~: gs) <- return $ fromMaybe (error "service ID mismatch in global map") eqT
let ( gs', res ) = f gs
return ( M.insert svc (SomeServiceGlobalState (Proxy @s) gs') global, res )
putTMVar (serverServiceStates server) global'
return res
Nothing -> do
throwOtherError $ "unhandled service '" ++ show (toUUID svc) ++ "'"
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 ()
joinMulticast :: Socket -> IO [ Word32 ]
joinMulticast sock =
withFdSocket sock $ \fd ->
alloca $ \pcount -> do
ptr <- cJoinMulticast fd pcount
if ptr == nullPtr
then do
return []
else do
count <- fromIntegral <$> peek pcount
res <- forM [ 0 .. count - 1 ] $ \i ->
peekElemOff ptr i
cFree ptr
return res
getServerAddresses :: Server -> IO [ SockAddr ]
getServerAddresses Server {..} = do
alloca $ \pcount -> do
ptr <- cLocalAddresses pcount
if ptr == nullPtr
then do
return []
else do
count <- fromIntegral <$> peek pcount
res <- peekArray count ptr
cFree ptr
return $ map (inetToSockAddr . (, serverPort serverOptions )) res
getBroadcastAddresses :: PortNumber -> IO [SockAddr]
getBroadcastAddresses port = do
ptr <- cBroadcastAddresses
let parse i = do
w <- peekElemOff ptr i
if w == 0 then return []
else (SockAddrInet port w:) <$> parse (i + 1)
if ptr == nullPtr
then return []
else do
addrs <- parse 0
cFree ptr
return addrs