Merge f2932df into 545e286

Unity-Technologies · Feb 4, 2019 · 764caf1 · 764caf1
2 parents 545e286 + f2932df
commit 764caf1
Show file tree

Hide file tree

Showing 25 changed files with 724 additions and 141 deletions.
diff --git a/.gitignore b/.gitignore
@@ -7,4 +7,5 @@ coverage/
 local-production.yml
 .cache*/
 Dockerfile*
-.dockerignore
+.dockerignore
+diagnostics/
diff --git a/README.md b/README.md
@@ -25,6 +25,7 @@ This open-source repository is maintained separately from the Cache Server avail
   * [Function](#function)
   * [Configuration](#configuration)
 * [Unity project Library Importer](#unity-project-library-importer)
+* [Diagnostic Tools](#diagnostic-tools)
 * [Contributors](#contributors)
 * [License](#license)
 
@@ -52,14 +53,15 @@ npm install github:Unity-Technologies/unity-cache-server -g
 unity-cache-server [arguments]
 ```
 
-Command                          | Description
+Option                          | Description
 -------------------------------- | -----------
 `-V`, `--version`                | Show the version number of the Cache Server.
 `-p`, `--port <n>`               | The port on which the Cache Server listens. The default value is 8126.
 `-c`, `--cache-module [path]`    | The path to cache module. The Default path is 'cache_fs'.
 `-P`, `--cache-path [path]`      | The path of the cache directory.
 `-l`, `--log-level <n>`          | The level of log verbosity. Valid values are 0 (silent) through 5 (debug). The default is 3 (info).
 `-w`, `--workers <n>`            | The number of worker threads to spawn. The default is 0.
+`--diag-client-recorder`         | Record incoming client network stream to disk.
 `-m`, `--mirror <host:port>`     | Mirror transactions to another cache server. Repeat this option for multiple mirrors.
 `-W`, `--putwhitelist <host:port>`  | Only allow PUT transactions (uploads) from the specified client address. Repeat this option for multiple addresses.
 `--dump-config`                  | Write the active configuration to the console.
@@ -167,7 +169,7 @@ Due to performance considerations, the `cache_fs` module shipped with Cache Serv
 or
 `node cleanup.js [options]`
 
-Command                          | Description
+Option                          | Description
 -------------------------------- | -----------
 -V, --version                    | Show the version number of cleanup script.
 -c --cache-module [path]         | The path to the cache module.
@@ -243,6 +245,37 @@ Tools are provided to quickly seed a Cache Server from a fully imported Unity pr
 * The import process connects and uploads to the target host like any other Unity client, so it should be safe in a production environment.
 * Files are skipped if any changes were detected between when the JSON data was exported and when the importer tool is executed.
 
+## Diagnostic Tools
+
+Diagnostic tools are included for debugging and performance testing.
+
+### Client Recorder (--diag-client-recorder)
+
+Starting up the Cache Server with the `--diag-client-recorder` option will write to disk raw data from all incoming client connections (by default to the `diagnostics/client-recordings` folder). These client data sessions are consumed by the `stream_player.js` utility.
+
+### Client Stream Player
+
+#### Usage
+
+`stream_player.js [options] <filePath> [ServerAddress]`
+
+Option                          | Description
+------------------------------- | -----------
+  -i --iterations <n>           | Number of times to send the recorded session to the server (default: 1)
+  -c --max-concurrency <n>      | Number of concurrent connections to make to the server (default: 1)
+  -d --debug-protocol           | Print protocol stream debugging data to the console.
+  -q --no-verbose               | Do not show progress and result statistics.
+  -h, --help                    | Show usage information.
+
+#### Description
+
+The stream player can read a recorded client session(s) at `<filePath>`, optionally print the protocol stream to the console, and optionally send the protocol stream to a remote server at `[ServerAddress]` for e.g. performance load testing.
+
+#### Notes
+
+* If `<filePath>` is a directory, all files within the directory (recursively) will be read and played back. Some rudimentary validation is done on each file to detect whether or not it is a valid client session stream.
+* If `[ServerAddress]` is omitted, data will be sent to a temporary "no-op" TCP server. This is useful if you are only concerned with reading the debug protocol stream with the `-d` option.
+
 ## Contributors
 Contributions are welcome! Before submitting pull requests please note the Submission of Contributions section of the Apache 2.0 license.
 

diff --git a/config/default.yml b/config/default.yml
@@ -45,4 +45,9 @@ Server:
   options:
     allowIpv6: false
 Global:
-  logLevel: 3
+  logLevel: 3
+Diagnostics:
+  clientRecorder: false
+  clientRecorderOptions:
+    bufferSize: 10000000
+    saveDir: "diagnostics/client-recordings"
diff --git a/lib/cache/cache_fs.js b/lib/cache/cache_fs.js
@@ -10,23 +10,27 @@ const pick = require('lodash').pick;
 const crypto = require('crypto');
 const fileExtensions = require('lodash').invert(consts.FILE_TYPE);
 const cluster = require('cluster');
-const cm = require('../cluster_messages');
-
-const k_incrementGuidRef = "_incrementGuidRef";
-const k_releaseGuidRef = "_releaseGuidRef";
 
 class CacheFS extends CacheBase {
     constructor() {
         super();
+        this._id = ++CacheFS._idCounter;
+        this._kMsgIncrementGuidRef = `_incrementGuidRef.${this._id}`;
+        this._kMsgReleaseGuidRef = `_releaseGuidRef.${this._id}`;
         this._guidRefs = {};
 
-        cm.listenFor(k_incrementGuidRef, key => {
-            this._incrementGuidRef(key);
-        });
+        if(cluster.isMaster) {
+            cluster.on('message', (worker, msg) => {
+                if(!msg._msg) return;
 
-        cm.listenFor(k_releaseGuidRef, key => {
-            this._releaseGuidRef(key);
-        });
+                if(msg._msg === this._kMsgIncrementGuidRef) {
+                    this._incrementGuidRef(msg.key);
+                }
+                else if(msg._msg === this._kMsgReleaseGuidRef) {
+                    this._releaseGuidRef(msg.key);
+                }
+            });
+        }
     }
 
     get _optionsPath() {
@@ -72,7 +76,7 @@ class CacheFS extends CacheBase {
      */
     _incrementGuidRef(key) {
         if(cluster.isWorker) {
-            return cm.send(k_incrementGuidRef, key);
+            return process.send({_msg: this._kMsgIncrementGuidRef, key: key });
         }
 
         if(this._guidRefs.hasOwnProperty(key)) {
@@ -90,7 +94,7 @@ class CacheFS extends CacheBase {
      */
     _releaseGuidRef(key) {
         if(cluster.isWorker) {
-            return cm.send(k_releaseGuidRef, key);
+            return process.send({_msg: this._kMsgReleaseGuidRef, key: key });
         }
 
         if(this._guidRefs.hasOwnProperty(key)) {
@@ -226,11 +230,9 @@ class CacheFS extends CacheBase {
             spinnerMessage = 'Gathering files to delete to satisfy Max cache size';
 
             await helpers.readDir(this._cachePath, (item) => {
-                if (item.stats.isDirectory()) return next();
-
                 if (item.stats.atime < minFileAccessTime) {
-                    // already expired items are handled in the previous pass
-                    return next();
+                    // already expired items are handled in the previous pass (only relevant for dry-run)
+                    return;
                 }
 
                 item = {path: item.path, stats: pick(item.stats, ['atime', 'size'])};
@@ -297,6 +299,8 @@ class CacheFS extends CacheBase {
     }
 }
 
+CacheFS._idCounter = 0;
+
 class PutTransactionFS extends PutTransaction {
     /**
      *

diff --git a/lib/cache/reliability_manager.js b/lib/cache/reliability_manager.js
@@ -1,7 +1,6 @@
 const cluster = require('cluster');
 const path = require('path');
 const helpers = require('../helpers');
-const cm = require('../cluster_messages');
 const consts = require('../constants');
 const crypto = require('crypto');
 const _ = require('lodash');
@@ -15,6 +14,8 @@ const defaultOptions = {
 
 class ReliabilityManager {
     constructor(db, cachePath, options) {
+        this._id = ++ReliabilityManager._idCounter;
+        this._kMsgUpdateReliabilityFactor = `_updateReliabilityFactorForVersion.${this._id}`;
         this._db = db;
         this._cachePath = cachePath;
         this._options = options || {};
@@ -30,9 +31,13 @@ class ReliabilityManager {
             }
         }
 
-        cm.listenFor("_updateReliabilityFactorForVersion", (data) => {
-            return this._updateReliabilityFactorForVersion(data);
-        });
+        if(cluster.isMaster) {
+            cluster.on('message', (worker, msg) => {
+                if(msg._msg && msg._msg === this._kMsgUpdateReliabilityFactor) {
+                    return this._updateReliabilityFactorForVersion(msg);
+                }
+            });
+        }
     }
 
     /**
@@ -43,7 +48,8 @@ class ReliabilityManager {
      */
     async _updateReliabilityFactorForVersion(params) {
         if(cluster.isWorker) {
-            return cm.send('_updateReliabilityFactorForVersion', params);
+            params._msg = this._kMsgUpdateReliabilityFactor;
+            return process.send(params);
         }
 
         const entry = this.getEntry(params.guidStr, params.hashStr, true);
@@ -160,4 +166,6 @@ ReliabilityManager.reliabilityStates = {
     Reliable: 'Reliable'
 };
 
+ReliabilityManager._idCounter = 0;
+
 module.exports = ReliabilityManager;
diff --git a/lib/client/server_stream_processor.js b/lib/client/server_stream_processor.js
@@ -40,7 +40,8 @@ class ServerStreamProcessor extends Transform {
             }
 
             if(this._errState !== null) {
-                helpers.log(consts.LOG_ERR, this._errState.msg);
+                helpers.log(consts.LOG_ERR, this._errState);
+                break;
             }
         }
 
@@ -97,25 +98,28 @@ class ServerStreamProcessor extends Transform {
 
             // Read command
             if (!this.readState.didReadCmd) {
+                this.readState.didReadCmd = true;
+
                 const cmd = this._headerBuf.slice(0, consts.CMD_SIZE).toString('ascii');
+
                 this.readState.headerData.cmd = cmd;
+                this.readState.doReadId = true;
+                this.readState.headerSize += consts.ID_SIZE;
+
                 switch (cmd[0]) {
                     case '+': // file found
                         this.readState.doReadSize = true;
-                        this.readState.doReadId = true;
-                        this.readState.headerSize += consts.SIZE_SIZE + consts.ID_SIZE;
+                        this.readState.headerSize += consts.SIZE_SIZE;
                         break;
                     case '-': // file not found
-                        this.readState.doReadId = true;
-                        this.readState.headerSize += consts.ID_SIZE;
                         break;
                     default:
-                        this._errState = new Error("Unrecognized command response, aborting!");
+                        this._errState = new Error(`Unrecognized command response, aborting! (${cmd})`);
                 }
-            }
 
-            if(!fillBufferWithData())
-                break;
+                if(this._errState || !fillBufferWithData())
+                    break;
+            }
 
             let pos = consts.CMD_SIZE;
 

diff --git a/lib/cluster_messages.js b/lib/cluster_messages.js
diff --git a/lib/constants.js b/lib/constants.js
@@ -30,7 +30,8 @@ const constants = {
         MIRROR: "Mirror.addresses",
         ALLOW_IP_V6: "Server.options.allowIpv6",
         COMMAND_PROCESSOR: "Cache.options.processor",
-        PUTWHITELIST: "Cache.options.processor.putWhitelist"
+        PUTWHITELIST: "Cache.options.processor.putWhitelist",
+        CLIENT_RECORDER: "Diagnostics.clientRecorder"
     }
 };