Resolve increased time.

[tchaton]

What does this PR do?

There were a race condition. The .put takes a bit of time to trigger and the state sent was the entire state. A deepcopy is necessary.

Fixes #14070

# With this PR:

3478 last run took (ms): 52.05106735229492 89496 1296
3479 last run took (ms): 17.143964767456055 89560 1296
3480 last run took (ms): 55.02200126647949 89624 1296
3481 last run took (ms): 55.60493469238281 89624 1296
3482 last run took (ms): 55.991172790527344 89624 1296
3483 last run took (ms): 55.35006523132324 89624 1296
3484 last run took (ms): 55.64689636230469 90480 1296
3485 last run took (ms): 18.70274543762207 90480 1296
3486 last run took (ms): 60.84728240966797 90480 1296
3487 last run took (ms): 59.233903884887695 90480 1296
3488 last run took (ms): 54.7330379486084 90480 1296
3489 last run took (ms): 55.959224700927734 90480 1296
3490 last run took (ms): 174.79276657104492 88704 1296
3491 last run took (ms): 58.320045471191406 88704 1296
3492 last run took (ms): 57.69681930541992 88704 1296
3493 last run took (ms): 54.07524108886719 88704 1296
3494 last run took (ms): 55.40585517883301 88704 1296
3495 last run took (ms): 57.137250900268555 88704 1296
3496 last run took (ms): 54.77190017700195 89560 1296
3497 last run took (ms): 57.33180046081543 88704 1296
3498 last run took (ms): 55.570125579833984 88704 1296
3499 last run took (ms): 55.64689636230469 88704 1296
3500 last run took (ms): 55.315256118774414 88704 1296
3501 last run took (ms): 56.95080757141113 88704 1296
3502 last run took (ms): 54.476022720336914 88704 1296
3503 last run took (ms): 56.0910701751709 88704 1296
3504 last run took (ms): 66.6959285736084 87848 1296
3505 last run took (ms): 65.20390510559082 87848 1296
3506 last run took (ms): 82.81111717224121 87848 1296

import os
import time

from pympler import asizeof

import lightning as L
from lightning_app.utilities.enum import WorkStageStatus


class Work(L.LightningWork):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.seq = None

    def run(self, seq: int, *args, **kwargs):
        self.seq = seq
        time.sleep(int(os.getenv("WORK_DURATION", 0)))


class Flow(L.LightningFlow):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.work = Work(parallel=True, cache_calls=True)
        self.counter = 1
        self.seq = None
        self.starttime = None

    def run(self):

        if self.work.seq == self.seq:
            now = time.time()

            if self.starttime is not None:
                latest_call_hash = self.work._calls["latest_call_hash"]
                send_state = {k: v for k, v in self.work._calls.items() if k in (latest_call_hash, "latest_call_hash")}
                print(
                    self.counter,
                    "last run took (ms):",
                    (now - self.starttime) * 1000,
                    asizeof.asizeof(self.work.state),
                    asizeof.asizeof(send_state),
                )

            self.work.run(self.counter)
            self.seq = self.counter
            self.starttime = now
            self.counter += 1


if __name__ == "__main__":
    app = L.LightningApp(Flow())

Does your PR introduce any breaking changes? If yes, please list them.

Before submitting

• Was this discussed/approved via a GitHub issue? (not for typos and docs)
• Did you read the contributor guideline, Pull Request section?
• Did you make sure your PR does only one thing, instead of bundling different changes together?
• Did you make sure to update the documentation with your changes? (if necessary)
• Did you write any new necessary tests? (not for typos and docs)
• Did you verify new and existing tests pass locally with your changes?
• Did you list all the breaking changes introduced by this pull request?
• Did you update the CHANGELOG? (not for typos, docs, test updates, or minor internal changes/refactors)

PR review

Anyone in the community is welcome to review the PR.
Before you start reviewing, make sure you have read the review guidelines. In short, see the following bullet-list:

• Is this pull request ready for review? (if not, please submit in draft mode)
• Check that all items from Before submitting are resolved
• Make sure the title is self-explanatory and the description concisely explains the PR
• Add labels and milestones (and optionally projects) to the PR so it can be classified

Did you have fun?

Make sure you had fun coding 🙃

cc @Borda

[hhsecond]

How difficult is to add a test for this one?