push.rocks/smartdata
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(core): update

Browse Source
This commit is contained in:
Philipp Kunz
2023-08-15 01:01:16 +02:00
parent 0e55cd8876
commit 2ff3a4e0b7
8 changed files with 186 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
ts/00_commitinfo_data.ts
View File

@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@push.rocks/smartdata',
version: '5.0.26',
version: '5.0.27',
description: 'do more with data'
}

114
ts/smartdata.classes.distributedcoordinator.ts
View File

@ -13,6 +13,7 @@ export class DistributedClass extends SmartDataDbDoc<DistributedClass, Distribut
public data: {
status: 'bidding' | 'settled' | 'initializing' | 'stopped';
biddingShortcode?: string;
biddingStartTime?: number;
lastUpdated: number;
elected: boolean;
/**
@ -52,11 +53,17 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
}
public async stop() {
if (this.ownInstance?.data.elected) {
this.ownInstance.data.elected = false;
} else {
console.log(`can't stop a distributed instance that has not been started yet.`);
}
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
if (this.ownInstance?.data.elected) {
this.ownInstance.data.elected = false;
}
if (this.ownInstance?.data.status === 'stopped') {
console.log(`stopping a distributed instance that has not been started yet.`);
}
this.ownInstance.data.status = 'stopped';
await this.ownInstance.save();
console.log(`stopped ${this.ownInstance.id}`);
});
}
public id = plugins.smartunique.uni('distributedInstance');
@ -64,15 +71,32 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
private startHeartbeat = async () => {
while (this.ownInstance.data.status !== 'stopped') {
await this.sendHeartbeat();
await plugins.smartdelay.delayFor(10000);
await plugins.smartdelay.delayForRandom(5000, 10000);
}
};
public async sendHeartbeat() {
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
if (this.ownInstance.data.status === 'stopped') {
console.log(`aborted sending heartbeat because status is stopped`);
return;
}
await this.ownInstance.updateFromDb();
this.ownInstance.data.lastUpdated = Date.now();
await this.ownInstance.save();
console.log(`sent heartbeat for ${this.ownInstance.id}`);
const allInstances = DistributedClass.getInstances({});
});
if (this.ownInstance.data.status === 'stopped') {
console.log(`aborted sending heartbeat because status is stopped`);
return;
}
const eligibleLeader = await this.getEligibleLeader();
// not awaiting here because we don't want to block the heartbeat
this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
if (!eligibleLeader && this.ownInstance.data.status === 'settled') {
this.checkAndMaybeLead();
}
});
}
private async init() {
@ -90,6 +114,8 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
};
await this.ownInstance.save();
});
} else {
console.warn(`distributed instance already initialized`);
}
// lets enable the heartbeat
@ -101,42 +127,73 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
return this.ownInstance;
}
public async getEligibleLeader() {
return this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
const allInstances = await DistributedClass.getInstances({});
let leaders = allInstances.filter((instanceArg) => instanceArg.data.elected === true);
const eligibleLeader = leaders.find(
(leader) =>
leader.data.lastUpdated >=
Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 20 })
);
return eligibleLeader;
});
}
// --> leader election
public async checkAndMaybeLead() {
const allInstances = await DistributedClass.getInstances({});
let leader = allInstances.find((instanceArg) => instanceArg.data.elected === true);
if (
leader &&
leader.data.lastUpdated >=
Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
) {
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
this.ownInstance.data.status = 'initializing';
this.ownInstance.save();
});
if (await this.getEligibleLeader()) {
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
await this.ownInstance.updateFromDb();
this.ownInstance.data.status = 'settled';
await this.ownInstance.save();
console.log(`${this.ownInstance.id} settled as follower`);
});
return;
} else if (
(await DistributedClass.getInstances({})).find((instanceArg) => {
instanceArg.data.status === 'bidding' &&
instanceArg.data.biddingStartTime <= Date.now() - 4000 &&
instanceArg.data.biddingStartTime >= Date.now() - 30000;
})
) {
console.log('too late to the bidding party... waiting for next round.');
return;
} else {
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
await this.ownInstance.updateFromDb();
this.ownInstance.data.status = 'bidding';
this.ownInstance.data.biddingStartTime = Date.now();
this.ownInstance.data.biddingShortcode = plugins.smartunique.shortId();
await this.ownInstance.save();
console.log('bidding code stored.');
});
await plugins.smartdelay.delayFor(plugins.smarttime.getMilliSecondsFromUnits({ minutes: 2 }));
console.log(`bidding for leadership...`);
await plugins.smartdelay.delayFor(
plugins.smarttime.getMilliSecondsFromUnits({ seconds: 20 })
);
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
let biddingInstances = await DistributedClass.getInstances({});
biddingInstances = biddingInstances.filter(
(instanceArg) =>
!instanceArg.data.elected &&
instanceArg.data.status === 'bidding' &&
instanceArg.data.lastUpdated >=
Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 25 })
);
console.log(`found ${biddingInstances.length} bidding instances...`);
this.ownInstance.data.elected = true;
for (const biddingInstance of biddingInstances) {
if (biddingInstance.data.biddingShortcode < this.ownInstance.data.biddingShortcode) {
this.ownInstance.data.elected = false;
}
}
await plugins.smartdelay.delayFor(5000);
console.log(`settling with status elected = ${this.ownInstance.data.elected}`);
this.ownInstance.data.status = 'settled';
await this.ownInstance.save();
});
if (this.ownInstance.data.elected) {
@ -151,20 +208,20 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
* the leading is implemented here
*/
public async leadFunction() {
const ownInstance = await this.init();
const watcher = await DistributedClass.watch({});
/**
* this function is started once per unique job request
*/
const startResultTimer = async () => {};
watcher.changeSubject.subscribe({
next: async (distributedDoc) => {
distributedDoc;
},
});
while (this.ownInstance.data.status !== 'stopped') {
await plugins.smartdelay.delayFor(1000);
while (this.ownInstance.data.status !== 'stopped' && this.ownInstance.data.elected) {
const allInstances = await DistributedClass.getInstances({});
for (const instance of allInstances) {
if (instance.data.status === 'stopped') {
await instance.delete();
};
}
await plugins.smartdelay.delayFor(10000);
}
}
@ -172,8 +229,11 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
public async fireDistributedTaskRequest(
taskRequestArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest
): Promise<plugins.taskbuffer.distributedCoordination.IDistributedTaskRequestResult> {
const ownInstance = await this.init();
await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
if (!this.ownInstance) {
console.error('instance need to be started first...');
return;
}
this.ownInstance.data.taskRequests.push(taskRequestArg);
await this.ownInstance.save();
});
@ -190,6 +250,10 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
infoBasisItem.taskExecutionTime === infoBasisArg.taskExecutionTime
);
});
if (!existingInfoBasis) {
console.warn('trying to update a non existing task request... aborting!');
return;
}
Object.assign(existingInfoBasis, infoBasisArg);
await this.ownInstance.save();
plugins.smartdelay.delayFor(60000).then(() => {

2
ts/smartdata.classes.doc.ts
View File

@ -257,7 +257,7 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
* updates an object from db
*/
public async updateFromDb() {
const mongoDbNativeDoc = await this.collection.findOne(this.createIdentifiableObject());
const mongoDbNativeDoc = await this.collection.findOne(await this.createIdentifiableObject());
for (const key of Object.keys(mongoDbNativeDoc)) {
this[key] = mongoDbNativeDoc[key];
}