zstan commented on code in PR #5525:
URL: https://github.com/apache/ignite-3/pull/5525#discussion_r2024617143
##########
modules/core/src/main/java/org/apache/ignite/internal/util/subscription/OrderedMergePublisher.java:
##########
@@ -221,86 +241,124 @@ private void drain() {
// Frequently accessed fields.
Subscriber<? super T> downstream = this.downstream;
OrderedMergeSubscriber<T>[] subscribers = this.subscribers;
- int subsCnt = subscribers.length;
Object[] values = this.values;
+
long emitted = this.emitted;
+ // Retry loop.
for (; ; ) {
- long requested = (long) REQUESTED.getAcquire(this);
+ switch ((State) STATE.getAcquire(this)) {
+ case INITIAL: {
+ int waiting = this.waiting;
- for (; ; ) {
- if ((boolean) CANCELLED.getAcquire(this)) {
- Arrays.fill(values, null);
+ // Moves non-initialized sources to the beginning of
the array for faster array scans
+ // in the case of long initialization.
+ for (int i = 0; i < waiting; ) {
+ boolean innerDone = subscribers[0].done; // Read
before polling to preserve correct program order.
+ Object obj = subscribers[0].queue.poll();
- for (OrderedMergeSubscriber<T> inner : subscribers) {
- inner.queue.clear();
- }
+ int done = (obj == null && innerDone) ? 1 : 0; //
Flag has no effect if poll was successful.
+ int initialized = obj != null || innerDone ? 1 : 0;
- return;
- }
+ values[0] = done > 0 ? DONE : obj;
- int completed = 0;
- boolean waitResponse = false;
+ waiting -= initialized;
- for (int i = 0; i < subsCnt; i++) {
- Object obj = values[i];
+ int move = initialized * waiting; // No effect if
value wasn't initialized.
+ swap(values, 0, move);
Review Comment:
as you previously wrote - "No effect if value wasn't initialized" but we
also no need to call "swap" at all, i think jvm can`t optimize such logic, wdyt
?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
