diff options
8 files changed, 335 insertions, 91 deletions
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java index f539e3affc..318282fc27 100644 --- a/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java +++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventTransport.java @@ -11,24 +11,43 @@ // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. - package com.google.devtools.build.lib.buildeventstream; -import java.io.IOException; +import java.util.concurrent.Future; +import javax.annotation.concurrent.ThreadSafe; -/** Interface for transports of the build-event stream. */ +/** + * Transport interface for the build event protocol. + * + * <p>All implementations need to be thread-safe. All methods are expected to return quickly. + * + * <p>Notice that this interface does not provide any error handling API. A transport may choose + * to log interesting errors to the command line and/or abort the whole build. + */ +@ThreadSafe public interface BuildEventTransport { /** - * Ensure that the event will eventually be reported to the receiver this object is a transport - * for; the transport is responsible that events arrive at the endpoint in the order they are sent - * by invocations of this method. + * Writes a build event to an endpoint. This method will always return quickly and will not + * wait for the write to complete. + * + * <p>In case the transport is in error, this method still needs to be able to accept build + * events. It may choose to ignore them, though. + * + * <p>This method should not throw any exceptions. + * + * @param event the event to sendBuildEvent. */ - void sendBuildEvent(BuildEvent event) throws IOException; + void sendBuildEvent(BuildEvent event); /** - * Close all open resources, if any. This method will be called on the transport after all events - * have been sent. If a transport is stateless, it is correct to do nothing. + * Initiates a close. Callers may listen to the returned future to be notified when the close + * is complete i.e. wait for all build events to be sent. The future does not contain any + * information about possible transport errors. + * + * <p>This method might be called multiple times without any effect after the first call. + * + * <p>This method should not throw any exceptions. */ - void close() throws IOException; + Future<Void> close(); } diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java index 3356c230ca..ccfa14c0b2 100644 --- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java +++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java @@ -14,46 +14,45 @@ package com.google.devtools.build.lib.buildeventstream.transports; +import static com.google.common.base.Preconditions.checkNotNull; + import com.google.devtools.build.lib.buildeventstream.BuildEvent; -import com.google.devtools.build.lib.buildeventstream.BuildEventConverters; +import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos; import com.google.devtools.build.lib.buildeventstream.BuildEventTransport; import com.google.devtools.build.lib.buildeventstream.PathConverter; -import java.io.BufferedOutputStream; -import java.io.File; -import java.io.FileOutputStream; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.util.logging.Level; +import java.util.logging.Logger; /** - * A simple {@link BuildEventTransport} that writes varint delimited binary representation of event - * {@link BuildEvent} protocol-buffers to a file. Files written by this Transport can be read by - * successive calls of {code BuildEvent.Builder#mergeDelimitedFrom(InputStream)} (or the static - * method {@code BuildEvent.parseDelimitedFrom(InputStream)}). + * A simple {@link BuildEventTransport} that writes a varint delimited binary representation of + * {@link BuildEvent} protocol buffers to a file. */ -public final class BinaryFormatFileTransport implements BuildEventTransport { - private final BufferedOutputStream out; - private final PathConverter pathConverter; - - public BinaryFormatFileTransport(String path, PathConverter pathConverter) - throws IOException { - this.out = new BufferedOutputStream(new FileOutputStream(new File(path))); - this.pathConverter = pathConverter; - } +public final class BinaryFormatFileTransport extends FileTransport { - @Override - public synchronized void sendBuildEvent(BuildEvent event) throws IOException { - BuildEventConverters converters = - new BuildEventConverters() { - @Override - public PathConverter pathConverter() { - return pathConverter; - } - }; - event.asStreamProto(converters).writeDelimitedTo(out); - out.flush(); + private static final Logger log = Logger.getLogger(BinaryFormatFileTransport.class.getName()); + + private static final int MAX_VARINT_BYTES = 9; + + BinaryFormatFileTransport(String path, PathConverter pathConverter) { + super(path, pathConverter); } @Override - public void close() throws IOException { - out.close(); + public void sendBuildEvent(BuildEvent event) { + checkNotNull(event); + BuildEventStreamProtos.BuildEvent protoEvent = event.asStreamProto(converters); + + int maxSerializedSize = MAX_VARINT_BYTES + protoEvent.getSerializedSize(); + ByteArrayOutputStream out = new ByteArrayOutputStream(maxSerializedSize); + + try { + protoEvent.writeDelimitedTo(out); + writeData(out.toByteArray()); + } catch (IOException e) { + log.log(Level.SEVERE, e.getMessage(), e); + close(); + } } } diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java new file mode 100644 index 0000000000..e691df1819 --- /dev/null +++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java @@ -0,0 +1,161 @@ +// Copyright 2017 The Bazel Authors. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.google.devtools.build.lib.buildeventstream.transports; + +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.SettableFuture; +import com.google.devtools.build.lib.buildeventstream.BuildEvent; +import com.google.devtools.build.lib.buildeventstream.BuildEventConverters; +import com.google.devtools.build.lib.buildeventstream.BuildEventTransport; +import com.google.devtools.build.lib.buildeventstream.PathConverter; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.AsynchronousFileChannel; +import java.nio.channels.CompletionHandler; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.Future; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Non-blocking file transport. + * + * <p>Implementors of this class need to implement {@link #sendBuildEvent(BuildEvent)} which + * serializes the build event and writes it to file using {@link #writeData(byte[])}. + */ +abstract class FileTransport implements BuildEventTransport { + + /** + * We use an {@link AsynchronousFileChannel} to perform non-blocking writes to a file. It get's + * tricky when it comes to {@link #close()}, as we may only complete the returned future when + * all writes have completed (succeeded or failed). Thus, we use a field + * {@link #outstandingWrites} to keep track of the number of writes that have not completed yet. + * It's simply incremented before a new write and decremented after a write has completed. When + * it's {@code 0} it's safe to complete the close future. + */ + + private static final Logger log = Logger.getLogger(FileTransport.class.getName()); + + @VisibleForTesting + final AsynchronousFileChannel ch; + private final WriteCompletionHandler completionHandler = new WriteCompletionHandler(); + protected final BuildEventConverters converters; + // The offset in the file to begin the next write at. + private long writeOffset; + // Number of writes that haven't completed yet. + private long outstandingWrites; + // The future returned by close() + private SettableFuture<Void> closeFuture; + + FileTransport(String path, final PathConverter pathConverter) { + try { + ch = AsynchronousFileChannel.open(Paths.get(path), StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); + } catch (IOException e) { + throw new RuntimeException(e); + } + this.converters = new BuildEventConverters() { + @Override + public PathConverter pathConverter() { + return pathConverter; + } + }; + } + + synchronized void writeData(byte[] data) { + checkNotNull(data); + if (!ch.isOpen()) { + close(); + return; + } + if (closing()) { + return; + } + + outstandingWrites++; + + ch.write(ByteBuffer.wrap(data), writeOffset, null, completionHandler); + + writeOffset += data.length; + } + + @Override + public synchronized Future<Void> close() { + if (closing()) { + return closeFuture; + } + closeFuture = SettableFuture.create(); + + if (writesComplete()) { + doClose(); + } + + return closeFuture; + } + + private void doClose() { + try { + ch.force(true); + ch.close(); + } catch (IOException e) { + log.log(Level.SEVERE, e.getMessage(), e); + } finally { + closeFuture.set(null); + } + } + + private boolean closing() { + return closeFuture != null; + } + + private boolean writesComplete() { + return outstandingWrites == 0; + } + + /** + * Handler that's notified when a write completes. + */ + private final class WriteCompletionHandler implements CompletionHandler<Integer, Void> { + + @Override + public void completed(Integer result, Void attachment) { + countWriteAndTryClose(); + } + + @Override + public void failed(Throwable exc, Void attachment) { + log.log(Level.SEVERE, exc.getMessage(), exc); + countWriteAndTryClose(); + // There is no point in trying to continue. Close the transport. + close(); + } + + private void countWriteAndTryClose() { + synchronized (FileTransport.this) { + checkState(outstandingWrites > 0); + + outstandingWrites--; + + if (closing() && writesComplete()) { + doClose(); + } + } + } + } +} diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java index 54e979e739..e331c0c573 100644 --- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java +++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java @@ -15,49 +15,28 @@ package com.google.devtools.build.lib.buildeventstream.transports; import com.google.devtools.build.lib.buildeventstream.BuildEvent; -import com.google.devtools.build.lib.buildeventstream.BuildEventConverters; import com.google.devtools.build.lib.buildeventstream.BuildEventTransport; import com.google.devtools.build.lib.buildeventstream.PathConverter; import com.google.protobuf.TextFormat; -import java.io.File; -import java.io.FileOutputStream; import java.io.IOException; -import java.nio.charset.StandardCharsets; /** * A simple {@link BuildEventTransport} that writes the text representation of the protocol-buffer - * representation of the events to a file. This is mainly useful for debugging. + * representation of the events to a file. + * + * <p>This class is used for debugging. */ -public final class TextFormatFileTransport implements BuildEventTransport { - private FileOutputStream out; - private final PathConverter pathConverter; +public final class TextFormatFileTransport extends FileTransport { - public TextFormatFileTransport(String path, PathConverter pathConverter) - throws IOException { - this.out = new FileOutputStream(new File(path)); - this.pathConverter = pathConverter; + TextFormatFileTransport(String path, PathConverter pathConverter) throws IOException { + super(path, pathConverter); } @Override - public synchronized void sendBuildEvent(BuildEvent event) throws IOException { - if (out != null) { - BuildEventConverters converters = - new BuildEventConverters() { - @Override - public PathConverter pathConverter() { - return pathConverter; - } - }; - String protoTextRepresentation = TextFormat.printToString(event.asStreamProto(converters)); - out.write(("event {\n" + protoTextRepresentation + "}\n\n").getBytes(StandardCharsets.UTF_8)); - out.flush(); - } - } + public void sendBuildEvent(BuildEvent event) { + String protoTextRepresentation = TextFormat.printToString(event.asStreamProto(converters)); + String line = "event {\n" + protoTextRepresentation + "}\n\n"; - @Override - public void close() throws IOException { - if (out != null) { - out.close(); - } + writeData(line.getBytes()); } } diff --git a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java index 9901ceeb49..697cd155a2 100644 --- a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java +++ b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java @@ -34,14 +34,17 @@ import com.google.devtools.build.lib.buildtool.buildevent.BuildInterruptedEvent; import com.google.devtools.build.lib.events.Event; import com.google.devtools.build.lib.events.EventHandler; import com.google.devtools.build.lib.rules.extra.ExtraAction; -import java.io.IOException; +import java.util.ArrayList; import java.util.Collection; import java.util.HashSet; +import java.util.List; import java.util.Set; +import java.util.concurrent.Future; import java.util.logging.Logger; /** Listen for {@link BuildEvent} and stream them to the provided {@link BuildEventTransport}. */ public class BuildEventStreamer implements EventHandler { + private final Collection<BuildEventTransport> transports; private Set<BuildEventId> announcedEvents; private final Set<BuildEventId> postedEvents = new HashSet<>(); @@ -89,15 +92,10 @@ public class BuildEventStreamer implements EventHandler { } for (BuildEventTransport transport : transports) { - try { - if (linkEvent != null) { - transport.sendBuildEvent(linkEvent); - } - transport.sendBuildEvent(event); - } catch (IOException e) { - // TODO(aehlig): signal that the build ought to be aborted - log.severe("Failed to write to build event transport: " + e); + if (linkEvent != null) { + transport.sendBuildEvent(linkEvent); } + transport.sendBuildEvent(event); } } @@ -128,12 +126,18 @@ public class BuildEventStreamer implements EventHandler { } private void close() { + List<Future<Void>> shutdownFutures = new ArrayList<>(transports.size()); + for (BuildEventTransport transport : transports) { + shutdownFutures.add(transport.close()); + } + + // Wait for all transports to close. + for (Future<Void> f : shutdownFutures) { try { - transport.close(); - } catch (IOException e) { - // TODO(aehlig): signal that the build ought to be aborted - log.warning("Failure while closing build event transport: " + e); + f.get(); + } catch (Exception e) { + log.severe("Failed to close a build event transport: " + e); } } } diff --git a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java index 359f59ed93..1697bb9d1f 100644 --- a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java +++ b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java @@ -15,6 +15,8 @@ package com.google.devtools.build.lib.buildeventstream.transports; import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.when; import com.google.devtools.build.lib.buildeventstream.BuildEvent; @@ -26,8 +28,8 @@ import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.Tar import com.google.devtools.build.lib.buildeventstream.PathConverter; import java.io.File; import java.io.FileInputStream; -import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.Future; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -61,7 +63,7 @@ public class BinaryFormatFileTransportTest { } @Test - public void testCreatesFileAndWritesProtoTextFormat() throws IOException { + public void testCreatesFileAndWritesProtoBinaryFormat() throws Exception { File output = tmp.newFile(); BuildEventStreamProtos.BuildEvent started = @@ -85,7 +87,7 @@ public class BinaryFormatFileTransportTest { when(buildEvent.asStreamProto(Matchers.<BuildEventConverters>any())).thenReturn(completed); transport.sendBuildEvent(buildEvent); - transport.close(); + transport.close().get(); try (InputStream in = new FileInputStream(output)) { assertThat(BuildEventStreamProtos.BuildEvent.parseDelimitedFrom(in)).isEqualTo(started); assertThat(BuildEventStreamProtos.BuildEvent.parseDelimitedFrom(in)).isEqualTo(progress); @@ -93,4 +95,81 @@ public class BinaryFormatFileTransportTest { assertThat(in.available()).isEqualTo(0); } } + + @Test + public void testFileDoesNotExist() throws Exception { + // Get a file that doesn't exist by creating a new file and immediately deleting it. + File output = tmp.newFile(); + String path = output.getAbsolutePath(); + assertTrue(output.delete()); + + BuildEventStreamProtos.BuildEvent started = + BuildEventStreamProtos.BuildEvent.newBuilder() + .setStarted(BuildStarted.newBuilder().setCommand("build")) + .build(); + when(buildEvent.asStreamProto(Matchers.<BuildEventConverters>any())).thenReturn(started); + BinaryFormatFileTransport transport = new BinaryFormatFileTransport(path, pathConverter); + transport.sendBuildEvent(buildEvent); + + transport.close().get(); + try (InputStream in = new FileInputStream(output)) { + assertThat(BuildEventStreamProtos.BuildEvent.parseDelimitedFrom(in)).isEqualTo(started); + assertThat(in.available()).isEqualTo(0); + } + } + + @Test + public void testWriteWhenFileClosed() throws Exception { + File output = tmp.newFile(); + + BuildEventStreamProtos.BuildEvent started = + BuildEventStreamProtos.BuildEvent.newBuilder() + .setStarted(BuildStarted.newBuilder().setCommand("build")) + .build(); + when(buildEvent.asStreamProto(Matchers.<BuildEventConverters>any())).thenReturn(started); + + BinaryFormatFileTransport transport = + new BinaryFormatFileTransport(output.getAbsolutePath(), pathConverter); + + // Close the file. + transport.ch.close(); + assertFalse(transport.ch.isOpen()); + + // This should not throw an exception. + transport.sendBuildEvent(buildEvent); + transport.close().get(); + + // Also, nothing should have been written to the file + try (InputStream in = new FileInputStream(output)) { + assertThat(in.available()).isEqualTo(0); + } + } + + @Test + public void testWriteWhenTransportClosed() throws Exception { + File output = tmp.newFile(); + + BuildEventStreamProtos.BuildEvent started = + BuildEventStreamProtos.BuildEvent.newBuilder() + .setStarted(BuildStarted.newBuilder().setCommand("build")) + .build(); + when(buildEvent.asStreamProto(Matchers.<BuildEventConverters>any())).thenReturn(started); + + BinaryFormatFileTransport transport = + new BinaryFormatFileTransport(output.getAbsolutePath(), pathConverter); + + transport.sendBuildEvent(buildEvent); + Future<Void> closeFuture = transport.close(); + // This should not throw an exception, but also not perform any write. + transport.sendBuildEvent(buildEvent); + + closeFuture.get(); + assertFalse(transport.ch.isOpen()); + + // There should have only been one write. + try (InputStream in = new FileInputStream(output)) { + assertThat(BuildEventStreamProtos.BuildEvent.parseDelimitedFrom(in)).isEqualTo(started); + assertThat(in.available()).isEqualTo(0); + } + } } diff --git a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java index ac62c7476c..7a008fabb0 100644 --- a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java +++ b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java @@ -28,7 +28,6 @@ import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.Tar import com.google.devtools.build.lib.buildeventstream.PathConverter; import com.google.protobuf.TextFormat; import java.io.File; -import java.io.IOException; import java.nio.charset.StandardCharsets; import org.junit.After; import org.junit.Before; @@ -63,7 +62,7 @@ public class TextFormatFileTransportTest { } @Test - public void testCreatesFileAndWritesProtoTextFormat() throws IOException { + public void testCreatesFileAndWritesProtoTextFormat() throws Exception { File output = tmp.newFile(); BuildEventStreamProtos.BuildEvent started = @@ -87,7 +86,7 @@ public class TextFormatFileTransportTest { when(buildEvent.asStreamProto(Matchers.<BuildEventConverters>any())).thenReturn(completed); transport.sendBuildEvent(buildEvent); - transport.close(); + transport.close().get(); String contents = trimLines( Joiner.on(System.lineSeparator()) diff --git a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java index 6d5dd01119..8c6483d66d 100644 --- a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java +++ b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java @@ -18,6 +18,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import com.google.common.collect.ImmutableSet; +import com.google.common.util.concurrent.Futures; import com.google.devtools.build.lib.buildeventstream.BuildEvent; import com.google.devtools.build.lib.buildeventstream.BuildEventConverters; import com.google.devtools.build.lib.buildeventstream.BuildEventId; @@ -31,6 +32,7 @@ import com.google.devtools.build.lib.buildtool.buildevent.BuildCompleteEvent; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.concurrent.Future; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -52,7 +54,9 @@ public class BuildEventStreamerTest { } @Override - public void close() {} + public Future<Void> close() { + return Futures.immediateFuture(null); + } List<BuildEvent> getEvents() { return events; |