From 5e4c7ca992f899092880ea37fabbf07f5d606b07 Mon Sep 17 00:00:00 2001 From: nek0 Date: Thu, 31 Oct 2024 19:19:13 +0100 Subject: [PATCH] attempting to fix communication again --- src-client/Client/Communication.hs | 7 ++-- src-client/Client/Game.hs | 2 -- src-server/Server/Communication.hs | 24 ++++++++++--- src-server/Server/Communication/Handler.hs | 17 ++++----- src-server/Server/Communication/Receive.hs | 13 ++++--- src-server/Server/Communication/Send.hs | 40 +++++++++++++++------- src-server/Server/Types.hs | 2 ++ wizard-wipeout.cabal | 2 ++ 8 files changed, 69 insertions(+), 38 deletions(-) diff --git a/src-client/Client/Communication.hs b/src-client/Client/Communication.hs index 73ea769..3503f2e 100644 --- a/src-client/Client/Communication.hs +++ b/src-client/Client/Communication.hs @@ -35,7 +35,7 @@ connectSocket :: FilePath -> IO Socket connectSocket path = do - sock <- socket AF_UNIX Stream defaultProtocol + sock <- socket AF_UNIX Stream 0 setSocketOption sock KeepAlive 1 connect sock (SockAddrUnix path) pure sock @@ -122,7 +122,10 @@ receiveMessage sock queue st = do let mJsonMsg = A.decode' msg maybe (putStrLn $ "received garbled data: " <> B8.unpack (B8.fromStrict rawMsg)) - (STM.atomically . STM.writeTQueue queue) + (\jsonMsg -> do + print jsonMsg + STM.atomically $ STM.writeTQueue queue jsonMsg + ) mJsonMsg ) msgs diff --git a/src-client/Client/Game.hs b/src-client/Client/Game.hs index d043041..65d536e 100644 --- a/src-client/Client/Game.hs +++ b/src-client/Client/Game.hs @@ -2,8 +2,6 @@ module Client.Game where import Control.Concurrent.STM (atomically, readTMVar) -import Control.Monad.IO.Class (liftIO) - import Control.Monad.Loops import Control.Monad.RWS diff --git a/src-server/Server/Communication.hs b/src-server/Server/Communication.hs index 9879ae3..d741356 100644 --- a/src-server/Server/Communication.hs +++ b/src-server/Server/Communication.hs @@ -39,7 +39,7 @@ bindSocket path = do unless (isSupportedSockAddr sockAddr) (error $ "invalid socket path " <> path) -- removeIfExists path - sock <- socket AF_UNIX Stream defaultProtocol + sock <- socket AF_UNIX Stream 0 bind sock sockAddr Net.listen sock 5 pure sock @@ -77,7 +77,7 @@ disconnectClients clientList clients = do maybe (pure ()) (\uuid -> do - sendMessage ServerQuit uuid clientList + queueMessage ServerQuit uuid clientList dropClient clientList (ccSocket client) ) (ccUUID client) @@ -101,11 +101,25 @@ processRequests = do Left (_ :: SomeException) -> putStrLn "Main socket vanished!" Right (clientSock, _) -> do - clientThreadId <- liftIO $ forkIO $ forever $ do - receiveMessage clientSock queue socketList + messageQueue <- STM.newTQueueIO + sockCOntainer <- STM.newTMVarIO clientSock + clientWriteThreadId <- liftIO $ forkIO $ forever $ do + receiveMessage sockCOntainer queue socketList + clientListenThreadId <- liftIO $ forkIO $ forever $ do + sendMessageQueue sockCOntainer messageQueue + print clientSock liftIO $ STM.atomically $ do list <- STM.takeTMVar socketList - STM.putTMVar socketList (ClientComms Nothing clientSock clientThreadId : list) + STM.putTMVar + socketList + (ClientComms + Nothing + clientSock + messageQueue + clientWriteThreadId + clientListenThreadId + : list + ) putStrLn "accepted new connection" abortCondition <- serverStop <$> STM.atomically (STM.readTMVar st) unless abortCondition $ diff --git a/src-server/Server/Communication/Handler.hs b/src-server/Server/Communication/Handler.hs index 2f86775..7df8e03 100644 --- a/src-server/Server/Communication/Handler.hs +++ b/src-server/Server/Communication/Handler.hs @@ -37,7 +37,7 @@ handleMessage stateContainer readerContainer msg = do let clientIdx = findIndex (isNothing . ccUUID) clients clientSock = ccSocket $ clients !! fromJust clientIdx newClients = map - (\old@(ClientComms mUUID oldClientSock _) -> + (\old@(ClientComms mUUID oldClientSock _ _ _) -> if oldClientSock == clientSock && isNothing mUUID then old @@ -49,7 +49,7 @@ handleMessage stateContainer readerContainer msg = do clients void $ liftIO $ STM.atomically $ STM.swapTMVar clientList newClients putStrLn $ "Accepted Client with UUID " <> show clientId - sendMessage (AcceptClient clientId) clientId clientList + queueMessage (AcceptClient clientId) clientId clientList ClientMessage clientId payload -> case payload of ClientQuit -> do @@ -67,7 +67,7 @@ handleMessage stateContainer readerContainer msg = do currentPlayers <- STM.readTMVar (scPlayers stateContainer) void $ STM.swapTMVar (scPlayers stateContainer) $ Player clientId freshWizard False (now, uuid) : currentPlayers - sendMessage (ProvideInitialWizard freshWizard) clientId clientList + queueMessage (ProvideInitialWizard freshWizard) clientId clientList ClientReady -> do putStrLn $ "client " <> show clientId <> " is ready!" now <- getCurrentTime @@ -118,14 +118,9 @@ handleMessages = do serverState <- get readerContainer <- ask liftIO $ do - msgs <- STM.atomically $ do - emptyState <- STM.isEmptyTQueue queue - if emptyState - then - pure [] - else - STM.flushTQueue queue - void $ do + msgs <- STM.atomically $ + STM.flushTQueue queue + do mapM_ (handleMessage serverState readerContainer) msgs diff --git a/src-server/Server/Communication/Receive.hs b/src-server/Server/Communication/Receive.hs index 5813b3d..283d7be 100644 --- a/src-server/Server/Communication/Receive.hs +++ b/src-server/Server/Communication/Receive.hs @@ -11,10 +11,12 @@ import qualified Data.Aeson as A import qualified Data.ByteString as B import qualified Data.ByteString.Lazy.Char8 as B8 -import Foreign.Marshal +import Foreign.Marshal hiding (void) import Network.Socket +-- internal imports + import Library.Types import Server.Communication.Send @@ -22,19 +24,20 @@ import Server.Types -- | receive a 'ClientMessage' receiveMessage - :: Socket + :: STM.TMVar Socket -> STM.TQueue ClientMessage -> STM.TMVar [ClientComms] -> IO () -receiveMessage sock queue clientList = do +receiveMessage sockContainer queue clientList = do + sock <- STM.atomically $ STM.readTMVar sockContainer let maxBufferLength = 4096 mMsg <- do ptr <- mallocArray maxBufferLength eBufferLength <- try $ recvBuf sock ptr maxBufferLength bufferLength <- case eBufferLength of - Left (_ :: IOException) -> do - putStrLn "Socket vanished, cleaning up…" + Left (e :: IOException) -> do + putStrLn ("Socket vanished, cleaning up after " <> show e) dropClient clientList sock pure 0 Right len -> pure len diff --git a/src-server/Server/Communication/Send.hs b/src-server/Server/Communication/Send.hs index 4f3789f..bce6f50 100644 --- a/src-server/Server/Communication/Send.hs +++ b/src-server/Server/Communication/Send.hs @@ -42,32 +42,46 @@ import Library.Types import Server.Types -- | Sends a specified message through given socket to the client -sendMessage +queueMessage :: ServerMessage -> UUID -> STM.TMVar [ClientComms] -> IO () -sendMessage msg uuid clientList = do +queueMessage msg uuid clientList = do clients <- STM.atomically $ STM.readTMVar clientList - let mSock = ccSocket <$> find (\client -> ccUUID client == Just uuid) clients - msgJson = A.encode msg - msgVector = VS.fromList $ B.unpack $ B.toStrict ('<' `B8.cons` (msgJson `B8.snoc` '>')) - -- putStrLn $ "Sending: " <> B8.unpack msgJson + let mQueue = ccQueue <$> find (\client -> ccUUID client == Just uuid) clients maybe (putStrLn $ "unknown client UUID: " <> show uuid) - (\sock -> + (\queue -> + STM.atomically $ STM.writeTQueue queue msg + ) + mQueue + +sendMessageQueue + :: STM.TMVar Socket + -> STM.TQueue ServerMessage + -> IO () +sendMessageQueue sockContainer queue = do + sock <- STM.atomically $ STM.takeTMVar sockContainer + msgs <- STM.atomically $ STM.flushTQueue queue + mapM_ + (\msg -> do + print msg + let msgJson = A.encode msg + msgVector = VS.fromList $ B.unpack $ B.toStrict ('<' `B8.cons` (msgJson `B8.snoc` '>')) VS.unsafeWith msgVector (\ptr -> do eResult <- try $ void $ sendBuf sock ptr (VS.length msgVector) case eResult of - Left (_ :: IOException) -> - putStrLn $ "can't reach client " <> show uuid + Left (e :: IOException) -> + putStrLn $ "can't reach client after " <> show e Right _ -> pure () ) - ) - mSock + ) + msgs + STM.atomically $ STM.putTMVar sockContainer sock sendPings :: Game () sendPings = do @@ -91,7 +105,7 @@ sendPings = do else do random <- liftIO nextRandom let newPong = (lastPongTime, random) - liftIO $ sendMessage + liftIO $ queueMessage ( Ping random ) plId @@ -165,4 +179,4 @@ sendUpdate stateContainer tileMap player = do sendSlice slice (Player playerId wizard _ _) = do let msg = TickUpdate slice wizard print slice - liftIO $ sendMessage msg playerId (scClientSockets stateContainer) + liftIO $ queueMessage msg playerId (scClientSockets stateContainer) diff --git a/src-server/Server/Types.hs b/src-server/Server/Types.hs index 6cd9dee..6a35840 100644 --- a/src-server/Server/Types.hs +++ b/src-server/Server/Types.hs @@ -70,6 +70,8 @@ data StateContainer = StateContainer data ClientComms = ClientComms { ccUUID :: Maybe UUID , ccSocket :: Socket + , ccQueue :: STM.TQueue ServerMessage + , ccWriter :: ThreadId , ccListener :: ThreadId } deriving (Eq) diff --git a/wizard-wipeout.cabal b/wizard-wipeout.cabal index 393544f..9f8cfa4 100644 --- a/wizard-wipeout.cabal +++ b/wizard-wipeout.cabal @@ -15,6 +15,8 @@ extra-doc-files: CHANGELOG.md common warnings ghc-options: -Wall -threaded -rtsopts + -fwrite-ide-info + -hiedir=.hie library import: warnings