Issues with onTermination

[alerosmile]

I have this example code:

    Cancellable cancellable = Uni.createFrom().item(10)
        .onItem().transform(item ->
        {
          System.out.println("start computing");
          try
          {
            item *= 10;

            Thread.sleep(1000);
          }
          catch(InterruptedException e)
          {
          }
          System.out.println("end computing");
          return item;
        })
        .toMulti()
        .onCancellation().invoke(() -> System.out.println("onCancellation1"))
        .onTermination().invoke(() -> System.out.println("onTermination"))
        .onCancellation().invoke(() -> System.out.println("onCancellation2"))
        .runSubscriptionOn(Infrastructure.getDefaultExecutor())
        .onSubscription().invoke(() -> System.out.println("onSubscription"))
        .subscribe().with(item -> System.out.println("itemReceived: " + item));

    try
    {
      Thread.sleep(200);
    }
    catch(InterruptedException e)
    {
    }

    cancellable.cancel();

I've expected this output:

onSubscription
start computing
onCancellation2
onCancellation1
end computing
onTermination

But what I've got is this:

onSubscription
start computing
onCancellation2
onTermination
onCancellation1
end computing

There are two issues with this:

• The order of onTermination and onCancellation1 is wrong. Removing the .onMulti() line fixes this. Is this a bug in the Multi class?
• I've expected onTermination to be the last. The Multi or Uni might be in a computing process (like here) when it is cancelled and cannot be terminated immediately. In my production code I return a Multi on the line after end computing. I need this Multi to be cancelled too. I tried to cancel this Multi during the the onTermination event. But this does not work if the timing is similar to the one in this example. Is there any solution for this?

Thanks a lot.

[jponge]

The order is correct, each stage in the pipeline is an intermediate operation (and also a Multi / publisher on its own). What you see is correct wrt what the reactive streams protocol do.

The cancellation is eventual, so when an operator / publisher receives a cancel() request, then it is not required to cancel right away (see the reactive streams protocol specification).

I also think that you are expecting Mutiny to magically create threads, but it does not. When you have a Thread.sleep in the example above, it blocks everything.

If you want to make more sense of what happens at each stage, you might plug a few .log(...) calls. It will show you all the events between each publisher and the subscriber.

[alerosmile]

But why is the onCancellation1 event after the onTermination? If I remove the .toMulti() stage, the order is different (onCancellation1 before onTermination). The Thread.sleep is just to provoke this exact timing situation.

[jponge]

There's an ordering difference in how UniOnTerminationProcessor and MultiOnTerminationInvokeProcessor forward cancel() signals, but can that be considered any bug?

[jponge]

In one case the callback is called before cancelling upstream, and in one case it's the opposite.

It's still correct per reactive-streams semantics, but should we harmonize that? Good question.

/cc @cescoffier

[cescoffier]

Yes, we should harmonize. Now that the TCK has checked the order, I would use the Multi-semantic as the default.

[jponge]

See #1843

[alerosmile]

@cescoffier Wouldn't it make more sense for the onCancellation signal (direction: downstream -> upstream) to come before the onTermination signal (direction: upstream -> downstream), since it is the 'reason' for the latter?

[alerosmile]

Back to my first post: For cleanup purposes, I need a signal which is fired when the stream is terminated completely. Now I know that onTermination cannot be used for this because it is fired even if the stream is still active. Is there a way to accomplish this using Mutiny, or is this outside of its scope?

[jponge]

onTermination does that, but you want to plug it at the very end of the pipeline. If you do the subscription yourself, then your subscriber can do it (or .subscribe().with(...) callbacks).

[alerosmile]

Upon cancellation, onTermination (as well as onCancellation) is invoked by the thread that called Cancellable.cancel(). In contrast, onItem and similar methods are called from a different thread. This discrepancy is the core issue I am facing. If my subscriber is processing an onItem signal and another thread cancels the stream at that precise moment, onTermination is triggered before onItem has completed. This situation requires synchronization, or alternatively, onTermination must be enqueued to the stream.

[jponge]

@alerosmile Back to this: yes if a thread cancels while another thread is busy with processing an item, the 2 events will be concurrent, and the termination callback will be called concurrently. This is by design of reactive streams.

What you could do is have some critical section in both the termination and item processing handler (say, in a onItem().transform(...)). But ideally whatever logic you have in a termination handler shall assume asynchronicity.

[alerosmile]

Ok. Thanks a lot.