How to implement a resumable Table?

Question

Motivation (my sob stories)

Too many times I've run a notebook for hours and then Mathematica crashes (FE or Kernel, mostly FE) and things are lost. I've made various attempts at custom solutions from time to time by serializing things intermittently though persistent constructs like LocalObject or PersistentValue, etc. but these stores can get corrupted during crashes! I've never found a good general solution to this, so I'm asking for one.

Here's a concrete example: say you're doing a curl for every file in some directory and collecting 100s of megabytes (or a few gigabytes) of results:

fns = FileNames["*.jpg", "~/work/"];
results = Table[remoteFunction @ f, {f, fns}]; (* crashes 3 hours in *)

Instead, I want for a drop-in replacement Table, that looks like this:

results = SafeTable[remoteFunction @ f, {f, fns}, "Monitor" -> True,
     "ParallelKernels" -> 4, "Checkpoint" -> Quantity[10, "Seconds"]
] (* after the crash I re-run and it resumes, e.g. like "wget -c ..." *)

Now, how could this be implemented? Well, perhaps it works like a cronjob or god task. Maybe it creates an association of asynchronous tasks and dispatches them. Or perhaps instead it uses a collection of PersistentObjects and ParallelTable under-the-hood... At different scales (assume we're handling multiple gigabytes) I'm not sure what advantages there are from using LocalCache or LocalValue vs just MX dumping every-so-often.

Problem Statement

I'm looking for help implementing a drop-in replacement for Table that is resumable after a crash. This new function SafeTable acts exactly like a normal Table, but with efficient and safe storing of temporary results to disk, so that it can be easily resumed after the assured crash and kernel restart occurs.

Here's some code to get started with (although it's an analog for Map, not Table), ...and I'm not even sure if LocalSymbol is a good thing to use!

ClearAll @ SafeMap;
SafeMap[id_String, f_, list_] := Module[
{  idx, status, len = Length @ list, b = Internal`Bag[], 
   jobDone = StringJoin[id, "_done"], jobIter = StringJoin[id, "_iter"],
   jobResult = StringJoin[id, "_result"] },
   status = LocalSymbol[jobDone];
   Switch[
       status, _LocalSymbol,
       Print @ "Start anew";
          LocalSymbol[jobDone] = False;
          idx = LocalSymbol[jobIter] = 0;
          LocalSymbol[jobResult] = b
       , True, Print @ "Already finished";
          b = LocalSymbol[jobResult];
          Return[Internal`BagPart[b, All], Module];
       , False, 
          idx = LocalSymbol[jobIter];
          Print["Resuming at index ", idx];
          b = LocalSymbol[jobResult]; 
     ];

  With[{s = LocalSymbol[jobIter]+1},
   Table[
        With[{x=f[list[[j]]]},
         Internal`StuffBag[b, x]
         ];
        LocalSymbol[jobIter] = j;
        LocalSymbol[jobResult] = b;
        If[j==Length @ list,
           LocalSymbol[jobDone] = True;
           Return[Internal`BagPart[LocalSymbol[jobResult], All], Module]
         ];
         Nothing
     , {j, s, len}
    ]
  ]
]

You give it a seed and if it crashes, re-evaluate the definition and re-execute and it will resume:

SafeMap["task_abcd", (Pause[1]; RandomChoice[{Quit, 1, 2}][]; #) &, 
 Range@5]

So this sort of works for small tasks, but is ugly, sequential, slow (doesn't save incremental chunks), and messy (with creating local symbols which don't clean themselves up). Moreover, using string seeds seems weird to me but I'm not sure what would be a better design. Anyhow, a cleaner implementation that is faster, parallelizable, and monitored would greatly help restore some of my sanity!

Updates

• To give some idea of the headaches I'm having - I experience frontend crashes at least 5 times per day. When I'm doing a complicated workload, e.g. running code for more than an hour, about 50% of the time MMA crashes out and I lose kernel data and notebook changes. This is killing my soul.
• @Roman gives a nice technique in his answer that uses memoization, but what I'm after is something that has similar semantics to Table - a construct that can be easily re-applied everywhere. Also, for longer lists with elements of larger sizes, saving every single element in separate MX files might be a very inefficient granularity, there's overhead manually changing names and directories, things quickly get messy trying to re-use this technique many times. Typically you don't want to pollute your NotebookDirectory, so you would use CreateDirectory but when the FE crashes, this path is lost. Also having millions of downvalues might not be optimal.

Answer 1

The function that does the actual calculation:

fcalc[n_] := n^2

make file names:

SetDirectory[NotebookDirectory[]];
makeFileName[n_] := "cachefile_" <> ToString[n] <> ".wl"

The function that you call from the rest of the code:

Clear[f];
f[n_] := f[n] = Module[{name, F},
  (* is there a cached version? *)
  name = makeFileName[n];
  If[FileExistsQ[name], Return[Get[name]]];
  (* if not, calculate *)
  Echo[n, "calc"];
  F = fcalc[n];
  Save[name, F];
  F]

Now you can do

Table[f[n], {n, 3}]

and if there's a crash you can manually debug the saved files.

---

new version, closer to new spec

A function that evaluates func[n] if the result isn't available in a file yet:

evalsave[id_String, func_Function, n_Integer] := 
  Module[{filename, F},
    (* construct file name *)
    filename = id <> ToString[n] <> ".wl";
    (* is there a cached version? *)
    If[FileExistsQ[filename], Return[Get[filename]]];
    (* if not, calculate *)
    F = func[n];
    Save[filename, F];
    F]

Map the function func onto a list of n-values, and show progress:

safeMap[id_String, func_Function, nlist_List] := Module[{progress}, 
  Monitor[
    MapIndexed[(progress = {##}; evalsave[id, func, #1]) &, nlist], 
    Row[{id <> ": ", progress[[1]], ProgressIndicator[(progress[[2,1]]-1)/Length[nlist]]}]]]

Try it out: this can be restarted and will pick off where the last call quit:

SetDirectory["/tmp"];
safeMap["task_abcd",
        (Pause[1]; RandomChoice[{Quit, 1, 2}][]; #) &, 
        Range@5]

If you're worried about having millions of files bogging down the computer, then you could switch to a better file system where this is not a problem, see for example here.

parallelization

The above code can be parallelized without problem:

parallelSafeMap[id_String, func_Function, nlist_List] :=
  ParallelMap[evalsave[id, func, #] &, nlist]
(* maybe experiment with Method option of ParallelMap *)

SetDirectory["/tmp"];
parallelSafeMap["task_abcd",
                (Pause[1]; RandomChoice[{Quit, 1, 2}][]; $KernelID) &,
                Range@20]

Parallelization has the advantage that a crashed kernel will be auto-restarted and the crashed evaluation auto-requeued. Running parallelSafeMap is therefore very easy and hands-off. You can see in the above results the $KernelID values that increase every time a kernel crashes and is restarted. If you'd rather not see all the warnings that this generates, run

Off[LinkObject::linkd, Kernels::rdead, Parallel`Developer`QueueRun::req, LaunchKernels::clone];

before doing the calculation.

As for progress indication, it's a bit harder to do in a parallel setting. Maybe you can use the file system for this task: check how many files have been produced. Otherwise, there are SE questions dealing specifically with this topic: 1548 153333