Refactored code base

Author: arunoda

lib/server/balancer/route.js

@@ -1,20 +1,11 @@
 var Cookies = Npm.require('cookies');
 
-var workers = null;
-
-// We need to start the worker pool after the server binded
-// This allow use to play nicely with tools like userdown
-WebApp.onListening(function() {
-  var workersCount = Balancer._getWorkersCount();
-  workers = new WorkerPool(workersCount);
-});
-
 Balancer.handleHttp = function handleHttp(req, res) {
-  if(!Cluster.discovery) return processHereHTTP();
+  if(!Cluster.discovery) return Balancer._processHereHTTP(req, res);
 
   // if this is from a balance, we don't need to proxy it
   if(req.headers['from-balancer']) {
-    return processHereHTTP();
+    return Balancer._processHereHTTP(req, res);
   }
 
   var cookies = new Cookies(req, res);
@@ -47,11 +38,11 @@ Balancer.handleHttp = function handleHttp(req, res) {
     // we can get the endpointHash from the cookie
     var endpointHash = cookies.get('cluster-endpoint');
     var endpoint = Balancer._pickEndpoint(endpointHash, cookies);
-    if(!endpoint) return processHereHTTP();
+    if(!endpoint) return Balancer._processHereHTTP(req, res);
   }
 
   if(endpoint === Cluster._endpoint) {
-    return processHereHTTP();
+    return Balancer._processHereHTTP(req, res);
   }
 
   Balancer._setFromBalanceUrlHeader(req);
@@ -60,11 +51,11 @@ Balancer.handleHttp = function handleHttp(req, res) {
 };
 
 Balancer.handleWs = function handleWs(req, socket, head) {
-  if(!Cluster.discovery) return processHereWS(req, socket, head);
+  if(!Cluster.discovery) return Balancer._processHereWS(req, socket, head);
 
   if(req.headers['from-balancer']) {
     // if this is from a balance, we don't need to proxy it
-    return processHereWS(req, socket, head)
+    return Balancer._processHereWS(req, socket, head)
   }
 
   // try to get endpointHash from the our cluster-ddp url
@@ -86,16 +77,16 @@ Balancer.handleWs = function handleWs(req, socket, head) {
       // just process here. We don't need to route it to a random web service
       // because, it is possible that this endpoint is for some other service
       // than web.
-      return processHereWS(req, socket, head);
+      return Balancer._processHereWS(req, socket, head);
     }
   }
 
   if(!endpoint) {
-    return processHereWS(req, socket, head);
+    return Balancer._processHereWS(req, socket, head);
   }
 
   if(endpoint === Cluster._endpoint) {
-    return processHereWS(req, socket, head);
+    return Balancer._processHereWS(req, socket, head);
   }
 
   Balancer._setFromBalanceUrlHeader(req);
@@ -104,25 +95,4 @@ Balancer.handleWs = function handleWs(req, socket, head) {
 };
 
 OverShadowServerEvent('request', Balancer.handleHttp);
-OverShadowServerEvent('upgrade', Balancer.handleWs);
-
-// Process locally. If there are any workers running, proxy DDP traffic to them
-
-function processHereHTTP() {
-  return false;
-}
-
-function processHereWS(req, socket, head) {
-  if(process.env['CLUSTER_WORKER_ID']) return;
-
-  var worker = workers && workers.pickWorker();
-  // No worker, can't proxy. So process here.
-  if(!worker) return false;
-
-  var target = {host: "127.0.0.1", port: worker.port};
-  Balancer.proxy.ws(req, socket, head, {target: target}, function(error) {
-    // not sure we can re-try websockets, simply log it
-    console.error("Cluster: WS proxying to the worker:", worker, error.message);
-  });
-  return true;
-}
+OverShadowServerEvent('upgrade', Balancer.handleWs);

lib/server/balancer/workers.js

@@ -0,0 +1,70 @@
+var workerMapping = {};
+var workers = null;
+
+// We need to start the worker pool after the server binded
+// This allow use to play nicely with tools like userdown
+WebApp.onListening(function() {
+  var workersCount = Balancer._getWorkersCount();
+  workers = new WorkerPool(workersCount);
+});
+
+Balancer._pickWorker = function() {
+  return workers && workers.pickWorker();
+};
+
+Balancer._processHereWS = function _processHereWS(req, socket, head) {
+  if(process.env['CLUSTER_WORKER_ID']) return;
+
+  var worker = Balancer._pickWorker();
+  // No worker, can't proxy. So process here.
+  if(!worker) return false;
+
+  var target = {host: "127.0.0.1", port: worker.port};
+  Balancer.proxy.ws(req, socket, head, {target: target}, function(error) {
+    // not sure we can re-try websockets, simply log it
+    console.error("Cluster: WS proxying to the worker:", worker, error.message);
+  });
+  return true;
+};
+
+
+Balancer._processHereHTTP = function _processHereHTTP(req, res) {
+  if(process.env['CLUSTER_WORKER_ID']) return;
+
+  var longPollingMatcher =  /^\/sockjs\/([0-9]+)\/(\w+)\/xhr/;
+  var match = req.url.match(longPollingMatcher);
+  if(match) {
+    var id = match[1] + match[2];
+    if(!workerMapping[id]) {
+      var worker = Balancer._pickWorker();
+      if(worker) {
+        workerMapping[id] = {worker: worker, lastUpdate: Date.now()}
+      }
+    }
+
+    if(workerMapping[id]) {
+      workerMapping[id].lastUpdate = Date.now();
+      var target = {host: "127.0.0.1", port: workerMapping[id].worker.port}
+      // Make sure we support long polling
+      res.setTimeout(2 * 60 * 1000);
+      Balancer.proxy.web(req, res, {target: target}, function(err) {
+        res.writeHead(500);
+        res.end();
+        // Since this is long polling, error can happen even if user close the
+        // session. That's why we don't print the message.
+        //
+        // Now we also delete the mapping. Then we don't have a leak on
+        // workerMapping.
+        // XXX come up with a better plan to clean up workerMapping
+        delete workerMapping[id];
+      });
+      return true;
+    } else {
+      // Do not proxy, if we can't find out a worker
+      return false;
+    }
+  } else {
+    // Do not proxy for other files
+    return false;
+  }
+}

package.js

@@ -52,6 +52,7 @@ function configurePackage(api) {
     'lib/server/worker_pool.js',
     'lib/server/balancer/namespace.js',
     'lib/server/balancer/utils.js',
+    'lib/server/balancer/workers.js',
     'lib/server/balancer/route.js',
     'lib/server/auto_connect.js'
   ], ['server']);

todo.md

@@ -10,8 +10,8 @@ test new changes to the balancer
 test store
 backend registrations
 auto config
-- make cookie reset time to be around 1hour
-- remove the mongo-livedata dependency and use mongodb node driver
+make cookie reset time to be around 1hour
+remove the mongo-livedata dependency and use mongodb node driver
 - add the support to query expired entries locally - to handle when the MongoDB goes down
 - verbose mode for logging
 -   so we can assume better distribution
@@ -20,3 +20,16 @@ auto config
 - support websocket URL based service delivery
 - load balance direct websocket connections (both web and discovery)
 - add an special api to reblance requests
+
+# Multi Core Support
+
+Add proxy support for static files
+
+* If the file has a .xxx extension then don't proxy
+* For other, if there is fast-render then proxy 
+* Othewise don't
+
+Add proxy support for SockJS
+Add better worker management
+    * If exit okay, then simply simply do a messgae and don't start the worker
+    * Add a retry logic to restart the worker