Skip to content

Commit

Permalink
refactor: core - switch from basic arrays to linked lists to improve …
Browse files Browse the repository at this point in the history
…performance
  • Loading branch information
Kos-M committed Dec 6, 2022
1 parent 3f5e0db commit 75ddb84
Show file tree
Hide file tree
Showing 5 changed files with 224 additions and 30 deletions.
24 changes: 16 additions & 8 deletions __tests__/index.spec.js
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@ describe('pushNewJob method', () => {

expect(encryptedPayload).toHaveProperty('iv');
expect(encryptedPayload).toHaveProperty('encryptedData');
expect(master.jobs).toHaveLength(1);
expect(JSON.parse(master.jobs[0])).toEqual(encryptedPayload);
expect(master.jobs.size).toEqual(1);
expect(JSON.parse(master.jobs.dequeue().value)).toEqual(
encryptedPayload
);
});
it('Push job succesfully and enrcypted , if job is array', async () => {
const encryptedPayload = master.crypt.encrypt(
Expand All @@ -35,8 +37,10 @@ describe('pushNewJob method', () => {

expect(encryptedPayload).toHaveProperty('iv');
expect(encryptedPayload).toHaveProperty('encryptedData');
expect(master.jobs).toHaveLength(1);
expect(JSON.parse(master.jobs[0])).toEqual(encryptedPayload);
expect(master.jobs.size).toEqual(1);
expect(JSON.parse(master.jobs.dequeue().value)).toEqual(
encryptedPayload
);
});
it('Push job succesfully and enrcypted , if job is string', async () => {
const jobData = '50';
Expand All @@ -45,8 +49,10 @@ describe('pushNewJob method', () => {

expect(encryptedPayload).toHaveProperty('iv');
expect(encryptedPayload).toHaveProperty('encryptedData');
expect(master.jobs).toHaveLength(1);
expect(JSON.parse(master.jobs[0])).toEqual(encryptedPayload);
expect(master.jobs.size).toEqual(1);
expect(JSON.parse(master.jobs.dequeue().value)).toEqual(
encryptedPayload
);
});
it('Push job succesfully and enrcypted , if job is number', async () => {
const jobData = 50;
Expand All @@ -55,7 +61,9 @@ describe('pushNewJob method', () => {

expect(encryptedPayload).toHaveProperty('iv');
expect(encryptedPayload).toHaveProperty('encryptedData');
expect(master.jobs).toHaveLength(1);
expect(JSON.parse(master.jobs[0])).toEqual(encryptedPayload);
expect(master.jobs.size).toEqual(1);
expect(JSON.parse(master.jobs.dequeue().value)).toEqual(
encryptedPayload
);
});
});
6 changes: 6 additions & 0 deletions src/Helper.js
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,12 @@ class Helper {
}
});
}

static sleep(ms) {
return new Promise((resolve) => {
setTimeout(resolve, ms);
});
}
}

module.exports = {
Expand Down
9 changes: 9 additions & 0 deletions src/Node.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
class Node {
constructor(value) {
this.value = value;
this.next = null;
this.prev = null;
}
}

module.exports = Node;
156 changes: 156 additions & 0 deletions src/Queue.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
const Node = require('./Node');

class Queue {
constructor() {
this.first = null;
this.last = null;
this.size = 0;
}

isEmpty() {
return !this.size;
}

enqueue(item) {
// Create node
const newNode = new Node(item);
/**
* * If our list is empty than both our
* * first item and last item is going to point the new node.
*/
if (this.isEmpty()) {
this.first = newNode;
this.last = newNode;
newNode.prev = null;
} else {
newNode.prev = this.last;

this.last.next = newNode;
this.last = newNode;
}
this.size += 1;
return this;
}
/**
*
* @returns
*/

dequeue() {
//* if our queue is empty we return null
if (this.isEmpty()) return null;
const itemToBeRemoved = this.first;
/**
* * if both our first and last node are pointing the same item
* * we dequeued our last node.
*/
if (this.first === this.last) {
this.last = null;
}
this.first = this.first.next;
this.size -= 1;
return itemToBeRemoved;
}

check() {
let current = this.first;
while (current) {
// while not null
current = current.next;
}
}

/**
* Searches inside list in linear fashion ,
* Time complexity O(n)
* @param {*} item String , Number , Object
* @returns Node , null
*/
search(item) {
let current = this.first;
if (typeof item !== typeof current.value) {
return null;
}
while (current !== null) {
if (current.value === item) {
return current;
}
if (typeof item === 'object') {
const itemKeys = Object.keys(item);
for (let i = 0; i < itemKeys.length; i += 1) {
// check all item properties agains current.value properties
if (
Object.prototype.hasOwnProperty.call(
current.value,
itemKeys[i]
)
) {
if (current.value[itemKeys[i]] === item[itemKeys[i]]) {
return current;
}
}
}
}
current = current.next;
}
return null;
}

bubbleSort() {
let { last } = this;
while (last) {
let node = this.first;
while (node !== last) {
const { next } = node;
if (typeof node.value === 'object') {
// object comparision in first matched (common) property
const itemKeys = Object.keys(node.value);
for (let i = 0; i < itemKeys.length; i += 1) {
if (
Object.prototype.hasOwnProperty.call(
next.value,
itemKeys[i]
)
) {
if (
node.value[itemKeys[i]] >
next.value[itemKeys[i]]
) {
// swap
[node.value, next.value] = [
next.value,
node.value,
];
break;
}
}
}
} else if (typeof node.value === 'number') {
if (node.value > next.value) {
// swap
[node.value, next.value] = [next.value, node.value];
}
} else if (typeof node.value === 'string') {
// Not Implemented yet
}

node = next;
}
last = last.prev; // shorten the range that must be bubbled through
}
}

/**
* * Returns the next element to be dequeued.
* @returns
*/
peek() {
return this.first;
}

tail() {
return this.last;
}
}

module.exports = Queue;
59 changes: 37 additions & 22 deletions src/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ const fs = require('fs');
const { Logger } = require('./Logger');
const { Helper } = require('./Helper');
const Crypt = require('./Crypt');
const Queue = require('./Queue');

const defaultExecAssets = [
{
Expand All @@ -26,7 +27,7 @@ class Master {
transferEncryptToken = null,
} = {}) {
this.availableWorkers = [];
this.jobs = [];
this.jobs = new Queue();

this.event = new events.EventEmitter();
this.log = new Logger({ level: loglevel || process.env.LOG_LEVEL });
Expand All @@ -53,6 +54,7 @@ class Master {
// this.peer.on('left', (address)=> );

this.event.addListener('init', this.init);
this.event.addListener('resultsShared', this.onResults);
this.event.emit('init', transferEncryptToken);
}

Expand Down Expand Up @@ -102,33 +104,47 @@ class Master {
});
this.peer.register('requestWork', (pk, args, cb) => {
switch (true) {
case this.jobs.length > 1: {
if (args.getBatch && this.jobs.length > args.batchSize) {
args.batchTasks = this.jobs.splice(0, args.batchSize); // splice mutates original array which slices it
case this.jobs.size > 1: {
if (args.getBatch) {
args.batchTasks = [];
let totalJobsSend = args.batchSize;
if (this.jobs.size <= args.batchSize) totalJobsSend = 1; // get only available
for (let i = 0; i < totalJobsSend; i += 1) {
const queuedJob = this.jobs.dequeue();
if (!queuedJob) break;
args.batchTasks.push(queuedJob.value);
}
// args.batchTasks = this.jobs.splice(0, args.batchSize); // splice mutates original array which slices it
this.log.debug(
`task queue reduced:${this.jobs.length} - ${args.batchSize}`
`task queue reduced:${this.jobs.size} - ${args.batchSize}`
);
break;
}
args.task = this.jobs.shift();
this.log.debug(`task queue reduced:${this.jobs.length}`);
const queuedJob = this.jobs.dequeue();
args.task = null;
if (queuedJob) {
args.task = queuedJob.value;
this.log.debug(`task queue reduced:${this.jobs.size}`);
}
break;
}
case this.jobs.length === 1: {
// shift leaves array undefined on last element
[args.task] = this.jobs; // this case is to avoid that
this.jobs = [];
this.log.debug(`task queue finished:${this.jobs.length}`);
case this.jobs.size === 1: {
const queuedJob = this.jobs.dequeue();
args.task = null;
if (queuedJob) {
args.task = queuedJob.value;
this.log.debug(`task queue reduced:${this.jobs.size}`);
}
break;
}
case this.jobs.length === 0: {
case this.jobs.size === 0: {
args.task = null;
this.log.debug(`task queue is empty:${this.jobs.length}`);
this.log.debug(`task queue is empty:${this.jobs.size}`);
break;
}
default: {
args.task = null;
this.log.warning(`task queue is empty:${this.jobs.length}`);
this.log.warning(`task queue is empty:${this.jobs.size}`);

break;
}
Expand All @@ -137,7 +153,7 @@ class Master {
});
this.peer.register('shareResults', (pk, args) => {
const results = JSON.parse(this.crypt.decrypt(JSON.parse(args)));
this.onResults(results);
this.event.emit('resultsShared', results);
});
this.peer.register('requestExecAssets', (pk, args, cb) => {
const currentHash = args?.currentHash; // hash of current assets array
Expand Down Expand Up @@ -179,14 +195,13 @@ class Master {

const encryptedPayload = this.crypt.encrypt(payloadJson);
this.log.debug('pushNewJob payload: ', payload);
if (this.jobs.length > 20) {
// resolve();
setTimeout(() => {
this.jobs.push(JSON.stringify(encryptedPayload));
if (this.jobs.size >= 1000) {
Helper.sleep(this.jobs.size * 0.3).then(() => {
this.jobs.enqueue(JSON.stringify(encryptedPayload));
resolve();
}, this.jobs.length * 3);
});
} else {
this.jobs.push(JSON.stringify(encryptedPayload));
this.jobs.enqueue(JSON.stringify(encryptedPayload));
resolve();
}
});
Expand Down

0 comments on commit 75ddb84

Please sign in to comment.