WIP: Add JobManager

[BalmungSan]

Implements the JobManager idea proposed in #1345

---

Roadmap

• Define the API.
• Implement it.
• Add tests.
• Add docs.

[BalmungSan]

On one had, I like the idea of users being able to use their own Ids.
On the other, I think most users would benefit from a default using automatically generated UUIDs.

Should we provide such default in some way?

[BalmungSan]

I somewhat dislike the idea that both bad id and already finished return in a None.
But, otherwise, the map will grow indefinitely.

Any ideas?

[durban]

By fully controlling IDs, we could have startJob return a fresh ID, so "bad ID" would never happen. (But then we'd lose the ability of users to have their own IDs...)

[mtomko]

Can it not return Either[Unit, Option[S]], where Left is "unknown ID" and Right(None) is finished?

Alternatively, I guess you could have an ADT with Unknown | Completed | Running(x)

I don't fully understand the comment about the map growing indefinitely, though, so my suggestion may not improve that situation. Also, making this distinction turns a simple API into one that has a lot of ceremony for the caller, who perhaps doesn't care or can guess. I'm wondering if what I'm missing is a detail in how MapRef works, that a lookup implictly fills a default value in?

[BalmungSan]

I don't fully understand the comment about the map growing indefinitely

Yeah, the question wasn't really about how to model it but rather whether it should.
In order to be able to return a Completed status, we would need to preserve all IDs that have run, thus making the internal Map "grow indefinitely", which smells like a memory leak.

I guess we could let the user decide which of the two options they want. But I do not want to complicate the API and implementation without satisfying a real use case.

In other words, the question was actually:
"Would anyone actually need to distinguish completed from bad ID. And, if so, should we provide that for them instead of letting them build that for themselves?"

[mtomko]

I don't think I can answer that question in general. I will say that in the use case I had in mind, the universe of possible IDs is not that big (maybe 100ish?) We might cancel a job and put in a replacement that would end up with the same ID though.

[BalmungSan]

Users would then implement this trait for their own Jobs.

[BalmungSan]

We don't wait for the Job to start before returning to the user.
But, that means there is a brief delay between starting the Job and users being able to query its status.

[durban]

I don't think that's okay. I think the point of std is to solve tricky race conditions like this for users.

[BalmungSan]

Fair point.
What do you think would be the best way to solve that? A Deferred a CountdownLatch? Other thing?
Or maybe, simply run the runJob logic there rather that sending it to the Supervisor?

[durban]

Yeah, I'm a little confused by why it's sent to the Supervisor... (I'm sure there is a reason, I just don't know it).

[BalmungSan]

The main thing is that if the same id was already found we cancel it, which is costly.
But, I think I could just send that to the Supervisor as well.

[BalmungSan]

Oh, looking at the code, I realized that we also need to run the job setup (Resource.acquire) before being able to run the job itself.
So that was also part of what was being done in the background, but it also means that the short delay could be bigger.

Thus, I decided to use a Deferred to wait until the job has been properly registered before returning. But that may take a while.
Another idea that I just had would be to have an Initializing status that could be used meanwhile. However, that would complicate the logic quite a bit.

[BalmungSan]

In case the same id was already used, we cancel the previous Job.

[djspiewak]

I'm finally reviewing this!

So I like where this is going quite a bit. One thing I'm not as fond of is the fact that it's built around F[Unit]. I could be persuaded otherwise, but I tend to think that it's better to allow users to return values and then pick them up later. Probably with something like join(id: Id): F[Option[Outcome[F, Throwable, R]]] (or get) where R is the result type.

On that note, I actually wonder about the usefulness of the status stuff. Any user who is doing something really robust there is probably going to want something like an Fs2 Signal or similar, while someone who isn't thinking to deeply about it is probably only going to want Unstarted/Started/Completed.

I do actually like the parametric ids. It imposes some limitations obviously but it keeps us a bit more honest, and in many applications, users will in fact have a natural Id type that they're going to want to work with. We can have a utility constructor which instantiates Id = Uuid for convenience or something.

Needs tests!

[mtomko]

Can it not return Either[Unit, Option[S]], where Left is "unknown ID" and Right(None) is finished?

Alternatively, I guess you could have an ADT with Unknown | Completed | Running(x)

I don't fully understand the comment about the map growing indefinitely, though, so my suggestion may not improve that situation. Also, making this distinction turns a simple API into one that has a lot of ceremony for the caller, who perhaps doesn't care or can guess. I'm wondering if what I'm missing is a detail in how MapRef works, that a lookup implictly fills a default value in?

[mtomko]

Would it work to track the job status using a Deferred as well? That would let you also provide a semantically blocking API to wait for the completion of the job (under the hood it would just await the Deferred). Otherwise, callers who need S are going to have to implement their own polling loops and while they could use a Deferred, it'll have to wake a fiber periodically to do it.

[BalmungSan]

Sorry, not sure I follow the wording of the question?

Are you asking if a user would be able to implement getStatus using a Deferred? If so, yes, I do not see why they couldn't; I also do not see why they would need to be constantly pooling?
If you are suggesting that we should provide a helper that would use a Deferred for them? Then, as with any other helper discussion, the question would be if the use case is common enough to deserve being put in the library.

[mtomko]

Maybe I'm missing a detail in the implementation, but currently the way startJob works is that it launches job.run in the background using a supervisor and then places a JobStatus in the MapRef; there's a Deferred here for tracking when job registration is complete, but the job could still be running.

Callers who just want S can only get it by calling getJobStatus over and over until it is Some. That's the polling I was referring to.

But I think you could also do something where the MapRef was MapRef[F, Id, (F[Unit], Deferred[F, S])] where the F[Unit] was the cancellation program. Then the existing getJobStatus could be served by tryGet on the Deferred. But you could also provide a variant that used get on the Deferred which would block until job.run completed the Deferred. That semantic blocking can't be done by a caller with the current API without the user implementing their own polling, I don't think.

Sorry, I haven't actually implemented the change yet, so the shape of these might not be 100% right.

[mtomko]

I can try to implement it tomorrow and see if the types work out. It's possible not everything can exist in the right order the way I was thinking of it.

[BalmungSan]

Sorry, I still do not follow what would be the purpose here?

Status is not really meant as a way to get a result out of the Job, although it certainly could be used for that. Rather, it is intended to be able to inspect the progress of a Job.

Callers who just want S can only get it by calling getJobStatus over and over until it is Some.

The very first call to getJobStatus would return Some, unless it is already completed by that time or you run the pooling and the start in parallel.

But you could also provide a variant that used get on the Deferred which would block until job.run completed the Deferred.

I am not sure why anyone would want that? The whole point of this is to run something in the background, not to wait for it.

---

However, this does connect with the point Daniel Spiewak made in his original review, that maybe we should not swallow the results and rather provide the users a handle over them. Similar to how Supervisor gives you the started Fiber.

So I am thinking that the correct API is something like this:

trait JobManager[F[_], Id, S, R] {
  // Starts job in the background.
  // Semantically blocks until the job starts, but does not wait for it to complete.
  def startJob(id: Id, job: Resource[F, JobManager.Job[F, S, R]]): F[Fiber[F, Throwable, R]]

  // Semantically blocks until the job associated with the given id is completed.
  // If the id is not associated with any job, completes immediately with a `None`.
  def join(id: Id): F[Option[Outcome[F, Throwable, R]]]

  // Polls for the status of a job.
  // If the id is not associated with any job, returns a `None`.
  def getJobStatus(id: Id): F[Option[S]]

  // Semantically blocks until the job is cancelled.
  // If the id is not associated with any job, it is a noop.
  def cancelJob(id: Id): F[Unit]
}

WDYT @djspiewak?

[BalmungSan]

While thinking about this idea, I found that the previous implementation had a bug where non-cancelled jobs were never cleaned up from the jobsMap.
Thus, I decided to fix that and adapt the API, since I am now convinced this is the most complete and flexible one.

[mtomko]

I think the confusion here may be somewhat related to the varying purposes we may have for this kind of implementation. In the Discord I published a trait that I had implemented and a description of the scenario I had in mind. In my case I'm not worrying about running jobs in the background at all, but managing the lifecycle of Resources that are long lived, but lazily initialized - in our case this could be things like connection pools or clients of upstream services.