diff options
Diffstat (limited to 'src/main/java/com/google/devtools')
4 files changed, 84 insertions, 10 deletions
diff --git a/src/main/java/com/google/devtools/build/lib/remote/BUILD b/src/main/java/com/google/devtools/build/lib/remote/BUILD index 18e38edb9e..5c4ef45f49 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/BUILD +++ b/src/main/java/com/google/devtools/build/lib/remote/BUILD @@ -37,6 +37,7 @@ java_library( "//src/main/java/com/google/devtools/common/options", "//third_party:auth", "//third_party:guava", + "//third_party:netty", "//third_party/grpc:grpc-jar", "//third_party/protobuf:protobuf_java", "//third_party/protobuf:protobuf_java_util", diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java index bb61082a3d..90f42077c4 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java +++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteOptions.java @@ -36,6 +36,17 @@ public final class RemoteOptions extends OptionsBase { public String remoteHttpCache; @Option( + name = "remote_cache_proxy", + defaultValue = "null", + documentationCategory = OptionDocumentationCategory.UNCATEGORIZED, + effectTags = {OptionEffectTag.UNKNOWN}, + help = + "Connect to the remote cache through a proxy. Currently this flag can only be used to " + + "configure a Unix domain socket (unix:/path/to/socket) for the HTTP cache." + ) + public String remoteCacheProxy; + + @Option( name = "remote_max_connections", defaultValue = "100", documentationCategory = OptionDocumentationCategory.UNCATEGORIZED, diff --git a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java index 47a14bffdd..ef8ffe22f7 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java +++ b/src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreFactory.java @@ -22,6 +22,7 @@ import com.google.devtools.build.lib.remote.blobstore.SimpleBlobStore; import com.google.devtools.build.lib.remote.blobstore.http.HttpBlobStore; import com.google.devtools.build.lib.vfs.Path; import com.google.devtools.build.lib.vfs.PathFragment; +import io.netty.channel.unix.DomainSocketAddress; import java.io.IOException; import java.net.URI; import java.util.concurrent.TimeUnit; @@ -37,11 +38,20 @@ public final class SimpleBlobStoreFactory { public static SimpleBlobStore createRest(RemoteOptions options, Credentials creds) { try { - return new HttpBlobStore( - URI.create(options.remoteHttpCache), - (int) TimeUnit.SECONDS.toMillis(options.remoteTimeout), - options.remoteMaxConnections, - creds); + URI uri = URI.create(options.remoteHttpCache); + int timeoutMillis = (int) TimeUnit.SECONDS.toMillis(options.remoteTimeout); + + if (options.remoteCacheProxy != null) { + if (options.remoteCacheProxy.startsWith("unix:")) { + return HttpBlobStore.create( + new DomainSocketAddress(options.remoteCacheProxy.replaceFirst("^unix:", "")), + uri, timeoutMillis, options.remoteMaxConnections, creds); + } else { + throw new Exception("Remote cache proxy unsupported: " + options.remoteCacheProxy); + } + } else { + return HttpBlobStore.create(uri, timeoutMillis, options.remoteMaxConnections, creds); + } } catch (Exception e) { throw new RuntimeException(e); } diff --git a/src/main/java/com/google/devtools/build/lib/remote/blobstore/http/HttpBlobStore.java b/src/main/java/com/google/devtools/build/lib/remote/blobstore/http/HttpBlobStore.java index a017d2a9c5..0bcc68bd64 100644 --- a/src/main/java/com/google/devtools/build/lib/remote/blobstore/http/HttpBlobStore.java +++ b/src/main/java/com/google/devtools/build/lib/remote/blobstore/http/HttpBlobStore.java @@ -24,12 +24,20 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.epoll.Epoll; +import io.netty.channel.epoll.EpollDomainSocketChannel; +import io.netty.channel.epoll.EpollEventLoopGroup; +import io.netty.channel.kqueue.KQueue; +import io.netty.channel.kqueue.KQueueDomainSocketChannel; +import io.netty.channel.kqueue.KQueueEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.pool.ChannelPool; import io.netty.channel.pool.ChannelPoolHandler; import io.netty.channel.pool.FixedChannelPool; import io.netty.channel.pool.SimpleChannelPool; import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.channel.unix.DomainSocketAddress; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpObjectAggregator; @@ -53,11 +61,14 @@ import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.URI; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; import java.util.regex.Pattern; import javax.annotation.Nullable; import javax.annotation.concurrent.GuardedBy; @@ -92,7 +103,7 @@ public final class HttpBlobStore implements SimpleBlobStore { private static final Pattern INVALID_TOKEN_ERROR = Pattern.compile("\\s*error\\s*=\\s*\"?invalid_token\"?"); - private final NioEventLoopGroup eventLoop = new NioEventLoopGroup(2 /* number of threads */); + private final EventLoopGroup eventLoop; private final ChannelPool channelPool; private final URI uri; private final int timeoutMillis; @@ -105,10 +116,44 @@ public final class HttpBlobStore implements SimpleBlobStore { @GuardedBy("credentialsLock") private long lastRefreshTime; - @SuppressWarnings("FutureReturnValueIgnored") - public HttpBlobStore( + public static HttpBlobStore create(URI uri, int timeoutMillis, + int remoteMaxConnections, @Nullable final Credentials creds) + throws Exception { + return new HttpBlobStore( + NioEventLoopGroup::new, + NioSocketChannel.class, + uri, timeoutMillis, remoteMaxConnections, creds, + null); + } + + public static HttpBlobStore create( + DomainSocketAddress domainSocketAddress, URI uri, int timeoutMillis, int remoteMaxConnections, @Nullable final Credentials creds) throws Exception { + + if (KQueue.isAvailable()) { + return new HttpBlobStore( + KQueueEventLoopGroup::new, + KQueueDomainSocketChannel.class, + uri, timeoutMillis, remoteMaxConnections, creds, + domainSocketAddress); + } else if (Epoll.isAvailable()) { + return new HttpBlobStore( + EpollEventLoopGroup::new, + EpollDomainSocketChannel.class, + uri, timeoutMillis, remoteMaxConnections, creds, + domainSocketAddress); + } else { + throw new Exception("Unix domain sockets are unsupported on this platform"); + } + } + + private HttpBlobStore( + Function<Integer, EventLoopGroup> newEventLoopGroup, + Class<? extends Channel> channelClass, + URI uri, int timeoutMillis, int remoteMaxConnections, @Nullable final Credentials creds, + @Nullable SocketAddress socketAddress) + throws Exception { boolean useTls = uri.getScheme().equals("https"); if (uri.getPort() == -1) { int port = useTls ? 443 : 80; @@ -123,6 +168,10 @@ public final class HttpBlobStore implements SimpleBlobStore { uri.getFragment()); } this.uri = uri; + if (socketAddress == null) { + socketAddress = new InetSocketAddress(uri.getHost(), uri.getPort()); + } + final SslContext sslCtx; if (useTls) { // OpenSsl gives us a > 2x speed improvement on fast networks, but requires netty tcnative @@ -132,12 +181,15 @@ public final class HttpBlobStore implements SimpleBlobStore { } else { sslCtx = null; } + + this.eventLoop = newEventLoopGroup.apply(2); Bootstrap clientBootstrap = new Bootstrap() - .channel(NioSocketChannel.class) + .channel(channelClass) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeoutMillis) .group(eventLoop) - .remoteAddress(uri.getHost(), uri.getPort()); + .remoteAddress(socketAddress); + ChannelPoolHandler channelPoolHandler = new ChannelPoolHandler() { @Override |