Manage previous outdated running jobs.

Author: T4rk1n

dash/_callback.py

@@ -339,6 +339,7 @@ def add_context(*args, **kwargs):
                 progress_outputs = long.get("progress")
                 cache_key = flask.request.args.get("cacheKey")
                 job_id = flask.request.args.get("job")
+                old_job = flask.request.args.getlist("oldJob")
 
                 current_key = callback_manager.build_cache_key(
                     func,
@@ -347,6 +348,10 @@ def add_context(*args, **kwargs):
                     long.get("cache_args_to_ignore", []),
                 )
 
+                if old_job:
+                    for job in old_job:
+                        callback_manager.terminate_job(job)
+
                 if not cache_key:
                     cache_key = current_key
 

dash/dash-renderer/src/actions/callbacks.ts

@@ -93,6 +93,7 @@ const updateResourceUsage = createAction('UPDATE_RESOURCE_USAGE');
 
 const addCallbackJob = createAction('ADD_CALLBACK_JOB');
 const removeCallbackJob = createAction('REMOVE_CALLBACK_JOB');
+const setCallbackJobOutdated = createAction('CALLBACK_JOB_OUTDATED');
 
 function unwrapIfNotMulti(
     paths: any,
@@ -333,8 +334,10 @@ function handleServerside(
     config: any,
     payload: any,
     paths: any,
-    long?: LongCallbackInfo,
-    additionalArgs?: [string, string][]
+    long: LongCallbackInfo | undefined,
+    additionalArgs: [string, string, boolean?][] | undefined,
+    getState: any,
+    output: string
 ): Promise<CallbackResponse> {
     if (hooks.request_pre) {
         hooks.request_pre(payload);
@@ -346,6 +349,7 @@ function handleServerside(
     let job: string;
     let runningOff: any;
     let progressDefault: any;
+    let moreArgs = additionalArgs;
 
     const fetchCallback = () => {
         const headers = getCSRFHeader() as any;
@@ -365,8 +369,9 @@ function handleServerside(
             addArg('job', job);
         }
 
-        if (additionalArgs) {
-            additionalArgs.forEach(([key, value]) => addArg(key, value));
+        if (moreArgs) {
+            moreArgs.forEach(([key, value]) => addArg(key, value));
+            moreArgs = moreArgs.filter(([_, __, single]) => !single);
         }
 
         return fetch(
@@ -383,6 +388,14 @@ function handleServerside(
         const handleOutput = (res: any) => {
             const {status} = res;
 
+            if (job) {
+                const callbackJob = getState().callbackJobs[job];
+                if (callbackJob?.outdated) {
+                    dispatch(removeCallbackJob({jobId: job}));
+                    return resolve({});
+                }
+            }
+
             function recordProfile(result: any) {
                 if (config.ui) {
                     // Callback profiling - only relevant if we're showing the debug ui
@@ -462,7 +475,8 @@ function handleServerside(
                             jobId: data.job,
                             cacheKey: data.cacheKey as string,
                             cancelInputs: data.cancel,
-                            progressDefault: data.progressDefault
+                            progressDefault: data.progressDefault,
+                            output
                         };
                         dispatch(addCallbackJob(jobInfo));
                         job = data.job;
@@ -635,9 +649,19 @@ export function executeCallback(
                 let newHeaders: Record<string, string> | null = null;
                 let lastError: any;
 
-                const additionalArgs: [string, string][] = [];
+                const additionalArgs: [string, string, boolean?][] = [];
+                console.log(cb.callback.output, getState().callbackJobs);
                 values(getState().callbackJobs).forEach(
                     (job: CallbackJobPayload) => {
+                        if (cb.callback.output === job.output) {
+                            // Terminate the old jobs that are not completed
+                            // set as outdated for the callback promise to
+                            // resolve and remove after.
+                            additionalArgs.push(['oldJob', job.jobId, true]);
+                            dispatch(
+                                setCallbackJobOutdated({jobId: job.jobId})
+                            );
+                        }
                         if (!job.cancelInputs) {
                             return;
                         }
@@ -667,7 +691,9 @@ export function executeCallback(
                             payload,
                             paths,
                             long,
-                            additionalArgs.length ? additionalArgs : undefined
+                            additionalArgs.length ? additionalArgs : undefined,
+                            getState,
+                            cb.callback.output
                         );
 
                         if (newHeaders) {

dash/dash-renderer/src/reducers/callbackJobs.ts

@@ -1,4 +1,4 @@
-import {assoc, dissoc} from 'ramda';
+import {assoc, assocPath, dissoc} from 'ramda';
 import {ICallbackProperty} from '../types/callbacks';
 
 type CallbackJobState = {[k: string]: CallbackJobPayload};
@@ -8,17 +8,21 @@ export type CallbackJobPayload = {
     cacheKey: string;
     jobId: string;
     progressDefault?: any;
+    output?: string;
+    outdated?: boolean;
 };
 
 type CallbackJobAction = {
-    type: 'ADD_CALLBACK_JOB' | 'REMOVE_CALLBACK_JOB';
+    type: 'ADD_CALLBACK_JOB' | 'REMOVE_CALLBACK_JOB' | 'CALLBACK_JOB_OUTDATED';
     payload: CallbackJobPayload;
 };
 
 const setJob = (job: CallbackJobPayload, state: CallbackJobState) =>
     assoc(job.jobId, job, state);
 const removeJob = (jobId: string, state: CallbackJobState) =>
     dissoc(jobId, state);
+const setOutdated = (jobId: string, state: CallbackJobState) =>
+    assocPath([jobId, 'outdated'], true, state);
 
 export default function (
     state: CallbackJobState = {},
@@ -29,6 +33,8 @@ export default function (
             return setJob(action.payload, state);
         case 'REMOVE_CALLBACK_JOB':
             return removeJob(action.payload.jobId, state);
+        case 'CALLBACK_JOB_OUTDATED':
+            return setOutdated(action.payload.jobId, state);
         default:
             return state;
     }