diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for alpaca-netcode + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/alpaca-netcode.cabal b/alpaca-netcode.cabal new file mode 100644 --- /dev/null +++ b/alpaca-netcode.cabal @@ -0,0 +1,75 @@ +cabal-version: 1.12 +name: alpaca-netcode +version: 0.1.0.0 +synopsis: Rollback/replay NetCode for realtime, deterministic, multiplayer games. +description: + A rollback/replay client-server system for realtime multiplayer games. The API + only requires you to express your game as a pure, deterministic function. + . + See "Alpaca.NetCode" to get started. + . + === Advantages + * Simple code. Your game logic contains no NetCode. + * Low bandwidth. Only inputs are shared. + * Zero latency. Player's own inputs affect their game immediatly. + * UDP based. Unordered and using redundancy to mitigate packet loss. + * Lightweight server. The server does not run the game logic, it only relays and tracks user inputs. + * Cheating. Only inputs are shared which eliminates a whole class state manipulation cheats. + . + === Disadvantages + * Increased CPU usage. Rollback/replay means that clients must run the game step function multiple times per frame. + * Not suitable for large numbers of players. Tens of players is likey reasonable. + . + === Disclaimer + This is an initial release with minimal functionality and still very + experimental. Use at your own risk. + +-- bug-reports: +license: Apache-2.0 +license-file: LICENSE +author: David Eichmann +maintainer: davide@well-typed.com +copyright: 2021 David Eichmann +category: Network, Game Engine +build-type: Simple +extra-source-files: CHANGELOG.md + +source-repository head + type: git + location: https://github.com/DavidEichmann/alpaca-netcode + +library + other-modules: + Alpaca.NetCode.Internal.Client + , Alpaca.NetCode.Internal.ClockSync + , Alpaca.NetCode.Internal.Common + , Alpaca.NetCode.Internal.Server + exposed-modules: + Alpaca.NetCode + , Alpaca.NetCode.Advanced + build-depends: base >= 4.9 && < 4.16 + , bytestring + , containers + , flat + , hashable + , network >= 3.1 && < 3.2 + , network-run + , random + , stm + , time + + hs-source-dirs: src + ghc-options: -Wall + default-language: Haskell2010 + +test-suite test + Type: exitcode-stdio-1.0 + hs-source-dirs: test + Main-is: Test.hs + Build-depends: base >= 4.9 && < 4.16 + , alpaca-netcode + , containers + , random + , tasty + , tasty-hunit + default-language: Haskell2010 diff --git a/src/Alpaca/NetCode.hs b/src/Alpaca/NetCode.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode.hs @@ -0,0 +1,165 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +{-# OPTIONS_HADDOCK not-home #-} + +-- | This module should be all you need to get started writing multiplayer +-- games. See "Alpaca.NetCode.Advanced" for more advanced usage. +module Alpaca.NetCode + ( runServer, + runClient, + Client, + clientPlayerId, + clientSample, + clientSetInput, + clientStop, + -- ** Types + Tick (..), + PlayerId (..), + HostName, + ServiceName, + ) where + +import Alpaca.NetCode.Advanced +import qualified Data.Map as M +import Flat ( + Flat, + ) + +-- | Start a client. This blocks until the initial handshake with the server is +-- finished. You must call 'clientSetInput' on the returned client to submit new +-- inputs. +-- +-- Think of @world@ as shared state between all clients. Alpaca NetCode takes +-- care of synchronizing and predicting the @world@ state across all clients. +-- Additionally, clock synchronization is done with the server and the "current" +-- tick is decided for you when sampling with `clientSample`. +-- +-- Typical usage looks like this: +-- +-- @ +-- main :: IO () +-- main = do +-- myClient <- runClient "localhost" "8111" 30 myInput0 myWorld0 worldStep +-- let myPlayerId = clientPlayerId myClient +-- +-- mainGameLoop $ do +-- myInput <- pollInput -- Poll inputs from some other library +-- clientSetInput myClient -- Push inputs to Alpaca NetCode +-- world <- clientSample -- Sample the current (predicted) world +-- renderWorld myPlayerId world -- Render the world +-- +-- -- You're free to do IO and maintain state local to the client. +-- +-- return (gameIsOver world) -- Return True to keep looping +-- +-- clientStop myClient +-- +-- -- Given +-- data World = World { .. } +-- data Input = Input { .. } deriving (Generic, Eq, Flat) +-- myWorld0 :: World +-- gameIsOver :: World -> Bool +-- myInput0 :: Input +-- worldStep :: Map PlayerId Input -> Tick -> World -> World +-- renderWorld :: PlayerId -> World -> IO () +-- pollInput :: IO Input +-- mainGameLoop :: IO Bool -> IO () +-- @ +runClient :: + forall world input. + Flat input => + -- | The server's host name or IP address e.g. @"localhost"@. + HostName -> + -- | The server's port number e.g. @"8111"@. + ServiceName -> + -- | Tick rate (ticks per second). Typically @30@ or @60@. Must be the same + -- across all clients and the server. Packet rate and hence network bandwidth + -- will scale linearly with the tick rate. + Int -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. + -- + -- Note that the client and server do input "prediction" by assuming @input@s + -- do not change. It is important to design your @input@ type accordingly. For + -- example, Do NOT store a @Bool@ indicating that a button has been clicked. + -- Instead, store a @Bool@ indicating if that button is currently held down. + -- Then, store enough information in the @world@ state to identify a click. + input -> + -- | Initial world state. Must be the same across all clients. + world -> + -- | A deterministic stepping function (for a single tick). In practice you + -- can choose to use whatever monad stack within as long as you (un)wrap into + -- a pure function e.g. you can use `ST` as long as you wrap it in `runST`. + -- Must be the same across all clients and the server. Takes: + -- + -- * a map from PlayerId to current input. You can use the key set as the set + -- of all connected players. + -- * current game tick. + -- * previous tick's world state. + -- + -- It is important that this is deterministic else clients' states will + -- diverge. Beware of floating point non-determinism! + ( M.Map PlayerId input -> + Tick -> + world -> + world + ) -> + IO (Client world input) +runClient + serverHostName + serverPort + tickRate + input0 + world0 + stepOneTick + = runClientWith + serverHostName + serverPort + Nothing + (defaultClientConfig tickRate) + input0 + world0 + stepOneTick + +-- | Run a server for a single game. This will block until the game ends, +-- specifically when all players have disconnected. +runServer :: + forall input. + (Eq input, Flat input) => + -- | The server's port number e.g. @"8111"@. + ServiceName -> + -- | Tick rate (ticks per second). Typically @30@ or @60@. Must be the same + -- across all clients and the server. Packet rate and hence network bandwidth + -- will scale linearly with the tick rate. + Int -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. + input -> + IO () +runServer + serverPort + tickRate + input0 + = runServerWith + serverPort + Nothing + (defaultServerConfig tickRate) + input0 diff --git a/src/Alpaca/NetCode/Advanced.hs b/src/Alpaca/NetCode/Advanced.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode/Advanced.hs @@ -0,0 +1,243 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Rollback and replay based game networking +module Alpaca.NetCode.Advanced + ( -- * Server + runServerWith, + module Alpaca.NetCode.Internal.Server, + -- * Client + runClientWith, + module Alpaca.NetCode.Internal.Client, + -- * Common Types + SimNetConditions (..), + Tick (..), + PlayerId (..), + NetMsg, + HostName, + ServiceName, + ) where + +import Alpaca.NetCode.Internal.Common +import Alpaca.NetCode.Internal.Client +import Alpaca.NetCode.Internal.Server +import Control.Concurrent ( + Chan, + forkIO, + newChan, + readChan, + writeChan, + ) +import qualified Control.Exception as E +import Control.Monad +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as BSL +import qualified Data.Map as M +import Flat ( + DecodeException (BadEncoding), + Flat, + flat, + unflat, + ) +import Network.Run.UDP (runUDPServer) +import Network.Socket ( + AddrInfo ( + addrAddress, + addrFamily, + addrFlags, + addrProtocol, + addrSocketType + ), + AddrInfoFlag (AI_PASSIVE), + HostName, + ServiceName, + SockAddr, + Socket, + SocketType (Datagram), + close, + connect, + defaultHints, + getAddrInfo, + socket, + withSocketsDo, + ) +import qualified Network.Socket.ByteString as NBS + +-- | Start a client. This blocks until the initial handshake with the server is +-- finished. +runClientWith :: + forall world input. + Flat input => + -- | The server's host name or IP address e.g. @"localhost"@. + HostName -> + -- | The server's port number e.g. @"8111"@. + ServiceName -> + -- | Optional simulation of network conditions. In production this should be + -- `Nothing`. May differ between clients. + Maybe SimNetConditions -> + -- | The 'defaultClientConfig' works well for most cases. + ClientConfig -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. See 'Alpaca.NetCode.runClient'. + input -> + -- | Initial world state. Must be the same across all clients. + world -> + -- | A deterministic stepping function (for a single tick). Must be the same + -- across all clients and the server. See 'Alpaca.NetCode.runClient'. + ( M.Map PlayerId input -> + Tick -> + world -> + world + ) -> + IO (Client world input) +runClientWith + serverHostName + serverPort + simNetConditionsMay + clientConfig + input0 + world0 + stepOneTick = do + sendChan <- newChan + recvChan <- newChan + + -- UDP + _ <- forkIO $ do + runUDPClient' serverHostName serverPort $ \sock server -> do + _ <- + forkIO $ + writeDatagramContentsAsNetMsg (Just server) fst recvChan sock + forever $ do + msg <- readChan sendChan + NBS.sendAllTo sock (flat msg) server + + runClientWith' + (writeChan sendChan) + (readChan recvChan) + simNetConditionsMay + clientConfig + input0 + world0 + stepOneTick + where + -- + -- Coppied from network-run + -- + + runUDPClient' :: + HostName -> ServiceName -> (Socket -> SockAddr -> IO a) -> IO a + runUDPClient' host port client = withSocketsDo $ do + addr <- resolve Datagram (Just host) port False + let sockAddr = addrAddress addr + E.bracket (openSocket addr) close $ \sock -> client sock sockAddr + + resolve :: SocketType -> Maybe HostName -> ServiceName -> Bool -> IO AddrInfo + resolve socketType mhost port passive = + head + <$> getAddrInfo (Just hints) mhost (Just port) + where + hints = + defaultHints + { addrSocketType = socketType + , addrFlags = if passive then [AI_PASSIVE] else [] + } + + openSocket :: AddrInfo -> IO Socket + openSocket addr = do + sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) + connect sock (addrAddress addr) + return sock + +-- | Run a server for a single game. This will block until the game ends, +-- specifically when all players have disconnected. +runServerWith :: + forall input. + (Eq input, Flat input) => + -- | The server's port number e.g. @"8111"@. + ServiceName -> + -- | Optional simulation of network conditions. In production this should be + -- `Nothing`. + Maybe SimNetConditions -> + -- | The 'defaultServerConfig' works well for most cases. + ServerConfig -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. + input -> + IO () +runServerWith serverPort tickRate netConfig input0 = do + sendChan <- newChan + recvChan <- newChan + + -- UDP + _ <- forkIO $ do + runUDPServer Nothing serverPort $ \sock -> do + _ <- forkIO $ writeDatagramContentsAsNetMsg Nothing id recvChan sock + forever $ do + (msg, addr) <- readChan sendChan + NBS.sendAllTo sock (flat msg) addr + + runServerWith' + (curry (writeChan sendChan)) + (readChan recvChan) + tickRate + netConfig + input0 + +-- Forever decode messages from the input socket using the given decoding +-- function and writing it to the given chan. Loops forever. +writeDatagramContentsAsNetMsg :: + forall input a. + (Flat input) => + -- | Just the sender if alwalys receiving from the same address (used in the client case where we only receive from the server) + (Maybe SockAddr) -> + -- | Decode the messages + ((NetMsg input, SockAddr) -> a) -> + -- | Write decoded msgs to this chan + Chan a -> + -- | Read from this socket + Socket -> + IO () +writeDatagramContentsAsNetMsg constSenderMay f chan sock = go + where + go = do + let maxBytes = 4096 + (bs, sender) <- case constSenderMay of + Nothing -> NBS.recvFrom sock maxBytes + Just s -> (,s) <$> NBS.recv sock maxBytes + if BS.length bs == maxBytes + then + error $ + "TODO support packets bigger than " + ++ show maxBytes + ++ " bytes." + else + if BS.length bs == 0 + then debugStrLn "Received 0 bytes from socket. Stopping." + else do + case unflat @(NetMsg input) (BSL.fromStrict bs) of + Left err -> do + debugStrLn $ + "Error decoding message: " ++ case err of + BadEncoding env errStr -> + "BadEncoding " ++ show env ++ "\n" ++ errStr + _ -> show err + Right msg -> writeChan chan (f (msg, sender)) + go diff --git a/src/Alpaca/NetCode/Internal/Client.hs b/src/Alpaca/NetCode/Internal/Client.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode/Internal/Client.hs @@ -0,0 +1,443 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Rollback and replay based game networking +module Alpaca.NetCode.Internal.Client ( + runClientWith', + ClientConfig (..), + defaultClientConfig, + Client, + clientPlayerId, + clientSample, + clientSample', + clientSetInput, + clientStop, +) where + +import Alpaca.NetCode.Internal.ClockSync +import Alpaca.NetCode.Internal.Common +import Control.Concurrent (forkIO, killThread, threadDelay) +import Control.Concurrent.STM as STM +import Control.Monad +import Data.Int (Int64) +import Data.IntMap (IntMap) +import qualified Data.IntMap as IM +import qualified Data.Map as M +import Data.Maybe (catMaybes, fromMaybe, isJust) +import qualified Data.Set as S +import Flat + + +-- | A Client. You'll generally obtain this via 'Alpaca.NetCode.runClient'. +data Client world input = Client + { -- | The client's @PlayerId@ + clientPlayerId :: PlayerId + , -- | Sample the world state. First, This will estimate the current tick + -- based on ping and clock synchronization with the server. Then, the world + -- state will be rollback and inputs replayed as necessary. This returns: + -- + -- * New authoritative world states in chronological order since the last + -- sample time. These world states are the True world states at each tick. + -- This list will be empty if no new authoritative world states have been + -- derived since that last call to this sample function. Though it's often + -- simpler to just use the predicted world state, you can use these + -- authoritative world states to render output when you're not willing to + -- miss-predict but are willing to have greater latency. If the client has + -- been stopped, this will be an empty list. + -- + -- * The predicted current world state. This extrapolates past the latest + -- know authoritative world state by assuming no user inputs have changed + -- (unless otherwise known e.g. our own player's inputs are known). If the + -- client has been stopped, this will return the last predicted world. + clientSample' :: IO ([world], world) + , -- | Set the client's current input. + clientSetInput :: input -> IO () + , -- | Stop the client. + clientStop :: IO () + } + + +-- | Sample the current world state. +-- +-- . First, This will estimate the current tick based on ping and clock +-- synchronization with the server. Then, this extrapolates past the latest know +-- authoritative world state by assuming no user inputs have changed (unless +-- otherwise known e.g. our own player's inputs are known). If the client has +-- been stopped, this will return the last predicted world. +clientSample :: Client world input -> IO world +clientSample client = snd <$> clientSample' client + + +-- | Configuration options specific to clients. +data ClientConfig = ClientConfig + { -- | Tick rate (ticks per second). Typically @30@ or @60@. Must be the same + -- across all clients and the server. Packet rate and hence network bandwidth + -- will scale linearly with this the tick rate. + ccTickRate :: Int + , -- | Add this constant amount of latency (in seconds) to this client's inputs. + -- A good value is @0.03@ or something between @0@ and @0.1@. May differ + -- between clients. + -- + -- Too high of a value and the player will get annoyed at the extra input + -- latency. On the other hand, a higher value means less miss-predictions of + -- other clients. In the extreme case, set to something higher than ping, + -- there will be no miss predictions: all clients will receive inputs before + -- rendering the corresponding tick. + ccFixedInputLatency :: Float + , -- | Maximum number of ticks to predict when sampling. 'defaultClientConfig' + -- uses @ccTickRate / 2@. If the client is this many ticks behind the current + -- tick, it will simply stop at an earlier tick. You may want to scale this + -- value along with the tick rate. May differ between clients. + ccMaxPredictionTicks :: Int + , -- | If the client's latest known authoritative world is this many ticks + -- behind the current tick, no prediction will be done at all when sampling. + -- 'defaultClientConfig' uses @ccTickRate * 3@. Useful because want to save + -- CPU cycles for catching up with the server. You may want to scale this + -- value along with the tick rate. May differ between clients. + ccResyncThresholdTicks :: Int + , -- | When submitting inputs to the server, we also send a copy of + -- @ccSubmitInputDuplication@ many recently submitted inputs in order to + -- mittigate the effect for dropped packets. 'defaultClientConfig' + -- uses @15@. + ccSubmitInputDuplication :: Int + } deriving (Show, Read, Eq, Ord) + + +-- | Sensible defaults for @ClientConfig@ based on the tick rate. +defaultClientConfig :: + -- | Tick rate (ticks per second). Must be the same across all clients and the + -- server. Packet rate and hence network bandwidth will scale linearly with + -- this the tick rate. + Int -> + ClientConfig +defaultClientConfig tickRate = + ClientConfig + { ccTickRate = tickRate + , ccFixedInputLatency = 0.03 + , ccMaxPredictionTicks = tickRate `div` 2 + , ccResyncThresholdTicks = tickRate * 3 + , ccSubmitInputDuplication = 15 + } + + +-- | Start a client. This blocks until the initial handshake with the +-- server is finished. +runClientWith' :: + forall world input. + Flat input => + -- | Function to send messages to the server. The underlying communication + -- protocol need only guarantee data integrity but is otherwise free to drop + -- and reorder packets. Typically this is backed by a UDP socket. + (NetMsg input -> IO ()) -> + -- | Blocking function to receive messages from the server. Has the same + -- reliability requirements as the send function. + (IO (NetMsg input)) -> + -- | Optional simulation of network conditions. In production this should be + -- `Nothing`. May differ between clients. + Maybe SimNetConditions -> + -- | The 'defaultClientConfig' works well for most cases. + ClientConfig -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. See 'Alpaca.NetCode.runClient'. + input -> + -- | Initial world state. Must be the same across all clients. + world -> + -- | A deterministic stepping function (for a single tick). Must be the same + -- across all clients and the server. See 'Alpaca.NetCode.runClient'. + ( M.Map PlayerId input -> + Tick -> + world -> + world + ) -> + IO (Client world input) +runClientWith' sendToServer' rcvFromServer' simNetConditionsMay clientConfig input0 world0 stepOneTick = playCommon (ccTickRate clientConfig) $ \tickTime getTime _resetTime -> do + (sendToServer, rcvFromServer) <- + simulateNetConditions + sendToServer' + rcvFromServer' + simNetConditionsMay + + -- Authoritative Map from tick and PlayerId to inputs. The inner map is + -- always complete (e.g. if we have the IntMap for tick i, then it contains + -- the inputs for *all* known players) + authInputsTVar :: TVar (IntMap (M.Map PlayerId input)) <- newTVarIO (IM.singleton 0 M.empty) + + -- Tick to authoritative world state. + authWorldsTVar :: TVar (IntMap world) <- newTVarIO (IM.singleton 0 world0) + + -- Max known auth inputs tick without any prior missing ticks. + maxAuthTickTVar :: TVar Tick <- newTVarIO 0 + + -- This client/host's PlayerId. Initially nothing, then set to Just the + -- player ID on connection to the server. This is a constant thereafter. + myPlayerIdTVar <- newTVarIO (Nothing :: Maybe PlayerId) + + -- Non-authoritative Map from tick and PlayerId to inputs. The inner map + -- is NOT always complete (e.g. if we have the IntMap for tick i, then + -- it may or may not yet contain all the inputs for *all* known players). + hintInputsTVar :: TVar (IntMap (M.Map PlayerId input)) <- newTVarIO (IM.singleton 0 M.empty) + + -- Clock Sync + (estimateServerTickPlusLatencyPlusBufferPlus, recordClockSyncSample, clockAnalytics) <- initializeClockSync tickTime getTime + let estimateServerTickPlusLatencyPlusBuffer = estimateServerTickPlusLatencyPlusBufferPlus 0 + + -- Keep trying to connect to the server. + heartbeatTid <- forkIO $ + forever $ do + clientSendTime <- getTime + isConnected <- isJust <$> atomically (readTVar myPlayerIdTVar) + sendToServer ((if isConnected then Msg_Heartbeat else Msg_Connect) clientSendTime) + isClockReady <- isJust <$> clockAnalytics + threadDelay $ + if isClockReady + then 500000 -- 0.5 seconds + else 50000 -- 0.05 seconds + + -- Main message processing loop + msgLoopTid <- forkIO $ + forever $ do + msg <- rcvFromServer + case msg of + Msg_Connect{} -> debugStrLn "Client received unexpected Msg_Connect from the server. Ignoring." + Msg_Connected playerId -> do + join $ + atomically $ do + playerIdMay <- readTVar myPlayerIdTVar + case playerIdMay of + Nothing -> do + writeTVar myPlayerIdTVar (Just playerId) + return (debugStrLn $ "Connected! " ++ show playerId) + Just playerId' -> return $ debugStrLn $ "Got Msg_Connected " ++ show playerId' ++ "but already connected (with " ++ show playerId + Msg_SubmitInput{} -> debugStrLn "Client received unexpected Msg_SubmitInput from the server. Ignoring." + Msg_Ack{} -> + debugStrLn "Client received unexpected Msg_Ack from the server. Ignoring." + Msg_Heartbeat{} -> + debugStrLn "Client received unexpected Msg_Heartbeat from the server. Ignoring." + Msg_HeartbeatResponse clientSendTime serverReceiveTime -> do + -- Record times for ping/clock sync. + clientReceiveTime <- getTime + recordClockSyncSample clientSendTime serverReceiveTime clientReceiveTime + Msg_AuthInput headTick authInputssCompact hintInputssCompact -> do + let authInputss = fromCompactMaps authInputssCompact + let hintInputss = fromCompactMaps hintInputssCompact + resMsgs <- do + -- Update maxAuthTickTVar if needed and send heartbeat + ackMsg <- atomically $ do + maxAuthTick <- readTVar maxAuthTickTVar + let newestTick = headTick + fromIntegral (length authInputss) - 1 + maxAuthTick' = + if headTick <= maxAuthTick + 1 && maxAuthTick < newestTick + then newestTick + else maxAuthTick + writeTVar maxAuthTickTVar maxAuthTick' + return (Msg_Ack maxAuthTick') + sendToServer ackMsg + + -- Save new auth inputs + let newAuthTickHi = headTick + Tick (fromIntegral $ length authInputss) + resMsg <- forM (zip [headTick ..] authInputss) $ \(tick, inputs) -> do + atomically $ do + authInputs <- readTVar authInputsTVar + -- when (tickInt `mod` 100 == 0) (putStrLn $ "Received auth tick: " ++ show tickInt) + case authInputs IM.!? fromIntegral tick of + Just _ -> return $ Just $ "Received a duplicate Msg_AuthInput for " ++ show tick ++ ". Ignoring." + Nothing -> do + -- New auth inputs + writeTVar authInputsTVar (IM.insert (fromIntegral tick) inputs authInputs) + return (Just $ "Got auth-inputs for " ++ show tick) + + -- Save new hint inputs, Excluding my own! + forM_ (zip [succ newAuthTickHi ..] hintInputss) $ \(tick, newHintinputs) -> + atomically $ do + myPlayerIdMay <- readTVar myPlayerIdTVar + modifyTVar hintInputsTVar $ + IM.alter + ( \case + Just oldHintinputs + | Just myPlayerId <- myPlayerIdMay -> + Just (M.restrictKeys oldHintinputs (S.singleton myPlayerId) <> newHintinputs <> oldHintinputs) + _ -> Just newHintinputs + ) + (fromIntegral tick) + + return resMsg + mapM_ debugStrLn (catMaybes resMsgs) + Msg_HintInput tick playerId inputs -> do + res <- atomically $ do + hintInputs <- readTVar hintInputsTVar + let hintInputsAtTick = fromMaybe M.empty (hintInputs IM.!? fromIntegral tick) + writeTVar hintInputsTVar (IM.insert (fromIntegral tick) (M.insert playerId inputs hintInputsAtTick) hintInputs) + return (Just $ "Got hint-inputs for " ++ show tick) + mapM_ debugStrLn res + + -- Wait to be connected. + myPlayerId <- atomically $ do + myPlayerIdMay <- readTVar myPlayerIdTVar + maybe retry return myPlayerIdMay + + -- Recently submitted inputs and their tick in reverse chronological order. + recentSubmittedInputsTVar <- newTVarIO [(Tick 0, input0)] + -- last returned auth world tick (inclusive) from the sampling function + lastSampledAuthWorldTickTVar :: TVar Tick <- newTVarIO 0 + -- last returned predicted world from the sampling function + lastSampledPredictedWorldTVar :: TVar world <- newTVarIO world0 + -- Is the client Stopped? + stoppedTVar :: TVar Bool <- newTVarIO False + + return $ + Client + { clientPlayerId = myPlayerId + , clientSample' = do + stopped <- atomically $ readTVar stoppedTVar + if stopped + then do + lastPredictedWorld <- atomically $ readTVar lastSampledPredictedWorldTVar + return ([], lastPredictedWorld) + else do + -- TODO we're just resimulating from the last snapshot every + -- time. We may be able to reuse past simulation data if + -- snapshot / inputs haven't changed. + + -- Since we are sending inputs for tick + -- estimateServerTickPlusLatencyPlusBuffer and we want to minimize + -- perceived input latency, we should target that same tick + targetTick <- estimateServerTickPlusLatencyPlusBuffer + (inputs, hintInputs, startTickInt, startWorld) <- atomically $ do + (startTickInt, startWorld) <- + fromMaybe (error $ "No authoritative world found <= " ++ show targetTick) -- We have at least the initial world + . IM.lookupLE (fromIntegral targetTick) + <$> readTVar authWorldsTVar + inputs <- readTVar authInputsTVar + hintInputs <- readTVar hintInputsTVar + return (inputs, hintInputs, startTickInt, startWorld) + let startInputs = + fromMaybe + (error $ "Have auth world but no authoritative inputs at " ++ show startTick) -- We assume that we always have auth inputs on ticks where we have auth worlds. + (IM.lookup startTickInt inputs) + startTick = Tick (fromIntegral startTickInt) + + predict :: + Int64 -> -- How many ticks of prediction to allow + Tick -> -- Some tick i + M.Map PlayerId input -> -- inputs at tick i + world -> -- world at tick i if simulated + Bool -> -- Is the world authoritative? + IO world -- world at targetTick (or latest tick if predictionAllowance ran out) + predict predictionAllowance tick tickInputs world isWAuth = case compare tick targetTick of + LT -> do + let tickNext = tick + 1 + + inputsNextAuthMay = inputs IM.!? (fromIntegral tickNext) -- auth input + isInputsNextAuth = isJust inputsNextAuthMay + isWNextAuth = isWAuth && isInputsNextAuth + if isWNextAuth || predictionAllowance > 0 + then do + let inputsNextHintPart = fromMaybe M.empty (hintInputs IM.!? (fromIntegral tickNext)) -- partial hint inputs + inputsNextHintFilled = inputsNextHintPart `M.union` tickInputs -- hint input (filled with previous input) + inputsNext = fromMaybe inputsNextHintFilled inputsNextAuthMay + wNext = stepOneTick inputsNext tickNext world + + pruneOldAuthWorlds = True + -- TODO ^^ in the future we may wan to keep all auth + -- worlds to implement a time traveling debugger + when isWNextAuth $ + atomically $ do + modifyTVar authWorldsTVar (IM.insert (fromIntegral tickNext) wNext) + when pruneOldAuthWorlds $ do + -- We keep all new authworlds as we used them in + -- `newAuthWorlds` and ultimately return them on + -- sample. + lastSampledAuthWorldTick <- readTVar lastSampledAuthWorldTickTVar + modifyTVar authWorldsTVar (snd . IM.split (fromIntegral lastSampledAuthWorldTick)) + + let predictionAllowance' = if isWNextAuth then predictionAllowance else predictionAllowance - 1 + predict predictionAllowance' tickNext inputsNext wNext isWNextAuth + else return world + EQ -> return world + GT -> error "Impossible! simulated past target tick!" + + -- If very behind the server, we want to do 0 prediction + maxAuthTick <- atomically $ readTVar maxAuthTickTVar + let predictionAllowance = + if targetTick - maxAuthTick > Tick (fromIntegral $ ccResyncThresholdTicks clientConfig) + then 0 + else fromIntegral (ccMaxPredictionTicks clientConfig) + + predictedTargetW <- predict predictionAllowance startTick startInputs startWorld True + atomically $ writeTVar lastSampledPredictedWorldTVar predictedTargetW + newAuthWorlds :: [world] <- atomically $ do + lastSampledAuthWorldTick <- readTVar lastSampledAuthWorldTickTVar + authWorlds <- readTVar authWorldsTVar + let latestAuthWorldTick = Tick $ fromIntegral $ fst $ IM.findMax authWorlds + writeTVar lastSampledAuthWorldTickTVar latestAuthWorldTick + return ((authWorlds IM.!) . fromIntegral <$> [lastSampledAuthWorldTick + 1 .. latestAuthWorldTick]) + + return (newAuthWorlds, predictedTargetW) + , clientSetInput = + -- TODO We can send (non-auth) inputs p2p! + + -- TODO this mechanism minimizes latency when `targetTick > lastTick` by + -- sending the input to the server immediately, but when `targetTick <= + -- lastTick`, then the input will be ghosted! + \newInput -> do + stopped <- atomically $ readTVar stoppedTVar + when (not stopped) $ do + -- We submit events as soon as we expect the server to be on a future + -- tick. Else we just store the new input. + targetTick <- estimateServerTickPlusLatencyPlusBufferPlus (ccFixedInputLatency clientConfig) + join $ + atomically $ do + lastTick <- + ( \case + [] -> Tick 0 + (t, _) : _ -> t + ) + <$> readTVar recentSubmittedInputsTVar + if targetTick > lastTick + then do + -- Store our own inputs as a hint so we get 0 latency. This is + -- only a hint and not authoritative as it's still possible that + -- submitted inputs are dropped or rejected by the server. If + -- we've jumped a few ticks forward than we keep we don't attempt + -- to submit inputs to "fill in the gap". We assume constant as + -- the server and other clients predicted those inputs as constant + -- anyway. + modifyTVar hintInputsTVar $ + IM.alter + (Just . M.insert myPlayerId newInput . fromMaybe M.empty) + (fromIntegral targetTick) + + modifyTVar recentSubmittedInputsTVar $ + take (ccSubmitInputDuplication clientConfig) + . ((targetTick, newInput) :) + inputsToSubmit <- readTVar recentSubmittedInputsTVar + return (sendToServer (Msg_SubmitInput inputsToSubmit)) + else pure (return ()) + , clientStop = do + stopped <- atomically (readTVar stoppedTVar) + when (not stopped) $ do + killThread msgLoopTid + killThread heartbeatTid + atomically $ do + writeTVar stoppedTVar True + } diff --git a/src/Alpaca/NetCode/Internal/ClockSync.hs b/src/Alpaca/NetCode/Internal/ClockSync.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode/Internal/ClockSync.hs @@ -0,0 +1,194 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Rollback and replay based game networking +module Alpaca.NetCode.Internal.ClockSync where + +import Alpaca.NetCode.Internal.Common +import Control.Concurrent.STM +import Data.Maybe (fromMaybe) + +import Data.Int (Int64) + +-- TODO make all these constants part of ClientConfig + +-- Min/Max Time dilation. This is the maximum speedup of our own clock that +-- we'll allow to catch up to the estimated server clock. Note that the min is +-- greater than 0 meaning that we never stop or reverse time. + +minTimeDilation :: Float +minTimeDilation = 0.9 + +maxTimeDilation :: Float +maxTimeDilation = 1.1 + +-- Number of ping samples to maintain +pingSamples :: Int +pingSamples = 8 + +-- Number of timing samples to maintain +timingSamples :: Int +timingSamples = 40 + +-- Some state for managing clock synchronization +data ClockSync = ClockSync + -- On the last server time estimate: (client's local time, estimated server's local time) + { csLastSample :: Maybe (Time, Time), + -- Last few samples of point times + csPingSamples :: [Duration], + -- Last few samples of: (server time, estimated corresponding client time) + -- relative to base. + csTimingSamples :: [(Time, Time)] + } + +csEstPing :: ClockSync -> Duration +csEstPing (ClockSync {csPingSamples = xs}) = sum xs / (realToFrac $ length xs) + +-- | returns (off, drift) sutch that serverTime = (drift * clientTime) + offset +csEstOffsetAndDrift :: ClockSync -> Maybe (Time, Time) +csEstOffsetAndDrift (ClockSync {csTimingSamples = xs}) + | nInt < pingSamples || slopDenom == 0 = Nothing + -- TODO perhaps it's more efficient to just use https://en.wikipedia.org/wiki/Simple_linear_regression#Fitting_the_regression_line + + | otherwise = Just (offset, slope) + where + nInt = length xs + n = fromIntegral nInt + avg xs' = sum xs' / n + avgServer = avg (fst <$> xs) + avgClient = avg (snd <$> xs) + slopNumer = sum [(s - avgServer) * (c - avgClient) | (s, c) <- xs] + slopDenom = sum [(c - avgClient) ^ (2 :: Int64) | (_, c) <- xs] + slope = slopNumer / slopDenom + offset = avgServer - (slope * avgClient) + +-- | Initialize clock synchronization. +initializeClockSync :: + -- | Tick time (time per tick in seconds) + Float -> + -- | Get the current time from the system in seconds. + IO Float -> + -- | Returns: + -- + -- * Given some @extraTime@, Estimate the tick on the server when a message + -- sent at @now + extraTime@ is received by the server plus some extraTime + -- time. + -- + -- * Record a clock sync event. Given a heartbeat meassge, this is: client + -- send time, server receive time, client receive (of the heart beat + -- response) time) + -- + -- * analytics returns: + -- + -- * Ping + -- + -- * Estimated error from the server clock. This error occurs when we've + -- committed to some time samples then realize that our measurements are + -- off. Instead of immediately correcting, we simply dilate time (speeding + -- up a bit or slowing down a bit) until the "effective" clock is + -- corrected (see min/maxTimeDilation). On till corrected, our time + -- estimates differ from what we really think the time is on the server, + -- and that difference is the "estimated error". Specifically `error = + -- servertime - effective time` + IO (Float -> IO Tick, Float -> Float -> Float -> IO (), IO (Maybe (Float, Float))) +initializeClockSync tickTime getTime = do + clockSyncTVar :: TVar ClockSync <- newTVarIO (ClockSync Nothing [] []) + let -- Estimate the tick on the server when a message sent at `now + extraTime` is + -- received by the server plus some extraTime time. + estimateServerTickPlusLatencyPlusBufferPlus :: Float -> IO Tick + estimateServerTickPlusLatencyPlusBufferPlus extraTime = do + clientTime <- getTime + atomically $ do + cs <- readTVar clockSyncTVar + anaMay <- analytics' cs clientTime extraTime + case anaMay of + Nothing -> retry + Just (_estServerTime, dilatedEstServerTime, _ping, newCS) -> do + writeTVar clockSyncTVar newCS + return (floor (dilatedEstServerTime / tickTime)) + + analytics :: IO (Maybe (Float, Float)) + analytics = do + clientTime <- getTime + atomically $ do + cs <- readTVar clockSyncTVar + anaMay <- analytics' cs clientTime 0 + case anaMay of + Nothing -> return Nothing + Just (estServerTime, dilatedEstServerTime, ping, _newCS) -> do + return $ Just (ping, estServerTime - dilatedEstServerTime) + + -- (estimated server time, estimated server time clamping time dilation, ping, ClockSync with the new sample point) + analytics' :: ClockSync -> Time -> Float -> STM (Maybe (Float, Float, Float, ClockSync)) + analytics' cs clientTime extraTime = do + let offDriftMay = csEstOffsetAndDrift cs + case offDriftMay of + Nothing -> return Nothing + Just (offset, drift) -> do + let estServerTime = (drift * clientTime) + offset + clampedEstServerTime = fromMaybe estServerTime $ + do + (lastClientTime, lastEstServerTime) <- csLastSample cs + let targetTimeDilation = + (estServerTime - lastEstServerTime) + / (clientTime - lastClientTime) + clampedTimeDilation = + min (realToFrac maxTimeDilation) $ + max (realToFrac minTimeDilation) $ + targetTimeDilation + return $ lastEstServerTime + (clampedTimeDilation * (clientTime - lastClientTime)) + + -- For now we're just on local host, so just add a small delay + -- to the current time to estimate the server time. + let elapsedTime = clampedEstServerTime + latency = csEstPing newCS / 2 -- TODO I think adding latency is probably causing some annoying preceived input latency variablility. Rethink this! + dilatedEstServerTime = (elapsedTime + latency + bufferTime + extraTime) + newCS = cs {csLastSample = Just (clientTime, clampedEstServerTime)} + ping = csEstPing newCS + return $ Just (estServerTime + latency + bufferTime, dilatedEstServerTime, ping, newCS) + + recordClockSyncSample :: Float -> Float -> Float -> IO () + recordClockSyncSample clientSendTime serverTime clientReceiveTime = do + let pingSample = clientReceiveTime - clientSendTime + latency = pingSample / 2 + timingSample = + ( serverTime, + latency + clientSendTime + ) + + _cs' <- atomically $ do + cs <- readTVar clockSyncTVar + let cs' = + ClockSync + { csLastSample = csLastSample cs, + csPingSamples = take pingSamples (pingSample : csPingSamples cs), + csTimingSamples = take timingSamples (timingSample : csTimingSamples cs) + } + writeTVar clockSyncTVar cs' + return cs' + + -- putStrLn $ "Ping: " ++ show (csEstPing cs') + -- forM_ (csEstOffsetAndDrift cs') $ \(off, drift) -> do + -- putStrLn $ "Offset: " ++ show off + -- putStrLn $ "Drift: " ++ show drift + return () + + return (estimateServerTickPlusLatencyPlusBufferPlus, recordClockSyncSample, analytics) diff --git a/src/Alpaca/NetCode/Internal/Common.hs b/src/Alpaca/NetCode/Internal/Common.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode/Internal/Common.hs @@ -0,0 +1,259 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE ViewPatterns #-} + +-- | Rollback and replay based game networking +module Alpaca.NetCode.Internal.Common where + +import Control.Concurrent (forkIO, newChan, readChan, threadDelay, writeChan) +import Control.Concurrent.STM as STM +import Control.Monad (forever, when) +import Data.Hashable (Hashable) +import Data.Int (Int64) +import qualified Data.List as L +import Data.Map (Map) +import qualified Data.Map as M +import Data.Time.Clock +import Data.Word (Word8) +import Flat +import System.Random (randomRIO) +import Prelude + + +-- Constants + +-- Note above, we don't actually step the simulation here. We leave +-- that all up to the draw function. All we need to do is submit +-- inputs once per tick to the server. + +-- | How many missing inputs to request at a time +maxRequestAuthInputs :: Int +maxRequestAuthInputs = 100 + + +-- | TODO I need some proper logging mechanism. +debugStrLn :: String -> IO () +debugStrLn _ = return () + + +-- This can be thought of as how far the authoritative simulation is behind the +-- clients. Making this large does NOT affect latency. It DOES affect how far +-- back clients might need to roll back their simulation. Too small of a buffer +-- time means inputs will tend to be dropped (not made authoritative) because +-- they arrived a bit late. Too high of a buffer time means clients can +-- experience more pronounced popping/corrections due to large rollback. +-- +-- TODO This seems like a bit of a hack. We could instead use a buffer based on +-- out jitter. On the other hand we want to avoid time dilation, so this should +-- not be overly dynamic. +bufferTime :: Duration +bufferTime = 0.03 -- seconds + + +type Time = Float -- seconds + + +type Duration = Float -- seconds + + +-- | The game is broken into discrete ticks starting from 0. +newtype Tick = Tick Int64 + deriving stock (Show) + deriving newtype (Eq, Ord, Num, Enum, Real, Integral, Hashable, Flat) + + +newtype PlayerId = PlayerId {unPlayerId :: Word8} + deriving stock (Show) + deriving newtype (Eq, Ord, Num, Hashable) + + +deriving newtype instance (Flat PlayerId) + + +-- | Settings for simulating network conditions. Packets in both the send and +-- receive directions are randomly dropped or delayed by `simPing/2` plus some +-- random duration between `-simJitter` and `simJitter`. +data SimNetConditions = SimNetConditions + { -- | Extra ping (seconds) + simPing :: Float + , -- | Extra jitter (seconds). Should be less than simPing. + simJitter :: Float + , -- | Package loss (0 = no packet loss, 1 = 100% packet loss). + simPackageLoss :: Float + } deriving (Show, Read, Eq, Ord) + + +-- data NetConfig = NetConfig +-- { -- | Add this latency (in seconds) to all input. Players will experience +-- -- this latency even during perfect prediction, but the latency will be +-- -- consistent and reduces artifacts because input messages will be received +-- -- earlier (at least relative to their intended tick). In the extream case, +-- -- if this is set to something higher than ping, there will be no miss +-- -- predictions: all clients will receive inputs before rendering their +-- -- corresponding tick. +-- inputLatency :: Float +-- , -- | Simulate: +-- -- * Ping (seconds) +-- -- * Jitter (seconds) +-- -- * Percentage Package loss (0 = no packet loss, 1 = 100% packet loss) +-- simulatedNetConditions :: Maybe (Float, Float, Float) +-- -- -- | number of times to duplicate unreliable messages (e.g. input messages) +-- -- -- to make them more reliable. +-- -- msgDuplication :: Int64 +-- } + +simulateNetConditions :: + -- | Send function + (msg -> IO ()) -> + -- | Receive function (blocking) + (IO msg) -> + -- | Simulated ping/jitter/packetloss[0-1] + Maybe SimNetConditions -> + -- | New send and receive functions. + IO + ( msg -> IO () + , IO msg + ) +simulateNetConditions doSendMsg doRecvMsg simMay = case simMay of + Nothing -> return (doSendMsg, doRecvMsg) + Just (SimNetConditions ping jitter loss) -> do + -- Start a thread that just writes received messages into a chan + recvChan <- newChan + _recvThreadId <- forkIO $ + forever $ do + msg <- doRecvMsg + dropPacket <- (<= loss) <$> randomRIO (0, 1) + when (not dropPacket) $ do + _ <- forkIO $ do + jitterT <- randomRIO (negate jitter, jitter) + let latency = max 0 ((ping / 2) + jitterT) + threadDelay (round $ latency * 1000000) + writeChan recvChan msg + return () + return + ( -- Sending a message just starts a thread that delays the send. + \msg -> do + dropPacket <- (< loss) <$> randomRIO (0, 1) + when (not dropPacket) $ do + jitterT <- randomRIO (negate jitter, jitter) + let latency = max 0 ((ping / 2) + jitterT) + _ <- forkIO $ do + threadDelay (round $ latency * 1000000) + doSendMsg msg + return () + , readChan recvChan + ) + + +playCommon :: + Real a => + a -> + ( Float -> -- seconds per tick + IO Float -> -- get time + (UTCTime -> STM ()) -> -- Reset timer to 0 at the given time + IO b + ) -> + IO b +playCommon + tickFreq + go = + do + let tickTime :: Float + tickTime = 1 / realToFrac tickFreq + + tick0SysTimTVar <- newTVarIO undefined + + let getTime :: IO Float + getTime = do + tick0SysTime <- atomically $ readTVar tick0SysTimTVar + timeUTC <- getCurrentTime + return $ realToFrac $ timeUTC `diffUTCTime` tick0SysTime + + resetTime :: UTCTime -> STM () + resetTime = writeTVar tick0SysTimTVar + + currentTime <- getCurrentTime + atomically $ resetTime currentTime + + go tickTime getTime resetTime + + +data NetMsg input + = -- Client -> Server + Msg_Connect + Float -- Client's local time (used for initial clock sync). + | -- Server -> Client + Msg_Connected PlayerId + | -- | Client -> Server: Regularly sent. Used for clock sync and to acknowledge receiving auth ticks up to a given point. + Msg_Heartbeat + Float -- Client's local time (used for clock sync). + | -- Client -> server + Msg_Ack + Tick -- Client's max known auth inputs tick such that there are no missing ticks before it. + | -- | Server -> Client: Sent in response to Msg_Connect. This indicates the + -- clients PlayerId + Msg_HeartbeatResponse + -- Clock time on the server at Tick 0 is alwyas just 0. + Float -- Clock time on the client when the connect message was sent. + Float -- Clock time on the server when the connect message was received. + | -- | Server -> Client: complete authoritative inputs for a run of ticks + Msg_AuthInput + Tick -- Start tick (inclusive) + (CompactMaps PlayerId input) -- auth ticks starting at the given tick + (CompactMaps PlayerId input) -- non-auth ticks (hints) starting after the auth ticks + | -- | A non-authoritative hint for some input. + Msg_HintInput Tick PlayerId input + | Msg_SubmitInput [(Tick, input)] + deriving stock (Show, Generic) + + +deriving instance Flat input => Flat (NetMsg input) + + +newtype CompactMaps key value = CompactMaps [([key], [[value]])] + deriving stock (Generic, Show) + + +deriving newtype instance (Flat key, Flat value) => Flat (CompactMaps key value) + + +-- | Convert a list of maps to a datastructure that is more compact when +-- serialized by flat. This is more compact assuming that many subsequent maps +-- have the same key set. +{-# SPECIALIZE toCompactMaps :: [Map PlayerId input] -> CompactMaps PlayerId input #-} +toCompactMaps :: Eq key => [Map key value] -> CompactMaps key value +toCompactMaps maps = + CompactMaps + [ (runKeys, M.elems <$> run) + | run <- L.groupBy (\a b -> M.keysSet a == M.keysSet b) maps + , let runKeys = M.keys (head run) + ] + + +-- | Inverse of toCompactMaps +{-# SPECIALIZE fromCompactMaps :: CompactMaps PlayerId input -> [Map PlayerId input] #-} +fromCompactMaps :: Eq key => CompactMaps key value -> [Map key value] +fromCompactMaps (CompactMaps runs) = + [ M.fromAscList (zip keys values) + | (keys, valuess) <- runs + , values <- valuess + ] diff --git a/src/Alpaca/NetCode/Internal/Server.hs b/src/Alpaca/NetCode/Internal/Server.hs new file mode 100644 --- /dev/null +++ b/src/Alpaca/NetCode/Internal/Server.hs @@ -0,0 +1,363 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE DerivingVia #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Rollback and replay based game networking +module Alpaca.NetCode.Internal.Server + ( runServerWith' + , ServerConfig (..) + , defaultServerConfig + ) where + +import Control.Applicative +import Control.Concurrent (forkIO, killThread, threadDelay) +import Control.Concurrent.STM as STM +import Control.Monad (forM_, forever, join, when, forM) +import Data.Coerce (coerce) +import Data.IntMap (IntMap) +import qualified Data.IntMap as IM +import Data.List (dropWhileEnd, foldl') +import qualified Data.Map as M +import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing) +import Data.Time (getCurrentTime) +import Flat +import Prelude + +import Alpaca.NetCode.Internal.Common + +-- | Configuration options specific to the server. +data ServerConfig = ServerConfig + { + -- | Tick rate (ticks per second). Typically @30@ or @60@. Must be the same + -- across all clients and the server. Packet rate and hence network bandwidth + -- will scale linearly with this the tick rate. + scTickRate :: Int + -- | Seconds of not receiving packets from a client before disconnecting that + -- client. + , scClientTimeout :: Float + } deriving (Show, Read, Eq, Ord) + +-- | Sensible defaults for @ServerConfig@ based on the tick rate. +defaultServerConfig :: + -- | Tick rate (ticks per second). Typically @30@ or @60@. Must be the same + -- across all clients and the server. Packet rate and hence network bandwidth + -- will scale linearly with this the tick rate. + Int + -> ServerConfig +defaultServerConfig tickRate = ServerConfig + { scTickRate = tickRate + , scClientTimeout = 5 + } + +-- | Run a server for a single game. This will block until the game ends, +-- specifically when all players have disconnected. +runServerWith' :: + forall input clientAddress. + ( Eq input + , Flat input + , Show clientAddress + , Ord clientAddress + ) => + -- | Function to send messages to clients. The underlying communication + -- protocol need only guarantee data integrity but is otherwise free to drop + -- and reorder packets. Typically this is backed by a UDP socket. + (NetMsg input -> clientAddress -> IO ()) -> + -- | Blocking function to receive messages from the clients. Has the same + -- reliability requirements as the send function. + (IO (NetMsg input, clientAddress)) -> + -- | Optional simulation of network conditions. In production this should be + -- `Nothing`. May differ between clients. + Maybe SimNetConditions -> + -- | The 'defaultServerConfig' works well for most cases. + ServerConfig -> + -- | Initial input for new players. Must be the same across all clients and + -- the server. See 'Alpaca.NetCode.runClient'. + input -> + IO () +runServerWith' sendToClient' recvFromClient' simNetConditionsMay serverConfig input0 = playCommon (scTickRate serverConfig) $ \tickTime getTime resetTime -> forever $ do + (sendToClient'', recvFromClient) <- simulateNetConditions + (uncurry sendToClient') + recvFromClient' + simNetConditionsMay + let sendToClient = curry sendToClient'' + debugStrLn "Waiting for clients" + + -- Authoritative Map from tick and PlayerId to inputs. The inner map is + -- always complete (e.g. if we have the IntMap for tick i, then it contains + -- the inputs for *all* known players) + authInputsTVar :: TVar (IntMap (M.Map PlayerId input)) <- newTVarIO (IM.singleton 0 M.empty) + + -- The next Tick i.e. the first non-frozen tick. All ticks before this + -- one have been frozen (w.r.t authInputsTVar). + nextTickTVar :: TVar Tick <- newTVarIO 1 + + -- Known players as of now. Nothing means the host (me). + playersTVar :: TVar (M.Map clientAddress PlayerData) <- newTVarIO M.empty + -- Known Players ( + -- , last time for which a message was received + -- ) + + -- Next available PlayerId + nextPlayerIdTVar :: TVar PlayerId <- newTVarIO 0 + + -- As the host we're authoritative and always simulating significantly + -- behind clients. This allows for ample time to receive inputs even + -- with large ping and jitter. Although the authoritative simulation is + -- significantly behind clients, we send input hints eagerly, and that + -- allows clients to make accurate predictions and hence they don't + -- perceive the lag in authoritative inputs. + + -- Main message processing loop + msgProcessingTID <- forkIO $ + forever $ do + (msg, sender) <- recvFromClient + + -- Handle the message + serverReceiveTimeMay <- case msg of + Msg_Connected{} -> do + debugStrLn $ "Server received unexpected Msg_Connected from " ++ show sender ++ ". Ignoring." + return Nothing + Msg_AuthInput{} -> do + debugStrLn $ "Server received unexpected Msg_AuthInput from " ++ show sender ++ ". Ignoring." + return Nothing + Msg_HeartbeatResponse{} -> do + debugStrLn $ "Server received unexpected Msg_HeartbeatResponse from " ++ show sender ++ ". Ignoring." + return Nothing + Msg_HintInput{} -> do + debugStrLn $ "Server received unexpected Msg_HintInput from " ++ show sender ++ ". Perhaps you meant to send a Msg_SubmitInput. Ignoring." + return Nothing + Msg_Connect clientSendTime -> do + -- new client connection + currentTimeUTC <- getCurrentTime + currentTime <- getTime + join $ + atomically $ do + playerMay <- M.lookup sender <$> readTVar playersTVar + (pid, debugMsg, serverReceiveTime) <- case playerMay of + Nothing -> do + -- New player + pid <- readTVar nextPlayerIdTVar + writeTVar nextPlayerIdTVar (pid + 1) + players <- readTVar playersTVar + let isFirstConnection = M.null players + -- We only start the game on first connection, so must reset the timer + serverReceiveTime <- + if isFirstConnection + then do + resetTime currentTimeUTC + return 0 + else return currentTime + writeTVar playersTVar (M.insert sender (PlayerData{playerId = pid, maxAuthTick = 0, lastMesgRcvTime = serverReceiveTime}) players) + return (pid, Just ("Connected " ++ show sender ++ " as " ++ show pid), serverReceiveTime) + Just PlayerData{..} -> do + -- Existing player + return (playerId, Nothing, currentTime) + return $ do + sendToClient (Msg_Connected pid) sender + sendToClient (Msg_HeartbeatResponse clientSendTime serverReceiveTime) sender + mapM_ debugStrLn debugMsg + return (Just serverReceiveTime) + Msg_Heartbeat clientSendTime -> do + serverReceiveTime <- getTime + isConnected <- atomically (isJust . M.lookup sender <$> readTVar playersTVar) + when isConnected $ sendToClient (Msg_HeartbeatResponse clientSendTime serverReceiveTime) sender + return (Just serverReceiveTime) + Msg_Ack clientMaxAuthTick -> do + atomically $ modifyTVar playersTVar (M.update (\pd -> Just $ pd{maxAuthTick = clientMaxAuthTick}) sender) + Just <$> getTime + Msg_SubmitInput submittedInputs -> do + msgMay <- atomically $ do + -- Check that the sender is connected. + playerMay <- M.lookup sender <$> readTVar playersTVar + case playerMay of + Nothing -> return [Just $ "Got Msg_SubmitInput from client that is not yet connected " ++ show sender] + Just PlayerData{..} -> forM submittedInputs $ \(tick, input) -> do + -- Check that the tick time has not already been simulated. + nextTick <- readTVar nextTickTVar + -- TODO upper bound on allowed tick time. + if tick < nextTick + then + return $ + Just $ + "Late Msg_Input from " ++ show playerId + ++ " for " + ++ show tick + ++ " but already simulated up to " + ++ show (nextTick - 1) + ++ ". Ignoring." + else do + inputs <- readTVar authInputsTVar + let inptsAtTick = fromMaybe M.empty (inputs IM.!? fromIntegral tick) + case inptsAtTick M.!? playerId of + Just existingInput + -- Duplicate message. Silently ignore + | existingInput == input -> return Nothing + -- Different input for the same tick! + | otherwise -> + return $ + Just $ + "Received inputs from " ++ show playerId ++ " for " ++ show tick + ++ " but already have inputs for that time with a DIFFERENT value! Ignoring." + -- First time we're hearing of this input. Store it. + Nothing -> do + writeTVar authInputsTVar $ + IM.insert + (fromIntegral tick) + (M.insert playerId input inptsAtTick) + inputs + + return Nothing + mapM_ debugStrLn (catMaybes msgMay) + Just <$> getTime + + -- set receive time for players + forM_ serverReceiveTimeMay $ \serverReceiveTime -> + atomically $ + modifyTVar + playersTVar + ( M.update + (\player -> Just player{lastMesgRcvTime = serverReceiveTime}) + sender + ) + + -- Wait for a connection + atomically $ do + players <- readTVar playersTVar + STM.check $ not $ M.null players + + debugStrLn "Client connected. Starting game." + + -- Disconnect players after a timeout + disconnectTID <- forkIO $ + forever $ do + -- Find next possilbe time to disconnect a player + oldestMsgRcvTime <- atomically (minimum . fmap lastMesgRcvTime . M.elems <$> readTVar playersTVar) + let disconnectTime = oldestMsgRcvTime + scClientTimeout serverConfig + + -- Wait till the disconnect time (plus a bit to really make sure we pass the threshold) + t <- getTime + when (t < disconnectTime) $ + threadDelay (round (((disconnectTime - t) + 0.01) * 1000000)) + + -- Kick players as needed + currentTime <- getTime + kickedPlayers <- atomically $ do + players <- readTVar playersTVar + let (retainedPlayers, kickedPlayers) = M.partition (\PlayerData{..} -> lastMesgRcvTime + scClientTimeout serverConfig > currentTime) players + writeTVar playersTVar retainedPlayers + return kickedPlayers + when (not (M.null kickedPlayers)) $ debugStrLn $ "Disconnect players due to timeout: " ++ show [pid | PlayerData{playerId = PlayerId pid} <- M.elems kickedPlayers] + + -- Main "simulation" loop + simTID <- forkIO $ + forever $ do + -- Calculate target tick according to current time + currTime <- getTime + let targetTick = floor $ currTime / tickTime + + -- Fill auth inputs + atomically $ do + nextAuthTick <- readTVar nextTickTVar + + -- Freeze ticks. + writeTVar nextTickTVar (targetTick + 1) + + -- Advance auth inputs up to target tick. + knownPlayers <- readTVar playersTVar + authInputs <- readTVar authInputsTVar + let nextAuthTickInputs = authInputs IM.! fromIntegral (nextAuthTick - 1) + writeTVar authInputsTVar $ + fst $ + foldl' + ( \(authInputs', prevInputs) currTick -> + let -- Fill inputs for the current tick. + currInputsRaw = fromMaybe M.empty (IM.lookup (fromIntegral currTick) authInputs) + currInputs = + M.fromList + [ ( pidInt + , fromMaybe + input0 + ( currInputsRaw M.!? pid + <|> prevInputs M.!? pid + ) + ) + | pid <- playerId <$> M.elems knownPlayers + , let pidInt = coerce pid + ] + in (IM.insert (fromIntegral currTick) currInputs authInputs', currInputs) + ) + (authInputs, nextAuthTickInputs) + [nextAuthTick .. targetTick] + + -- broadcast some auth inputs + knownPlayers <- atomically $ readTVar playersTVar + (authInputs, nextAuthTick) <- atomically $ do + authInputs <- readTVar authInputsTVar + nextAuthTick <- readTVar nextTickTVar + return (authInputs, nextAuthTick) + forM_ (M.assocs knownPlayers) $ \(sock, playerData) -> do + let lastAuthTick = maxAuthTick playerData + (_, _, inputsToSendIntMap') = IM.splitLookup (fromIntegral lastAuthTick) authInputs + (inputsToSendIntMap, firstHint, _) = IM.splitLookup (fromIntegral nextAuthTick) inputsToSendIntMap' + inputsToSend = take maxRequestAuthInputs $ IM.elems inputsToSendIntMap + hintsToSendCount = maxRequestAuthInputs - IM.size inputsToSendIntMap + hintsToSend = + fmap (fromMaybe M.empty) $ + dropWhileEnd isNothing $ + take hintsToSendCount $ + firstHint : + [ authInputs IM.!? fromIntegral hintTick + | hintTick <- [succ nextAuthTick ..] + ] + when (not $ null inputsToSend) $ + sendToClient + ( Msg_AuthInput + (lastAuthTick + 1) + (toCompactMaps inputsToSend) + (toCompactMaps hintsToSend) + ) + sock + + -- Sleep thread till the next tick. + currTime' <- getTime + let nextTick = targetTick + 1 + nextTickTime = fromIntegral nextTick * tickTime + timeTillNextTick = nextTickTime - currTime' + threadDelay $ round $ 1000000 * timeTillNextTick + + -- Wait till all players quit + atomically $ do + players <- readTVar playersTVar + STM.check $ M.null players + + debugStrLn "No more clients, Stopping game!" + + mapM_ killThread [msgProcessingTID, disconnectTID, simTID] + +-- | Per player info stored by the server +data PlayerData = PlayerData + { -- | last tick for which auth inputs were sent from the server + playerId :: PlayerId + , -- | Client's max known auth inputs tick such that there are no missing + -- ticks before it. + maxAuthTick :: Tick + , -- | Last server time at which a message was received from this player. + lastMesgRcvTime :: Float + } diff --git a/test/Test.hs b/test/Test.hs new file mode 100644 --- /dev/null +++ b/test/Test.hs @@ -0,0 +1,196 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +import Test.Tasty +import Test.Tasty.HUnit +import Control.Concurrent +import Control.Monad (forever, when) +import Data.Bits +import Data.Map (Map) +import Data.Maybe (isNothing) +import qualified Data.Map as M +import Data.Int (Int64) +import System.Random (randomIO) +import System.Timeout (timeout) +import Alpaca.NetCode +import Alpaca.NetCode.Advanced +import Data.Maybe (fromMaybe) +import Data.List (foldl') + +main :: IO () +main = defaultMain $ testGroup "alpaca-netcode" $ let + tickRate = 1000 + tickRate32 = fromIntegral 1000 + + initialInput :: Int64 + initialInput = 123456789 + + inputLatency :: Float + inputLatency = 0.1 + + -- Step of the world does a simple hashes all the inputs. + stepWorld :: Map PlayerId Int64 -> Tick -> (Int64, Int64) -> (Int64, Int64) + stepWorld playerInputs (Tick t) (_numPlayersOld, hash) = + ( fromIntegral $ M.size playerInputs + , foldl' + (\hash' x -> (shiftL hash' 1) `xor` x) + (shiftL hash 1 `xor` t) + (concat [[fromIntegral i, j] | (PlayerId i, j) <- M.toList playerInputs]) + ) + + -- (number of players on this tick, hash over past states/inputs) + initialWorld :: (Int64, Int64) + initialWorld = (0, 12345654321) + + simulateClient :: (Int64 -> IO ()) -> IO ThreadId + simulateClient setInput = forkIO $ forever $ do + threadDelay (1000000 `div` tickRate) + setInput =<< randomIO + + test :: + ( Maybe SimNetConditions + -> ServerConfig + -> Int64 + -> IO () + ) + -> (Maybe SimNetConditions + -> ClientConfig + -> Int64 + -> (Int64, Int64) + -> (Map PlayerId Int64 -> Tick -> (Int64, Int64) -> (Int64, Int64)) -> IO (Client (Int64, Int64) Int64) + ) + -> (Maybe SimNetConditions + -> ClientConfig + -> Int64 + -> (Int64, Int64) + -> (Map PlayerId Int64 -> Tick -> (Int64, Int64) -> (Int64, Int64)) -> IO (Client (Int64, Int64) Int64) + ) + -> IO () + test runServerWith' runClient0With' runClient1With' = do + x <- timeout (15 * 1000000) $ do + -- Run a server + tidServer <- forkIO $ runServerWith' + Nothing + (defaultServerConfig tickRate32) + initialInput + + -- A client with Perfect network conditions + client0 <- runClient0With' + Nothing + (defaultClientConfig tickRate32) + initialInput + initialWorld + stepWorld + tid0 <- simulateClient (clientSetInput client0) + + -- A client with very poor network conditions + client1 <- runClient1With' + (Just (SimNetConditions 0.2 0.1 0.5)) + (defaultClientConfig tickRate32) + initialInput + initialWorld + stepWorld + tid1 <- simulateClient (clientSetInput client1) + + -- Let the game play for a bit + threadDelay (4 * 1000000) + + -- Collect auth worlds from both clients + let n = 2000 + auths0 <- take n . fst <$> clientSample' client0 + auths1 <- take n . fst <$> clientSample' client1 + + length auths0 >= n @? "Expected at least " ++ show n ++ " auth worlds but client 0 got " ++ show (length auths0) + length auths1 >= n @? "Expected at least " ++ show n ++ " auth worlds but client 1 got " ++ show (length auths1) + + (auths0 == auths1) @? "Auth worlds do not match between clients" + + let k = 100 + length (filter ((>0) . fst) auths0) > k @? "Expected at least " ++ show k ++ " tick with more that 0 players" + + killThread tidServer + clientStop client0 + killThread tid0 + clientStop client1 + killThread tid1 + + return () + when (isNothing x) (assertFailure "Timeout!") + in + [ testCase "Core" $ do + -- Use `Chan` to communicate + toServer <- newChan + toClient0 <- newChan + toClient1 <- newChan + + test + (runServerWith' + (\msg (client :: Int64) -> case client of + 0 -> writeChan toClient0 msg + 1 -> writeChan toClient1 msg + _ -> error $ "Test error! unknown client: " ++ show client + ) + (readChan toServer) + ) + ( runClientWith' + (\msg -> writeChan toServer (msg, 0)) + (readChan toClient0) + ) + (runClientWith' + (\msg -> writeChan toServer (msg, 1)) + (readChan toClient1) + ) + , testCase "UDP [NOCI]" $ do + let port = "8888" + test + (runServerWith port) + (runClientWith "localhost" port) + (runClientWith "localhost" port) + , testCase "clientStop" $ do + toServer <- newChan + toClient <- newChan + + -- Run a server + tidServer <- forkIO $ runServerWith' + (\msg 0 -> writeChan toClient msg) + (readChan toServer) + Nothing + (defaultServerConfig tickRate32) + initialInput + + -- A client with Perfect network conditions + client <- runClientWith' + (\msg -> writeChan toServer (msg, 0)) + (readChan toClient) + Nothing + (defaultClientConfig tickRate32) + initialInput + initialWorld + stepWorld + tidClient <- simulateClient (clientSetInput client) + + threadDelay (2 * 1000000) + clientStop client + w <- clientSample client + threadDelay (1 * 1000000) + (authWs', w') <- clientSample' client + assertEqual + "Sample after clientStop should return the last sampled world" + w w' + assertEqual + "Sample after clientStop should return no new auth worlds" + authWs' [] + + threadDelay (1 * 1000000) + clientStop client + (authWs'', w'') <- clientSample' client + assertEqual + "Sample after SECOND clientStop should return the last sampled world" + w w'' + assertEqual + "Sample after clientStop should return no new auth worlds" + authWs'' [] + + killThread tidServer + killThread tidClient + + ]