How to control the number of Future / threads in a confined memory that only allow, say, 2 Future-s?

futures.scala

// Previous revision does not reflect the original intention as the title stated

import concurrent.Future
import concurrent.ExecutionContext.Implicits.global

// this will cause OOM Exception
(1l to 10000000).grouped(4).grouped(4).foldLeft(Future.successful(0l))((num0, xss0) =>
  num0.flatMap(x0 => xss0.foldLeft(Future.successful(x0))((num1, xss1) => 
    num1.flatMap(x1 => Future(xss1.sum).map(x1 + _))).map(_ + x0))).foreach(println)

// solution using semaphore to limit 2 Futures at a time. It will not create new Futures unless 
// they are completed i.e. foreach(...)

val semaphore = new java.util.concurrent.Semaphore(2)

(1l to 1000000).grouped(4).grouped(4).foldLeft(Future.successful(0l))((num0, xss0) =>
  num0.flatMap(x0 => 
    Future.sequence(xss0.map(xs => {
      semaphore.acquire()
      val f = Future(xs.sum)
      f.foreach(_ => semaphore.release())
      f  
    })).map(_.sum + x0))).foreach(println)  // 500000500000

RunManyIO.scala

package org.teckhooi

import java.util.concurrent.{Executors, ThreadFactory}
import java.util.concurrent.atomic.AtomicInteger

import cats.effect.{ExitCode, IO, IOApp}

import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
import cats.effect._
import cats.implicits._
import fs2.Stream
import io.chrisdavenport.log4cats.{Logger, SelfAwareStructuredLogger}
import io.chrisdavenport.log4cats.slf4j.Slf4jLogger

import scala.language.postfixOps

object RunManyIO extends IOApp {
  val MAX_THREAD = 4

  val ec: ExecutionContextExecutor =
    ExecutionContext.fromExecutor(
      Executors.newFixedThreadPool(
        MAX_THREAD,
        new ThreadFactory {
          val ctr = new AtomicInteger(0)
          def newThread(r: Runnable): Thread = {
            val back = new Thread(r)
            back.setName(s"for-ioapp-${ctr.getAndIncrement()}")
            back.setDaemon(true)
            back
          }
        }
      ))

  override protected implicit def contextShift: ContextShift[IO] =
    IO.contextShift(ec)

  override protected implicit def timer: Timer[IO] = IO.timer(ec)

  def writeToDatabase[F[_]: Sync: Timer: Logger](chunk: Int): F[Long] =
    Logger[F].info(s"Counting $chunk by ${Thread.currentThread().getName}") *>
      Sync[F].delay(chunk.toLong)

  override def run(args: List[String]): IO[ExitCode] = {
    implicit val unsafeLogger: SelfAwareStructuredLogger[IO] = Slf4jLogger.getLogger[IO]

    def writeToDb(threads: Int, i: Int): IO[List[Long]] =
      Stream
        .emits(1 to i)
        //    .chunkN(10)
        .covary[IO]
        .parEvalMap(threads)(writeToDatabase[IO])
        //    .evalMap(writeToDatabase[IO])
        .reduce(_ + _)
        .compile
        .toList

    (writeToDb(2, 40000) >>=
      ((x: List[Long]) => IO(println(x.head)))) *>
      IO(ExitCode.Success)
  }
}

ThrottlingFuture.scala

/**
  *
  * Limit the JVM heap to 8MB or less using the VM option i.e. -Xmx8M. The default ExecutionContext
  * does not limit the creation of new Future instances. Therefore, it will continue to create Future instances
  * until out of memory.
  *
  * On the other hand, com.mchange.v3.concurrent.BoundedExecutorService will limit the number of
  * Future instances created based on "blockBound" setting. The call to get a Thread will be
  * suspended if the block bound is exhausted. The call will be resumed when there is a Thread
  * returns to the pool and available.
  *
  */

import java.util.concurrent.Executors

import scala.concurrent.ExecutionContext
import com.mchange.v3.concurrent.BoundedExecutorService

import scala.io.Source
import concurrent.Future

object ThrottlingFuture {
  val MAX_INSTANCES = 50000

  def main(args: Array[String]): Unit = {
    success()
    println("Completed. Press any key to proceed to next test sequence")
    Source.stdin.next()
    failsWithOOM()
  }

  private def success(): Unit = {
    val es = new BoundedExecutorService(Executors.newFixedThreadPool(4), 100, 50)

    implicit val ec = ExecutionContext.fromExecutorService(es)

    (1 to MAX_INSTANCES).foreach(x => {
      Future(println(s"Starting $x"))
    })

    ec.shutdown()
  }

  private def failsWithOOM(): Unit = {
    import concurrent.ExecutionContext.Implicits.global

    (1 to MAX_INSTANCES).foreach(x => {
      Future(println(s"Starting $x"))
    })
  }
}

Revision 1 solution does not work as designed