You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@lucene.apache.org by GitBox <gi...@apache.org> on 2020/04/30 08:54:17 UTC
[GitHub] [lucene-solr] CaoManhDat opened a new pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
CaoManhDat opened a new pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470
# Description
Sending request in an async way inside HttpShardHandler
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat merged pull request #1470: SOLR-14354: HttpShardHandler send requests in async
Posted by GitBox <gi...@apache.org>.
CaoManhDat merged pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418628204
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -54,6 +53,10 @@
import org.apache.solr.util.tracing.GlobalTracer;
import org.apache.solr.util.tracing.SolrRequestCarrier;
+/**
+ * Submit requests in async manner.
+ * This class is not thread-safe so all methods should be called in a same thread.
Review comment:
Use `@SolrSingleThreaded`
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -259,12 +261,10 @@ private ShardResponse take(boolean bailOnError) {
@Override
public void cancelAll() {
- for (Cancellable cancellable : requests) {
+ for (Cancellable cancellable : responseCancellableMap.values()) {
Review comment:
And clear the map?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] shalinmangar commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
shalinmangar commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r445295319
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/util/OnComplete.java
##########
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.solr.client.solrj.util;
+
+/**
+ * Listener for async requests
+ */
+public interface OnComplete<T> {
Review comment:
Perhaps rename to `RequestLifecycleListener`?
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -66,16 +66,20 @@
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
private HttpShardHandlerFactory httpShardHandlerFactory;
- private CompletionService<ShardResponse> completionService;
- private Set<Future<ShardResponse>> pending;
+ private Map<ShardResponse, Cancellable> responseCancellableMap;
+ private BlockingQueue<ShardResponse> responses;
+ private AtomicInteger pending;
private Map<String, List<String>> shardToURLs;
private Http2SolrClient httpClient;
Review comment:
It seems that this `Http2SolrClient` instance can be removed. It is being used in the request method but that method is not used anywhere anymore.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -130,77 +134,64 @@ public void submit(final ShardRequest sreq, final String shard, final Modifiable
final Tracer tracer = GlobalTracer.getTracer();
final Span span = tracer != null ? tracer.activeSpan() : null;
- Callable<ShardResponse> task = () -> {
+ params.remove(CommonParams.WT); // use default (currently javabin)
+ params.remove(CommonParams.VERSION);
+ QueryRequest req = makeQueryRequest(sreq, params, shard);
+ req.setMethod(SolrRequest.METHOD.POST);
- ShardResponse srsp = new ShardResponse();
- if (sreq.nodeName != null) {
- srsp.setNodeName(sreq.nodeName);
- }
- srsp.setShardRequest(sreq);
- srsp.setShard(shard);
- SimpleSolrResponse ssr = new SimpleSolrResponse();
- srsp.setSolrResponse(ssr);
- long startTime = System.nanoTime();
+ LBSolrClient.Req lbReq = httpShardHandlerFactory.newLBHttpSolrClientReq(req, urls);
+
+ ShardResponse srsp = new ShardResponse();
+ if (sreq.nodeName != null) {
+ srsp.setNodeName(sreq.nodeName);
+ }
+ srsp.setShardRequest(sreq);
+ srsp.setShard(shard);
+ SimpleSolrResponse ssr = new SimpleSolrResponse();
+ srsp.setSolrResponse(ssr);
+
+ pending.incrementAndGet();
+ // if there are no shards available for a slice, urls.size()==0
+ if (urls.size() == 0) {
+ // TODO: what's the right error code here? We should use the same thing when
+ // all of the servers for a shard are down.
+ SolrException exception = new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE, "no servers hosting shard: " + shard);
+ srsp.setException(exception);
+ srsp.setResponseCode(exception.code());
+ responses.add(srsp);
+ return;
+ }
- try {
- params.remove(CommonParams.WT); // use default (currently javabin)
- params.remove(CommonParams.VERSION);
+ // all variables that set inside this listener must be at least volatile
+ responseCancellableMap.put(srsp, this.lbClient.asyncReq(lbReq, new OnComplete<>() {
+ long startTime = System.nanoTime();
Review comment:
This needs to be volatile or AtomicLong because it is accessed from other threads that call onSuccess and onFailure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418163481
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBSolrClient.java
##########
@@ -136,6 +137,91 @@ public boolean equals(Object obj) {
}
}
+ protected static class ServerIterator {
Review comment:
Why is this in LBSolrClient instead of LBHttp2SolrClient?
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -64,18 +62,23 @@
* by the RealtimeGet handler, since other types of replicas shouldn't respond to RTG requests
*/
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
+ private static final ShardResponse END_QUEUE = new ShardResponse();
private HttpShardHandlerFactory httpShardHandlerFactory;
- private CompletionService<ShardResponse> completionService;
- private Set<Future<ShardResponse>> pending;
+ private LinkedList<Cancellable> requests;
Review comment:
How many requests are we planning on handling? LinkedList is not great for performance due to node allocation overhead.
I also don't see us removing requests anywhere?
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
Review comment:
Might be worthwhile to create a singleton instance and reuse that, since we can give it a descriptive name like alreadyFailedRequest, assuming I understand the intent here.
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttp2SolrClient.java
##########
@@ -66,4 +80,132 @@ public LBHttp2SolrClient(Http2SolrClient httpClient, String... baseSolrUrls) {
protected SolrClient getClient(String baseUrl) {
return httpClient;
}
+
+ public interface OnComplete {
Review comment:
There is already an interface OnComplete in Http2SolrClient, can we either reuse that or maybe we can use CompletableFuture or something here? Have two similarly named interfaces is going to get confusing I think.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -243,12 +233,13 @@ public ShardResponse takeCompletedOrError() {
}
private ShardResponse take(boolean bailOnError) {
+ try {
+ while (pending.get() > 0) {
+ ShardResponse rsp = responses.take();
+ if (rsp == END_QUEUE)
Review comment:
Do we need to reinsert the END_QUEUE element in case there are other threads?
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
+ }
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
- if (onComplete != null) {
- // This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
-
- @Override
- public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
+ req.onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(() -> {
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ try {
+ NamedList<Object> body = processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(body);
+ } catch (RemoteSolrException e) {
+ if (SolrException.getRootCause(e) != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(e);
+ }
+ } catch (SolrServerException e) {
+ onComplete.onFailure(e);
+ }
+ });
}
- NamedList<Object> rsp;
- try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ @Override
+ public void onFailure(Response response, Throwable failure) {
+ super.onFailure(response, failure);
+ if (failure != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(createException(req, failure));
+ }
}
- }
- });
- return null;
- } else {
- try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
- Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
- InputStream is = listener.getInputStream();
- assert ObjectReleaseTracker.track(is);
- return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (TimeoutException e) {
- throw new SolrServerException(
- "Timeout occured while waiting response from server at: " + req.getURI(), e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ConnectException) {
- throw new SolrServerException("Server refused connection at: " + req.getURI(), cause);
- }
- if (cause instanceof SolrServerException) {
- throw (SolrServerException) cause;
- } else if (cause instanceof IOException) {
- throw new SolrServerException(
- "IOException occured when talking to server at: " + getBaseURL(), cause);
- }
- throw new SolrServerException(cause.getMessage(), cause);
+ });
+ return () -> req.abort(CANCELLED_EXCEPTION);
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest solrRequest, String collection) throws SolrServerException, IOException {
+ Request req = makeRequest(solrRequest, collection);
+ final ResponseParser parser = solrRequest.getResponseParser() == null
+ ? this.parser: solrRequest.getResponseParser();
+
+ try {
+ InputStreamResponseListener listener = new InputStreamResponseListener();
+ req.send(listener);
+ Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Throwable e) {
+ throw createException(req, e);
Review comment:
Catching Throwable seems very broad to me, can we do something less aggressive here?
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
+ }
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
- if (onComplete != null) {
- // This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
-
- @Override
- public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
+ req.onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(() -> {
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ try {
+ NamedList<Object> body = processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(body);
+ } catch (RemoteSolrException e) {
+ if (SolrException.getRootCause(e) != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(e);
+ }
+ } catch (SolrServerException e) {
+ onComplete.onFailure(e);
+ }
+ });
}
- NamedList<Object> rsp;
- try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ @Override
+ public void onFailure(Response response, Throwable failure) {
+ super.onFailure(response, failure);
+ if (failure != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(createException(req, failure));
+ }
}
- }
- });
- return null;
- } else {
- try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
- Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
- InputStream is = listener.getInputStream();
- assert ObjectReleaseTracker.track(is);
- return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (TimeoutException e) {
- throw new SolrServerException(
- "Timeout occured while waiting response from server at: " + req.getURI(), e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ConnectException) {
- throw new SolrServerException("Server refused connection at: " + req.getURI(), cause);
- }
- if (cause instanceof SolrServerException) {
- throw (SolrServerException) cause;
- } else if (cause instanceof IOException) {
- throw new SolrServerException(
- "IOException occured when talking to server at: " + getBaseURL(), cause);
- }
- throw new SolrServerException(cause.getMessage(), cause);
+ });
+ return () -> req.abort(CANCELLED_EXCEPTION);
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest solrRequest, String collection) throws SolrServerException, IOException {
+ Request req = makeRequest(solrRequest, collection);
+ final ResponseParser parser = solrRequest.getResponseParser() == null
+ ? this.parser: solrRequest.getResponseParser();
+
+ try {
+ InputStreamResponseListener listener = new InputStreamResponseListener();
+ req.send(listener);
+ Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Throwable e) {
+ throw createException(req, e);
+ }
+ }
+
+ private SolrServerException createException(Request req, Throwable throwable) {
Review comment:
This whole method seems problematic, but I can't immediately think of what the alternative should look like.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat edited a comment on pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat edited a comment on pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#issuecomment-649949789
Note that this PR including changes for removing the ability to use HttpShardHandler with a custom HttpClient. My reasons for doing that is
* In current code
* `HttpShardHandlerFactory.LBClient` which can't be changed will be used when requests to multiple replicas (replicas are interchangeable to other, i.e for query cases),
* `HttpShardHandler.HttpClient` which can be `HttpShardHandlerFactory.HttpClient` or some custom client by calling `HttpShardHandlerFactory.getShardHandler(HttpClient)` will be used for query to single replica
* This is quite counter-intuitive because when people look into `getShardHandler(HttpClient)` method they will assume that all requests (single or multiple replicas) will use the provided client.
* The client that provided to `getShardHandler` is all places is `updateShardHandler.defaultClient`. That does not make sense since in the ref-guide mentioned that
> **shardHandlerFactory**
> socketTimeout
The read timeout for intra-cluster query and administrative requests. The default is the same as the distribUpdateSoTimeout specified in the <solrcloud> section.
therefore the client of httpShardHandlerFactory must be used instead of `updateShardHandler.defaultClient`
* The changes also lead to less connections usage by switching from Http/1.1 client to Http/2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418391517
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBSolrClient.java
##########
@@ -136,6 +137,91 @@ public boolean equals(Object obj) {
}
}
+ protected static class ServerIterator {
Review comment:
I do not see this class very relates to `LBHttp2SolrClient` only and most of the code in LBSolrClient.req() can be replaced by this class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] anshumg commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
anshumg commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r420457814
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -130,77 +134,64 @@ public void submit(final ShardRequest sreq, final String shard, final Modifiable
final Tracer tracer = GlobalTracer.getTracer();
final Span span = tracer != null ? tracer.activeSpan() : null;
- Callable<ShardResponse> task = () -> {
+ params.remove(CommonParams.WT); // use default (currently javabin)
+ params.remove(CommonParams.VERSION);
+ QueryRequest req = makeQueryRequest(sreq, params, shard);
+ req.setMethod(SolrRequest.METHOD.POST);
- ShardResponse srsp = new ShardResponse();
- if (sreq.nodeName != null) {
- srsp.setNodeName(sreq.nodeName);
- }
- srsp.setShardRequest(sreq);
- srsp.setShard(shard);
- SimpleSolrResponse ssr = new SimpleSolrResponse();
- srsp.setSolrResponse(ssr);
- long startTime = System.nanoTime();
+ LBSolrClient.Req lbReq = httpShardHandlerFactory.newLBHttpSolrClientReq(req, urls);
+
+ ShardResponse srsp = new ShardResponse();
+ if (sreq.nodeName != null) {
+ srsp.setNodeName(sreq.nodeName);
+ }
+ srsp.setShardRequest(sreq);
+ srsp.setShard(shard);
+ SimpleSolrResponse ssr = new SimpleSolrResponse();
+ srsp.setSolrResponse(ssr);
+
+ pending.incrementAndGet();
+ // if there are no shards available for a slice, urls.size()==0
+ if (urls.size() == 0) {
+ // TODO: what's the right error code here? We should use the same thing when
Review comment:
Service unavailable error code seems reasonable to me here. Do you have something else you were debating to switch this out to?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418416881
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
Review comment:
+1
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r445955649
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -130,77 +134,64 @@ public void submit(final ShardRequest sreq, final String shard, final Modifiable
final Tracer tracer = GlobalTracer.getTracer();
final Span span = tracer != null ? tracer.activeSpan() : null;
- Callable<ShardResponse> task = () -> {
+ params.remove(CommonParams.WT); // use default (currently javabin)
+ params.remove(CommonParams.VERSION);
+ QueryRequest req = makeQueryRequest(sreq, params, shard);
+ req.setMethod(SolrRequest.METHOD.POST);
- ShardResponse srsp = new ShardResponse();
- if (sreq.nodeName != null) {
- srsp.setNodeName(sreq.nodeName);
- }
- srsp.setShardRequest(sreq);
- srsp.setShard(shard);
- SimpleSolrResponse ssr = new SimpleSolrResponse();
- srsp.setSolrResponse(ssr);
- long startTime = System.nanoTime();
+ LBSolrClient.Req lbReq = httpShardHandlerFactory.newLBHttpSolrClientReq(req, urls);
+
+ ShardResponse srsp = new ShardResponse();
+ if (sreq.nodeName != null) {
+ srsp.setNodeName(sreq.nodeName);
+ }
+ srsp.setShardRequest(sreq);
+ srsp.setShard(shard);
+ SimpleSolrResponse ssr = new SimpleSolrResponse();
+ srsp.setSolrResponse(ssr);
+
+ pending.incrementAndGet();
+ // if there are no shards available for a slice, urls.size()==0
+ if (urls.size() == 0) {
+ // TODO: what's the right error code here? We should use the same thing when
+ // all of the servers for a shard are down.
+ SolrException exception = new SolrException(SolrException.ErrorCode.SERVICE_UNAVAILABLE, "no servers hosting shard: " + shard);
+ srsp.setException(exception);
+ srsp.setResponseCode(exception.code());
+ responses.add(srsp);
+ return;
+ }
- try {
- params.remove(CommonParams.WT); // use default (currently javabin)
- params.remove(CommonParams.VERSION);
+ // all variables that set inside this listener must be at least volatile
+ responseCancellableMap.put(srsp, this.lbClient.asyncReq(lbReq, new OnComplete<>() {
+ long startTime = System.nanoTime();
Review comment:
right, thank you Shalin.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418627559
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttp2SolrClient.java
##########
@@ -66,4 +80,132 @@ public LBHttp2SolrClient(Http2SolrClient httpClient, String... baseSolrUrls) {
protected SolrClient getClient(String baseUrl) {
return httpClient;
}
+
+ public interface OnComplete {
Review comment:
Yea, having a generic interface sounds good.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r422459139
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -130,77 +134,64 @@ public void submit(final ShardRequest sreq, final String shard, final Modifiable
final Tracer tracer = GlobalTracer.getTracer();
final Span span = tracer != null ? tracer.activeSpan() : null;
- Callable<ShardResponse> task = () -> {
+ params.remove(CommonParams.WT); // use default (currently javabin)
+ params.remove(CommonParams.VERSION);
+ QueryRequest req = makeQueryRequest(sreq, params, shard);
+ req.setMethod(SolrRequest.METHOD.POST);
- ShardResponse srsp = new ShardResponse();
- if (sreq.nodeName != null) {
- srsp.setNodeName(sreq.nodeName);
- }
- srsp.setShardRequest(sreq);
- srsp.setShard(shard);
- SimpleSolrResponse ssr = new SimpleSolrResponse();
- srsp.setSolrResponse(ssr);
- long startTime = System.nanoTime();
+ LBSolrClient.Req lbReq = httpShardHandlerFactory.newLBHttpSolrClientReq(req, urls);
+
+ ShardResponse srsp = new ShardResponse();
+ if (sreq.nodeName != null) {
+ srsp.setNodeName(sreq.nodeName);
+ }
+ srsp.setShardRequest(sreq);
+ srsp.setShard(shard);
+ SimpleSolrResponse ssr = new SimpleSolrResponse();
+ srsp.setSolrResponse(ssr);
+
+ pending.incrementAndGet();
+ // if there are no shards available for a slice, urls.size()==0
+ if (urls.size() == 0) {
+ // TODO: what's the right error code here? We should use the same thing when
Review comment:
I do not, just copied and pasted from the old code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat edited a comment on pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat edited a comment on pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#issuecomment-649949789
Note that this PR including changes for removing the ability to use HttpShardHandler with a custom HttpClient. My reasons for doing that is
* In current code, we use HttpShardHandlerFactory.LBClient (which can't be changed) when requests to multiple replicas (replicas are interchangeable to other, i.e for query cases), for query to single replica we will use provided HttpClient which can be changed by calling `HttpShardHandlerFactory.getShardHandler(HttpClient)`. But when people look into `getShardHandler(HttpClient)` method they will assume that all requests (single or multiple replicas) will use the provided client.
* The client that provided to `getShardHandler` is all places is updateShardHandler.defaultClient. That does not make sense since in the ref-guide mentioned that
> **shardHandlerFactory**
> socketTimeout
The read timeout for intra-cluster query and administrative requests. The default is the same as the distribUpdateSoTimeout specified in the <solrcloud> section.
therefore the client of httpShardHandlerFactory must be used instead of `updateShardHandler.defaultClient`
* The changes also lead to less connections usage by switching from Http/1.1 client to Http/2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418399765
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
+ }
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
- if (onComplete != null) {
- // This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
-
- @Override
- public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
+ req.onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(() -> {
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ try {
+ NamedList<Object> body = processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(body);
+ } catch (RemoteSolrException e) {
+ if (SolrException.getRootCause(e) != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(e);
+ }
+ } catch (SolrServerException e) {
+ onComplete.onFailure(e);
+ }
+ });
}
- NamedList<Object> rsp;
- try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ @Override
+ public void onFailure(Response response, Throwable failure) {
+ super.onFailure(response, failure);
+ if (failure != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(createException(req, failure));
+ }
}
- }
- });
- return null;
- } else {
- try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
- Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
- InputStream is = listener.getInputStream();
- assert ObjectReleaseTracker.track(is);
- return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (TimeoutException e) {
- throw new SolrServerException(
- "Timeout occured while waiting response from server at: " + req.getURI(), e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ConnectException) {
- throw new SolrServerException("Server refused connection at: " + req.getURI(), cause);
- }
- if (cause instanceof SolrServerException) {
- throw (SolrServerException) cause;
- } else if (cause instanceof IOException) {
- throw new SolrServerException(
- "IOException occured when talking to server at: " + getBaseURL(), cause);
- }
- throw new SolrServerException(cause.getMessage(), cause);
+ });
+ return () -> req.abort(CANCELLED_EXCEPTION);
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest solrRequest, String collection) throws SolrServerException, IOException {
+ Request req = makeRequest(solrRequest, collection);
+ final ResponseParser parser = solrRequest.getResponseParser() == null
+ ? this.parser: solrRequest.getResponseParser();
+
+ try {
+ InputStreamResponseListener listener = new InputStreamResponseListener();
+ req.send(listener);
+ Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Throwable e) {
+ throw createException(req, e);
Review comment:
Right, I think it is worth to seperate the exception handling logic of asyncReq() and req().
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418394732
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
+ }
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
- if (onComplete != null) {
- // This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
-
- @Override
- public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
+ req.onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(() -> {
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ try {
+ NamedList<Object> body = processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(body);
+ } catch (RemoteSolrException e) {
+ if (SolrException.getRootCause(e) != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(e);
+ }
+ } catch (SolrServerException e) {
+ onComplete.onFailure(e);
+ }
+ });
}
- NamedList<Object> rsp;
- try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ @Override
+ public void onFailure(Response response, Throwable failure) {
+ super.onFailure(response, failure);
+ if (failure != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(createException(req, failure));
+ }
}
- }
- });
- return null;
- } else {
- try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
- Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
- InputStream is = listener.getInputStream();
- assert ObjectReleaseTracker.track(is);
- return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (TimeoutException e) {
- throw new SolrServerException(
- "Timeout occured while waiting response from server at: " + req.getURI(), e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ConnectException) {
- throw new SolrServerException("Server refused connection at: " + req.getURI(), cause);
- }
- if (cause instanceof SolrServerException) {
- throw (SolrServerException) cause;
- } else if (cause instanceof IOException) {
- throw new SolrServerException(
- "IOException occured when talking to server at: " + getBaseURL(), cause);
- }
- throw new SolrServerException(cause.getMessage(), cause);
+ });
+ return () -> req.abort(CANCELLED_EXCEPTION);
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest solrRequest, String collection) throws SolrServerException, IOException {
+ Request req = makeRequest(solrRequest, collection);
+ final ResponseParser parser = solrRequest.getResponseParser() == null
+ ? this.parser: solrRequest.getResponseParser();
+
+ try {
+ InputStreamResponseListener listener = new InputStreamResponseListener();
+ req.send(listener);
+ Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Throwable e) {
+ throw createException(req, e);
+ }
+ }
+
+ private SolrServerException createException(Request req, Throwable throwable) {
Review comment:
Yeah, the current error handling in SolrClient is quite complex and variety. That is what I can come up at the moment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418394131
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -64,18 +62,23 @@
* by the RealtimeGet handler, since other types of replicas shouldn't respond to RTG requests
*/
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
+ private static final ShardResponse END_QUEUE = new ShardResponse();
private HttpShardHandlerFactory httpShardHandlerFactory;
- private CompletionService<ShardResponse> completionService;
- private Set<Future<ShardResponse>> pending;
+ private LinkedList<Cancellable> requests;
Review comment:
I think most of the time, max size of the list will equals to number of shards, so I think it won't be very different here but I can change it back to ArrayList.
> I also don't see us removing requests anywhere?
That is right, we need to remove elements from this list on the corresponding result arrived.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418394732
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
##########
@@ -359,65 +366,95 @@ public void send(OutStream outStream, SolrRequest req, String collection) throws
outStream.flush();
}
- public NamedList<Object> request(SolrRequest solrRequest,
- String collection,
- OnComplete onComplete) throws IOException, SolrServerException {
- Request req = makeRequest(solrRequest, collection);
+ private static final Exception CANCELLED_EXCEPTION = new Exception();
+
+ public Cancellable asyncRequest(SolrRequest solrRequest, String collection, OnComplete onComplete) {
+ Request req;
+ try {
+ req = makeRequest(solrRequest, collection);
+ } catch (SolrServerException | IOException e) {
+ onComplete.onFailure(e);
+ return () -> {};
+ }
final ResponseParser parser = solrRequest.getResponseParser() == null
? this.parser: solrRequest.getResponseParser();
-
- if (onComplete != null) {
- // This async call only suitable for indexing since the response size is limited by 5MB
- req.onRequestQueued(asyncTracker.queuedListener)
- .onComplete(asyncTracker.completeListener).send(new BufferingResponseListener(5 * 1024 * 1024) {
-
- @Override
- public void onComplete(Result result) {
- if (result.isFailed()) {
- onComplete.onFailure(result.getFailure());
- return;
+ req.onRequestQueued(asyncTracker.queuedListener)
+ .onComplete(asyncTracker.completeListener)
+ .send(new InputStreamResponseListener() {
+ @Override
+ public void onHeaders(Response response) {
+ super.onHeaders(response);
+ InputStreamResponseListener listener = this;
+ executor.execute(() -> {
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ try {
+ NamedList<Object> body = processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ onComplete.onSuccess(body);
+ } catch (RemoteSolrException e) {
+ if (SolrException.getRootCause(e) != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(e);
+ }
+ } catch (SolrServerException e) {
+ onComplete.onFailure(e);
+ }
+ });
}
- NamedList<Object> rsp;
- try {
- InputStream is = getContentAsInputStream();
- assert ObjectReleaseTracker.track(is);
- rsp = processErrorsAndResponse(result.getResponse(),
- parser, is, getEncoding(), isV2ApiRequest(solrRequest));
- onComplete.onSuccess(rsp);
- } catch (Exception e) {
- onComplete.onFailure(e);
+ @Override
+ public void onFailure(Response response, Throwable failure) {
+ super.onFailure(response, failure);
+ if (failure != CANCELLED_EXCEPTION) {
+ onComplete.onFailure(createException(req, failure));
+ }
}
- }
- });
- return null;
- } else {
- try {
- InputStreamResponseListener listener = new InputStreamResponseListener();
- req.send(listener);
- Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
- InputStream is = listener.getInputStream();
- assert ObjectReleaseTracker.track(is);
- return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (TimeoutException e) {
- throw new SolrServerException(
- "Timeout occured while waiting response from server at: " + req.getURI(), e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof ConnectException) {
- throw new SolrServerException("Server refused connection at: " + req.getURI(), cause);
- }
- if (cause instanceof SolrServerException) {
- throw (SolrServerException) cause;
- } else if (cause instanceof IOException) {
- throw new SolrServerException(
- "IOException occured when talking to server at: " + getBaseURL(), cause);
- }
- throw new SolrServerException(cause.getMessage(), cause);
+ });
+ return () -> req.abort(CANCELLED_EXCEPTION);
+ }
+
+ @Override
+ public NamedList<Object> request(SolrRequest solrRequest, String collection) throws SolrServerException, IOException {
+ Request req = makeRequest(solrRequest, collection);
+ final ResponseParser parser = solrRequest.getResponseParser() == null
+ ? this.parser: solrRequest.getResponseParser();
+
+ try {
+ InputStreamResponseListener listener = new InputStreamResponseListener();
+ req.send(listener);
+ Response response = listener.get(idleTimeout, TimeUnit.MILLISECONDS);
+ InputStream is = listener.getInputStream();
+ assert ObjectReleaseTracker.track(is);
+ return processErrorsAndResponse(response, parser, is, getEncoding(response), isV2ApiRequest(solrRequest));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Throwable e) {
+ throw createException(req, e);
+ }
+ }
+
+ private SolrServerException createException(Request req, Throwable throwable) {
Review comment:
Yeah, the current error handling in SolrClient is quite complex and variety. That is what I can come up at the moment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r445955727
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -66,16 +66,20 @@
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
private HttpShardHandlerFactory httpShardHandlerFactory;
- private CompletionService<ShardResponse> completionService;
- private Set<Future<ShardResponse>> pending;
+ private Map<ShardResponse, Cancellable> responseCancellableMap;
+ private BlockingQueue<ShardResponse> responses;
+ private AtomicInteger pending;
private Map<String, List<String>> shardToURLs;
private Http2SolrClient httpClient;
Review comment:
right, I removed that in this commit https://github.com/apache/lucene-solr/pull/1470/commits/0a02baad8587ff6c4f6fce10f06aeeaf88dd8fb2.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r445955598
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/util/OnComplete.java
##########
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.solr.client.solrj.util;
+
+/**
+ * Listener for async requests
+ */
+public interface OnComplete<T> {
Review comment:
changed, thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418394131
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -64,18 +62,23 @@
* by the RealtimeGet handler, since other types of replicas shouldn't respond to RTG requests
*/
public static String ONLY_NRT_REPLICAS = "distribOnlyRealtime";
+ private static final ShardResponse END_QUEUE = new ShardResponse();
private HttpShardHandlerFactory httpShardHandlerFactory;
- private CompletionService<ShardResponse> completionService;
- private Set<Future<ShardResponse>> pending;
+ private LinkedList<Cancellable> requests;
Review comment:
I think most of the time, max size of the list will equals to number of shards, so I think it won't be very different here but I can change it back to ArrayList.
> I also don't see us removing requests anywhere?
That is right, we need to remove elements from this list on the corresponding result arrived.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418416790
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttp2SolrClient.java
##########
@@ -66,4 +80,132 @@ public LBHttp2SolrClient(Http2SolrClient httpClient, String... baseSolrUrls) {
protected SolrClient getClient(String baseUrl) {
return httpClient;
}
+
+ public interface OnComplete {
Review comment:
Hmm, so the to OnComplete different in the parameter of `onSuccess` one is
* NamedList
* another is LBSolrClient.Rsp
Should we making them as template then move it to util package of solrj?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#issuecomment-649949789
Note that this PR including changes for removing the ability to use HttpShardHandler with a custom HttpClient. My reasons for doing that is
* In current code, we use HttpShardHandlerFactory.LBClient (which can't be changed) when requests to multiple replicas (replicas are interchangeable to other, i.e for query cases), for query to single replica we will use provided HttpClient which can be changed by calling `HttpShardHandlerFactory.getShardHandler(HttpClient)`. But when people look into `getShardHandler(HttpClient)` method they will assume that all requests (single or multiple replicas) will use the provided client.
* The client that provided to `getShardHandler` is all places is updateShardHandler.defaultClient. That does not make sense since in the ref-guide mentioned that
> **shardHandlerFactory**
> socketTimeout
The read timeout for intra-cluster query and administrative requests. The default is the same as the distribUpdateSoTimeout specified in the <solrcloud> section.
therefore the client of httpShardHandlerFactory must be used instead of `updateShardHandler.defaultClient`
* The changes also lead to less connections usage by switching from Http/1.1 client to Http/2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r418404455
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -243,12 +233,13 @@ public ShardResponse takeCompletedOrError() {
}
private ShardResponse take(boolean bailOnError) {
+ try {
+ while (pending.get() > 0) {
+ ShardResponse rsp = responses.take();
+ if (rsp == END_QUEUE)
Review comment:
So all usage of HttpShardHandler is single-thread. Submit tasks, takeResponse then cancelAll() if there are anyerror. Therefore even END_QUEUE is not needed here. I may add some comments to indicate that HttpShardHandler is not thread-safe.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] CaoManhDat commented on a change in pull request #1470: SOLR-14354: Async or using threads in better way for HttpShardHandler
Posted by GitBox <gi...@apache.org>.
CaoManhDat commented on a change in pull request #1470:
URL: https://github.com/apache/lucene-solr/pull/1470#discussion_r419198072
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -259,12 +261,10 @@ private ShardResponse take(boolean bailOnError) {
@Override
public void cancelAll() {
- for (Cancellable cancellable : requests) {
+ for (Cancellable cancellable : responseCancellableMap.values()) {
Review comment:
+1
##########
File path: solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
##########
@@ -54,6 +53,10 @@
import org.apache.solr.util.tracing.GlobalTracer;
import org.apache.solr.util.tracing.SolrRequestCarrier;
+/**
+ * Submit requests in async manner.
+ * This class is not thread-safe so all methods should be called in a same thread.
Review comment:
+1
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org