From c1d0404f236a2424544d4445a2e80370f565b1d0 Mon Sep 17 00:00:00 2001 From: Daniel Date: Wed, 9 Jun 2021 19:13:13 +0200 Subject: [PATCH] Made inventories download in chunks to prevent blocking rcon --- plugins/inventory_sync/README.md | 32 +++++++++ plugins/inventory_sync/info.js | 7 ++ plugins/inventory_sync/instance.js | 72 ++++++------------- .../inventory_sync/module/inventory_sync.lua | 41 ++++++++--- 4 files changed, 90 insertions(+), 62 deletions(-) diff --git a/plugins/inventory_sync/README.md b/plugins/inventory_sync/README.md index 8c2067a2..e01bec8a 100644 --- a/plugins/inventory_sync/README.md +++ b/plugins/inventory_sync/README.md @@ -10,3 +10,35 @@ Run the following commands in the folder Clusterio is installed to: npx clusteriomaster plugin add @clusterio/plugin-inventory_sync Substitute clusteriomaster with clusterioslave or clusterioctl if this a dedicate slave or ctl installation respectively. + +## Method of operation + +This plugin does even based synchronization of inventories. The data flow is as follows: + +1. Player joins server, gets put in spectator mode + +2. Lua script asks instance for inventory + +3. Instance asks master for inventory + +4. Master does not find inventory, and responds to instance with undefined + +5. Instance sets player to survival + +6. Player keeps playing, and eventually leaves. Inventory is serialized and sent to instance through stdout/file + +7. Instance sends inventory to master + +8. Player joins a different server. Lua sends a request to instance, puts player into spectator mode + +9. Instance sends request to master + +10. Master responds with json data + +11. Instance calls command in scenario script using rcon + +12. Once the entire command is transfered, the scenario script clears the old inventory, applies the new inventory from the scenario script and sets the player to survival mode. + +Communication between the server and instance goes over stdout or rcon, depending on the size of the data. +Communication between the instance and the master server goes over websockets. +Overall, we are able to achieve a latency between 3 and infinite ticks from server join, depending on the size of the inventory. The major limiter is rcon transfer speeds with larger inventories, especially if they contain blueprints. diff --git a/plugins/inventory_sync/info.js b/plugins/inventory_sync/info.js index 03fecb81..d0001caa 100644 --- a/plugins/inventory_sync/info.js +++ b/plugins/inventory_sync/info.js @@ -39,6 +39,13 @@ InstanceConfigGroup.define({ type: "boolean", initial_value: true, }); +InstanceConfigGroup.define({ + name: "rcon_chunk_size", + title: "Rcon inventory chunk size", + description: "Divide inventories into multiple chunks before sending with rcon to prevent blocking the pipe", + type: "number", + initial_value: 1000, +}); InstanceConfigGroup.finalize(); libUsers.definePermission({ diff --git a/plugins/inventory_sync/instance.js b/plugins/inventory_sync/instance.js index fadd0198..d1d5c881 100644 --- a/plugins/inventory_sync/instance.js +++ b/plugins/inventory_sync/instance.js @@ -8,36 +8,14 @@ const libLuaTools = require("@clusterio/lib/lua_tools"); class InstancePlugin extends libPlugin.BaseInstancePlugin { async init() { this.inventoryQueue = []; - - // setInterval(() => { - - // console.log("Connected") - // this.info.messages.upload.send(this.instance, { - // instance_id: this.instance.id, - // player: "danielv", - // inventory: "stuff", - // }) - // },5000) - - - this.instance.server.on("ipc-inventory_sync_upload", content => this.handleUpload(content).catch(err => this.logger.error( - `Error handling ipc-inventory_sync_upload:\n${err.stack}` - )) - ); - this.instance.server.on("ipc-inventory_sync_download", content => this.handleDownload(content).catch(err => this.logger.error( - `Error handling ipc-inventory_sync_download:\n${err.stack}` - )) - ); - } - - onMasterConnectionEvent(event) { - console.log("Connectionevent", event); - if (event === "connect") { - } + this.instance.server.on("ipc-inventory_sync_upload", content => this.handleUpload(content) + .catch(err => this.logger.error(`Error handling ipc-inventory_sync_upload:\n${err.stack}`))); + this.instance.server.on("ipc-inventory_sync_download", content => this.handleDownload(content) + .catch(err => this.logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`))); } async handleUpload(player) { - console.log("Uploading", player); + this.logger.verbose(`Uploading ${player.name} (${JSON.stringify(player).length / 1000}kB)`); this.info.messages.upload.send(this.instance, { instance_id: this.instance.id, instance_name: this.instance.name, @@ -47,39 +25,29 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } async handleDownload(player) { - console.log("Downloading", player.player_name); + this.logger.verbose(`Downloading ${player.player_name}`); let response = await this.info.messages.download.send(this.instance, { player_name: player.player_name, }); if (response.inventory) { - console.log("Sending command"); - await this.sendRcon(`/sc inventory_sync.downloadInventory('${response.player_name}', '${libLuaTools.escapeString(response.inventory)}')`); - } - } - - async chatEventHandler(message) { - // TODO check if cross server chat is enabled - let content = `[${message.data.instance_name}] ${removeTags(message.data.content)}`; - await this.sendRcon(`/sc game.print('${libLuaTools.escapeString(content)}')`, true); - } - - sendChat(message) { - this.info.messages.chat.send(this.instance, { - instance_name: this.instance.name, - content: message, - }); - } - - async onOutput(output) { - if (output.type === "action" && output.action === "INVENTORY") { - if (this.slave.connector.connected) { - this.sendChat(output.message); - } else { - this.inventoryQueue.push(output.message); + const chunkSize = this.instance.config.get("inventory_sync.rcon_chunk_size"); + const chunks = chunkify(chunkSize, response.inventory) + this.logger.verbose(`Sending inventory for ${player.player_name} in ${chunks.length} chunks`); + for (let i = 0; i < chunks.length; i++) { + // this.logger.verbose(`Sending chunk ${i+1} of ${chunks.length}`) + await this.sendRcon(`/sc inventory_sync.downloadInventory('${response.player_name}', '${libLuaTools.escapeString(response.inventory)}', ${i + 1}, ${chunks.length})`); } } } } +function chunkify(chunkSize, string) { + if (string.length <= chunkSize) { + return [string] + } else { + let chunk = string.substr(0, chunkSize) + return [chunk, ...chunkify(chunkSize, string.replace(chunk, ""))] + } +} module.exports = { InstancePlugin, diff --git a/plugins/inventory_sync/module/inventory_sync.lua b/plugins/inventory_sync/module/inventory_sync.lua index ebce407e..d0050cc4 100644 --- a/plugins/inventory_sync/module/inventory_sync.lua +++ b/plugins/inventory_sync/module/inventory_sync.lua @@ -34,24 +34,45 @@ end function inventory_sync.initiateInventoryDownload(playerIndex) game.players[playerIndex].print("Initiating inventory download...") global["inv_sync_download_start_tick "..game.players[playerIndex].name] = game.tick + if global.download_cache == nil then + global.download_cache = {} + end + global.download_cache[playerIndex] = "" clusterio_api.send_json("inventory_sync_download", { player_name = game.players[playerIndex].name }) end -- This function is called by instance.js as /sc inventory_sync.downloadInventory("danielv", Escaped JSON string) with data from the master -function inventory_sync.downloadInventory(playerName, data) - local serialized_player = game.json_to_table(data) - local player = game.players[playerName] - -- game.print(serpent.block(serialized_player)) - for _, inv in pairs(inventories) do - local inventory = player.get_inventory(inv) - if inventory ~= nil and serialized_player.inventories[tostring(inv)] ~= nil then - serialize.deserialize_inventory(inventory, serialized_player.inventories[tostring(inv)]) +function inventory_sync.downloadInventory(playerName, data, number, total) + if number then + if global.download_cache[playerName] == nil then + global.download_cache[playerName] = "" end + -- Append data to download cache + global.download_cache[playerName] = global.download_cache[playerName] .. data + end + + if number ~= total then + -- Show progress in GUI + game.players[playerName].print("Downloaded "..number.."/"..total.." parts") + + -- Request next segment + rcon.print("Segment downloaded") + else + local serialized_player = game.json_to_table(global.download_cache[playerName]) + global.download_cache[playerName] = nil -- remove data to lower mapsize + local player = game.players[playerName] + -- game.print(serpent.block(serialized_player)) + for _, inv in pairs(inventories) do + local inventory = player.get_inventory(inv) + if inventory ~= nil and serialized_player.inventories[tostring(inv)] ~= nil then + serialize.deserialize_inventory(inventory, serialized_player.inventories[tostring(inv)]) + end + end + local startTick = global["inv_sync_download_start_tick "..game.players[playerName].name] + game.print("Imported inventory for "..playerName.." in "..game.tick - startTick.." ticks") end - local startTick = global["inv_sync_download_start_tick "..game.players[playerName].name] - game.print("Imported inventory for "..playerName.." in "..game.tick - startTick.." ticks") end -- Upload inventory when a player leaves the game. Triggers on restart after crash if player was online during crash.