fez5587/Superalgos
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Re-enabling multiple process instances executing in parallel and integrating them with the system event handler

Browse Source
This commit is contained in:
Luis Fernando Molina committed 2019-09-11 19:07:51 +02:00
1 parent 0e606c971b
commit f18f86fa17
5 files changed
+64 -101
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Integrations/FileStorage.js
+1 -1
View File
@@ -40,7 +40,7 @@ exports.newFileStorage = function newFileStorage(logger) {
if (logger) {
logger.write(MODULE_NAME, message);
} else {
console.log("['" + new Date().toISOString() + "', 0,'" + MODULE_NAME + "','" + message + "']")
//console.log("['" + new Date().toISOString() + "', 0,'" + MODULE_NAME + "','" + message + "']")
}
}
Project.njsproj
-2
View File
@@ -73,8 +73,6 @@
<ItemGroup>
<Content Include=".env" />
<Content Include=".gitignore" />
<Content Include="sequence - all processes.json" />
<Content Include="sequence.json" />
</ItemGroup>
<ItemGroup>
<Folder Include="Integrations\" />
Root.js
+19 -33
View File
@@ -201,7 +201,6 @@
let processConfig = botConfig.processes[i];
processConfig.startMode = global.EXECUTION_CONFIG.startMode // Override file storage configuration
if(global.EXECUTION_CONFIG.timePeriod){
processConfig.timePeriod = global.EXECUTION_CONFIG.timePeriod
}
@@ -447,6 +446,7 @@
function runSensorBot(pBotConfig, pProcessConfig, pMonth, pYear) {
try {
global.TOTAL_PROCESS_INSTANCES_CREATED++
const DEBUG_MODULE = require(ROOT_DIR + 'DebugLog');
let logger;
@@ -498,7 +498,7 @@
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runSensorBot -> onInitializeReady -> whenStartFinishes -> Bot execution was aborted.");
logger.persist();
}
exitProcess()
exitProcessInstance()
}
} else {
@@ -507,19 +507,21 @@
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runSensorBot -> onInitializeReady -> err = "+ err.stack);
logger.persist();
exitProcess()
exitProcessInstance()
}
}
}
catch (err) {
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runSensorBot -> err = " + err.stack);
exitProcess()
exitProcessInstance()
}
}
function runIndicatorBot(pBotConfig, pProcessConfig, pMonth, pYear) {
try {
global.TOTAL_PROCESS_INSTANCES_CREATED++
const DEBUG_MODULE = require(ROOT_DIR + 'DebugLog');
let logger;
@@ -568,7 +570,7 @@
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runIndicatorBot -> onInitializeReady -> whenStartFinishes -> Bot execution finished with errors. Please check the logs.");
logger.persist();
}
exitProcess()
exitProcessInstance()
}
} else {
@@ -576,18 +578,20 @@
logger.write(MODULE_NAME, "[ERROR] start -> findProcess -> runIndicatorBot -> onInitializeReady -> Failed to initialize the bot. ");
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runIndicatorBot -> onInitializeReady -> err = "+ err.stack);
logger.persist();
exitProcess()
exitProcessInstance()
}
}
}
catch (err) {
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runIndicatorBot -> err = "+ err.stack);
exitProcess()
exitProcessInstance()
}
}
function runTradingEngine(pBotConfig, pProcessConfig) {
global.TOTAL_PROCESS_INSTANCES_CREATED++
try {
const DEBUG_MODULE = require(ROOT_DIR + 'DebugLog');
let logger;
@@ -632,7 +636,7 @@
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runTradingEngine -> onInitializeReady -> whenStartFinishes -> Bot execution finished with errors. Please check the logs.");
logger.persist();
}
exitProcess()
exitProcessInstance()
}
} else {
@@ -640,47 +644,29 @@
logger.write(MODULE_NAME, "[ERROR] start -> findProcess -> runTradingEngine -> onInitializeReady -> Failed to initialize the bot. ");
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runTradingEngine -> onInitializeReady -> err = "+ err.stack);
logger.persist();
exitProcess()
exitProcessInstance()
}
}
}
catch (err) {
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> runTradingEngine -> err = "+ err.stack);
exitProcess()
exitProcessInstance()
}
}
}
catch (err) {
console.log(logDisplace + "Root : [ERROR] start -> findProcess -> err.message = " + err.message);
exitProcess()
exitProcessInstance()
}
}
function exitProcess() {
function exitProcessInstance() {
global.ENDED_PROCESSES_COUNTER++
console.log("[INFO] Task Server -> exitProcess -> Process #" + global.ENDED_PROCESSES_COUNTER + " exiting.");
console.log("[INFO] Task Server -> exitProcessInstance -> Process #" + global.ENDED_PROCESSES_COUNTER + " from " + global.TOTAL_PROCESS_INSTANCES_CREATED + ' exiting.');
if (global.ENDED_PROCESSES_COUNTER === global.USER_DEFINITION.bot.processes.length) {
/* Cleaning Before Exiting. */
clearInterval(global.HEARTBEAT_INTERVAL_HANDLER)
for (let i = 0; i < global.USER_DEFINITION.bot.processes.length; i++) {
let code = global.USER_DEFINITION.bot.processes[i].code
/* Delete the event handler for each process. */
let key = code.devTeam + "-" + code.codeName + "-" + code.process
let event = {
reason: 'Signal Received to Terminate this Process.'
}
global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated', event)
global.SYSTEM_EVENT_HANDLER.deleteEventHandler(key)
}
global.SYSTEM_EVENT_HANDLER.finalize()
global.SYSTEM_EVENT_HANDLER = undefined
console.log("[INFO] Task Server -> exitProcess -> Task Server Stopped.");
if (global.ENDED_PROCESSES_COUNTER === global.TOTAL_PROCESS_INSTANCES_CREATED) {
global.EXIT_NODE_PROCESS()
}
}
}
StatusReport.js
+5 -1
View File
@@ -58,8 +58,12 @@
if (owner.waitUntilNextUpdate === true) {
/* This forces this process to wait until the process that this one depends on, updates its status report. */
let extraCallerId = ''
if (pMonth) { extraCallerId = extraCallerId + '-' + pMonth }
if (pYear) { extraCallerId = extraCallerId + '-' + pYear }
let key = owner.devTeam + "-" + owner.bot + "-" + owner.process
let callerId = bot.devTeam + "-" + bot.codeName + "-" + bot.process
let callerId = bot.devTeam + "-" + bot.codeName + "-" + bot.process + extraCallerId
let subscriptionIdStatusReport
server.js
+39 -64
View File
@@ -84,7 +84,15 @@ global.WRITE_LOGS_TO_FILES = process.env.WRITE_LOGS_TO_FILES
global.EXCHANGE_NAME = process.env.EXCHANGE_NAME
global.MARKET = { assetA: 'USDT', assetB: 'BTC' }
global.CLONE_EXECUTOR = { codeName: 'AACloud', version: '1.1' }
/*
We need to count how many process instances we deployd and how many of them have already finished their job, either
because they just finished or because there was a request to stop the proceses. In this way, once we reach the
amount of instances started, we can safelly destroy the rest of the objects running and let this nodejs process die.
*/
global.ENDED_PROCESSES_COUNTER = 0
global.TOTAL_PROCESS_INSTANCES_CREATED = 0
/* Here we listen for the message to stop this Task / Process comming from the Task Manager, which is the paret of this node js process. */
process.on('message', message => {
@@ -92,9 +100,40 @@ process.on('message', message => {
process.env.STOP_GRACEFULLY = true;
/*
There are some process that might no be able to end grafully, for example the ones schedulle to process information in a future day or month.
In order to be sure that the process will be terminated, we schedulle one forced exit in 2 minutes from now.
*/
console.log('[INFO] Task Server -> server -> process.on -> Executing order received from Task Manager to Stop this Task. Nodejs process will be exited in less than 2 minutes.')
setTimeout(global.EXIT_NODE_PROCESS, 120000);
}
});
global.EXIT_NODE_PROCESS = function exitProcess() {
/* Cleaning Before Exiting. */
clearInterval(global.HEARTBEAT_INTERVAL_HANDLER)
for (let i = 0; i < global.USER_DEFINITION.bot.processes.length; i++) {
let code = global.USER_DEFINITION.bot.processes[i].code
/* Delete the event handler for each process. */
let key = code.devTeam + "-" + code.codeName + "-" + code.process
let event = {
reason: 'Signal Received to Terminate this Process.'
}
global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated', event)
global.SYSTEM_EVENT_HANDLER.deleteEventHandler(key)
}
global.SYSTEM_EVENT_HANDLER.finalize()
global.SYSTEM_EVENT_HANDLER = undefined
console.log("[INFO] Task Server -> exitProcessInstance -> Task Server Stopped.");
process.exit()
}
let notFirstSequence = false;
/* Setting up the global Event Handler */
@@ -293,73 +332,10 @@ async function readExecutionConfiguration(execution, processIndex) {
}
}
let startMode
// General Financial Being Configuration
global.DEV_TEAM = process.env.DEV_TEAM
global.CURRENT_BOT_REPO = process.env.BOT + "-" + process.env.TYPE + "-Bot"
if (process.env.TYPE === 'Trading' || process.env.TYPE === 'Trading-Engine') {
let live = {
run: 'false',
resumeExecution: execution.resumeExecution,
beginDatetime: new Date(global.DEFINITION.simulationParams.timestamp).toISOString(),
endDatetime: finalDatetime
}
let backtest = {
run: 'false',
resumeExecution: execution.resumeExecution,
beginDatetime: initialDatetime,
endDatetime: finalDatetime
}
let competition = {
run: 'false',
resumeExecution: execution.resumeExecution,
beginDatetime: initialDatetime,
endDatetime: finalDatetime
}
startMode = {
live: live,
backtest: backtest,
competition: competition
}
} else if (process.env.TYPE === 'Indicator' || process.env.TYPE === 'Sensor') {
let allMonths = {
run: "false",
minYear: process.env.MIN_YEAR,
maxYear: process.env.MAX_YEAR
}
let oneMonth = {
run: "false",
year: process.env.MIN_YEAR,
month: process.env.MONTH
}
let noTime = {
run: "false",
beginDatetime: process.env.BEGIN_DATE_TIME,
resumeExecution: execution.resumeExecution
}
let fixedInterval = {
run: "false",
interval: process.env.INTERVAL
}
startMode = {
allMonths: allMonths,
oneMonth: oneMonth,
noTime: noTime,
fixedInterval: fixedInterval
}
} else {
console.log("[ERROR] readExecutionConfiguration -> Bot Type is invalid.");
throw new Error("readExecutionConfiguration -> Bot Type is invalid.")
}
startMode[process.env.START_MODE].run = "true"
if (botProcess === undefined) { botProcess = process.env.PROCESS } // Only use the .env when nothing comes at Definition.json
let cloneToExecute = {
enabled: "true",
@@ -378,7 +354,6 @@ async function readExecutionConfiguration(execution, processIndex) {
global.EXECUTION_CONFIG = {
cloneToExecute: cloneToExecute,
startMode: startMode,
timePeriod: getTimePeriod(timePeriod),
timePeriodFileStorage: timePeriod,
timePeriodFilter: timePeriodFilter,