Make the fastPublisher in the FlowInterop benchmark a sequential one

benchmark/src/main/scala/fs2/benchmark/FlowInteropBenchmark.scala

@@ -37,18 +37,17 @@ import org.openjdk.jmh.annotations.{
 
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.Flow.{Publisher, Subscriber, Subscription}
-
-import scala.concurrent.Future
+import scala.annotation.elidable
 
 @State(Scope.Thread)
 @BenchmarkMode(Array(Mode.Throughput))
 @OutputTimeUnit(TimeUnit.SECONDS)
 class FlowInteropBenchmark {
   @Param(Array("1024", "5120", "10240"))
-  var totalElements: Int = _
+  var totalElements: Long = _
 
   @Param(Array("1000"))
-  var iterations: Int = _
+  var iterations: Long = _
 
   @Benchmark
   def fastPublisher(): Unit = {
@@ -57,25 +56,21 @@ class FlowInteropBenchmark {
         override final def subscribe(subscriber: Subscriber[? >: Unit]): Unit =
           subscriber.onSubscribe(
             new Subscription {
-              @volatile var i: Int = 0
+              var i: Long = 0
               @volatile var canceled: Boolean = false
 
+              // Sequential fast Publisher.
               override final def request(n: Long): Unit = {
-                Future {
-                  var j = 0
-                  while ((j < n) && (i < totalElements) && !canceled) {
-                    subscriber.onNext(())
-                    i += 1
-                    j += 1
-                  }
+                val elementsToProduce = math.min(i + n, totalElements)
 
-                  if (i == totalElements || canceled) {
-                    subscriber.onComplete()
-                  }
-                }(global.compute)
+                while (i < elementsToProduce) {
+                  subscriber.onNext(())
+                  i += 1
+                }
 
-                // Discarding the Future so it runs in the background.
-                ()
+                if (i == totalElements || canceled) {
+                  subscriber.onComplete()
+                }
               }
 
               override final def cancel(): Unit =