example workflow with high memory use

[oliver-sanders]

We have occasionally observed workflows gradually accumulating large quantities of memory.

Reproducible Example

Working with a recent workflow, Dave has managed to extract a reproducible example:

[task parameters]
    origins = ADD, ARN, ATH, ATL, AUH, BAH, BCN, BDA, BER, BEY, BGI, BKK, BLQ, BLR, BOG, BOM, BOS, CAI, CAN, CLT, CMB, CMN, CPH, CPT, CTU, DAC, DEL, DEN, DFW, DOH, DTW, DXB, EWR, EZE, FCO, FNC, GIB, GOT, GRU, GYD, HAN, HEL, HKG, HND, IAD, IAH, ICN, IST, JED, JFK, JNB, KBP, KEF, KUL, KWI, LAX, LIM, LIN, LIS, LOS, MAD, MEX, MIA, MLA, MNL, MUC, NBO, NCE, NQZ, ORD, OSL, OTP, PDL, PEK, PER, PHL, PHX, PRG, PVG, RUH, SCL, SEA, SFO, SIN, SVO, TIA, TLV, TPE, VIE, WAW, YHZ, YUL, YYC, YYZ, ZAG
    mogreps_timesteps = 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96, 102, 108, 114, 120, 126, 132, 138
    ecmwf_timesteps = 144, 150, 156, 162, 168, 174, 180, 186, 192, 198, 204, 210, 216, 222, 228, 234, 240, 246, 252, 258, 264
    ecmwf_resil_timesteps = 270, 276, 282, 288
    [[templates]]
        origins = _%(origins)s
        mogreps_timesteps = _t+%(mogreps_timesteps)03d
        ecmwf_timesteps = _t+%(ecmwf_timesteps)03d
        ecmwf_resil_timesteps = _t+%(ecmwf_resil_timesteps)03d
[scheduling]
    initial cycle point = previous(T00)
    runahead limit = P0
    [[queues]]
        [[[default]]]
           limit = 5
    [[graph]]
        PT24H = '''
                natseg_res_create => nats_start_00 => nats_start => nats_run_astar_mogreps<origins, mogreps_timesteps>
                nats_start => nats_prepare_ecmwf
                nats_prepare_ecmwf => nats_run_astar_ecmwf<origins, ecmwf_timesteps>
                nats_run_astar_ecmwf<origins, ecmwf_timesteps> & nats_run_astar_mogreps<origins, mogreps_timesteps> => nats_genxml
                nats_run_astar_ecmwf<origins, ecmwf_timesteps> => nats_run_astar_ecmwf_resil<origins, ecmwf_resil_timesteps>
                nats_run_astar_mogreps<origins, mogreps_timesteps> & nats_run_astar_ecmwf_resil<origins, ecmwf_resil_timesteps> => natseg_res_delete
                nats_genxml & nats_run_astar_ecmwf_resil<origins, ecmwf_resil_timesteps> => nats_preprocess_for_archive
                nats_preprocess_for_archive => nats_archiving
                nats_archiving & housekeep[-P1D] => housekeep
                housekeep => archive_logs
                '''
[runtime]
    [[root]]
        run mode = skip
    [[nats_archiving]]
    [[nats_genxml]]
    [[housekeep]]
    [[nats_prepare_ecmwf]]
    [[nats_preprocess_for_archive]]
    [[nats_run_astar_set<origins>]]
    [[nats_run_astar_ecmwf<origins, ecmwf_timesteps>]]
        inherit = nats_run_astar_set<origins>
    [[nats_run_astar_ecmwf_resil<origins, ecmwf_resil_timesteps>]]
        inherit = nats_run_astar_set<origins>
    [[nats_run_astar_mogreps<origins, mogreps_timesteps>]]
        inherit = nats_run_astar_set<origins>
    [[nats_start]]
    [[nats_start_00]]
    [[natseg_res_create]]
    [[natseg_res_delete]]
    [[archive_logs]]

Note, this workflow contains many-to-many triggers so the number of edges is extreme. It is likely these are triggering the issue (pun intended).

Example Results:

Example memory trace:

[oliver-sanders]

A funky pattern to this memory trace:

Those two initial memory drops can also be seen in the image on the OP. From this point on, it's gradual memory accumulation all the way...

[oliver-sanders]

It looks like this may be related to data store / deltas:

Memray is pretty clear about where it thinks the memory is going (apply_deltas):

And tracemalloc identifies this particular line within the apply_delta method allocating ~1MB of RAM with each sample (PT1M sampling):

cylc-flow/cylc/flow/data_store_mgr.py

Line 418 in 5fff43f

data[key].update({e.id: e for e in delta.added})

  /home/users/oliver.sanders/cylc-flow/cylc/flow/data_store_mgr.py:418: size=3765 KiB (+1115 B), count=29010 (+10), average=133 B
  /home/users/oliver.sanders/cylc-flow/cylc/flow/data_store_mgr.py:418: size=3766 KiB (+1115 B), count=29020 (+10), average=133 B
  /home/users/oliver.sanders/cylc-flow/cylc/flow/data_store_mgr.py:418: size=3767 KiB (+1115 B), count=29030 (+10), average=133 B
  /home/users/oliver.sanders/cylc-flow/cylc/flow/data_store_mgr.py:418: size=3768 KiB (+1115 B), count=29040 (+10), average=133 B

So presumably, this means the data store itself is using that memory? Which isn't showing with ordinary profiling tools because it's protobuf allocated memory?

[dpmatthews]

Now that I've done some more testing, it looks like most (all?) of the memory is freed when a cycle completes. This example workflow runs 4285 tasks per cycle, limited to a single cycle by the runahead limit. Running in skip mode I'm seeing memory peak at 1.9G at the end of a cycle but then drop to ~480M at the beginning of the next cycle. This appears to have been consistent over several cycles.

Replacing run mode = skip with script = sleep 5 (i.e. running local tasks rather than skipping) I've managed to complete one cycle so far. The memory peaked at 5.9G and then dropped to ~420M at the beginning of the second cycle.

[dwsutherland]

data[key].update({e.id: e for e in delta.added})

That is just a straight python dictionary of all non-workflow protobuf objects (i.e. PbEdge, PbTask, PbTaskProxy), the accumulation of which will increase memory until they're pruned...
Updating these short lived objects will also incur a memory growth protobuf penalty until they are pruned.. Long lived protobuf objects are reset on update for this reason. (see #6727)

There also was that fast job memory accumulation problem I never got to the bottom of.. (see #6751 )

Does the problem exist to the same extent running at n=0 window size? (just to try rule things out)

[dpmatthews]

Using the python protobuf implementation (export PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python) avoids the high memory usage (but is slow and uses more CPU):

• Running in skip mode I got a memory peak of ~530 M (down from 1.9G) at the end of a cycle (with only a small drop at the beginning of the next cycle).
• Running with local tasks I got a memory peak of ~560 M (down from 5.9G) at the end of a cycle.

[dwsutherland]

Using the python protobuf implementation

Interesting.. this suggests the build up is in the protobuf objects

The memory peaked at 5.9G and then dropped to ~420M at the beginning of the second cycle.

And this suggests it's an accumulation in the "short-lived" objects (i.e. w/e a delta happens, i.e. task or family proxy change of state/state-totals)

I'll run a few more scenario.. at present I'm able to reproduce the memory build-up:

(flow-mem) sutherlander@cortex-hyper:cylc-flow$ free -h
               total        used        free      shared  buff/cache   available
Mem:            11Gi       2.6Gi       7.2Gi        81Mi       2.2Gi       9.1Gi
Swap:          4.0Gi          0B       4.0Gi

~15min later

(flow-mem) sutherlander@cortex-hyper:cylc-flow$ free -h
               total        used        free      shared  buff/cache   available
Mem:            11Gi       6.1Gi       2.8Gi        80Mi       3.0Gi       5.6Gi
Swap:          4.0Gi          0B       4.0Gi

(using 5sec script, n=1, UIS & WUI tree view)

Will try with 5sec, n=0, UIS&WUI next... Which should also indicate whether it's in the "short lived" objects...
(quotes on "short-lived" because with the example workflow not a lot is pruned within the one cycle, and the cycle lasts a long time with many deltas against some of these)

Then I'll try resetting proto objects that receive any delta (like we do currently with Workflow).. I suspect it's PbFamilyProxy

[dwsutherland]

Yes, so I was able to produce the same:

The big jump back down is the end of the first cycle when it gets pruned, I started the second cycle with n=0, so it appears to be family objects..

I will test resetting PbFamilyProxy because it would accumulate just like this throughout a single cycle, then report back
(should be an easy fix... but we'll see)

[dwsutherland]

So WORKFLOW, TASK, FAMILY are the current reset, with FAMILY_PROXIES in addition:

with TASK_PROXIES, FAMILY_PROXIES, JOBS:

(changed the queue size to 10 to get through it quicker)

[dwsutherland]

With TASK_PROXIES (in addition):

It makes sense because of the number of tasks and updates with prerequisites, states, outputs ... etc

I think I'll go for WORKFLOW, TASK, FAMILY, TASK_PROXIES, FAMILY_PROXIES as the types to reset on update... Which is:

def reset_protobuf_object(msg_class, msg_orig):
    """Reset object to clear memory build-up."""
    # See: https://github.com/protocolbuffers/protobuf/issues/19674
    # The new message instantiation needs happen on a separate line.
    new_msg = msg_class()
    new_msg.CopyFrom(msg_orig)
    return new_msg

This too will add a little CPU.. But isn't blocking, and only with deltas (although I could profile)..

RESET_PROTOBUF_TYPES = {
    TASKS,
    TASK_PROXIES,
    FAMILIES,
    FAMILY_PROXIES,
    WORKFLOW,
}

(still a touch upward on the resident, but not heap)