From 41fc5e8622d2c1e43ffa9c7f48fdbcfbe492c19c Mon Sep 17 00:00:00 2001 From: Vincent Latombe Date: Wed, 25 Mar 2020 21:38:41 +0100 Subject: [PATCH] [JENKINS-61409] Websockets: Use AbstractByteBufferCommandTransport to transport messages (#4596) * [JENKINS-61409] Websockets: Use AbstractByteBufferCommandTransport to transport messages * Depend on timestamped snapshot since incrementals is broken * Fix reviews * Update snapshot dep * Print directly --- .../java/jenkins/agents/WebSocketAgents.java | 40 +++++++------------ .../jenkins/websocket/WebSocketSession.java | 10 +++++ pom.xml | 2 +- .../jenkins/agents/WebSocketAgentsTest.java | 18 ++++++++- 4 files changed, 43 insertions(+), 27 deletions(-) diff --git a/core/src/main/java/jenkins/agents/WebSocketAgents.java b/core/src/main/java/jenkins/agents/WebSocketAgents.java index 965c3b67eb..4cbfd9465f 100644 --- a/core/src/main/java/jenkins/agents/WebSocketAgents.java +++ b/core/src/main/java/jenkins/agents/WebSocketAgents.java @@ -30,17 +30,16 @@ import hudson.ExtensionList; import hudson.model.Computer; import hudson.model.InvisibleAction; import hudson.model.UnprotectedRootAction; -import hudson.remoting.AbstractByteArrayCommandTransport; +import hudson.remoting.AbstractByteBufferCommandTransport; import hudson.remoting.Capability; -import hudson.remoting.Channel; import hudson.remoting.ChannelBuilder; +import hudson.remoting.ChunkHeader; import hudson.remoting.Engine; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.ClosedChannelException; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; -import java.util.Arrays; import java.util.Collections; import java.util.logging.Level; import java.util.logging.Logger; @@ -106,7 +105,7 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte private final JnlpConnectionState state; private final String agent; private final Capability remoteCapability; - private AbstractByteArrayCommandTransport.ByteArrayReceiver receiver; + private Transport transport; Session(JnlpConnectionState state, String agent, Capability remoteCapability) { this.state = state; @@ -119,7 +118,8 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte Computer.threadPoolForRemoting.submit(() -> { LOGGER.fine(() -> "setting up channel for " + agent); state.fireBeforeChannel(new ChannelBuilder(agent, Computer.threadPoolForRemoting)); - state.fireAfterChannel(state.getChannelBuilder().build(new Transport())); + transport = new Transport(); + state.fireAfterChannel(state.getChannelBuilder().build(transport)); LOGGER.fine(() -> "set up channel for " + agent); return null; }); @@ -128,10 +128,10 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte @Override protected void binary(byte[] payload, int offset, int len) { LOGGER.finest(() -> "reading block of length " + len + " from " + agent); - if (offset == 0 && len == payload.length) { - receiver.handle(payload); - } else { - receiver.handle(Arrays.copyOfRange(payload, offset, offset + len)); + try { + transport.receive(ByteBuffer.wrap(payload, offset, len)); + } catch (IOException | InterruptedException e) { + error(e); } } @@ -139,7 +139,7 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte protected void closed(int statusCode, String reason) { LOGGER.finest(() -> "closed " + statusCode + " " + reason); IOException x = new ClosedChannelException(); - receiver.terminate(x); + transport.terminate(x); state.fireChannelClosed(x); state.fireAfterDisconnect(); } @@ -149,22 +149,13 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte LOGGER.log(Level.WARNING, null, cause); } - class Transport extends AbstractByteArrayCommandTransport { + class Transport extends AbstractByteBufferCommandTransport { @Override - public void setup(AbstractByteArrayCommandTransport.ByteArrayReceiver bar) { - receiver = bar; - } - - @Override - public void writeBlock(Channel chnl, byte[] bytes) throws IOException { - LOGGER.finest(() -> "writing block of length " + bytes.length + " to " + agent); - try { - sendBinary(ByteBuffer.wrap(bytes)).get(); - } catch (Exception x) { - x.printStackTrace(); - throw new IOException(x); - } + protected void write(ByteBuffer header, ByteBuffer data) throws IOException { + LOGGER.finest(() -> "sending message of length " + ChunkHeader.length(ChunkHeader.peek(header))); + sendBinary(header, false); + sendBinary(data, true); } @Override @@ -183,7 +174,6 @@ public final class WebSocketAgents extends InvisibleAction implements Unprotecte LOGGER.finest(() -> "closeRead"); close(); } - } } diff --git a/core/src/main/java/jenkins/websocket/WebSocketSession.java b/core/src/main/java/jenkins/websocket/WebSocketSession.java index 7330bd6cef..78400a77a7 100644 --- a/core/src/main/java/jenkins/websocket/WebSocketSession.java +++ b/core/src/main/java/jenkins/websocket/WebSocketSession.java @@ -60,6 +60,7 @@ public abstract class WebSocketSession { private static final Logger LOGGER = Logger.getLogger(WebSocketSession.class.getName()); private Object session; + // https://www.eclipse.org/jetty/javadoc/9.4.24.v20191120/org/eclipse/jetty/websocket/common/WebSocketRemoteEndpoint.html private Object remoteEndpoint; private ScheduledFuture pings; @@ -130,6 +131,15 @@ public abstract class WebSocketSession { } } + @SuppressWarnings("unchecked") + protected final void sendBinary(ByteBuffer partialByte, boolean isLast) { + try { + remoteEndpoint.getClass().getMethod("sendPartialBytes", ByteBuffer.class, boolean.class).invoke(remoteEndpoint, partialByte, isLast); + } catch (Exception x) { + throw new RuntimeException(x); + } + } + @SuppressWarnings("unchecked") protected final Future sendText(String text) { try { diff --git a/pom.xml b/pom.xml index 99b41ae91f..311779ba35 100755 --- a/pom.xml +++ b/pom.xml @@ -101,7 +101,7 @@ THE SOFTWARE. 3.2.3 - 4.2 + 4.3-20200323.083513-2 3.14 diff --git a/test/src/test/java/jenkins/agents/WebSocketAgentsTest.java b/test/src/test/java/jenkins/agents/WebSocketAgentsTest.java index 37c47dcfd3..009a99963e 100644 --- a/test/src/test/java/jenkins/agents/WebSocketAgentsTest.java +++ b/test/src/test/java/jenkins/agents/WebSocketAgentsTest.java @@ -35,6 +35,7 @@ import hudson.slaves.SlaveComputer; import hudson.tasks.BatchFile; import hudson.tasks.Shell; import java.io.File; +import java.util.Random; import java.util.concurrent.atomic.AtomicReference; import java.util.logging.Level; import java.util.logging.Logger; @@ -99,9 +100,10 @@ public class WebSocketAgentsTest { ).stdout(System.out).start()); r.waitOnline(s); assertEquals("response", s.getChannel().call(new DummyTask())); + assertNotNull(s.getChannel().call(new FatTask())); FreeStyleProject p = r.createFreeStyleProject(); p.setAssignedNode(s); - p.getBuildersList().add(Functions.isWindows() ? new BatchFile("echo hello") : new Shell("echo hello")); + p.getBuildersList().add(Functions.isWindows() ? new BatchFile("echo hello") : new Shell("dd if=/dev/random count=1024 bs=200")); r.buildAndAssertSuccess(p); s.toComputer().getLogText().writeLogTo(0, System.out); } finally { @@ -122,4 +124,18 @@ public class WebSocketAgentsTest { } } + private static class FatTask extends SlaveToMasterCallable { + private byte[] payload; + + private FatTask() { + payload = new byte[1024 * 1024]; + new Random().nextBytes(payload); + } + + @Override + public String call() { + return new String(payload); + } + } + }