You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2019/12/16 18:38:48 UTC
[tinkerpop] 01/01: TINKERPOP-2175 Better manage the executor thread
on close.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2175
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 02cee17db879521ef465a21676a0481cf8cd81a0
Author: stephen <sp...@gmail.com>
AuthorDate: Mon Dec 16 13:37:20 2019 -0500
TINKERPOP-2175 Better manage the executor thread on close.
---
CHANGELOG.asciidoc | 1 +
.../apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java | 8 +++++++-
.../gremlin/server/op/traversal/TraversalOpProcessor.java | 8 +++++++-
3 files changed, 15 insertions(+), 2 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 83cab12..486e7d8 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -28,6 +28,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Bump to Tornado 5.x for gremlin-python.
* Deprecated `TraversalStrategies.applyStrategies()`.
* Deprecated Jython support in `gremlin-python`.
+* Improved Gremlin Server executor thread handling on client close requests.
* Reverted: Modified Java driver to use IP address rather than hostname to create connections.
[[release-3-3-9]]
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
index 0cd2d9e..03bf6fd 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
@@ -146,7 +146,13 @@ public abstract class AbstractOpProcessor implements OpProcessor {
// send back a page of results if batch size is met or if it's the end of the results being iterated.
// also check writeability of the channel to prevent OOME for slow clients.
- if (nettyContext.channel().isWritable()) {
+ //
+ // clients might decide to close the Netty channel to the server with a CloseWebsocketFrame after errors
+ // like CorruptedFrameException. On the server, although the channel gets closed, there might be some
+ // executor threads waiting for watermark to clear which will not clear in these cases since client has
+ // already given up on these requests. This leads to these executors waiting for the client to consume
+ // results till the timeout. checking for isActive() should help prevent that.
+ if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
if (forceFlush || aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
final ResponseStatusCode code = itty.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
index 7b7dbdb..8737a9a 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
@@ -561,7 +561,13 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
// send back a page of results if batch size is met or if it's the end of the results being iterated.
// also check writeability of the channel to prevent OOME for slow clients.
- if (nettyContext.channel().isWritable()) {
+ //
+ // clients might decide to close the Netty channel to the server with a CloseWebsocketFrame after errors
+ // like CorruptedFrameException. On the server, although the channel gets closed, there might be some
+ // executor threads waiting for watermark to clear which will not clear in these cases since client has
+ // already given up on these requests. This leads to these executors waiting for the client to consume
+ // results till the timeout. checking for isActive() should help prevent that.
+ if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
if (forceFlush || aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
final ResponseStatusCode code = itty.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;