Merge branch 'master' into scheduler-buffered-write · jruby/jruby@a739758 · GitHub
Skip to content

Commit a739758

Browse files
headiusheadius
authored
Merge branch 'master' into scheduler-buffered-write
2 parents 47b2e95 + 97e6805 commit a739758

35 files changed

Lines changed: 988 additions & 50 deletions

‎.github/workflows/ci.yml‎

Lines changed: 19 additions & 3 deletions

‎core/src/main/java/org/jruby/RubyBinding.java‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -143,31 +143,51 @@ public IRubyObject local_variable_defined_p(ThreadContext context, IRubyObject s
143143
@JRubyMethod
144144
public IRubyObject local_variable_get(ThreadContext context, IRubyObject symbol) {
145145
String id = checkLocalId(context, symbol);
146-
DynamicScope evalScope = binding.getEvalScope(context.runtime);
147-
StaticScope staticScope = evalScope.getStaticScope();
148-
int slot = staticScope.isDefinedOrImplicit(id);
146+
DynamicScope scope = findLocalScope(binding.getEvalScope(context.runtime), id);
147+
int slot = scope == null ? -1 : scope.getStaticScope().existsOrImplicit(id);
149148

150149
if (slot < 0) throw undefinedVariableError(context, symbol);
151150

152-
return evalScope.getValueOrNil(unpackLowChar(slot), unpackHighChar(slot), context.nil);
151+
return scope.getValueOrNil(slot, 0, context.nil);
153152
}
154153

155154
@JRubyMethod
156155
public IRubyObject local_variable_set(ThreadContext context, IRubyObject symbol, IRubyObject value) {
157156
String id = checkLocalId(context, symbol);
158157
DynamicScope evalScope = binding.getEvalScope(context.runtime);
159-
StaticScope staticScope = evalScope.getStaticScope();
160-
int slot = staticScope.isDefinedOrImplicit(id);
158+
DynamicScope scope = findLocalScope(evalScope, id);
159+
160+
if (scope == null) { // Yay! New variable associated with this binding
161+
int slot = evalScope.getStaticScope().addVariable(id.intern());
162+
evalScope.growIfNeeded();
163+
164+
return evalScope.setValue(unpackLowChar(slot), value, unpackHighChar(slot));
165+
}
166+
167+
int slot = scope.getStaticScope().existsOrImplicit(id);
161168

162169
// for the implicit "it" variable
163170
if (slot == StaticScope.IMPLICIT) throw undefinedVariableError(context, symbol);
164171

165-
if (slot == -1) { // Yay! New variable associated with this binding
166-
slot = evalScope.getStaticScope().addVariable(id.intern());
167-
evalScope.growIfNeeded();
172+
return scope.setValue(slot, value, 0);
173+
}
174+
175+
/**
176+
* Find the dynamic scope that holds a local variable, or null if none does.
177+
*
178+
* This walks the dynamic scopes rather than taking a depth from the static ones, because the two
179+
* chains don't always line up: in the interpreter, a for loop body gets a dynamic scope of its own,
180+
* but a block inside it has the enclosing method's static scope as its parent.
181+
*/
182+
private static DynamicScope findLocalScope(DynamicScope scope, String id) {
183+
for (DynamicScope current = scope; current != null; current = current.getParentScope()) {
184+
StaticScope staticScope = current.getStaticScope();
185+
186+
if (staticScope.existsOrImplicit(id) != -1) return current;
187+
if (staticScope.getType() == StaticScope.Type.LOCAL) break;
168188
}
169189

170-
return evalScope.setValue(unpackLowChar(slot), value, unpackHighChar(slot));
190+
return null;
171191
}
172192

173193
@JRubyMethod(name = "implicit_parameter_defined?")

‎core/src/main/java/org/jruby/RubyString.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7191,7 +7191,7 @@ public static IRubyObject includeRange(ThreadContext context, RubyString _beg, R
71917191
(e = eByteList.get(0)) < 128 &&
71927192
(v = vByteList.get(0)) < 128) {
71937193
if (b <= v && v < e) return context.tru;
7194-
if (!exclusive && v == e) return context.tru;
7194+
if (!exclusive && b <= v && v == e) return context.tru;
71957195
return context.fals;
71967196
}
71977197
}

‎core/src/main/java/org/jruby/RubyThread.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1945,8 +1945,9 @@ public Long run(ThreadContext context, Object data) throws InterruptedException
19451945

19461946
@Override
19471947
public void wakeup(RubyThread thread, Object data) {
1948+
// notify could wake another thread waiting on the same object instead of this one, so wake them all
19481949
synchronized (data) {
1949-
data.notify();
1950+
data.notifyAll();
19501951
}
19511952
}
19521953
}

‎core/src/main/java/org/jruby/ext/socket/RubyBasicSocket.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import jnr.posix.Timeval;
4242
import jnr.posix.POSIX;
4343
import jnr.unixsocket.UnixSocketAddress;
44+
import org.jruby.FiberScheduler;
4445
import org.jruby.Ruby;
4546
import org.jruby.RubyArray;
4647
import org.jruby.RubyBasicObject;
@@ -967,6 +968,23 @@ protected void initSocket(ChannelFD fd) {
967968
setAscii8bitBinmode();
968969
}
969970

971+
/**
972+
* Wait through the fiber scheduler for a connection, after a nonblocking accept found none pending.
973+
* Call it without holding any lock on this server, since io_wait runs sibling fibers that may accept on
974+
* or close it. Returns when the caller should try accepting again.
975+
*/
976+
// MRI: rsock_s_accept on EAGAIN, which waits with rb_io_maybe_wait_readable and RUBY_IO_TIMEOUT_DEFAULT
977+
protected void schedulerWaitAcceptable(ThreadContext context, IRubyObject scheduler) {
978+
IRubyObject result = FiberScheduler.ioWaitReadable(context, scheduler, this);
979+
980+
if (result == context.fals) {
981+
throw RubyIOTimeoutError.newIOTimeoutError(context.runtime, "Timed out waiting for IO to become readable!").toThrowable();
982+
}
983+
984+
// no events ready, so MRI gives up with the EAGAIN it got from accept(2)
985+
if (result.isNil() || toInt(context, result) == 0) throw context.runtime.newErrnoFromErrno(Errno.EAGAIN, "accept(2)");
986+
}
987+
970988
private Channel getOpenChannel() {
971989
return getOpenFileChecked().channel();
972990
}

‎core/src/main/java/org/jruby/ext/socket/RubyServerSocket.java‎

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@
2929
package org.jruby.ext.socket;
3030

3131
import jnr.constants.platform.Sock;
32+
import org.jruby.FiberScheduler;
3233
import org.jruby.Ruby;
34+
import org.jruby.RubyArray;
3335
import org.jruby.RubyClass;
3436
import org.jruby.RubyFixnum;
3537
import org.jruby.anno.JRubyClass;
@@ -49,6 +51,7 @@
4951
import java.nio.channels.Channel;
5052
import java.nio.channels.IllegalBlockingModeException;
5153
import java.nio.channels.SelectableChannel;
54+
import java.nio.channels.SelectionKey;
5255
import java.nio.channels.ServerSocketChannel;
5356
import java.nio.channels.SocketChannel;
5457

@@ -112,7 +115,30 @@ public IRubyObject bind(ThreadContext context, IRubyObject addr, IRubyObject bac
112115

113116
@JRubyMethod()
114117
public IRubyObject accept(ThreadContext context) {
115-
return doAccept(this, context, true);
118+
IRubyObject scheduler = FiberScheduler.current(context);
119+
if (scheduler != null) return schedulerAccept(context, scheduler);
120+
121+
while (true) {
122+
// :wait_readable when another thread or fiber has the channel nonblocking, so wait instead of raising EAGAIN
123+
IRubyObject result = doAccept(this, context, false);
124+
if (result instanceof RubyArray) return result;
125+
126+
if (!context.getThread().select(this, SelectionKey.OP_ACCEPT)) {
127+
// we were woken up without being selected...poll for thread events and go back to sleep
128+
context.pollThreadEvents();
129+
}
130+
}
131+
}
132+
133+
// MRI: rsock_s_accept, which retries a nonblocking accept(2) whenever io_wait reports the server readable
134+
private IRubyObject schedulerAccept(ThreadContext context, IRubyObject scheduler) {
135+
while (true) {
136+
// [Socket, Addrinfo], or :wait_readable with the channel's blocking lock already released
137+
IRubyObject result = doAcceptNonblock(this, context, false);
138+
if (result instanceof RubyArray) return result;
139+
140+
schedulerWaitAcceptable(context, scheduler);
141+
}
116142
}
117143

118144
@JRubyMethod()

‎core/src/main/java/org/jruby/ext/socket/RubyTCPServer.java‎

Lines changed: 48 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828

2929
package org.jruby.ext.socket;
3030

31+
import org.jruby.FiberScheduler;
3132
import org.jruby.Ruby;
3233
import org.jruby.RubyBasicObject;
3334
import org.jruby.RubyClass;
@@ -145,6 +146,9 @@ public IRubyObject accept(ThreadContext context) {
145146
RubyTCPSocket socket = new RubyTCPSocket(context.runtime, Access.getClass(context, "TCPSocket"));
146147

147148
try {
149+
IRubyObject scheduler = FiberScheduler.current(context);
150+
if (scheduler != null) return initAcceptedSocket(context, socket, schedulerAccept(context, scheduler));
151+
148152
RubyThread thread = context.getThread();
149153

150154
while (true) {
@@ -160,16 +164,8 @@ public IRubyObject accept(ThreadContext context) {
160164

161165
connected.finishConnect();
162166

163-
// Force the client socket to be blocking
164-
synchronized (connected.blockingLock()) {
165-
connected.configureBlocking(false);
166-
connected.configureBlocking(true);
167-
}
168-
169167
// otherwise one key has been selected (ours) so we get the channel and hand it off
170-
socket.initSocket(newChannelFD(context.runtime, connected));
171-
172-
return socket;
168+
return initAcceptedSocket(context, socket, connected);
173169
}
174170
}
175171

@@ -178,6 +174,46 @@ public IRubyObject accept(ThreadContext context) {
178174
}
179175
}
180176

177+
private static RubyTCPSocket initAcceptedSocket(ThreadContext context, RubyTCPSocket socket, SocketChannel connected) throws IOException {
178+
// Force the client socket to be blocking
179+
synchronized (connected.blockingLock()) {
180+
connected.configureBlocking(false);
181+
connected.configureBlocking(true);
182+
}
183+
184+
socket.initSocket(newChannelFD(context.runtime, connected));
185+
186+
return socket;
187+
}
188+
189+
// MRI: rsock_s_accept, which retries a nonblocking accept(2) whenever io_wait reports the server readable
190+
private SocketChannel schedulerAccept(ThreadContext context, IRubyObject scheduler) throws IOException {
191+
while (true) {
192+
ServerSocketChannel ssc = getServerSocketChannel();
193+
SocketChannel connected;
194+
195+
synchronized (ssc.blockingLock()) {
196+
boolean oldBlocking = ssc.isBlocking();
197+
198+
try {
199+
ssc.configureBlocking(false);
200+
connected = ssc.accept();
201+
} finally {
202+
try {ssc.configureBlocking(oldBlocking);} catch (IOException ioe) {}
203+
}
204+
}
205+
206+
if (connected != null) {
207+
connected.finishConnect();
208+
209+
return connected;
210+
}
211+
212+
// outside the blocking lock, which a sibling fiber may need to accept on this server
213+
schedulerWaitAcceptable(context, scheduler);
214+
}
215+
}
216+
181217
@JRubyMethod(name = "accept_nonblock")
182218
public IRubyObject accept_nonblock(ThreadContext context) {
183219
return accept_nonblock(context, context.runtime, true);
@@ -235,6 +271,9 @@ public IRubyObject accept_nonblock(ThreadContext context, Ruby runtime, boolean
235271
@JRubyMethod(name = "sysaccept")
236272
public IRubyObject sysaccept(ThreadContext context) {
237273
try {
274+
IRubyObject scheduler = FiberScheduler.current(context);
275+
if (scheduler != null) return asFixnum(context, newChannelFD(context.runtime, schedulerAccept(context, scheduler)).bestFileno());
276+
238277
RubyThread thread = context.getThread();
239278

240279
while (true) {

‎core/src/main/java/org/jruby/ext/socket/RubyUNIXServer.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import jnr.unixsocket.UnixServerSocketChannel;
3333
import jnr.unixsocket.UnixSocketAddress;
3434
import jnr.unixsocket.UnixSocketChannel;
35+
import org.jruby.FiberScheduler;
3536
import org.jruby.Ruby;
3637
import org.jruby.RubyClass;
3738
import org.jruby.anno.JRubyClass;
@@ -83,6 +84,9 @@ public IRubyObject accept(ThreadContext context) {
8384
}
8485

8586
private UnixSocketChannel acceptChannel(ThreadContext context) {
87+
IRubyObject scheduler = FiberScheduler.current(context);
88+
if (scheduler != null) return schedulerAcceptChannel(context, scheduler);
89+
8690
try {
8791

8892
while (true) { // select loop to allow interrupting
@@ -106,6 +110,35 @@ private UnixSocketChannel acceptChannel(ThreadContext context) {
106110
}
107111
}
108112

113+
// MRI: rsock_s_accept, which retries a nonblocking accept(2) whenever io_wait reports the server readable
114+
private UnixSocketChannel schedulerAcceptChannel(ThreadContext context, IRubyObject scheduler) {
115+
while (true) {
116+
SelectableChannel selectable = (SelectableChannel) getChannel();
117+
UnixSocketChannel socketChannel;
118+
119+
synchronized (selectable.blockingLock()) {
120+
boolean oldBlocking = selectable.isBlocking();
121+
122+
try {
123+
selectable.configureBlocking(false);
124+
socketChannel = ((UnixServerSocketChannel) selectable).accept();
125+
} catch (IOException ioe) {
126+
if (!"accept failed: Resource temporarily unavailable".equals(ioe.getMessage())) {
127+
throw context.runtime.newIOErrorFromException(ioe);
128+
}
129+
socketChannel = null;
130+
} finally {
131+
try {selectable.configureBlocking(oldBlocking);} catch (IOException ioe) {}
132+
}
133+
}
134+
135+
if (socketChannel != null) return socketChannel;
136+
137+
// outside the blocking lock, which a sibling fiber may need to accept on this server
138+
schedulerWaitAcceptable(context, scheduler);
139+
}
140+
}
141+
109142
@JRubyMethod
110143
public IRubyObject accept_nonblock(ThreadContext context) {
111144
return accept_nonblock(context, context.runtime, true);

‎core/src/main/java/org/jruby/ir/builder/IRBuilder.java‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2604,15 +2604,29 @@ protected Operand buildRange(U beginNode, U endNode, boolean isExclusive) {
26042604
end = build(endNode);
26052605
}
26062606

2607-
if (begin instanceof ImmutableLiteral && end instanceof ImmutableLiteral) {
2607+
if (begin instanceof ImmutableLiteral b && end instanceof ImmutableLiteral e && isValidRange(b, e)) {
26082608
// endpoints are free of side effects, cache the range after creation
2609-
return new Range((ImmutableLiteral) begin, (ImmutableLiteral) end, isExclusive);
2609+
return new Range(b, e, isExclusive);
26102610
}
26112611

26122612
// must be built every time
26132613
return addResultInstr(new BuildRangeInstr(temp(), begin, end, isExclusive));
26142614
}
26152615

2616+
// Whether a range of these literals can be built without raising. A Range operand is built wherever its value is
2617+
// used, which can be outside a rescue or after the scope has popped its frame, so one that raises (like 1.."3",
2618+
// whose endpoints aren't comparable) has to be built by an instruction instead.
2619+
private static boolean isValidRange(ImmutableLiteral begin, ImmutableLiteral end) {
2620+
if (begin instanceof Nil || end instanceof Nil) return true;
2621+
if (isRealNumber(begin) && isRealNumber(end)) return true;
2622+
if (begin instanceof FrozenString && end instanceof FrozenString) return true;
2623+
return begin instanceof Symbol && end instanceof Symbol;
2624+
}
2625+
2626+
private static boolean isRealNumber(ImmutableLiteral literal) {
2627+
return literal instanceof Fixnum || literal instanceof Bignum || literal instanceof org.jruby.ir.operands.Float || literal instanceof Rational;
2628+
}
2629+
26162630
protected Operand buildRational(U numerator, U denominator) {
26172631
return new Rational((ImmutableLiteral) build(numerator), (ImmutableLiteral) build(denominator));
26182632
}

‎core/src/main/java/org/jruby/ir/instructions/IntegerMathInstr.java‎

Lines changed: 2 additions & 0 deletions

0 commit comments

Comments
 (0)