Add onRequest() hook for propagating request from downstream

[chenggangpro]

Try to fix #34178
After reproducing and debugging the issue, I found the onRequest() hook is missing. After the fix, the MultipartParser is functioning properly, and the upload hang issue has not occurred again. I tested this locally.

[sdeleuze]

I have the same analysis than @bclozel in #34178 (comment), I think this change, while probably fixing the issue, is going to over-request. We need to explore and understand what triggers this more in detail before crafting a fix. As a consequence, I will close this PR unmerged in favor of #34178.

[sdeleuze]

Reopening this PR based on #34178 (comment).

[sdeleuze]

Reopening to refine in order to prevent over-requesting based on a discussion with @chemicL.

[sdeleuze]

@bclozel @chemicL @rstoyanchev I tried to add a private final AtomicBoolean waitingForRequest = new AtomicBoolean(); field to MultipartParser and changed

private void requestBuffer() {
		if (upstream() != null &&
				!this.sink.isCancelled() &&
				this.sink.requestedFromDownstream() > 0 &&
				this.requestOutstanding.compareAndSet(false, true)) {
			request(1);
		}
	}

To:

private void requestBuffer() {
		if (upstream() != null && !this.sink.isCancelled() && this.requestOutstanding.compareAndSet(false, true)) {
			if (this.sink.requestedFromDownstream() > 0 || this.waitingForRequest.compareAndSet(true, false)) {
				request(1);
			}
			else {
				System.out.println("Waiting for request");
				this.waitingForRequest.set(true);
			}
		}
	}

After 20-30 tries, I saw the Waiting for request message printed but this version that tries to prevent over-requested does not fix the hanging issue anymore unlike what is currently committed with 4606337.

I am wondering if there is really over-requesting, especially with the requestOutstanding field and related conditional requesting. Any guidance on how I could check if we are really over-requesting with the new implementation?

[chemicL]

I think the above conditions won't match in case of resumption.

The problematic scenario seems to be the following:

1. Downstream issued request and is now at requestedFromDownstream = 1, requestOutstanding is flipped to true
2. Upstream delivers the requested value satisfying the demand (requestOutstanding = false, requestedFromDownstream = 0)
3. requestBuffer resulting from upstream's logic to re-request internally (notice now there can be 2 calls - the other potentially following from the onNext in the downstream)
4. requestOutstanding flipped to true, but requestedFromDownstream = 0, fail
5. downstream requests, but can't flip requestOutstanding to true, since it already is true and no request is made -> stalemate.

Usually the downstream requests a higher number so the initial value is not 1 and progress can be made for some time.

I think the requestOutstanding flag present in the existing implementation is enough to protect against over-requesting. I'm sorry if my concerns have caused unnecessary confusion.

[sdeleuze]

No problem @chemicL, thanks a lot for taking the time to share your detailed feedback, I will close the PR and keep what is committed then.