diff --git a/source/core/CMakeLists.txt b/source/core/CMakeLists.txt index 8c2964e..bbad6f4 100644 --- a/source/core/CMakeLists.txt +++ b/source/core/CMakeLists.txt @@ -97,6 +97,7 @@ SET (star_core_HEADERS StarRandomPoint.hpp StarRect.hpp StarRpcPromise.hpp + StarRpcThreadPromise.hpp StarSectorArray2D.hpp StarSecureRandom.hpp StarSet.hpp diff --git a/source/core/StarRpcThreadPromise.hpp b/source/core/StarRpcThreadPromise.hpp new file mode 100644 index 0000000..bae509c --- /dev/null +++ b/source/core/StarRpcThreadPromise.hpp @@ -0,0 +1,164 @@ +#ifndef STAR_RPC_THREAD_PROMISE_HPP +#define STAR_RPC_THREAD_PROMISE_HPP + +#include "StarEither.hpp" +#include "StarString.hpp" +#include "StarThread.hpp" + +// A thread-safe version of RpcPromise. +// This is just a copy-paste with a Mutex tacked on. I don't like that, but it's 11 PM. + +namespace Star { + +STAR_EXCEPTION(RpcThreadPromiseException, StarException); + +template +class RpcThreadPromiseKeeper { +public: + void fulfill(Result result); + void fail(Error error); + +private: + template + friend class RpcThreadPromise; + + function m_fulfill; + function m_fail; +}; + +template +class RpcThreadPromise { +public: + static pair> createPair(); + static RpcThreadPromise createFulfilled(Result result); + static RpcThreadPromise createFailed(Error error); + + // Has the respoonse either failed or succeeded? + bool finished() const; + // Has the response finished with success? + bool succeeded() const; + // Has the response finished with failure? + bool failed() const; + + // Returns the result of the rpc call on success, nothing on failure or when + // not yet finished. + Maybe result() const; + + // Returns the error of a failed rpc call. Returns nothing if the call is + // successful or not yet finished. + Maybe error() const; + +private: + template + friend class RpcThreadPromise; + + struct Value { + Mutex mutex; + + Maybe result; + Maybe error; + }; + + RpcThreadPromise() = default; + + function m_getValue; +}; + +template +void RpcThreadPromiseKeeper::fulfill(Result result) { + m_fulfill(move(result)); +} + +template +void RpcThreadPromiseKeeper::fail(Error error) { + m_fail(move(error)); +} + +template +pair, RpcThreadPromiseKeeper> RpcThreadPromise::createPair() { + auto valuePtr = make_shared(); + + RpcThreadPromise promise; + promise.m_getValue = [valuePtr]() { + return valuePtr.get(); + }; + + RpcThreadPromiseKeeper keeper; + keeper.m_fulfill = [valuePtr](Result result) { + MutexLocker lock(valuePtr->mutex); + if (valuePtr->result || valuePtr->error) + throw RpcThreadPromiseException("fulfill called on already finished RpcThreadPromise"); + valuePtr->result = move(result); + }; + keeper.m_fail = [valuePtr](Error error) { + MutexLocker lock(valuePtr->mutex); + if (valuePtr->result || valuePtr->error) + throw RpcThreadPromiseException("fail called on already finished RpcThreadPromise"); + valuePtr->error = move(error); + }; + + return {move(promise), move(keeper)}; +} + +template +RpcThreadPromise RpcThreadPromise::createFulfilled(Result result) { + auto valuePtr = make_shared(); + valuePtr->result = move(result); + + RpcThreadPromise promise; + promise.m_getValue = [valuePtr]() { + return valuePtr.get(); + }; + return promise; +} + +template +RpcThreadPromise RpcThreadPromise::createFailed(Error error) { + auto valuePtr = make_shared(); + valuePtr->error = move(error); + + RpcThreadPromise promise; + promise.m_getValue = [valuePtr]() { + return valuePtr.get(); + }; + return promise; +} + +template +bool RpcThreadPromise::finished() const { + auto val = m_getValue(); + MutexLocker lock(val->mutex); + return val->result || val->error; +} + +template +bool RpcThreadPromise::succeeded() const { + auto val = m_getValue(); + MutexLocker lock(val->mutex); + return val->result.isValid(); +} + +template +bool RpcThreadPromise::failed() const { + auto val = m_getValue(); + MutexLocker lock(val->mutex); + return val->error.isValid(); +} + +template +Maybe RpcThreadPromise::result() const { + auto val = m_getValue(); + MutexLocker lock(val->mutex); + return val->result; +} + +template +Maybe RpcThreadPromise::error() const { + auto val = m_getValue(); + MutexLocker lock(val->mutex); + return val->error; +} + +} + +#endif diff --git a/source/game/StarUniverseServer.cpp b/source/game/StarUniverseServer.cpp index 7742d1e..dcf76e3 100644 --- a/source/game/StarUniverseServer.cpp +++ b/source/game/StarUniverseServer.cpp @@ -258,6 +258,13 @@ void UniverseServer::setPvp(ConnectionId clientId, bool pvp) { } } +RpcThreadPromise UniverseServer::sendWorldMessage(WorldId const& worldId, String const& message, JsonArray const& args) { + auto pair = RpcThreadPromise::createPair(); + RecursiveMutexLocker locker(m_mainLock); + m_pendingWorldMessages[worldId].push_back({ message, args, pair.second }); + return pair.first; +} + void UniverseServer::clientWarpPlayer(ConnectionId clientId, WarpAction action, bool deploy) { RecursiveMutexLocker locker(m_mainLock); m_pendingPlayerWarps[clientId] = pair(move(action), move(deploy)); @@ -502,6 +509,7 @@ void UniverseServer::run() { sendClientContextUpdates(); respondToCelestialRequests(); clearBrokenWorlds(); + handleWorldMessages(); shutdownInactiveWorlds(); doTriggeredStorage(); } catch (std::exception const& e) { @@ -1029,6 +1037,23 @@ void UniverseServer::clearBrokenWorlds() { } } +void UniverseServer::handleWorldMessages() { + RecursiveMutexLocker locker(m_mainLock); + ReadLocker clientsLocker(m_clientsLock); + + auto it = m_pendingWorldMessages.begin(); + while (it != m_pendingWorldMessages.end()) { + auto& worldId = it->first; + if (auto worldPtr = triggerWorldCreation(worldId).value()) { + worldPtr->passMessages(move(it->second)); + it = m_pendingWorldMessages.erase(it); + } + else + ++it; + + } +} + void UniverseServer::shutdownInactiveWorlds() { RecursiveMutexLocker locker(m_mainLock); ReadLocker clientsLocker(m_clientsLock); @@ -1040,7 +1065,7 @@ void UniverseServer::shutdownInactiveWorlds() { world->stop(); Logger::error("UniverseServer: World {} has stopped due to an error", worldId); worldDiedWithError(world->worldId()); - } else if (world->clients().empty()) { + } else if (world->noClients()) { bool anyPendingWarps = false; for (auto const& p : m_pendingPlayerWarps) { if (resolveWarpAction(p.second.first, p.first, p.second.second).world == world->worldId()) { @@ -1048,7 +1073,8 @@ void UniverseServer::shutdownInactiveWorlds() { break; } } - if (!anyPendingWarps) { + + if (!anyPendingWarps && world->shouldExpire()) { Logger::info("UniverseServer: Stopping idle world {}", worldId); world->stop(); } diff --git a/source/game/StarUniverseServer.hpp b/source/game/StarUniverseServer.hpp index da6fc9c..33b0838 100644 --- a/source/game/StarUniverseServer.hpp +++ b/source/game/StarUniverseServer.hpp @@ -78,6 +78,8 @@ public: bool isPvp(ConnectionId clientId) const; void setPvp(ConnectionId clientId, bool pvp); + RpcThreadPromise sendWorldMessage(WorldId const& worldId, String const& message, JsonArray const& args = {}); + void clientWarpPlayer(ConnectionId clientId, WarpAction action, bool deploy = false); void clientFlyShip(ConnectionId clientId, Vec3I const& system, SystemLocation const& location); WorldId clientWorld(ConnectionId clientId) const; @@ -127,6 +129,7 @@ private: void respondToCelestialRequests(); void processChat(); void clearBrokenWorlds(); + void handleWorldMessages(); void shutdownInactiveWorlds(); void doTriggeredStorage(); @@ -242,6 +245,7 @@ private: List> m_pendingFlagActions; HashMap>> m_pendingChat; Maybe> m_nextRandomizedStarterWorld; + Map> m_pendingWorldMessages; List m_tempBans; }; diff --git a/source/game/StarWorldServer.cpp b/source/game/StarWorldServer.cpp index 3c5fbfe..83e5cea 100644 --- a/source/game/StarWorldServer.cpp +++ b/source/game/StarWorldServer.cpp @@ -41,6 +41,7 @@ WorldServer::WorldServer(WorldTemplatePtr const& worldTemplate, IODevicePtr stor m_tileProtectionEnabled = true; m_universeSettings = make_shared(); m_worldId = worldTemplate->worldName(); + m_expiryTimer = GameTimer(0.0f); init(true); writeMetadata(); @@ -514,6 +515,10 @@ List WorldServer::getOutgoingPackets(ConnectionId clientId) { return move(clientInfo->outgoingPackets); } +Maybe WorldServer::receiveMessage(ConnectionId fromConnection, String const& message, JsonArray const& args) { + return "what a wonderful world"; +} + WorldServerFidelity WorldServer::fidelity() const { return m_fidelity; } @@ -523,6 +528,19 @@ void WorldServer::setFidelity(WorldServerFidelity fidelity) { m_fidelityConfig = m_serverConfig.get("fidelitySettings").get(WorldServerFidelityNames.getRight(m_fidelity)); } +bool WorldServer::shouldExpire() { + if (!m_clientInfo.empty()) { + m_expiryTimer.reset(); + return false; + } + + return m_expiryTimer.ready(); +} + +void WorldServer::setExpiryTime(float expiryTime) { + m_expiryTimer = GameTimer(expiryTime); +} + void WorldServer::update(float dt) { ++m_currentStep; for (auto const& pair : m_clientInfo) @@ -627,6 +645,8 @@ void WorldServer::update(float dt) { for (auto& pair : m_clientInfo) pair.second->pendingForward = false; + m_expiryTimer.tick(dt); + LogMap::set(strf("server_{}_entities", m_worldId), strf("{} in {} sectors", m_entityMap->size(), m_tileArray->loadedSectorCount())); LogMap::set(strf("server_{}_time", m_worldId), strf("age = {:4.2f}, day = {:4.2f}/{:4.2f}s", epochTime(), timeOfDay(), dayLength())); LogMap::set(strf("server_{}_active_liquid", m_worldId), m_liquidEngine->activeCells()); diff --git a/source/game/StarWorldServer.hpp b/source/game/StarWorldServer.hpp index 66a21d3..b8f3965 100644 --- a/source/game/StarWorldServer.hpp +++ b/source/game/StarWorldServer.hpp @@ -14,6 +14,7 @@ #include "StarLuaRoot.hpp" #include "StarWorldRenderData.hpp" #include "StarWarping.hpp" +#include "StarRpcThreadPromise.hpp" namespace Star { @@ -98,6 +99,8 @@ public: void handleIncomingPackets(ConnectionId clientId, List const& packets); List getOutgoingPackets(ConnectionId clientId); + Maybe receiveMessage(ConnectionId fromConnection, String const& message, JsonArray const& args); + void startFlyingSky(bool enterHyperspace, bool startInWarp); void stopFlyingSkyAt(SkyParameters const& destination); void setOrbitalSky(SkyParameters const& destination); @@ -106,6 +109,9 @@ public: WorldServerFidelity fidelity() const; void setFidelity(WorldServerFidelity fidelity); + bool shouldExpire(); + void setExpiryTime(float expiryTime); + void update(float dt); ConnectionId connection() const override; @@ -379,6 +385,8 @@ private: List m_forceRegions; String m_worldId; + + GameTimer m_expiryTimer; }; } diff --git a/source/game/StarWorldServerThread.cpp b/source/game/StarWorldServerThread.cpp index e06221e..6928456 100644 --- a/source/game/StarWorldServerThread.cpp +++ b/source/game/StarWorldServerThread.cpp @@ -13,7 +13,8 @@ WorldServerThread::WorldServerThread(WorldServerPtr server, WorldId worldId) m_worldServer(move(server)), m_worldId(move(worldId)), m_stop(false), - m_errorOccurred(false) { + m_errorOccurred(false), + m_shouldExpire(true) { if (m_worldServer) m_worldServer->setWorldId(printWorldId(m_worldId)); } @@ -50,6 +51,10 @@ bool WorldServerThread::serverErrorOccurred() { return m_errorOccurred; } +bool WorldServerThread::shouldExpire() { + return m_shouldExpire; +} + bool WorldServerThread::spawnTargetValid(SpawnTarget const& spawnTarget) { try { RecursiveMutexLocker locker(m_mutex); @@ -115,6 +120,12 @@ bool WorldServerThread::hasClient(ConnectionId clientId) const { return m_clients.contains(clientId); } +bool WorldServerThread::noClients() const { + RecursiveMutexLocker locker(m_mutex); + return m_clients.empty(); +} + + List WorldServerThread::erroredClients() const { RecursiveMutexLocker locker(m_mutex); auto unerroredClients = HashSet::from(m_worldServer->clientIds()); @@ -165,6 +176,11 @@ void WorldServerThread::setUpdateAction(WorldServerAction updateAction) { m_updateAction = updateAction; } +void WorldServerThread::passMessages(List&& messages) { + RecursiveMutexLocker locker(m_messageMutex); + m_messages.appendAll(move(messages)); +} + WorldChunks WorldServerThread::readChunks() { try { RecursiveMutexLocker locker(m_mutex); @@ -256,12 +272,26 @@ void WorldServerThread::update(WorldServerFidelity fidelity) { if (!m_pause || *m_pause == false) m_worldServer->update(dt); + List messages; + { + RecursiveMutexLocker locker(m_messageMutex); + messages = move(m_messages); + } + for (auto& message : messages) { + if (auto resp = m_worldServer->receiveMessage(ServerConnectionId, message.message, message.args)) + message.promise.fulfill(*resp); + else + message.promise.fail("Message not handled by world"); + } + for (auto& clientId : unerroredClientIds) { auto outgoingPackets = m_worldServer->getOutgoingPackets(clientId); RecursiveMutexLocker queueLocker(m_queueMutex); m_outgoingPacketQueue[clientId].appendAll(move(outgoingPackets)); } + m_shouldExpire = m_worldServer->shouldExpire(); + if (m_updateAction) m_updateAction(this, m_worldServer.get()); } diff --git a/source/game/StarWorldServerThread.hpp b/source/game/StarWorldServerThread.hpp index bd13c02..d61c3dc 100644 --- a/source/game/StarWorldServerThread.hpp +++ b/source/game/StarWorldServerThread.hpp @@ -3,6 +3,7 @@ #include "StarWorldServer.hpp" #include "StarThread.hpp" +#include "StarRpcThreadPromise.hpp" namespace Star { @@ -13,6 +14,12 @@ STAR_CLASS(WorldServerThread); // the error and trigger the WorldServerThread error state. class WorldServerThread : public Thread { public: + struct Message { + String message; + JsonArray args; + RpcThreadPromiseKeeper promise; + }; + typedef function WorldServerAction; WorldServerThread(WorldServerPtr server, WorldId worldId); @@ -28,6 +35,7 @@ public: // An exception occurred from the actual WorldServer itself and the // WorldServerThread has stopped running. bool serverErrorOccurred(); + bool shouldExpire(); bool spawnTargetValid(SpawnTarget const& spawnTarget); @@ -37,6 +45,7 @@ public: List clients() const; bool hasClient(ConnectionId clientId) const; + bool noClients() const; // Clients that have caused an error with incoming packets are removed from // the world and no further packets are handled from them. They are still @@ -61,6 +70,9 @@ public: // also in a thread safe context. void setUpdateAction(WorldServerAction updateAction); + // + void passMessages(List&& messages); + // Syncs all active sectors to disk and reads the full content of the world // into memory, useful for the ship. WorldChunks readChunks(); @@ -84,9 +96,13 @@ private: Map> m_incomingPacketQueue; Map> m_outgoingPacketQueue; + mutable RecursiveMutex m_messageMutex; + List m_messages; + atomic m_stop; shared_ptr> m_pause; mutable atomic m_errorOccurred; + mutable atomic m_shouldExpire; }; } diff --git a/source/game/scripting/StarLuaGameConverters.hpp b/source/game/scripting/StarLuaGameConverters.hpp index ff84792..6188bf1 100644 --- a/source/game/scripting/StarLuaGameConverters.hpp +++ b/source/game/scripting/StarLuaGameConverters.hpp @@ -10,6 +10,7 @@ #include "StarBehaviorState.hpp" #include "StarSystemWorld.hpp" #include "StarDrawable.hpp" +#include "StarRpcThreadPromise.hpp" namespace Star { @@ -33,6 +34,14 @@ struct LuaUserDataMethods> { static LuaMethods> make(); }; +template +struct LuaConverter> : LuaUserDataConverter> {}; + +template +struct LuaUserDataMethods> { + static LuaMethods> make(); +}; + template <> struct LuaConverter { static LuaValue from(LuaEngine& engine, PlatformerAStar::Path const& path); @@ -115,6 +124,16 @@ LuaMethods> LuaUserDataMethods>::make() { return methods; } +template +LuaMethods> LuaUserDataMethods>::make() { + LuaMethods> methods; + methods.template registerMethodWithSignature&>("finished", mem_fn(&RpcThreadPromise::finished)); + methods.template registerMethodWithSignature&>("succeeded", mem_fn(&RpcThreadPromise::succeeded)); + methods.template registerMethodWithSignature, RpcThreadPromise&>("result", mem_fn(&RpcThreadPromise::result)); + methods.template registerMethodWithSignature, RpcThreadPromise&>("error", mem_fn(&RpcThreadPromise::error)); + return methods; +} + template <> struct LuaConverter { static LuaValue from(LuaEngine& engine, Collection const& c); diff --git a/source/game/scripting/StarUniverseServerLuaBindings.cpp b/source/game/scripting/StarUniverseServerLuaBindings.cpp index b055b76..9101be5 100644 --- a/source/game/scripting/StarUniverseServerLuaBindings.cpp +++ b/source/game/scripting/StarUniverseServerLuaBindings.cpp @@ -18,6 +18,7 @@ LuaCallbacks LuaBindings::makeUniverseServerCallbacks(UniverseServer* universe) callbacks.registerCallbackWithSignature("isAdmin", bind(UniverseServerCallbacks::isAdmin, universe, _1)); callbacks.registerCallbackWithSignature("isPvp", bind(UniverseServerCallbacks::isPvp, universe, _1)); callbacks.registerCallbackWithSignature("setPvp", bind(UniverseServerCallbacks::setPvp, universe, _1, _2)); + callbacks.registerCallbackWithSignature, LuaEngine&, String, String, LuaVariadic>("sendWorldMessage", bind(UniverseServerCallbacks::sendWorldMessage, universe, _1, _2, _3, _4)); return callbacks; } @@ -107,4 +108,8 @@ void LuaBindings::UniverseServerCallbacks::setPvp(UniverseServer* universe, Conn universe->setPvp(client, setPvpTo); } +RpcThreadPromise LuaBindings::UniverseServerCallbacks::sendWorldMessage(UniverseServer* universe, LuaEngine& engine, String const& worldId, String const& message, LuaVariadic args) { + return universe->sendWorldMessage(parseWorldId(worldId), message, JsonArray::from(move(args))); +} + } diff --git a/source/game/scripting/StarUniverseServerLuaBindings.hpp b/source/game/scripting/StarUniverseServerLuaBindings.hpp index f94f285..5eb1674 100644 --- a/source/game/scripting/StarUniverseServerLuaBindings.hpp +++ b/source/game/scripting/StarUniverseServerLuaBindings.hpp @@ -3,6 +3,7 @@ #include "StarLua.hpp" #include "StarGameTypes.hpp" +#include "StarRpcThreadPromise.hpp" namespace Star { @@ -22,6 +23,7 @@ namespace LuaBindings { bool isAdmin(UniverseServer* universe, ConnectionId arg1); bool isPvp(UniverseServer* universe, ConnectionId arg1); void setPvp(UniverseServer* universe, ConnectionId arg1, Maybe arg2); + RpcThreadPromise sendWorldMessage(UniverseServer* universe, LuaEngine& engine, String const& worldId, String const& message, LuaVariadic args); } } } diff --git a/source/game/scripting/StarWorldLuaBindings.cpp b/source/game/scripting/StarWorldLuaBindings.cpp index 8d7d1bd..da4f7c8 100644 --- a/source/game/scripting/StarWorldLuaBindings.cpp +++ b/source/game/scripting/StarWorldLuaBindings.cpp @@ -390,6 +390,9 @@ namespace LuaBindings { callbacks.registerCallbackWithSignature("setSkyTime", [serverWorld](double skyTime) { return serverWorld->sky()->setEpochTime(skyTime); }); + + callbacks.registerCallback("setExpiryTime", [serverWorld](float expiryTime) { serverWorld->setExpiryTime(expiryTime); }); + callbacks.registerCallback("flyingType", [serverWorld]() -> String { return FlyingTypeNames.getRight(serverWorld->sky()->flyingType()); }); callbacks.registerCallback("warpPhase", [serverWorld]() -> String { return WarpPhaseNames.getRight(serverWorld->sky()->warpPhase()); }); callbacks.registerCallback("setUniverseFlag", [serverWorld](String flagName) { return serverWorld->universeSettings()->setFlag(flagName); });