Simplified integrated WSTP timer

[TimWSpence]

Please do tell me if I've just completely missed the point but it seems to me that work on a more performant timer has somewhat stalled and/or been rolled into the integrated runtime discussion. While this is super exciting, it feels like there's more design work to do there which might take quite a while. In the meantime, it seems to me that a first version of a timer should be relatively straightforward, super valuable and compatible with future runtime work.

My current understanding is that we could do the following:

1. Keep the current scheduled executor service for the sake of sleep when running on a thread other than the WSTP (done via a Thread.currentThread().isInstanceOf just like we do for scheduling currently)
2. Introduce a heap per WSTP thread to track the wakeup time of fibers which have gone to sleep on that thread
3. Every n iterations we dequeue fibers whose wait time has elapsed and add them to the queue of runnable fibers
4. Profit

From my understanding, this would have the following advantages over the current scheduled executor service design:

1. No kernel-level context switching from the timer thread awaking
2. More efficient as it approximates a hashed wheel timer, rather than trying to awaken each fiber at exactly the right time
3. No need for locking on the heap as it is effectively thread-local (I think sleep currently actually blocks on the WSTP trying to acquire a lock to enqueue on the scheduled executor service)

Time for everyone to pile on. What have I missed? 😅

[vasilmkd]

Thanks for starting this discussion Tim. You're right, work on this has somewhat fizzled out, because there's one issue that we haven't found a good solution for. What you described has already been implemented once, but we didnt go through with it all the way because of the possibility of some sleeping fibers to lag behind a CPU intensive task. Imagine that a fiber sleeps, and as you said, it gets put on the thread local heap. Then that same thread goes on to execute a long CPU intensive task, pushing way past the wakeup time of the sleeping fiber. We haven't yet thought of a way for another free thread to steal the sleeping fiber and wake it up on time. Or rather, we don't yet have a good concurrent heap implementation that would allow us to do that. Daniel has some ideas, but maybe the best thing to do would be to find a white paper/research paper/library and essentially reimplement that. Our usual source of inspiration, JCTools is also lacking in this department.

[vasilmkd]

Here's the PR with the implementation that was scrapped. #2252

[TimWSpence]

Thanks Vasil, this is super helpful!! I should have mentioned that on my original writeup - I thought that we'd decided that if you have tasks like that (or at least if you have enough of them for one per thread) then your timer is screwed anyway so it doesn't matter. But obviously I got that wrong 😅 I guess this is more sensitive to even one CPU hogging fiber.
Thanks for the PR link, I'll have a look through that and do some googling on concurrent heaps :)

[armanbilge]

@vasilmkd I took a look at #2252 and at a cursory glance it seems really really good actually. Not perfect maybe wrt to the stealing issue, but maybe the good first version Tim is suggesting we do!

[TimWSpence]

I'll have a look on Monday. If we want to take this idea further and Vasil doesn't have time then I'm happy to pick it up and give it a go (and then others can fix it for me 😅)

[armanbilge]

Then that same thread goes on to execute a long CPU intensive task, pushing way past the wakeup time of the sleeping fiber.

It's worth noting that long CPU intensive tasks are an anti-pattern for Cats Effect (hence ideas like the starvation detector in #3090). That doesn't mean it's unrealistic in practice of course. But it seems to complicate the implementation exponentially ... and even more so if we get into the integrated I/O runtime idea and have to start thinking about stealing I/O events.

See discussion in Discord where Daniel seems to think it's okay to not have timer stealing.

In the meantime, it seems to me that a first version of a timer should be relatively straightforward, super valuable and compatible with future runtime work.

💯 to this. At least, starting with a non-stealing version and maybe making it opt-in we could get folks trying it out and find out how important stealing is or isn't. Coupled with Tim's starvation detector we might actually be able to collect decent empirical data on this.

Keep the current scheduled executor service for the sake of sleep when running on a thread other than the WSTP

Do we really need this? Seems like a waste of a thread, why not make it an externally-scheduled sleep on the WSTP (just like an externally queued task).

[djspiewak]

See discussion in Discord where Daniel seems to think it's okay to not have timer stealing.

Yeeeaaaahhhhh… I think @vasilmkd convinced me that I'm wrong. :-) We should support stealing, and I don't think it'll be that horrific to do so.

Do we really need this? Seems like a waste of a thread, why not make it an externally-scheduled sleep on the WSTP (just like an externally queued task).

Surprisingly, a single parked thread isn't really that much of a cost. It would be if we had thousands of them, but just one is not that terrible, and it gives us a good solution to "what do we do when someone evalOns? If we find out later we can get rid of it, then so much the better.

[TimWSpence]

😂 what did Vasil say to change your mind? Even if we make it opt-in and with a caveat emptor in size 100 bold?

Does stealing literally just require a performant concurrent heap or are there other subtleties that I've missed?

[djspiewak]

It just requires a performant heap. I think this whole thing is a data structure design exercise.

[TimWSpence]

making it opt-in

This is a very good idea!! If people believe that their app doesn't have long cpu intensive tasks then they can try it out and hopefully get a much better timer and give us some feedback! As you say, monlithic cpu intensive tasks are an anti-pattern anway so I could even live with us supporting both options and potentially flipping the default at some point in the future (but hopefully we'll have worked the stealing version out by then anyway)

Do we really need this?

Maybe not! I thought that we still needed to keep the existing scheduled executor for bincompat anyway and just suggested it as an easy way to support a relative edgecase. Probably we can do better!

[djspiewak]

Definitely should be a configurable thing! IOApp and system properties, just like other tuning params.

[TimWSpence]

I've done a brief survey of the literature and from what I can see, the state of the art seems to be a variant of a concurrent skip list. Those implementations all exploit marking pointers and various low-level things which are probably nasty to do on the JVM, but might be achievable with AtomicMarkableReference. Alternatively we might be able to use j.u.c.ConcurrentSkipListMap but I think we probably need low-level access to the internals to achieve the results that those papers present.

However these are all solving a more general problem than we have. AFAICT all we need is to integrate a hashed wheel timer into the WSTP. There is some prior art here as well which uses combining funnels. They are still solving a more general problem however - we want to do batch reads on a bucket so our read contention should be very low (or none if we maintain an AtomicInteger to track the last slow that was read from) and we hope that writes should be reasonably well distributed as well if we have enough buckets.

TLDR: my proposal is that we build our own hashed wheel timer, that would basically be Array[AtomicMarkableReference[Cell]] where Cell is similar to the one from UnsafeUnbounded queue. Markable lets us do soft deletes instead of having to iterate the bucket in order to cancel a timer.

Pseudocode (don't read anything into the details 😅):

val lastIdx = new AtomicInteger(hash(nanoTime()))

def enqueue(task, time) =
  val bucket = hash(time)
  val cell = new Cell(task)
  var current = buckets[bucket].get
  cell.lazySet(current)
  while (!buckets[bucket].compareAndSet(current, cell)) {
    var current = buckets[bucket].get
    cell.lazySet(current)
  }

def dequeue: List[Task] =
  val to = hash(nanoTime())
  val from = lastIdx.getAndSet(to)
  val buf = new ListBuffer()
  for (i = from; i <to; i++) {
    val cells = buckets[bucket].getAndSet(null)
    //iterate cells, collecting non-cancelled tasks whose timer has expired
    //re-insert tasks which are not cancelled and whose timer has not expired
  }
  buf.toList