asapScheduler not working as expected

[leebenmorris]

I'm doing some simple experiments with RxJS in a simple NodeJS Express server where I am comparing different approaches to handling and processing requests (based on this post https://snyk.io/blog/nodejs-how-even-quick-async-functions-can-block-the-event-loop-starve-io/). This is the basic setup:

const express = require('express')
const crypto = require('crypto')
const { asyncScheduler, asapScheduler, range } = require('rxjs')
const { promisify } = require('util')

const setImmediatePromise = promisify(setImmediate)

const PID = process.pid

function log(msg) {
	console.log(`[${PID}]`, new Date(), msg)
}

const app = express()

function randomString() {
	return crypto.randomBytes(100).toString('hex')
}

app.get('/compute-sync', function computeSync(req, res) {
	log('computing sync!')
	const hash = crypto.createHash('sha256')
	for (let i = 0; i < 1e6; i++) {
		hash.update(randomString())
	}
	res.send(hash.digest('hex') + '\n')
})

app.get('/compute-immediate', function computeImmediate(req, res) {
	log('computing immediate!')

	const hash = crypto.createHash('sha256')

	for (let i = 0; i < 1e6; i++) {
		await setImmediatePromise(hash.update, randomString())
	}

	res.send(hash.digest('hex') + '\n')
})

app.get('/compute-rxjs', async function computeRxjs(req, res) {
	log('computing Rxjs!')

	const hash = crypto.createHash('sha256')

	range(0, 1e6, asapScheduler).subscribe({
		next() {
			hash.update(randomString())
		},
		complete() {
			res.send(hash.digest('hex') + '\n')
		},
	})
})

app.get('/healthcheck', function healthcheck(req, res) {
	log('they check my health')
	res.send('all good!\n')
})

const PORT = process.env.PORT || 1337
let server = app.listen(PORT, () => log('server listening on :' + PORT))

It was my understanding that the asapScheduler would use setImmediate under the hood, so why does the /compute-immediate endpoint NOT block the event loop (keeps the server responsive to new requests) but the /compute-rxjs does block and leads to server timeouts on the health endpoint?

I have also tried the asyncScheduler - this does not block, but it does take perhaps an order of magnitude longer to complete then the /compute-immediate endpoint.

I would really like to use RxJS for more complex processing of incoming requests, but feel this issue makes that choice undesirable. Is there something I am missing? Is there a way to get the RxJS solution to work in the same way as the setImmediate solution?

[kwonoj]

asap scheduler explcitly states this:

rxjs/src/internal/scheduler/asap.ts

Lines 11 to 12 in 9acb950

	* in time. If however you set delay to `0`, `asap` will wait for current synchronously executing
	* code to end and then it will try to execute given task as fast as possible.

asap scheduler behaves the same as {@link asyncScheduler} scheduler when you use it to delay task
in time. If however you set delay to 0, asap will wait for current synchronously executing
code to end and then it will try to execute given task as fast as possible.

[leebenmorris]

But from a quick read of the code the asapAction specifically seems to use setimmediate when the delay is 0, but it doesn’t behave as I would expect as detailed above. Could this be a bug of some sort?

[kwonoj]

I'm not sure where you referring, https://github.com/ReactiveX/rxjs/blob/master/src/internal/scheduler/AsapAction.ts#L15-L19

schedules to asyncaction only if delay > 0.

[kwonoj]

code seems not unfurled

 protected requestAsyncId(scheduler: AsapScheduler, id?: any, delay: number = 0): any {
    // If delay is greater than 0, request as an async action.
    if (delay !== null && delay > 0) {
      return super.requestAsyncId(scheduler, id, delay);
    }
    // Push the action to the end of the scheduler queue.
    scheduler.actions.push(this);

[leebenmorris]

rxjs/src/internal/scheduler/AsapAction.ts

Lines 23 to 25 in 873e52d

	return scheduler.scheduled || (scheduler.scheduled = immediateProvider.setImmediate(
	scheduler.flush.bind(scheduler, undefined)
	));

?

[cartant]

It doesn't call the runtime's setImmediate. It uses this microtask implementation:

rxjs/src/internal/util/Immediate.ts

Lines 22 to 36 in 91b5688

	export const Immediate = {
	setImmediate(cb: () => void): number {
	const handle = nextHandle++;
	activeHandles[handle] = true;
	if (!resolved) {
	resolved = Promise.resolve();
	}
	resolved.then(() => findAndClearHandle(handle) && cb());
	return handle;
	},
	clearImmediate(handle: number): void {
	findAndClearHandle(handle);
	},
	};

IIRC, Ben mentioned that we might want to rename this to something less confusing.