wip

examples/unfinished/benchmark/.meteor/release

@@ -1 +1 @@
-0.6.0
+0.6.5.1

examples/unfinished/benchmark/benchmark.js

@@ -3,6 +3,7 @@
 // scenarios.
 
 var PARAMS = {};
+// XXX settings now has public. could move stuff there and avoid this.
 if (Meteor.isServer) {
   if (!Meteor.settings.params)
     throw new Error("Must set scenario with Meteor.settings");
@@ -11,6 +12,11 @@ if (Meteor.isServer) {
   PARAMS = __meteor_runtime_config__.PARAMS;
 }
 
+// id for this client or server.
+var processId = Random.id();
+console.log("SSS", processId);
+
+
 //////////////////////////////
 // Helper Functions
 //////////////////////////////
@@ -36,9 +42,7 @@ var pickCollection = function () {
 
 var generateDoc = function () {
   var ret = {};
-  ret.bucket = random(PARAMS.numBuckets);
-  // XXX trusting client clock is wrong!!
-  ret.when = +(new Date);
+  ret.fromProcess = processId;
   _.times(PARAMS.documentNumFields, function (n) {
     ret['Field' + n] = randomString(PARAMS.documentSize/PARAMS.documentNumFields);
   });
@@ -59,18 +63,35 @@ _.times(PARAMS.numCollections, function (n) {
 
 
 if (Meteor.isServer) {
+  // init
   Meteor.startup(function () {
     // clear all the collections.
     _.each(Collections, function (C) {
       C.remove({});
     });
+  });
 
-    // insert initial docs
-    _.times(PARAMS.initialDocuments, function () {
-      pickCollection().insert(generateDoc());
+  // periodic db check. client list
+  var currentClients = [];
+  var totalDocs = 0;
+  Meteor.setInterval(function () {
+    var newClients = {};
+    var newTotal = 0;
+    // XXX explain 3
+    var since = +(new Date) - 1000*PARAMS.insertsPerSecond * 3;
+    _.each(Collections, function (C) {
+      _.each(C.find({}, {fields: {fromProcess: 1, when: 1}}).fetch(), function (d) {
+        newTotal += 1;
+        if (d.fromProcess && d.when > since)
+          newClients[d.fromProcess] = true;
+      });
     });
-  });
+    currentClients = _.keys(newClients);
+    totalDocs = newTotal;
+    console.log(totalDocs, currentClients.length);
+  }, 3*1000); // XXX hardcoded time
 
+  // periodic document cleanup.
   if (PARAMS.maxAgeSeconds) {
     Meteor.setInterval(function () {
       var when = +(new Date) - PARAMS.maxAgeSeconds*1000;
@@ -81,19 +102,52 @@ if (Meteor.isServer) {
     }, 1000*PARAMS.maxAgeSeconds / 20);
   }
 
-  Meteor.publish("data", function (collection, bucket) {
+  Meteor.publish("data", function (collection, process) {
+    check(collection, Number);
+    check(process, String);
     var C = Collections[collection];
-    return C.find({bucket: bucket});
+    return C.find({toProcess: process});
   });
 
+  Meteor.methods({
+    'insert': function (doc) {
+      check(doc, Object);
+      check(doc.fromProcess, String);
+      // pick a random destination. send to ourselves if there is no one
+      // else. by having an entry in the db, we'll end up in the target
+      // list.
+      doc.toProcess = Random.choice(currentClients) || doc.fromProcess;
+
+      doc.when = +(new Date);
+
+      var C = pickCollection();
+      C.insert(doc);
+    },
+    update: function (processId, field, value) {
+      check([processId, field, value], [String]);
+      var modifer = {};
+      modifer[field] = value; // XXX injection attack?
+
+      var C = pickCollection();
+      // update one message.
+      C.update({fromProcess: processId}, {$set: modifer}, {multi: false});
+    }
+  });
+
+
+  // XXX publish stats
+  // - currentClients.length
+  // - serverId
+  // - num ddp sessions
+  // - total documents
 }
 
 
 
 if (Meteor.isClient) {
   // sub to data
   _.times(PARAMS.numCollections, function (n) {
-    Meteor.subscribe("data", n, random(PARAMS.numBuckets));
+    Meteor.subscribe("data", n, processId);
   });
 
   // templates
@@ -108,17 +162,30 @@ if (Meteor.isClient) {
   };
 
   Template.status.updateRate = function () {
-    return Session.get('updateRate') + ", " + Session.get('updateAvg');
+    return (Session.get('updateAvgs') || []).join(", ");
   };
 
+  // XXX count of how many docs are in local collection. don't 
+
   // do stuff periodically
 
   if (PARAMS.insertsPerSecond) {
     Meteor.setInterval(function () {
-      pickCollection().insert(generateDoc());
+      Meteor.call('insert', generateDoc());
     }, 1000 / PARAMS.insertsPerSecond);
   }
 
+  if (PARAMS.updatesPerSecond) {
+    Meteor.setInterval(function () {
+      Meteor.call('update',
+                  processId,
+                  'Field' + random(PARAMS.documentNumFields),
+                  randomString(PARAMS.documentSize/PARAMS.documentNumFields)
+                 );
+    }, 1000 / PARAMS.updatesPerSecond);
+  }
+
+  // XXX make a method? don't use it for now.
   if (PARAMS.removesPerSecond) {
     Meteor.setInterval(function () {
       var C = pickCollection();
@@ -129,39 +196,29 @@ if (Meteor.isClient) {
     }, 1000 / PARAMS.removesPerSecond);
   }
 
-  if (PARAMS.updatesPerSecond) {
-    Meteor.setInterval(function () {
-      var C = pickCollection();
-      var docs = C.find({}).fetch();
-      var doc = Random.choice(docs);
-      if (doc) {
-        var field = 'Field' + random(PARAMS.documentNumFields);
-        var modifer = {};
-        modifer[field] =
-          randomString(PARAMS.documentSize/PARAMS.documentNumFields);
-        C.update(doc._id, {$set: modifer});
-      }
-    }, 1000 / PARAMS.updatesPerSecond);
-  }
 
 
   // XXX very rough per client update rate. we need to measure this
   // better. ideally, on the server we could get the global update rate
   var updateCount = 0;
-  var updateHistory = [];
+  var updateHistories = {1: [], 10: [], 100: [], 1000: []};
   var updateFunc = function () { updateCount += 1; };
   _.each(Collections, function (C) {
-    C.find({}).observe({
+    C.find({}).observeChanges({
       added: updateFunc, changed: updateFunc, removed: updateFunc
     });
   });
   Meteor.setInterval(function () {
-    updateHistory.push(updateCount);
-    if (updateHistory.length > 10)
-      updateHistory.shift();
-    Session.set('updateRate', updateCount);
-    Session.set('updateAvg', _.reduce(updateHistory, function(memo, num){
-      return memo + num; }, 0) / updateHistory.length);;
+    _.each(updateHistories, function (h, max) {
+      h.push(updateCount);
+      if (h.length > max)
+        h.shift();
+    });
+    Session.set('updateAvgs', _.map(updateHistories, function (h) {
+      return _.reduce(h, function(memo, num) {
+        return memo + num;
+      }, 0) / h.length;
+    }));;
     updateCount = 0;
   }, 1000);
 

examples/unfinished/benchmark/scenarios/default.json

@@ -1,12 +1,9 @@
 {
   "params": {
     "numCollections": 1,
-    "numBuckets": 3,
-    "initialDocuments": 1,
     "maxAgeSeconds": 60,
     "insertsPerSecond": 1,
     "updatesPerSecond": 1,
-    "removesPerSecond": 0.1,
     "documentSize": 1024,
     "documentNumFields": 8
   }