No initial emit under locks in share(replay: 1, scope: .whileConnected)

[isaac-weisberg]

UPDATE2: unit tests have uncovered another unrelated issue #2655 . This PR provides fix for that too.

UPDATE: turns out, Subjects are not broken, the issue is 100% centralized in the ShareReplay1WhileConnected

This pull request aims to fix #2653.

The original issue stems from the fact that:

• ShareWhileConnected, ShareReplay1WhileConnected ReplaySubject, BehaviorSubject and PublishSubject emit their initial events upon subscription while holding onto their internal locks that protect their mutable state
• Their results can be forwarded to any operators
• These same operators might take decision to dispose of the upstream observables
• And these operators might do this while retaining their own locks
• And these operators might perform the handling of upstream events under the same locks

The last 3 points perfectly describe the behavior of a zip operator, which:

• will try to dispose of the source, while retaining its own lock
• will receive an initial event upon subscription from ShareReplay1WhileConnect or any Subject, with these sources retaining their locks
• will try to acquire its own lock to handle these events

In a multithreaded environment, these conditions can race, and this can produce a deadlock. This PR positions "emission upon subscription under lock" as the core issue and tries to fix specifically this behavior.

Special considerations:

• The current circumstances of the immediate emissions happening under lock might be both deliberate or accidental
• And with these changes, the sequentiality of these emissions will not be protected by the state lock anymore.
• Maybe I should add a separate lock to sequentialize specifically the emission of the initial elements? (as prophesied by @firecore in Locks during Event Forwarding in RxSwift #2525)
• I should say that didn't try hard enough to come up with a case in which the sequentialization of the emissions is absolutely necessary across threads.
• But I did ensure that at least the read out of the current state, and preparation of the values for the initial emission is synchronized.

Additionally:

@freak4pc I would like to say sorry for this comment. I really wasn't planning to add anything valuable to the conversation, and my behavior was really counter productive here. This wasn't a troll comment, it was just - I was amused by the fact that RxSwift can break, I found it funny.

@danielt1263 Also I would like to say thank you for your time and the fact that you responded to me.

I would also like to say sorry to you both for this comment - the vagueness of my answer actually permitted an interpretation where I am not only posting nothing of value, but also deliberately try and insult you or the framework. This wasn't my intention, and I'm sorry that I came off this way.

Fixes #2655
Fixes #2653

[freak4pc]

I'll review this later, but wanted to say I highly appreciate your PR, and the apology 🙌
And even more appreciate the fact you submitted a fix to the issue that bothered you. Thank you!

[freak4pc]

Could you add a unit tests that validates the fix against the issue you found?

[isaac-weisberg]

I have prepared tests to test the specific issue, although, now that I think of it, they could be implemented better:

• Tecnically, the original problem really only offended share(replay: 1, scope: .whileConnected). However, I wrote the same tests for other combinations of parameters for the share operator.
• At the same time, I deleted the implementation for share(replay: 0, ...) because it doesn't make sense semantically with my reproduction method of using zip + take(1) - they proper way to test these would be to: a) take multicast of PublishSubject; b) subscribe from multiple threads; c) connect once; d) count all the elements

But the main take away I guess is that it checks for the original issue, despite it intrinsically being impossible to happen.
So what's up, what are gonna be the next steps?