fix sending from forked container

[szysad]

Before submitting checklist

• Did you update the CHANGELOG? (not for test updates, internal changes/refactors or CI/CD setup)
• Did you ask the docs owner to review all the user-facing changes?

[codecov]

Codecov Report

Attention: Patch coverage is 95.00000% with 1 lines in your changes are missing coverage. Please review.

Project coverage is 74.92%. Comparing base (a93ebc2) to head (31e502e).
Report is 5 commits behind head on master.

Files	Patch %	Lines
...ion_processors/lazy_operation_processor_wrapper.py	95.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1692      +/-   ##
==========================================
- Coverage   80.98%   74.92%   -6.07%     
==========================================
  Files         304      304              
  Lines       15480    15180     -300     
==========================================
- Hits        12536    11373    -1163     
- Misses       2944     3807     +863     

Flag	Coverage Δ
e2e	?
e2e-management	?
e2e-s3	?
e2e-s3-gcs	?
e2e-standard	?
macos	74.76% <95.00%> (-6.03%)	⬇️
py3.10	?
py3.11	?
py3.12	?
py3.7	75.03% <95.00%> (-5.39%)	⬇️
py3.7.16	?
py3.8	?
py3.9	?
ubuntu	74.85% <95.00%> (-5.93%)	⬇️
unit	75.03% <95.00%> (-0.52%)	⬇️
windows	73.95% <95.00%> (-6.41%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[szysad]

❓ E2E

[Raalsky]

Isn't it simpler to set it as a part of the constructor?

[szysad]

its actually a bit tricky to do it in a constructor. This PR actually moves post_trigger_side_effect assigment from constructor to setter function. Previously we had

# self._op_processor was already defined
self._op_processor = LazyOperationProcessorWrapper(
                operation_processor_getter=partial(
                    get_operation_processor,
                    mode=self._mode,
                    container_id=self._id,
                    container_type=self.container_type,
                    backend=self._backend,
                    lock=self._lock,
                    flush_period=self._flush_period,
                    queue=self._signals_queue,
                ),
                post_trigger_side_effect=self._op_processor.start,
            )

so in line post_trigger_side_effect=self._op_processor.start
self._op_processor was pointing to already defined op_processor(ex. AsyncOperationProcessor).
It was also used by parent process (fork model) and therefore both processes would reference the same consumer thread which caused errors down the line. (ex. one process when finishing was forced to join on already finished thread)

Now in fork childself._op_processor points to LazyOperationProcessorWrapper and when triggered will create new AsyncOperationProcessor instance with new consumer thread.

[Raalsky]

You don't have to provide it as a parameter. You can set up it internally in the LazyOperationProcessor as:
self._post_trigger_side_effect=self.start

[Raalsky]

I know that you cannot provide it as a parameter because you don't have an instance created 😉

[szysad]

but self in this scope points to MetadataContainer not LazyOperationProcessorWrapper.
If it wasn't the case this code would work as expected previously

[szysad]

In theory I could define post_trigger_side_effect as function that takes one argument - self of LazyOperationProcessorWrapper and work my way around that but I think current impl is cleaner

[Raalsky]

I mean that you can set this property internally as a part of a constructor of LazyOperationProcessorWrapper like: self._post_trigger_side_effect=self.start. Think that it needs to be always called and it's an "atomic" operation that you have to initialize OperationProcessor and once it's created, run start on init. I've prepared some suggestions here: #1705