Atomically downgrade lock when committing tx to prevent deadlock

[coolreader18]

Description of Changes

Also, move the logic with committing and taking the subscription lock into module_subscription_actor. Best reviewed ignoring whitespace.

before/after:

[Shubham8287]

Ack it. Here is another visualisation focusing on main part

add_subcription:  begin_tx()                 subscription.write()
reducer1:      subscription.read()              begin_tx()
reducer2:                           begin_mut_tx()

So, the reducer 2 lock will not be acquired but It will block the queue (further reads) due to fairness.

[Shubham8287]

EventStatus::Commit name confused me while reveiwing, It may be better choice to name it EventStatus::YetToCommit or EventStatus::ReducerSuccess now

[Shubham8287]

Suggested change

	let subscriptions = self.subscriptions.read();
	// Take a read lock on `subscriptions` before committing tx
	// else it can result in subscriber receiving duplicate updates.
	let subscriptions = self.subscriptions.read();

Will prefer not to drop this comment.

[Shubham8287]

nit: I would recommend to use name WriteConflict because by defination WriteSkew anomoly is something transaction doesn't really detect while commiting unless transaction is tracking its reads.