Reuse ThreadsafeFunction in EventQueue #739

indutny · 2021-05-18T23:28:47Z

Node.js optimizes subsequent ThreadsafeFunction invocations to happen
during the same event loop tick, but only if the same instance of
ThreadsafeFunction is used. The performance improvement is most
noticeable when used in Electron, because scheduling a new UV tick in
Electron is very costly.

With this change EventQueue will use an
existing instance of ThreadsafeTrampoline (wrapper around
ThreadsafeFunction) if compiled with napi-6 feature, or it will fallback
to creating a new ThreadsafeFunction per EventQueue instance.

Fix: #727

indutny · 2021-05-18T23:33:11Z

~~Converted to draft because of the reference counting issues. Will keep you posted.~~

Update: it is fixed now!

indutny-signal · 2021-05-19T01:11:45Z

(Force pushed to fix lint errors)

kjvalencik · 2021-05-19T12:46:00Z

I left a comment here mentioning a couple of my thoughts: #727 (comment)

indutny-signal · 2021-05-19T16:02:41Z

Replied.

src/event/event_queue.rs

jrose-signal

This is pretty much the implementation I was envisioning, yep. :-) I like the compromise of having cx.queue() return a shared queue but still allowing EventQueue::new to create a new one.

I'll leave organizational and documentation feedback to the Neon maintainers. (In particular, I wonder if trampoline.rs should be a submodule of event_queue, since it's not going to be reused for anything else.)

jrose-signal · 2021-05-19T19:29:48Z

src/trampoline.rs

+        if self.ref_count == 0 {
+            return;
+        }


This should probably panic, or however else Neon detects internal errors.

jrose-signal · 2021-05-19T19:33:04Z

src/lifecycle.rs

+    pub(crate) fn threadsafe_trampoline<'a, C: Context<'a>>(
+        cx: &mut C,
+    ) -> Arc<RwLock<ThreadsafeTrampoline>> {
+        Arc::clone(&InstanceData::get(cx).threadsafe_trampoline)


Nitpick: it might be nice to lazily initialize this for programs that don't use EventQueues.

It sounds like in order to that I'll have to store it in an Arc<RwLock<Option<Arc<RwLock<ThreadsafeTrampoline>>>>> which doesn't look to nice to me. Is there any better way to do this?

You have an exclusive (&mut) reference to InstanceData, so you don't need extra locking. Option<Arc<_>> is sufficient.

jrose-signal · 2021-05-19T19:34:49Z

src/trampoline.rs

+
+pub(crate) struct ThreadsafeTrampoline {
+    tsfn: ThreadsafeFunction<Callback>,
+    ref_count: u32,


Nitpick: if this were AtomicU32, you could avoid the lock. (You'd be able to use Relaxed operations safely because you're only checking the refcount, not relying on any other data to have been synchronized.)

Sadly, the lock is needed because .reference()/.unref() calls need mutable access to the ThreadsafeFunction.

Since they require &mut Context, it implies the operations are serialized. We could get away with a Cell instead.

But, that might be an optimization better left for when it's demonstrated to be a performance bottleneck.

Yeah, I don't imagine this being a bottleneck, and it is way too simple to do it with a mutex. I'm open to changing it, though.

jrose-signal · 2021-05-19T19:37:14Z

src/trampoline.rs

+    /// _Not idempotent_
+    pub(crate) fn reference(&mut self, env: Env) {


Nitpick: Maybe we should make this more obviously distinct from the idempotent nature of ref by calling it increment_reference_count or something.

jrose-signal · 2021-05-19T19:37:50Z

src/trampoline.rs

+        }
+    }
+
+    // Monomorphized trampoline funciton for calling the user provided closure


Typo: function

jrose-signal · 2021-05-19T19:45:46Z

src/event/event_queue.rs

-            tsfn,
-            has_ref: true,
-        }
+        Self::with_trampoline(cx, trampoline)


new returns a referenced trampoline, but with_trampoline expects an unreferenced one. I think the best answer here is to have with_trampoline take an already-referenced trampoline, which is consistent with fresh ones and not hard to implement for the shared one. Having it take a not-already-referenced one and immediately referencing it would also work, but would mean extra calls into Node for the fresh case.

In that case it should probably be called with_referenced_trampoline or something so people don't forget.

indutny-signal · 2021-05-19T20:14:33Z

@jrose-signal I believe I've addressed your feedback except for one suggestion. PTAL

jrose-signal

👍

jrose-signal · 2021-05-19T20:42:24Z

src/context/mod.rs

+        let shared_trampoline = {
+            let shared_trampoline = ThreadsafeTrampoline::new(self.env());
+            shared_trampoline.unref(self.env().to_raw());
+            shared_trampoline
+        };


This is still a little less efficient but it's harder to get wrong, so maybe that's worth it.

src/trampoline.rs

indutny-signal · 2021-05-20T00:30:16Z

The latest commit on this PR actually fixes a bug, but it doesn't look like it could be reproduced without the proposed changes so I'm not sure if we should split this commit into a separate PR or handle here as well. Any suggestions are welcome!

kjvalencik · 2021-05-20T17:47:57Z

That last bug looks interesting! It seems like it might be a bug in Node/N-API. I'll see if I can make a minimal non-Rust example and open an issue upstream.

indutny · 2021-05-20T18:32:19Z

Opened a separate PR to track it: #744

indutny-signal · 2021-05-20T18:38:22Z

crates/neon-runtime/src/napi/tsfn.rs

@@ -167,6 +179,12 @@ impl<T: Send + 'static> ThreadsafeFunction<T> {

 impl<T> Drop for ThreadsafeFunction<T> {
    fn drop(&mut self) {
+        // tsfn was already finalized by `Environment::CleanupHandles()` in
+        // Node.js
+        if *self.is_finalized.lock().unwrap() {


I'm a bit worried about the cases where we might end up running release concurrently to finalize callback. Maybe this has to be fixed in Node.js instead...

kjvalencik · 2021-06-01T21:43:37Z

I like the way this API is headed. One thing that I'm still not sure about is the asymmetry on EventQueue you get from EventQueue::new and ones you get from cx.queue(). The later shares a backing queue and the former doesn't, but it's not visible from the type signature and isn't super clear from usage.

Any ideas of how to make this more clear? I have a couple thoughts:

EventQueue could implement Clone in a way that shares the backing queue. This could be a way for users to leverage sharing outside of the global queue and maybe make things feel more symmetrical.
Introduce a new type for sharing. This is what tokio does, but increasing the API surface area might be confusing.

indutny-signal · 2021-06-01T21:44:25Z

@kjvalencik I like the Clone idea! Let me see if I can concoct something later today.

kjvalencik · 2021-06-01T21:46:26Z

@indutny Don't feel the need to jump on an implementation. Right now I'm trying to get a discussion going. I don't have a clear picture of what the API should be.

But, I also don't want to bikeshed too long because this optimization is very cool! 🚲 🏠

indutny-signal · 2021-06-01T22:04:49Z

To confirm, nothing actionable for me right now? If yes - what's the plan for this PR?

kjvalencik · 2021-06-01T22:54:41Z

Correct, nothing actionable. I'm going to think about the API a bit more and talk with @dherman while working on getting the other changes released.

Thanks!

dherman · 2021-06-02T06:31:20Z

I like the idea of using Clone as a way to connect sharing to a user-visible operation, which then makes it a little easier to explain that the cx method always returns a clone of the same shared queue.

In #734 we experimented with renaming EventQueue to Channel, which maybe also helps here. EventQueue is a little mysterious since the queue is internal and the user of the API isn't actually directly interacting with it. I like the idea that you can create new channels or you can clone an existing one to share its internal queue, and cx.channel() gives you a cloned channel so in the common case you get the sharing by default.

dherman · 2021-06-02T16:30:03Z

One more thought: I think just the one Channel type is sufficient even with cloning (as opposed to one type for the data structure that owns the queue and another type for the shared references). It's reminiscent of Rc and Arc, where you have a single type with interesting connections under the hood between the original and its clones, but the user doesn't generally need to worry about the difference; they all act as the same type.

kjvalencik · 2021-06-03T16:36:30Z

@indutny Thanks again so much for this PR. I spent a good amount of time thinking about this and also discussing it with @dherman.

For both of us, what makes this all fit is impl Clone for EventQueue. This makes it semantically identical to things in Rust std that share a backing store when cloned (e.g., std::sync::Arc, std::mpsc::Sender). I think it also makes cx.queue() easy to understand--it's a clone of the globally cached queue.

There are a few changes I would like to make to this PR:

We renamed EventQueue to Channel (with deprecated aliases, to minimize the user impact). This (hopefully) more clearly illustrates what it is. This PR needs to be updated to use Channel everywhere.
Implement Clone for Channel. This can't be derived, but should be fairly straight forward. Clone the fields and increment the reference count (if and only if the channel being cloned is referenced).
Replace the cx.channel() implementation with a clone() + reference(&mut cx) of the InstanceData queue.
Remove the ThreadsafeTrampoline abstraction. In my opinion, this makes things more complicated than necessary. It's a little easier to understand with the functionality inlined into Channel. Possibly with a ChannelState helper struct to store the data inside the RwLock.
Add some additional documentation explaining the new behavior
Document that if needed for performance the RwLock may be replaced with an AtomicUsize and a small amount of unsafe since &mut cx already enforces the necessary invariants.

I know this is asking a lot and you have already been incredibly generous with your time, so I am happy to take over this PR if you would like me to, but I can also help guide it through if you prefer.

Cheers!

indutny-signal · 2021-06-03T18:10:54Z

@kjvalencik I'll happily look into it. Will let you know how it goes!

src/event/event_queue.rs

kjvalencik · 2021-06-07T15:06:49Z

I performed some benchmarks with this and unfortunately it performs worse in most scenarios and only marginally better in the ideal situation. (I didn't test in Electron, but I suspect it's still a large improvement there because of the very high cost of synchronizing event loops at each tick.)

Poking around a bit, I think the primary issue is acquiring a lock to perform Channel::try_send contending with the lock in Drop. The lock in Channel::try_send shouldn't be necessary because we are guaranteed the channel won't be dropped (we're holding a reference to it) and sending is Sync.

Unfortunately, I think the best fix is a more complicated solution that uses an atomic for reference counting instead of a RwLock. It will share a very similar design to the internals of Arc. It would be nice if we could simply use an Arc over a sentinel with a clever Drop implementation, but I haven't been able to think of a way to do that.

jrose-signal · 2021-06-08T17:05:54Z

src/event/event_queue.rs

+}
+
+impl ChannelState {
+    fn reference<'a, C: Context<'a>>(self: &Self, cx: &mut C) {


Nitpick: self: &Self is idiomatically written &self.

src/context/mod.rs

src/event/event_queue.rs

indutny marked this pull request as draft May 18, 2021 23:32

indutny-signal force-pushed the feature/shared-event-queue branch 2 times, most recently from 4caa935 to b9864a4 Compare May 18, 2021 23:49

indutny marked this pull request as ready for review May 18, 2021 23:49

indutny force-pushed the feature/shared-event-queue branch from b9864a4 to 76853a9 Compare May 19, 2021 01:11

kjvalencik reviewed May 19, 2021

View reviewed changes

src/event/event_queue.rs Show resolved Hide resolved

indutny mentioned this pull request May 19, 2021

Shared ThreadsafeFunction in EventQueue RFC neon-bindings/rfcs#42

Closed

jrose-signal reviewed May 19, 2021

View reviewed changes

jrose-signal mentioned this pull request May 19, 2021

Node: Switch to our fork of Neon with faster EventQueue operations signalapp/libsignal#308

Merged

indutny mentioned this pull request May 20, 2021

Handle early finalization of TSFN by Node.js #744

Merged

indutny-signal reviewed May 20, 2021

View reviewed changes

indutny force-pushed the feature/shared-event-queue branch from c22f932 to 74a21cb Compare June 1, 2021 21:19

Trying out a pattern for sharing tsfn in channels

e6948b2

indutny-signal force-pushed the feature/shared-event-queue branch from 74a21cb to 3b8ec20 Compare June 3, 2021 22:46

kjvalencik reviewed Jun 4, 2021

View reviewed changes

src/event/event_queue.rs Show resolved Hide resolved

optimize dropping of non-cloned channels

4780c3b

indutny-signal force-pushed the feature/shared-event-queue branch from 0a662b0 to 4780c3b Compare June 8, 2021 16:55

jrose-signal reviewed Jun 8, 2021

View reviewed changes

indutny-signal added 6 commits June 8, 2021 10:16

use idiomatic rust

e1fbfb7

revert complicated abstraction, no benefit

532eec0

remove mutability requirements

9b1f125

small stylistic fixes

f8fc9f8

few more

0dfffbb

remove stale comment

740c4be

kjvalencik reviewed Jun 8, 2021

View reviewed changes

src/context/mod.rs Outdated Show resolved Hide resolved

src/event/event_queue.rs Outdated Show resolved Hide resolved

src/event/event_queue.rs Show resolved Hide resolved

src/event/event_queue.rs Show resolved Hide resolved

src/event/event_queue.rs Show resolved Hide resolved

indutny-signal added 4 commits June 8, 2021 15:01

add comments

bdd8ce2

extra comment

4e6e9eb

change order

5c8704c

fix comment

01f3bdc

kjvalencik mentioned this pull request Jun 29, 2021

Performance regression in Channel (EventQueue) for some usage patterns #762

Open

kjvalencik merged commit 65c0768 into neon-bindings:main Jun 29, 2021

renovate bot mentioned this pull request Aug 1, 2021

fix(deps): update dependency neon-cli to ^0.9.0 amilajack/node-disk-utility#75

Merged

1 task

		/// _Not idempotent_
		pub(crate) fn reference(&mut self, env: Env) {

Reuse ThreadsafeFunction in EventQueue #739

Reuse ThreadsafeFunction in EventQueue #739

Conversation

indutny commented May 18, 2021

indutny commented May 18, 2021 • edited Loading

indutny-signal commented May 19, 2021

kjvalencik commented May 19, 2021

indutny-signal commented May 19, 2021

jrose-signal left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

indutny-signal commented May 19, 2021

jrose-signal left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

indutny-signal commented May 20, 2021

kjvalencik commented May 20, 2021

indutny commented May 20, 2021

Choose a reason for hiding this comment

kjvalencik commented Jun 1, 2021

indutny-signal commented Jun 1, 2021

kjvalencik commented Jun 1, 2021

indutny-signal commented Jun 1, 2021

kjvalencik commented Jun 1, 2021

dherman commented Jun 2, 2021

dherman commented Jun 2, 2021 • edited Loading

kjvalencik commented Jun 3, 2021 • edited Loading

indutny-signal commented Jun 3, 2021

kjvalencik commented Jun 7, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

indutny commented May 18, 2021 •

edited

Loading

dherman commented Jun 2, 2021 •

edited

Loading

kjvalencik commented Jun 3, 2021 •

edited

Loading

kjvalencik commented Jun 7, 2021 •

edited

Loading