diff options
Diffstat (limited to 'container-search/src/main/java/com/yahoo/search')
3 files changed, 33 insertions, 78 deletions
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/CloseableChannel.java b/container-search/src/main/java/com/yahoo/search/dispatch/CloseableChannel.java index 643b8f81318..3f5ebe53d0d 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/CloseableChannel.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/CloseableChannel.java @@ -1,54 +1,41 @@ // Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.search.dispatch; -import com.yahoo.fs4.BasicPacket; -import com.yahoo.fs4.ChannelTimeoutException; -import com.yahoo.fs4.mplex.Backend; -import com.yahoo.fs4.mplex.FS4Channel; -import com.yahoo.fs4.mplex.InvalidChannelException; +import com.yahoo.fs4.QueryPacket; +import com.yahoo.prelude.fastsearch.CacheKey; import com.yahoo.search.Query; +import com.yahoo.search.Result; import java.io.Closeable; import java.io.IOException; -import java.util.Optional; /** + * CloseableChannel is an interface for running a search query and getting document summaries against some + * content node, node group or dispatcher while abstracting the specifics of the invocation target. + * * @author ollivir */ -public class CloseableChannel implements Closeable { - private FS4Channel channel; - private final Optional<Integer> distributionKey; +public abstract class CloseableChannel implements Closeable { + /** Retrieve the hits for the given {@link Query} */ + public abstract Result search(Query query, QueryPacket queryPacket, CacheKey cacheKey) throws IOException; - public CloseableChannel(Backend backend) { - this(backend, Optional.empty()); - } - - public CloseableChannel(Backend backend, Optional<Integer> distributionKey) { - this.channel = backend.openChannel(); - this.distributionKey = distributionKey; - } + /** Retrieve document summaries for the unfilled hits in the given {@link Result} */ + public abstract void partialFill(Result result, String summaryClass); - public void setQuery(Query query) { - channel.setQuery(query); - } + protected abstract void closeChannel(); - public boolean sendPacket(BasicPacket packet) throws InvalidChannelException, IOException { - return channel.sendPacket(packet); - } - - public BasicPacket[] receivePackets(long timeout, int packetCount) throws InvalidChannelException, ChannelTimeoutException { - return channel.receivePackets(timeout, packetCount); - } + private Runnable teardown = null; - public Optional<Integer> distributionKey() { - return distributionKey; + public void teardown(Runnable teardown) { + this.teardown = teardown; } @Override - public void close() { - if (channel != null) { - channel.close(); - channel = null; + public final void close() { + if (teardown != null) { + teardown.run(); + teardown = null; } + closeChannel(); } } diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/DispatchedChannel.java b/container-search/src/main/java/com/yahoo/search/dispatch/DispatchedChannel.java deleted file mode 100644 index d005d9491d5..00000000000 --- a/container-search/src/main/java/com/yahoo/search/dispatch/DispatchedChannel.java +++ /dev/null @@ -1,38 +0,0 @@ -// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. -package com.yahoo.search.dispatch; - -import com.yahoo.prelude.fastsearch.FS4ResourcePool; -import com.yahoo.search.dispatch.SearchCluster.Group; -import com.yahoo.search.dispatch.SearchCluster.Node; - -import java.util.Optional; - -/** - * An extension to CloseableChannel that encapsulates the release of a LoadBalancer group allocation. - * - * @author ollivir - */ -public class DispatchedChannel extends CloseableChannel { - private final SearchCluster.Group group; - private final LoadBalancer loadBalancer; - private boolean groupAllocated = true; - - public DispatchedChannel(FS4ResourcePool fs4ResourcePool, LoadBalancer loadBalancer, Group group, Node node) { - super(fs4ResourcePool.getBackend(node.hostname(), node.fs4port()), Optional.of(node.key())); - - this.loadBalancer = loadBalancer; - this.group = group; - } - - public DispatchedChannel(FS4ResourcePool fs4ResourcePool, LoadBalancer loadBalancer, Group group) { - this(fs4ResourcePool, loadBalancer, group, group.nodes().iterator().next()); - } - - public void close() { - if (groupAllocated) { - groupAllocated = false; - loadBalancer.releaseGroup(group); - } - super.close(); - } -} diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java index c383b681558..0cf18852dd3 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java @@ -8,18 +8,20 @@ import com.yahoo.compress.CompressionType; import com.yahoo.compress.Compressor; import com.yahoo.container.handler.VipStatus; import com.yahoo.container.protect.Error; -import com.yahoo.prelude.fastsearch.DocumentDatabase; -import com.yahoo.slime.ArrayTraverser; +import com.yahoo.data.access.Inspector; import com.yahoo.data.access.slime.SlimeAdapter; +import com.yahoo.prelude.fastsearch.DocumentDatabase; +import com.yahoo.prelude.fastsearch.FS4CloseableChannel; import com.yahoo.prelude.fastsearch.FS4ResourcePool; import com.yahoo.prelude.fastsearch.FastHit; import com.yahoo.prelude.fastsearch.TimeoutException; +import com.yahoo.prelude.fastsearch.VespaBackEndSearcher; import com.yahoo.search.Query; import com.yahoo.search.Result; import com.yahoo.search.query.SessionId; import com.yahoo.search.result.ErrorMessage; import com.yahoo.search.result.Hit; -import com.yahoo.data.access.Inspector; +import com.yahoo.slime.ArrayTraverser; import com.yahoo.slime.BinaryFormat; import com.yahoo.slime.Cursor; import com.yahoo.slime.Slime; @@ -52,7 +54,7 @@ public class Dispatcher extends AbstractComponent { /** A model of the search cluster this dispatches to */ private final SearchCluster searchCluster; - + /** Connections to the search nodes this talks to, indexed by node id ("partid") */ private final ImmutableMap<Integer, Client.NodeConnection> nodeConnections; @@ -84,7 +86,7 @@ public class Dispatcher extends AbstractComponent { this.fs4ResourcePool = null; this.loadBalancer = new LoadBalancer(searchCluster); } - + /** Returns the search cluster this dispatches to */ public SearchCluster searchCluster() { return searchCluster; } @@ -283,14 +285,18 @@ public class Dispatcher extends AbstractComponent { } - public Optional<CloseableChannel> getDispatchedChannel(Query query) { + public Optional<CloseableChannel> getDispatchedChannel(VespaBackEndSearcher searcher, Query query) { Optional<SearchCluster.Group> groupInCluster = loadBalancer.takeGroupForQuery(query); return groupInCluster.flatMap(group -> { if(group.nodes().size() == 1) { SearchCluster.Node node = group.nodes().iterator().next(); query.trace(false, 2, "Dispatching internally to ", group, " (", node.toString(), ")"); - return Optional.of(new DispatchedChannel(fs4ResourcePool, loadBalancer, group)); + CloseableChannel channel = new FS4CloseableChannel(searcher, query, fs4ResourcePool, node.hostname(), node.fs4port(), node.key()); + channel.teardown(() -> { + loadBalancer.releaseGroup(group); + }); + return Optional.of(channel); } else { loadBalancer.releaseGroup(group); return Optional.empty(); |