channel: call ops.unwait as soon as fiber is resumed/enqueued; fixes select segfault

[forksaber]

Fixes #3900

A select call registers multiple wait events (one for each of its branches) and yields. When any of the branches is ready, select fiber resumes and unwaits the remaining events by calling
ops.each &.unwait. This works correctly with unbuffered channels as upon readiness, they immediately switch to the waiting/selecting fiber.

However, since buffered channels don't block unless buffer is full(send blocks) or empty(receive blocks), the selecting fiber is not resumed upon readiness. Hence, unwait is not called on time. By the time unwait is called, the selecting fiber may be enqueued multiple times.

Resuming this fiber after it has finished executing causes a segfault.

So, the fix here to call ops.each &.unwait as soon as the first resume/enqueue happens.

[straight-shoota]

Hey @forksaber, thanks for your initiative!

This PR most likely conflicts with #7407 which also applies substantial changes to Channel.select. That PR has a far broader scope on multithreading concurrency, and it's quite heavy but definitely worth a look!

/cc @ysbaddaden

[ysbaddaden]

I think the approach is correct: the ready fiber must unwait/cancel other clauses before yielding to another fiber. The wrapper class should be a struct, thought, to avoid an allocation on each and every channel operation, which would noticeably slow down channels, even outside of select.

[ysbaddaden]

For example:

struct SelectFiber
  def initialize(@fiber = Fiber.current)
    @callback = nil
  end

  def initialize(@fiber = Fiber.current, &@callback : ->)
  end

  def resume
    @callback.try(&.call)
    @fiber.resume
  end

  def enqueue
    @callback.try(&.call)
    Crystal::Scheduler.enqueue(@fiber)
  end
end

f1 = SelectFiber.new
f2 = SelectFiber.new { p 123 }

Or maybe add it to Fiber itself, as a self erasing callback:

class Fiber
  # :nodoc:
  setter callback : Proc(Nil, Nil)?

  def resume
    exec_callback
    Crystal::Scheduler.resume(self)
  end

  def enqueue
    exec_callback
    Crystal::Scheduler.enqueue(self)
  end

  private def exec_callback
    if callback = @callback
      @callback = nil
      callback.call
    end
  end
end

f = Fiber.current
f.callback = -> { ops.each(&.unwait(f)) }

[forksaber]

Thanks @ysbaddaden for your input. I prefer adding the self erasing callback to Fiber itself. I also had the same approach in mind....but was reluctant to make a change to Fiber . Now with your go-ahead, I'll update this PR with changes to Fiber.

[ysbaddaden]

Yes, the patch will be smaller, and it's pushing more of the scheduling API to Fiber itself, which I think if good (less calls to Crystal::Scheduler).

[forksaber]

updated PR to store callback in Fiber object

[Sija]

Is this GTG? Would be nice to have it 🚢 :)

[forksaber]

It's GTG from my side.

[ysbaddaden]

👍 for me, but I'd like @RX14 or @bcardiff to review as well.

[straight-shoota]

Could you please add a comment to explain the purpose of this?
Comments are welcome in other places as well ;)

[asterite]

This forms a closure and so it allocates memory, which is not ideal. Please change it to store a reference to ops directly in Fiber. For example you can have record UnwaitCallback, ops : ... and then callback : UnwaitCallback or something like that (and if we need more callbacks we create a union and son on).

[forksaber]

@asterite I tried implementing your suggestion but I am not able to figure out the type for ops. If I use Tuple | Array, I get the following error:

can't use Array(T) in unions yet, use a more specific type

[asterite]

Oh, right, it can't be implemented. Nevermind.

[straight-shoota]

Would converting the tuple to an array be more efficient than a closure?

[asterite]

The problem is that w don't know the types inside the array

[straight-shoota]

Yeah, right...
Yet we know it's all Channel, we just don't know the generic arguments. But that's actually irrelevant for calling #unwait. Just thinking here, but we could hack around it by introducing a module which is included in Channel:

abstract class Channel(T)
  module Unwaitable
    def unwait(fiber : Fiber)
    end
  end
    
  include Unwaitable
end

record UnwaitCallback, ops : Array(Channel::Unwaitable)

It works. Question is, is it worth the effort of having an additional type and needing to explicitly cast tuple and array arguments to Array(Channel::Unwaitable)? Probably not.

[asterite]

Closing because #3900 was closed because of #8112