<commit_before>

/**
*
* The Server
*
**/

var main = function(){
	var col = {},
		couch_processing = {},
		couch_errors = {},
		ref_errors = [0],
		jobsCounter = [0],
		httpserver = require('./httpserver'),
		async = require("async"),
		hostJobs = {},
		httpCB = {
			syncData : function(cb){
				cb({
					processing:couch_processing,
					universe:jobsCounter,
					servers : hostJobs,
					errors : ref_errors
				});
			},
			resetJobs : function(){
				couch_errors = {};
				ref_errors[0] = 0;
				httpserver.syncData();
			}
		},
		nano     = require('nano'),
		jsonpack = require('jsonpack/main'),

		//Server side requirements
		os       = require("os"),
		couchdb  = null,
		couchque = null,
		tickreset = 5,
		justSent = {},
		zmq      = require('zmq'),
		s_wk = zmq.socket('push',{_maxListeners:50}), //socket Work Que

		s_wc = zmq.socket('push'), //socket Work CallBack
		s_wc_client = zmq.socket('pull',{_maxListeners:50}), //socket Work CallBack Client

		r = {
			"ready" : false,
			"listening" : false,
			"DBSyncInterval" : null,
			"initServer" : function (){
				var datalog = {gauge : function(){},legauge:function(){}}; //empty placeholder
				/**
				*
				* Set Up the Server that hosts the Ques
				*
				**/
				if (r.ready) return r; //already initialized
				var onReady;
				if (arguments && arguments.length>0){
					for (var x=0;x<arguments.length;x++){
						switch (typeof arguments[x]){
							case "function" : 
								onReady = arguments[x];
								break;
							case "object" : 
								if (!arguments[x].done) config = arguments[x];
						}
					}
					if (onReady) config.onReady = onReady;
				}else{
					throw new Error("Not enough arguments !");
				}

				if (config.datadog){
					var statsD = require("node-statsd");
					datalog = new statsD();
					datalog.socket.on("error", function(exception){
						console.log ("[datadog] error event in socked.send():"+exception);
					});
					datalog.legauges = {};
					datalog.legauge = function(type,host){
						var key = type+host;
						now = (~~((new Date()).getTime()/1000));
						slot = (Math.floor(now / 20) * 20)+"";
						if (!datalog.legauges[key]) datalog.legauges[key] = {};
						if (!datalog.legauges[key][slot]){
							datalog.legauges[key][slot] = 1;
						}else{
							datalog.legauges[key][slot]++;
						}
						jobsDone = 0;
						for (var ix in datalog.legauges[key]){
							if ((now-parseInt(ix)) > 300){
								delete datalog.legauges[key][ix];
							}else{
								jobsDone+=datalog.legauges[key][ix];
							}
						}
						datalog.gauge(type, jobsDone, ["worker:"+host]);
					};
				}
				
				var startClients = function(){
					//Setup zmq listening to push of commands (to clients)
					s_wk.bindSync('tcp://*:'+config.port);
					//Setup zmq for return data on the next port (from clients)
					s_wc_client.bindSync('tcp://*:'+(config.port+1));
				};
				/*==========  Setup DB Connection [CouchDB]  ==========*/
				couchdb = nano(config.couch.server);
				couchdb.db.create(config.couch.database,function(err){
					if (err && err.error != 'file_exists'){
						console.error(
							'unable to create the que database',
							err.message
						);
					}else{
						couchque = couchdb.db.use(config.couch.database);
						couchque.update = function(obj, key, callback) {
							couchque.get(key, function (error, existing) {
								if(!error){
									var newobj = existing;
									for (var attrname in obj) { newobj[attrname] = obj[attrname]; }
									couchque.insert(newobj,key, callback);
								}else{
									console.log("Unable to find job ",key);
									callback(new Error ("Unable to find job"),"");
								}
							});
						};
						r.ready = true;
						startClients();
						getKeysCargo.push({limit:config.readKeySize,descending:"true"});
						if (config.onReady) config.onReady();
						/**
						
							TODO:
							- Re-Que all Jobs, right now we only process the
							first dbIncremental, rather than all (which should be done by batches)
						
						**/
						httpserver.startHttp({
							cfg      : config,
							callback : httpCB
						});
						
					}
				});

				/*==========  ZmQ listening  ==========*/
				var couchDelCargoWorking = false;
				var couchDelCargo =  async.cargo(function (tasks, callback){
					if (couchDelCargoWorking){
						setImmediate(function(){couchDelCargo.push(tasks);});
						callback();
					}else{
						couchDelCargoWorking = true;
						var label = "dropping "+tasks.length+" job(s)";
						console.time(label);
						var docs = [];
						for (var i=0;i<tasks.length;i++){
							docs.push({
								"_id" : tasks[i].id,
								"_deleted" : true,
								"_rev" : tasks[i].c._rev
							});
						}
						couchque.bulk({docs:docs},function(err,resp){
							couchDelCargoWorking = false;
							console.timeEnd(label);
							if (err){
								console.log("error while dropping..",err);
								setImmediate(function(){couchDelCargo.push(tasks);});
								callback();
							}else{
								for (var x=0;x<resp.length;x++){
									if (!resp[x].error){
										//console.log(":) ",resp[x].id);
									}else{
										console.log(":| ",resp[x].id,resp[x].error);
									}
									delete couch_processing[resp[x].id];
									delete couch_errors[resp[x].id];
									ref_errors[0] = Object.keys(couch_errors).length;
								}
								callback();
							}
						});
					}
				},config.delBatchsize);
				s_wc_client.on('message', function(msg){
					var c = JSON.parse(msg);
					//set the lastseen flag and re-sync
					if (c && c.node && c.node.name){
						if (!hostJobs[c.node.name]) {
							hostJobs[c.node.name] = {
								lastseen : (~~((new Date()).getTime()/1000))
							};
						}else{
							hostJobs[c.node.name].lastseen = ~~((new Date()).getTime()/1000);
						}
						r.syncJobs(c.node.name,c.jobIds);
					}
					if (c.id){
						var now = ~~((new Date()).getTime()/1000);
						var id = (c.idtype == "string" ? c.id : ObjectID(c.id));
						delete c.id;
						if (c.node.name && hostJobs[c.node.name]){
							if (c.status != 3){
								hostJobs[c.node.name][id] = now;
							}else{
								delete hostJobs[c.node.name][id];
								datalog.legauge("task_completed",c.node.name);
							}
						}else if (c.node.name && c.status != 3){
							//the node took the job and it is not an end
							hostJobs[c.node.name][id] = now;
						}
						if (c.node.jobs && c.node.jobs==="0" && hostJobs[c.node.name] && Object.keys(hostJobs[c.node.name]).length > 0){
							//the node has orphan jobs
							for (var key in hostJobs[c.node.name]){
								if (key != "lastseen"){
									delete hostJobs[c.node.name][key];
									delete couch_processing[key];
								}
							}
						}
						var doUpdate = function(){
							couch_processing[id] = c.status;
							if (c.status==3 && c.error){
								console.log(":(",c.node?c.node.name:"?",id,c.error.message);
								if (couch_errors[id]) {
									couch_errors[id]++;
								}else{
									couch_errors[id] = 1;
								}
								ref_errors[0] = Object.keys(couch_errors).length;
								if (couch_errors[id] < config.retries) {
									delete couch_processing[id];
								}else{
									//error log
									if (config.errorlog){
										require("fs").appendFile(
											config.errorlog,
											JSON.stringify({id:id,error:c.error,timestamp:~~(new Date().getTime()/1000)})+"\n",
											function (apenderr) {
												if (apenderr) console.error("Unable to append to log file",apenderr);
											});
									}
								}
								datalog.legauge("task_error",c.node.name);
							}
						};
						if (config.dumpSuccess && c.status == 3 && !c.error){
							couch_processing[id] = 3;
							couchDelCargo.push({c:c,id:id});
						}else{
							doUpdate();
						}
					}else{
						if (c.startup){
							console.log("Server Reconnected ... ",c.startup);
							//erase all previous jobs from server
							for (var nodeJob in hostJobs[c.startup]){
								if (nodeJob !=="lastseen"){
									delete hostJobs[c.startup][nodeJob];
									delete couch_processing[nodeJob];
								}
							}
							/**
							
								TODO:
								- If the client deamon restart the jobs might still be running, need 
								to decide how to solve the issue first.
								- The best option is to chain the call back of the job finish to
								the job execution ?
							
							**/
						}else{
							console.log("Not sure what to do with ",c);
						}
					}
					if (c.node && c.node.name) datalog.gauge('task_processing', Object.keys(hostJobs[c.node.name]).length, ["worker:"+c.node.name]);
					datalog.gauge('task_errors', Object.keys(couch_errors).length);
				});
				/*==========  DB Syncing  ==========*/
				/**
				*
				* Functions that Monitors the DB for Jobs and adds them (or remove's them from the que)
				*
				**/
				// Cargo to gett all pending keys it auto updates when is drained
				var couch_cargo = {};
				var getKeysCargo = async.cargo(function (tasks, callback){
					if (!r.ready  || getDocsCargo.length()>config.readBatchsize || couchDelCargo.length() > config.delBatchsize || couchDelCargoWorking) {
						setImmediate(function(){getKeysCargo.push(tasks);});
						callback();
					}else{
						for(var i=0; i<tasks.length; i++){
							var task = tasks[i];
							console.time("getting "+task.limit+" keys");
							couchque.list(task,function(geterr,getbody){
								if (geterr){
									getKeysCargo.push(tasks);
								}else{
									jobsCounter[0] = getbody.total_rows;
									if (getbody.rows && getbody.rows.length>0){
										//removing all keys alredy in que
										var couch_toprocess = [];
										for (var x=0;x<getbody.rows.length;x++){
											if (!couch_processing[getbody.rows[x].id] && !couch_cargo[getbody.rows[x].id]){
												couch_toprocess.push([getbody.rows[x].id,getbody.rows[x].value.rev]);
												couch_cargo[getbody.rows[x].id] = true;
											}
										}
										if (couch_toprocess.length>0){ //sending workers
											console.timeEnd("getting "+config.readKeySize+" keys");
											console.log("["+s_wk._zmq.pending+"."+Object.keys(couch_processing).length+"]q->"+couch_toprocess.length);
											getDocsCargo.push(couch_toprocess);
										}
										//run itself again
										if (getbody.rows.length<config.readKeySize){
											setImmediate(function () {
												getKeysCargo.push({limit: config.readKeySize,descending:"true"});
											});
										}else{
											setImmediate(
												(function(gb) {
													return function () {
														getKeysCargo.push({
															limit: config.readKeySize,
															startkey :gb.rows[gb.rows.length - 1].key,
															descending:"true"
														});
													};
												})(getbody)
											);
										}
									}else{
										setImmediate(function(){getKeysCargo.push({limit:config.readKeySize,descending:"true"});});
									}
								}
								callback();
							});
						}
					}
				},1);
				var getDocsCargo = async.cargo(function(couch_toprocess, callback){
					if (!r.ready  || os.freemem() < config.minMem*1048576 || couchDelCargo.length() > config.delBatchsize || couchDelCargoWorking) {
						setImmediate(function(){
							getDocsCargo.push(couch_toprocess);
						});
						callback();
					}else{
						var keys = [], revs = {};
						for (var y=0;y<couch_toprocess.length;y++) {
							keys.push(couch_toprocess[y][0]);
							revs[couch_toprocess[y][0]] = couch_toprocess[y][1];
						}
						var label = "getting "+keys.length+" docs";
						console.time(label);
						couchque.list({
							keys:keys,
							include_docs : true
						},function(bulkerr,bulkbody){
							console.timeEnd(label);
							if (bulkerr) {
								console.log("error while getting DB data",bulkerr);
								setImmediate(function(){
									getDocsCargo.push(couch_toprocess);
								});
								callback();
							}
							
							for (var value in justSent){if (justSent[value]>tickreset) {delete justSent[value];}justSent[value]++;} //already timedout

							for (var i=0;i<bulkbody.rows.length;i++){
								if (!bulkbody.rows[i].doc || !bulkbody.rows[i].doc.exec) continue;
								var id = bulkbody.rows[i].doc._id;
								couch_processing[id] = 1;
								bulkbody.rows[i].doc["_rev"] = revs[id];
								bulkbody.rows[i].doc["idtype"] = "string";
								s_wk.send(config.jsonpack ? jsonpack.pack(bulkbody.rows[i].doc) : JSON.stringify(bulkbody.rows[i].doc));
								if (!justSent[id])justSent[id] = 1; //increment the waiting on one ir set it as one
							}
							console.log("["+s_wk._zmq.pending+"]<-d"+bulkbody.rows.length);
							//Reset jobs when the queue is 0
							if(s_wk._zmq.pending===0){
								for (var jobId in couch_processing){
									if(couch_processing[jobId]===1 && !justSent[jobId]) delete couch_processing[jobId];
								}
							}
							callback();
						});
					}
					
				},config.readBatchsize);
				getDocsCargo.drain = function(){
					couch_cargo = {}; //reset
				};
				return r;
			},

			"syncJobs" : function(host,jobs){
				if (host && Array.isArray(jobs)){
					var now = ~~((new Date()).getTime()/1000);
					//add all the jobs the client sent it already has
					if (!hostJobs[host]) hostJobs[host] = {};
					jobs.forEach(function(el){
						if (!hostJobs[host][el]) hostJobs[host][el] = now;
						if (!couch_processing[el]) couch_processing[el] = 1;
					});
					//Job Pruning
					for (var nodeJob in hostJobs[host]){
						if (jobs.indexOf(nodeJob) <0 ) delete hostJobs[host][nodeJob];
					}
					for (var cpJob in couch_processing){
						var found = false;
						for (var ilhost in hostJobs){
							for (var iljob in hostJobs[ilhost]){
								if (cpJob == iljob) found = true;
							}
						}
						if (!found) delete couch_processing[cpJob];
					}
					s_wk._outgoing = s_wk._outgoing.filter(function(item){
						jobs.forEach(function(el){
							return  (item.toString('utf8').indexOf(el) == -1);
						});
					});
				}
			},

			/*==========  Adding a Job  ==========*/

			/**
			*
			*	This Method is not to be used other than Unit Testing; 
			*	since this is ment to work solely from the DB collection
			*
			**/
<commit_msg>was removing all jobs not in the current host :( .. fixed :)<commit_after>

/**
*
* The Server
*
**/

var main = function(){
	var col = {},
		couch_processing = {},
		couch_errors = {},
		ref_errors = [0],
		jobsCounter = [0],
		httpserver = require('./httpserver'),
		async = require("async"),
		hostJobs = {},
		httpCB = {
			syncData : function(cb){
				cb({
					processing:couch_processing,
					universe:jobsCounter,
					servers : hostJobs,
					errors : ref_errors
				});
			},
			resetJobs : function(){
				couch_errors = {};
				ref_errors[0] = 0;
				httpserver.syncData();
			}
		},
		nano     = require('nano'),
		jsonpack = require('jsonpack/main'),

		//Server side requirements
		os       = require("os"),
		couchdb  = null,
		couchque = null,
		tickreset = 5,
		justSent = {},
		zmq      = require('zmq'),
		s_wk = zmq.socket('push',{_maxListeners:50}), //socket Work Que

		s_wc = zmq.socket('push'), //socket Work CallBack
		s_wc_client = zmq.socket('pull',{_maxListeners:50}), //socket Work CallBack Client

		r = {
			"ready" : false,
			"listening" : false,
			"DBSyncInterval" : null,
			"initServer" : function (){
				var datalog = {gauge : function(){},legauge:function(){}}; //empty placeholder
				/**
				*
				* Set Up the Server that hosts the Ques
				*
				**/
				if (r.ready) return r; //already initialized
				var onReady;
				if (arguments && arguments.length>0){
					for (var x=0;x<arguments.length;x++){
						switch (typeof arguments[x]){
							case "function" : 
								onReady = arguments[x];
								break;
							case "object" : 
								if (!arguments[x].done) config = arguments[x];
						}
					}
					if (onReady) config.onReady = onReady;
				}else{
					throw new Error("Not enough arguments !");
				}

				if (config.datadog){
					var statsD = require("node-statsd");
					datalog = new statsD();
					datalog.socket.on("error", function(exception){
						console.log ("[datadog] error event in socked.send():"+exception);
					});
					datalog.legauges = {};
					datalog.legauge = function(type,host){
						var key = type+host;
						now = (~~((new Date()).getTime()/1000));
						slot = (Math.floor(now / 20) * 20)+"";
						if (!datalog.legauges[key]) datalog.legauges[key] = {};
						if (!datalog.legauges[key][slot]){
							datalog.legauges[key][slot] = 1;
						}else{
							datalog.legauges[key][slot]++;
						}
						jobsDone = 0;
						for (var ix in datalog.legauges[key]){
							if ((now-parseInt(ix)) > 300){
								delete datalog.legauges[key][ix];
							}else{
								jobsDone+=datalog.legauges[key][ix];
							}
						}
						datalog.gauge(type, jobsDone, ["worker:"+host]);
					};
				}
				
				var startClients = function(){
					//Setup zmq listening to push of commands (to clients)
					s_wk.bindSync('tcp://*:'+config.port);
					//Setup zmq for return data on the next port (from clients)
					s_wc_client.bindSync('tcp://*:'+(config.port+1));
				};
				/*==========  Setup DB Connection [CouchDB]  ==========*/
				couchdb = nano(config.couch.server);
				couchdb.db.create(config.couch.database,function(err){
					if (err && err.error != 'file_exists'){
						console.error(
							'unable to create the que database',
							err.message
						);
					}else{
						couchque = couchdb.db.use(config.couch.database);
						couchque.update = function(obj, key, callback) {
							couchque.get(key, function (error, existing) {
								if(!error){
									var newobj = existing;
									for (var attrname in obj) { newobj[attrname] = obj[attrname]; }
									couchque.insert(newobj,key, callback);
								}else{
									console.log("Unable to find job ",key);
									callback(new Error ("Unable to find job"),"");
								}
							});
						};
						r.ready = true;
						startClients();
						getKeysCargo.push({limit:config.readKeySize,descending:"true"});
						if (config.onReady) config.onReady();
						/**
						
							TODO:
							- Re-Que all Jobs, right now we only process the
							first dbIncremental, rather than all (which should be done by batches)
						
						**/
						httpserver.startHttp({
							cfg      : config,
							callback : httpCB
						});
						
					}
				});

				/*==========  ZmQ listening  ==========*/
				var couchDelCargoWorking = false;
				var couchDelCargo =  async.cargo(function (tasks, callback){
					if (couchDelCargoWorking){
						setImmediate(function(){couchDelCargo.push(tasks);});
						callback();
					}else{
						couchDelCargoWorking = true;
						var label = "dropping "+tasks.length+" job(s)";
						console.time(label);
						var docs = [];
						for (var i=0;i<tasks.length;i++){
							docs.push({
								"_id" : tasks[i].id,
								"_deleted" : true,
								"_rev" : tasks[i].c._rev
							});
						}
						couchque.bulk({docs:docs},function(err,resp){
							couchDelCargoWorking = false;
							console.timeEnd(label);
							if (err){
								console.log("error while dropping..",err);
								setImmediate(function(){couchDelCargo.push(tasks);});
								callback();
							}else{
								for (var x=0;x<resp.length;x++){
									if (!resp[x].error){
										//console.log(":) ",resp[x].id);
									}else{
										console.log(":| ",resp[x].id,resp[x].error);
									}
									delete couch_processing[resp[x].id];
									delete couch_errors[resp[x].id];
									ref_errors[0] = Object.keys(couch_errors).length;
								}
								callback();
							}
						});
					}
				},config.delBatchsize);
				s_wc_client.on('message', function(msg){
					var c = JSON.parse(msg);
					//set the lastseen flag and re-sync
					if (c && c.node && c.node.name){
						if (!hostJobs[c.node.name]) {
							hostJobs[c.node.name] = {
								lastseen : (~~((new Date()).getTime()/1000))
							};
						}else{
							hostJobs[c.node.name].lastseen = ~~((new Date()).getTime()/1000);
						}
						r.syncJobs(c.node.name,c.jobIds);
					}
					if (c.id){
						var now = ~~((new Date()).getTime()/1000);
						var id = (c.idtype == "string" ? c.id : ObjectID(c.id));
						delete c.id;
						if (c.node.name && hostJobs[c.node.name]){
							if (c.status != 3){
								hostJobs[c.node.name][id] = now;
							}else{
								delete hostJobs[c.node.name][id];
								datalog.legauge("task_completed",c.node.name);
							}
						}else if (c.node.name && c.status != 3){
							//the node took the job and it is not an end
							hostJobs[c.node.name][id] = now;
						}
						if (c.node.jobs && c.node.jobs==="0" && hostJobs[c.node.name] && Object.keys(hostJobs[c.node.name]).length > 0){
							//the node has orphan jobs
							for (var key in hostJobs[c.node.name]){
								if (key != "lastseen"){
									delete hostJobs[c.node.name][key];
									delete couch_processing[key];
								}
							}
						}
						var doUpdate = function(){
							couch_processing[id] = c.status;
							if (c.status==3 && c.error){
								console.log(":(",c.node?c.node.name:"?",id,c.error.message);
								if (couch_errors[id]) {
									couch_errors[id]++;
								}else{
									couch_errors[id] = 1;
								}
								ref_errors[0] = Object.keys(couch_errors).length;
								if (couch_errors[id] < config.retries) {
									delete couch_processing[id];
								}else{
									//error log
									if (config.errorlog){
										require("fs").appendFile(
											config.errorlog,
											JSON.stringify({id:id,error:c.error,timestamp:~~(new Date().getTime()/1000)})+"\n",
											function (apenderr) {
												if (apenderr) console.error("Unable to append to log file",apenderr);
											});
									}
								}
								datalog.legauge("task_error",c.node.name);
							}
						};
						if (config.dumpSuccess && c.status == 3 && !c.error){
							couch_processing[id] = 3;
							couchDelCargo.push({c:c,id:id});
						}else{
							doUpdate();
						}
					}else{
						if (c.startup){
							console.log("Server Reconnected ... ",c.startup);
							//erase all previous jobs from server
							for (var nodeJob in hostJobs[c.startup]){
								if (nodeJob !=="lastseen"){
									delete hostJobs[c.startup][nodeJob];
									delete couch_processing[nodeJob];
								}
							}
							/**
							
								TODO:
								- If the client deamon restart the jobs might still be running, need 
								to decide how to solve the issue first.
								- The best option is to chain the call back of the job finish to
								the job execution ?
							
							**/
						}else{
							console.log("Not sure what to do with ",c);
						}
					}
					if (c.node && c.node.name) datalog.gauge('task_processing', Object.keys(hostJobs[c.node.name]).length, ["worker:"+c.node.name]);
					datalog.gauge('task_errors', Object.keys(couch_errors).length);
				});
				/*==========  DB Syncing  ==========*/
				/**
				*
				* Functions that Monitors the DB for Jobs and adds them (or remove's them from the que)
				*
				**/
				// Cargo to gett all pending keys it auto updates when is drained
				var couch_cargo = {};
				var getKeysCargo = async.cargo(function (tasks, callback){
					if (!r.ready  || getDocsCargo.length()>config.readBatchsize || couchDelCargo.length() > config.delBatchsize || couchDelCargoWorking) {
						setImmediate(function(){getKeysCargo.push(tasks);});
						callback();
					}else{
						for(var i=0; i<tasks.length; i++){
							var task = tasks[i];
							console.time("getting "+task.limit+" keys");
							couchque.list(task,function(geterr,getbody){
								if (geterr){
									getKeysCargo.push(tasks);
								}else{
									jobsCounter[0] = getbody.total_rows;
									if (getbody.rows && getbody.rows.length>0){
										//removing all keys alredy in que
										var couch_toprocess = [];
										for (var x=0;x<getbody.rows.length;x++){
											if (!couch_processing[getbody.rows[x].id] && !couch_cargo[getbody.rows[x].id]){
												couch_toprocess.push([getbody.rows[x].id,getbody.rows[x].value.rev]);
												couch_cargo[getbody.rows[x].id] = true;
											}
										}
										if (couch_toprocess.length>0){ //sending workers
											console.timeEnd("getting "+config.readKeySize+" keys");
											console.log("["+s_wk._zmq.pending+"."+Object.keys(couch_processing).length+"]q->"+couch_toprocess.length);
											getDocsCargo.push(couch_toprocess);
										}
										//run itself again
										if (getbody.rows.length<config.readKeySize){
											setImmediate(function () {
												getKeysCargo.push({limit: config.readKeySize,descending:"true"});
											});
										}else{
											setImmediate(
												(function(gb) {
													return function () {
														getKeysCargo.push({
															limit: config.readKeySize,
															startkey :gb.rows[gb.rows.length - 1].key,
															descending:"true"
														});
													};
												})(getbody)
											);
										}
									}else{
										setImmediate(function(){getKeysCargo.push({limit:config.readKeySize,descending:"true"});});
									}
								}
								callback();
							});
						}
					}
				},1);
				var getDocsCargo = async.cargo(function(couch_toprocess, callback){
					if (!r.ready  || os.freemem() < config.minMem*1048576 || couchDelCargo.length() > config.delBatchsize || couchDelCargoWorking) {
						setImmediate(function(){
							getDocsCargo.push(couch_toprocess);
						});
						callback();
					}else{
						var keys = [], revs = {};
						for (var y=0;y<couch_toprocess.length;y++) {
							keys.push(couch_toprocess[y][0]);
							revs[couch_toprocess[y][0]] = couch_toprocess[y][1];
						}
						var label = "getting "+keys.length+" docs";
						console.time(label);
						couchque.list({
							keys:keys,
							include_docs : true
						},function(bulkerr,bulkbody){
							console.timeEnd(label);
							if (bulkerr) {
								console.log("error while getting DB data",bulkerr);
								setImmediate(function(){
									getDocsCargo.push(couch_toprocess);
								});
								callback();
							}
							
							for (var value in justSent){if (justSent[value]>tickreset) {delete justSent[value];}justSent[value]++;} //already timedout

							for (var i=0;i<bulkbody.rows.length;i++){
								if (!bulkbody.rows[i].doc || !bulkbody.rows[i].doc.exec) continue;
								var id = bulkbody.rows[i].doc._id;
								couch_processing[id] = 1;
								bulkbody.rows[i].doc["_rev"] = revs[id];
								bulkbody.rows[i].doc["idtype"] = "string";
								s_wk.send(config.jsonpack ? jsonpack.pack(bulkbody.rows[i].doc) : JSON.stringify(bulkbody.rows[i].doc));
								if (!justSent[id])justSent[id] = 1; //increment the waiting on one ir set it as one
							}
							console.log("["+s_wk._zmq.pending+"]<-d"+bulkbody.rows.length);
							//Reset jobs when the queue is 0
							if(s_wk._zmq.pending===0){
								for (var jobId in couch_processing){
									if(couch_processing[jobId]===1 && !justSent[jobId]) delete couch_processing[jobId];
								}
							}
							callback();
						});
					}
					
				},config.readBatchsize);
				getDocsCargo.drain = function(){
					couch_cargo = {}; //reset
				};
				return r;
			},

			"syncJobs" : function(host,jobs){
				if (host && Array.isArray(jobs)){
					var now = ~~((new Date()).getTime()/1000);
					//add all the jobs the client sent it already has
					if (!hostJobs[host]) hostJobs[host] = {};
					jobs.forEach(function(el){
						if (!hostJobs[host][el]) hostJobs[host][el] = now;
						if (!couch_processing[el]) couch_processing[el] = 1;
					});
					//Job Pruning
					for (var nodeJob in hostJobs[host]){
						if (jobs.indexOf(nodeJob) <0 ) delete hostJobs[host][nodeJob];
					}
					for (var cpJob in couch_processing){
						var found = false;
						for (var ilhost in hostJobs){
							for (var iljob in hostJobs[ilhost]){
								if (cpJob == iljob) found = true;
							}
						}
						if (!found) delete couch_processing[cpJob];
					}
					s_wk._outgoing = s_wk._outgoing.filter(function(item){
						var found = false;
						for (var cpJob in couch_processing){
							if (item.toString('utf8').indexOf(cpJob) > -1) found = true;
						}
						return !found;
					});
				}
			},

			/*==========  Adding a Job  ==========*/

			/**
			*
			*	This Method is not to be used other than Unit Testing; 
			*	since this is ment to work solely from the DB collection
			*
			**/

{'fraction_non_alphanumeric': 0.16276711083132236, 'fraction_numerical': 0.0059455111390899875, 'mean_word_length': 2.226585420898217, 'pattern_counts': {'":': 6, '<': 27, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '564981', 'n_tokens_mistral': 14250, 'n_tokens_neox': 9461, 'n_words': 1857}