Thanks @TapanVaishnav Could you add a test for it? Something that if n tasks are running in parallel and any of them fails then all the others will be canceled as well. It should be quite simple to do with TestScheduler which can be used to manually tick time. e.g.

val failedTask = Task.raiseError(ex).delayExecution(1.second)
val otherTask = Task.sleep(2.second).doOnCancel(incrementCounter)

Observable(0, 1, 2, 3, 4, 5, 6).mapParallelOrdered(n)(if (i == 0) failedTask else otherTask)

and then we can do s.tick(1.second) to instantly complete failedTask and see if others were canceled

Thanks @TapanVaishnav Could you add a test for it? Something that if n tasks are running in parallel and any of them fails then all the others will be canceled as well. It should be quite simple to do with TestScheduler which can be used to manually tick time. e.g.

val failedTask = Task.raiseError(ex).delayExecution(1.second)
val otherTask = Task.sleep(2.second).doOnCancel(incrementCounter)

Observable(0, 1, 2, 3, 4, 5, 6).mapParallelOrdered(n)(if (i == 0) failedTask else otherTask)

and then we can do s.tick(1.second) to instantly complete failedTask and see if others were canceled

Got it, will add a test. Thanks

@Avasil Hi, I've added test for MapParallelOrderedSuite and was planning to add one for mapParallelUnorderedSuite but because the map is unordered(in the latter case) we can't ensure the value of the counter. For example

    test("should cancel the whole stream when if one fails") { implicit s =>
    val wasThrown: Throwable = null
    var received = 0

    val failedTask = Task.raiseError(wasThrown).delayExecution(1.second)
    val otherTask = Task.sleep(2.second).doOnCancel(Task { received += 1})

    Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => if (i == 0) failedTask else otherTask)
      .toListL.runToFuture

    s.tick(1.second)

    assertEquals(received, 3)
  }

in the above case, because the map is unordered test might pass/fail, is that right?

I feel like it should still work because it will preserve the order of the input (so failedTask should start as the first one and then 1,2,3)

Did you experience non-deterministic behavior here?

Did you experience non-deterministic behavior here?

Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => Task.now(println(i))) .toListL.runToFuture
Yes, I changed my task inside the map to print the order and it's varying.
Like in one case the order was 1->3->4->2->0->5->6 and in another was 1->4->3->2->0->6->5

Thought not sure if is this the input order or the order in which Tasks are getting completed in their respective threads.

Also, I tried running the #1064 (comment) test and out of three cases, it passed in one and failed in others. So it seems like the input order isn't preserved. (which I believe, should be the case)

In your example the tasks are "instant" so I can see the output happening in this way, I think it would be a little bit more consistent with a small delay

AFAIK mapParallelUnordered will start input tasks in the correct order but they can finish in different order and will send any result downstream as soon as it is done

Also, I tried running the #1064 (comment) test and out of three cases, it passed in one and failed in others. So it seems like the input order isn't preserved.

I wonder if we have a race condition, i.e.

Do you see any tasks successfully completing or just more cancels?
I think we could try guaranteeCase instead of doOnCancel to track all exit cases and see what's going on

I think it would be a little bit more consistent with a small delay
I think we could try guaranteeCase instead of doOnCancel to track all exit cases and see what's going on

I see, will try both, thanks for your input. 👍

Hi @Avasil, I changed my observable to this Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => Task(println(i)).delayExecution(1.second)).toListL.runToFuture
and following are the outputs for two different runs
0->3->2->1->5->6->4
1->2->0->3->6->4->5

Am I executing my tasks in the right manner??

Hi @Avasil, I changed my observable to this Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => Task(println(i)).delayExecution(1.second)).toListL.runToFuture
and following are the outputs for two different runs
0->3->2->1->5->6->4
1->2->0->3->6->4->5

Am I executing my tasks right in the right manner??

This is correct, note that the first 4 tasks are always from 0-3 range.
0, 1, 2, 3 start and wait 1 second. They run concurrently so they can print in a different order

This is correct, note that the first 4 tasks are always from 0-3 range.
0, 1, 2, 3 start and wait 1 second. They run concurrently so they can print in a different order

I see but if that is the case then the below code should produce the same output(=3) as failedTask would always be there in the first call, right? But in some cases, it's producing the received value as 4 and in some as 3.

    test("should cancel the whole stream when if one fails") { implicit s =>
    val wasThrown: Throwable = null
    var received = 0

    val failedTask = Task.raiseError(wasThrown).delayExecution(1.second)
    val otherTask = Task.sleep(2.second).doOnCancel(Task { received+= 1; println("otherTask")})

    Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => if (i == 0) failedTask else otherTask).toListL.runToFuture

    s.tick(1.second)
    assertEquals(received, 6)
  }

Also, will check for the race condition(which you mentioned earlier) and will update the comment.

Ok, so the output is not the same on each run but at most only the 5th Task is getting called i.e.
Observable(0,1,2,3,4,5,6).mapParallelUnordered(4)(i => if (i == 0) failedTask else { Task.now(println(i)) ; otherTask}) .toListL.runToFuture
running this produces two kind of outputs

3
1
2
// the above order can be different as you mentioned earlier
otherTask
otherTask
otherTask
4
otherTask

and

3 
1
2
// the above order can be different as you mentioned earlier
otherTask
otherTask
otherTask

will check more on the code.

@Avasil I think it was as you mentioned about the race condition,
failedTask was releasing the semaphore on cancellation and nextTask (element 4) was getting triggered (not sure if it making much sense), so I changed the doOnCancel function

Thank you @TapanVaishnav
This fix should be fine, I've left one suggestion (it might cause ordered version to need a similar treatment but I'm not sure)

@Avasil Thank you for your review. I've added composite.cancel() in both process and sendDownstreamOrdered method. Also, created a cancellable task instead of releasing semaphore.

All the tests are passing on my local machine and I didn't even touch the tests which are failing on CI(and don't have any dependency on MapParallel(Un)orderedSuite),
can you please have a look?

All the tests are passing on my local machine and I didn't even touch the tests which are failing on CI(and don't have any dependency on MapParallel(Un)orderedSuite),
can you please have a look?

Looks like a random error, I restarted CI

@Avasil hey, no worries.
I totally understand your concern, if you don’t nitpick now, the code will become cumbersome to understand later.
And I’m also in a learning phase so totally appreciate all the comments.