Skip to content

Optimize create-and-resume usage of Fiber in Enumerator #9552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
headius wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Optimize create-and-resume usage of Fiber in Enumerator #9552

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 53 additions & 3 deletions core/src/main/java/org/jruby/ext/fiber/ThreadFiber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@

import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.lang.ref.WeakReference;
import java.lang.reflect.Method;
import java.util.Map;
Expand All @@ -49,6 +50,7 @@ public class ThreadFiber extends RubyObject implements ExecutionContext {
private static final Logger LOG = LoggerFactory.getLogger(ThreadFiber.class);

private static final BiConsumer<Ruby, Runnable> FIBER_LAUNCHER;
private static final VarHandle INITIAL_BLOCK;
private static final String[] INITIALIZE_KWARGS = {"blocking", "pool", "storage"};

static {
Expand All @@ -61,6 +63,12 @@ public class ThreadFiber extends RubyObject implements ExecutionContext {
}

FIBER_LAUNCHER = fiberLauncher;

try {
INITIAL_BLOCK = MethodHandles.lookup().findVarHandle(ThreadFiber.class, "initialBlock", Block.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}

private static void nativeThreadLauncher(Ruby runtime, Runnable runnable) {
Expand Down Expand Up @@ -149,6 +157,23 @@ public IRubyObject initialize(ThreadContext context, IRubyObject _opts, Block bl
return context.nil;
}

/**
* Construct an internal Fiber whose thread will be launched by its first
* resume. The caller must resume the returned Fiber immediately.
*/
@JRubyMethod(name = "__create_raw__", meta = true, visibility = Visibility.PRIVATE)
public static IRubyObject createRaw(ThreadContext context, IRubyObject recv, Block block) {
if (!block.isGiven()) throw argumentError(context, "tried to create Proc object without block");

Ruby runtime = context.runtime;
ThreadFiber fiber = new ThreadFiber(runtime, (RubyClass) recv);

fiber.inheritFiberStorage(context);
fiber.data = new FiberData(new FiberQueue(runtime), context.getFiberCurrentThread(), fiber, false);
fiber.initialBlock = block;
return fiber;
}

// MRI: inherit_fiber_storage
public void inheritFiberStorage(ThreadContext context) {
RubyHash storage = context.getFiber().storage;
Expand All @@ -159,7 +184,7 @@ public void inheritFiberStorage(ThreadContext context) {
public IRubyObject resume(ThreadContext context, IRubyObject[] values) {
Ruby runtime = context.runtime;

if (!alive()) throw runtime.newFiberError("dead fiber called");
if (initialBlock == null && !alive()) throw runtime.newFiberError("dead fiber called");

final FiberData data = this.data;
FiberData currentFiberData = context.getFiber().data;
Expand Down Expand Up @@ -188,7 +213,13 @@ public IRubyObject resume(ThreadContext context, IRubyObject[] values) {

FiberRequest result;
try {
result = exchangeWithFiber(context, currentFiberData, data, val);
Block block = consumeInitialBlock();
if (block == null) {
result = exchangeWithFiber(context, currentFiberData, data, val);
} else {
createThread(context, data, currentFiberData.queue, block, val);
result = waitForFiber(context, currentFiberData, data);
}
} finally {
data.prev = null;
}
Expand All @@ -204,6 +235,11 @@ public IRubyObject resume(ThreadContext context, IRubyObject[] values) {
return processResultData(context, result);
}

private Block consumeInitialBlock() {
Block block = initialBlock;
return block != null && INITIAL_BLOCK.compareAndSet(this, block, null) ? block : null;
}

private static void fiberCalledAcrossThreads(Ruby runtime) {
throw runtime.newFiberError("fiber called across threads");
}
Expand Down Expand Up @@ -234,6 +270,10 @@ private static FiberRequest exchangeWithFiber(ThreadContext context, FiberData c
// push should not block and does not check interrupts
targetFiberData.queue.push(context, request);

return waitForFiber(context, currentFiberData, targetFiberData);
}

private static FiberRequest waitForFiber(ThreadContext context, FiberData currentFiberData, FiberData targetFiberData) {
// At this point we consider ourselves "in" the resume, so we need to make an effort to propagate any interrupt
// raise to the fiber. We forward the exception using the interrupt mechanism and then try once more to let the
// fiber deal with the exception and either re-raise it or handle it and give us a non-exceptional result. If
Expand Down Expand Up @@ -486,6 +526,11 @@ final boolean alive() {
}

static RubyThread createThread(ThreadContext context, final FiberData data, final FiberQueue queue, final Block block) {
return createThread(context, data, queue, block, null);
}

private static RubyThread createThread(ThreadContext context, final FiberData data, final FiberQueue queue,
final Block block, final FiberRequest initialRequest) {
final AtomicReference<RubyThread> fiberThread = new AtomicReference();

// retry with GC once
Expand All @@ -500,13 +545,17 @@ static RubyThread createThread(ThreadContext context, final FiberData data, fina
RubyThread rubyThread = ctxt.getThread();
fiberThread.set(rubyThread);
rubyThread.setFiberCurrentThread(data.parent);
if (initialRequest != null) {
ThreadFiber fiber = data.fiber.get();
if (fiber != null) fiber.thread = rubyThread;
}

Thread thread = Thread.currentThread();
String oldName = thread.getName();
thread.setName("Fiber thread for block at: " + block.getBody().getFile() + ":" + block.getBody().getLine());

try {
FiberRequest init = data.queue.pop(ctxt);
FiberRequest init = initialRequest == null ? data.queue.pop(ctxt) : initialRequest;

try {
FiberRequest result;
Expand Down Expand Up @@ -862,6 +911,7 @@ public ThreadFiber getPrev() {

volatile FiberData data;
volatile RubyThread thread;
private volatile Block initialBlock;
RubyHash storage;
final boolean root;
}
3 changes: 2 additions & 1 deletion core/src/main/ruby/jruby/kernel/enumerator.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ def next

def next_values
reset unless @fiber&.alive?
@fiber ||= Fiber.send(:__create_raw__, &@state)

vals, _ = @fiber.resume

Expand All @@ -129,7 +130,7 @@ def rewind
def reset
@state.done = false
@state.result = nil
@fiber = Fiber.new(&@state)
@fiber = nil
end

end
Expand Down
Loading