support for background tasks?

[larochef]

I want to run some possibly long running in the background. By long running, I mean that the rest of the code executing cannot wait for this code to complete.
The supervised blocks cancelling all that's running once they are over, what would be a good way to have tasks executing in the background?
So far, the only way I found is using futures, which I would really would like to avoid.

Is there a way to do it (I'm currently on 1.0.0-RC2)? If not, do you think it is something that would be added to Ox in the future?

[adamw]

Yes! If you have a task that should be running for the entire duration of your application, it should be placed in the proper scope. You can start a top-level scope by hand in your main, or just extend OxApp, which gives you a top-level scope (among other things). Any forks that you start within this scope, will live as long as the application. If this fork ends with an unhandled exception, your entire application will fold (which is a good thing: the error won't go unnoticed, if you don't want this behaviour, you must handle the error).

Within this top-level scopes, you can of course run nested supervised scopes. The code running them in fact is totally independent on what happens at the higher levels, due to the structured nature!

What might also be useful, is delegating starting the background tasks to a helper method. In such a case, add a (using Ox) parameter list, which when called from the top-level scope, will capture the top-level scope appropriately, and use it to create whatever fork invocations you have in that top-level method.

As an example, you might take a look at Bootzooka, where the application-wide background process is started in Main, and invokes a helper method as described above.

[larochef]

Thx for the answer, I already have such a scope, so I ended up writing some service for it:

class TaskService {

  private val queue = new ConcurrentLinkedQueue[() => Any]()

  // This method is blocking and will never end
  def start(): Unit = {
    supervised {
      Flow
        .tick[Unit](20.milliseconds)
        .filter(_ => !queue.isEmpty)
        .runForeach { _ =>
          while (!queue.isEmpty) {
            val task = queue.poll()
            fork(Try(task())).discard
          }
        }
    }
  }

  def runTask[T](task: () => T) = queue.offer(task)

}

where ConcurrentLinkedQueue is java.util.concurrent.ConcurrentLinkedQueue

The start method is then called in the main, using ox.par to run concurrently to the rest of the app.

it seems to be doing the job, the error management is missing though, if it can help others

[adamw]

I think you also need to run TaskService.start in the background? Plus, you can use a BlockingQueue to avoid poll + sleeping, but just waiting until an element is available (or Ox's Channel, if you need completion/error handling). So something like:

class TaskService:
  private val queue = new LinkedBlockingQueue[() => Any]();

  private def run(): Unit =
    supervised:
      forever:
        // any exceptions in the task propagate all the way up to the main fork, probably ending the application
        forkDiscard(queue.take()())

  // runs the main logic in the background, using the scope from the caller
  def start()(using Ox): Unit = fork(run())

[larochef]

Thx, this looks better for sure. The start was already called in the background in my main, yes.

If someone wants to duplicate this code, in order for it to work properly, the queue.take() must be called outside of the forkDiscard:

  private def run(): Unit = {
    supervised {
      forever {
        // any exceptions in the task propagate all the way up to the main fork, probably ending the application
        val task = queue.take()
        forkDiscard(task())
      }
    }
  }

with the previous version, it was creating a lot of forks and the rest of the app had a hard time finding some free thread to execute the code.

[larochef]

After some time injecting it in the code to replace the futures, I remembered another of the main reason I started on async tasks (this required some changes in my service): I wanted to have a flow where I have some long running task, so a function taking a long time to execute (can be > a few minutes) and have some status on the task running on a regular basis. I can create a flow from a future but not from a function (executing it in concurrency with the rest of the flow, like with the futures). Using fromFuture is not great since you have to handle some ExecutionContext and since I use ox from concurrency, I don't want to have to deal with platform threads or execution contexts. Since the returned flow is then used in an SSE endpoint with tapir, I have no control on the concurrency scope.

I came up with this version 2:

class TaskService(using Ox) {

  private val actor = Actor.create(ForkingActor())

  def runTask[T](task: => T): Fork[Try[T]] = {
    actor.ask(_.execute(Try(task)))
  }
}

class ForkingActor(using Ox) {
  def execute[T](task: => T) = fork(task)
}

so far tests are giving the expected results,

[adamw]

Thanks for the fix, indeed, I was writing without testing the code, and ended up posting an infinite loop ;)

Anyway - if you want to receive progress updates, maybe Channels are what you are looking for? I'm not sure I understand how granular you'd like to updates to be. Does the actor approach you posted solve any of the problems?

Also: is there something that's missing from Ox, that you think would be good to add to support long-running tasks? If not, I'll close the issue.

[adamw]

I'll close the issue for now, please reopen if you think we can somehow better support this use-case on Ox side