了解 Jetty 的“待处理/未就绪时关闭”警告答案

【问题标题】：Understanding Jetty's "Closed while Pending/Unready" warning了解 Jetty 的“待处理/未就绪时关闭”警告
【发布时间】：2018-12-03 12:37:12
【问题描述】：

我们有一个异步 servlet，它从 Jetty 生成以下警告日志：

java.io.IOException: Closed while Pending/Unready

启用调试日志后，我得到了以下堆栈跟踪：

WARN [jetty-25948] (HttpOutput.java:278)   - java.io.IOException: Closed while Pending/Unready
DEBUG [jetty-25948] (HttpOutput.java:279)   - 
java.io.IOException: Closed while Pending/Unready
        at org.eclipse.jetty.server.HttpOutput.close(HttpOutput.java:277) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.Response.closeOutput(Response.java:1044) [jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:488) [jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpChannel.run(HttpChannel.java:293) [jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:708) [jetty-util.jar:9.4.8.v20171121]
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:626) [jetty-util.jar:9.4.8.v20171121]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_151]

这并没有太大帮助。

警告仅在 Jetty 调用我们的 AsyncListener 的 onTimeout() 方法后出现。 QA 有时可以通过在客户端应用程序上使用 kill -9 来重现它。

如何使用示例 servlet 客户端代码重现此警告？我想在比我们的生产代码更简单的环境中理解这个问题，以便之后能够修复生产代码。示例 servlet 应该如何表现？是否可以在同一个 JUnit 测试中使用 Apache Commons HttpClient 客户端重现它？（这对于编写没有复杂网络黑客攻击的集成测试非常有用，例如kill -9。）

我尝试了一些方法来实现示例异步 servlet 和客户端，但均未成功。我认为附加此代码不会有太大帮助，但如果有人感兴趣，我可以这样做。

码头版本：9.4.8.v20171121

更新（2018-06-27）：

考虑到@Joakim Erdfelt 的有用回答，我没有在我们的代码中找到任何close() 调用，但发现了可疑的丢失同步。这是我们的异步轮询 servlet 的基础：

public class QueuePollServlet extends HttpServlet {

    public QueuePollServlet() {
    }

    @Override
    protected void doPost(final HttpServletRequest req, final HttpServletResponse resp)
            throws ServletException, IOException {
        resp.setContentType(MediaType.OCTET_STREAM.type());
        resp.setStatus(HttpServletResponse.SC_OK);
        resp.flushBuffer();
        final AsyncContext async = req.startAsync();
        async.setTimeout(30_000);
        final ServletOutputStream output = resp.getOutputStream();
        final QueueWriteListener writeListener = new QueueWriteListener(async, output);
        async.addListener(writeListener);
        output.setWriteListener(writeListener);
    }

    private static class QueueWriteListener implements AsyncListener, WriteListener {

        private final AsyncContext asyncContext;
        private final ServletOutputStream output;

        public QueueWriteListener(final AsyncContext asyncContext, final ServletOutputStream output) {
            this.asyncContext = checkNotNull(asyncContext, "asyncContext cannot be null");
            this.output = checkNotNull(output, "output cannot be null");
        }

        @Override
        public void onWritePossible() throws IOException {
            writeImpl();
        }

        private synchronized void writeImpl() throws IOException {
            while (output.isReady()) {
                final byte[] message = getNextMessage();
                if (message == null) {
                    output.flush();
                    return;
                }
                output.write(message);
            }
        }

        private void completeImpl() {
            asyncContext.complete();
        }

        public void dataArrived() {
            try {
                writeImpl();
            } catch (IOException e) {
                ...
            }
        }

        public void noMoreBuffers() {
            completeImpl();
        }

        @Override
        public void onTimeout(final AsyncEvent event) throws IOException {
            completeImpl();
        }

        @Override
        public void onError(final Throwable t) {
            logger.error("Writer.onError", t);
            completeImpl();
        }


        ...
    }
}

一个可能的竞争条件：

DataFeederThread：调用dataArrived() -> writeImpl()，然后得到output.isReady() 是true。
Jetty 调用 onTimeout() 完成上下文。
DataFeederThread：在 while 循环中调用 output.write()，但找到了完整的上下文。

这种情况会导致Closed while Pending/Unready 警告还是另一个问题？我是对的，制作completeImpl() synchronized 可以解决问题还是有其他需要关心的事情？

更新（2018-06-28）：

我们在QueueWriteListener 中也有类似的onError 实现，如下面的sn-p：

@Override
public void onError(final Throwable t) {
    logger.error("Writer.onError", t);
    completeImpl();
}

无论如何，Closed while Pending/Unready 日志消息周围没有 onError 错误日志（查看每个两个小时的时间范围），只有像我们的 DataFeederThread 中的以下 EOF：

DEBUG [DataFeederThread] (HttpOutput.java:224) - 
org.eclipse.jetty.io.EofException: null
        at org.eclipse.jetty.server.HttpConnection$SendCallback.reset(HttpConnection.java:704) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpConnection$SendCallback.access$300(HttpConnection.java:668) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpConnection.send(HttpConnection.java:526) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpChannel.sendResponse(HttpChannel.java:778) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpChannel.write(HttpChannel.java:834) ~[jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpOutput.write(HttpOutput.java:234) [jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpOutput.write(HttpOutput.java:218) [jetty-server.jar:9.4.8.v20171121]
        at org.eclipse.jetty.server.HttpOutput.flush(HttpOutput.java:392) [jetty-server.jar:9.4.8.v20171121]
        at com.example.QueuePollServlet$QueueWriteListener.writeImpl()
        at com.example.QueuePollServlet$QueueWriteListener.dataArrived()


DEBUG [DataFeederThread] (QueuePollServlet.java:217) - messageArrived exception
org.eclipse.jetty.io.EofException: Closed
        at org.eclipse.jetty.server.HttpOutput.write(HttpOutput.java:476) ~[jetty-server.jar:9.4.8.v20171121]
        at com.example.QueuePollServlet$QueueWriteListener.writeImpl()
        at com.example.QueuePollServlet$QueueWriteListener.dataArrived()

【问题讨论】：

实现WriteListener.onError() 并检查写入是否失败。这可能是您的虚假close() 来源的线索。（ServletOutputStream.close() 是在onError() 报告给应用程序之后调用的）
@JoakimErdfelt：我们有 onError 带有日志记录，但它没有被调用（我已经更新了这个问题）。
另外，与您的问题无关，而是you really should upgrade your version of Jetty being used in production。您还将获得其他错误修复，甚至围绕您的问题在代码库中进行改进。（提示：由于发现高负载 HTTP/2 系统，关闭的处理方式不同）
谢谢，@JoakimErdfelt！我已经更新了我们的码头。

标签： servlets jetty embedded-jetty servlet-3.1

【解决方案1】：

可以通过一些手动调试和普通的curl 客户端重现Closed while Pending/Unready 警告。我已经使用 Jetty 9.4.8.v20171121 和 Jetty 9.4.11.v20180605 对其进行了测试。您需要两个断点，因此在测试环境中可靠地重现并不容易。

The first breakpoint is in the HttpOutput.write() method right after it changes its state from READY to PENDING:

case READY:
    if (!_state.compareAndSet(OutputState.READY, OutputState.PENDING))
        continue;
    // put a breakpoint here

The second one in Response.closeOutput():

case STREAM:
    // put a breakpiont here
    if (!_out.isClosed())
        getOutputStream().close();
    break;

重现步骤：

[JettyThread1] 调用 QueueWriteListener.onWritePossible() 将几个字节写入输出然后返回（因为其输入缓冲区为空）。
等待 onTimeout 事件。
[JettyThread2] HttpChannelState.onTimeout() 调用 QueueWriteListener.onTimeout()，后者调用 asyncContext.complete()。
[JettyThread2] HttpChannelState.onTimeout() 在异步超时后安排调度。
[JettyThread2] 在第二个断点处暂停
[DFT] DataFeederThread 调用 writeImpl()
[DFT] DataFeederThread 调用 HttpOutput.write()（这是输出流）
[DFT] HttpOutput.write() 将其状态从 READY 更改为 PENDING
[DFT] DataFeederThread 在这里暂停，由于上面的断点
[JettyThread2] 计划调度关闭输出流并产生Closed while Pending/Unready 警告。

所以，实际上是 Jetty 关闭了此（Jetty 9.4.8.v20171121）堆栈上的输出流：

Thread [jetty-19] (Suspended)   
    Response.closeOutput() line: 1043   
    HttpChannelOverHttp(HttpChannel).handle() line: 488 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748

在侦听器中设置onTimeout() synchronized（以及writeImpl() 也是synchronized）无济于事，因为预定的关闭仍然能够与writeImpl 重叠（来自DataFeederThread）。考虑这种情况：

[JettyThread1] 调用 QueueWriteListener.onWritePossible() 将几个字节写入输出然后返回（因为其输入缓冲区为空）。
等待 onTimeout 事件。
[JettyThread2] HttpChannelState.onTimeout() 调用 QueueWriteListener.onTimeout()，后者调用 asyncContext.complete()。
[DFT] DataFeederThread 调用 writeImpl() （由于 onTimeout 尚未完成而被阻止）
[JettyThread2] QueueWriteListener.onTimeout() 完成
[DFT] writeImpl() 可以运行
[JettyThread2] HttpChannelState.onTimeout() 在异步超时后安排调度。
[JettyThread2] 在第二个断点处暂停
[DFT] DataFeederThread 调用 HttpOutput.write()（这是输出流）
[DFT] HttpOutput.write() 将其状态从 READY 更改为 PENDING
[DFT] DataFeederThread 在这里暂停，由于上面的断点
[JettyThread2] 计划调度关闭输出流并产生Closed while Pending/Unready 警告。

不幸的是，在asnyContext.complete() 之后检查output.isReady() 是不够的。它返回 true，因为 Jetty 重新打开了 HttpOutput（参见下面的堆栈），因此您需要在侦听器中为此设置一个单独的标志。

Thread [jetty-13] (Suspended (access of field _state in HttpOutput))    
    HttpOutput.reopen() line: 195   
    HttpOutput.recycle() line: 923  
    Response.recycle() line: 138    
    HttpChannelOverHttp(HttpChannel).recycle() line: 269    
    HttpChannelOverHttp.recycle() line: 83  
    HttpConnection.onCompleted() line: 424  
    HttpChannelOverHttp(HttpChannel).onCompleted() line: 695    
    HttpChannelOverHttp(HttpChannel).handle() line: 493 
    HttpChannelOverHttp(HttpChannel).run() line: 293    
    QueuedThreadPool.runJob(Runnable) line: 708 
    QueuedThreadPool$2.run() line: 626  
    Thread.run() line: 748

此外，isReady() 还会在输出仍关闭时（在回收/重新打开之前）返回 true。相关问题：isReady() returns true in closed state - why?

最终的实现是类似的：

@Override
protected void doPost(final HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {
    resp.setContentType(MediaType.OCTET_STREAM.type());
    resp.setStatus(HttpServletResponse.SC_OK);
    resp.setBufferSize(4096);
    resp.flushBuffer();
    final AsyncContext async = req.startAsync();
    async.setTimeout(5_000); // millis
    final ServletOutputStream output = resp.getOutputStream();
    final QueueWriteListener writeListener = new QueueWriteListener(async, output);
    async.addListener(writeListener);
    output.setWriteListener(writeListener);
}

private static class QueueWriteListener implements AsyncListener, WriteListener {

    private static final Logger logger = LoggerFactory.getLogger(QueueWriteListener.class);

    private final AsyncContext asyncContext;
    private final ServletOutputStream output;

    @GuardedBy("this")
    private boolean completed = false;

    public QueueWriteListener(final AsyncContext asyncContext, final ServletOutputStream output) {
        this.asyncContext = checkNotNull(asyncContext, "asyncContext cannot be null");
        this.output = checkNotNull(output, "output cannot be null");
    }

    @Override
    public void onWritePossible() throws IOException {
        writeImpl();
    }

    private synchronized void writeImpl() throws IOException {
        if (completed) {
            return;
        }
        while (output.isReady()) {
            final byte[] message = getNextMessage();
            if (message == null) {
                output.flush();
                return;
            }
            output.write(message);
        }
    }

    private synchronized void completeImpl() {
        // also stops DataFeederThread to call bufferArrived
        completed = true;
        asyncContext.complete();
    }

    @Override
    public void onError(final Throwable t) {
        logger.error("Writer.onError", t);
        completeImpl();
    }

    public void dataArrived() {
        try {
            writeImpl();
        } catch (RuntimeException | IOException e) {
            ...
        }
    }

    public void noMoreData() {
        completeImpl();
    }

    @Override
    public synchronized void onComplete(final AsyncEvent event) throws IOException {
        completed = true; // might not needed but does not hurt
    }

    @Override
    public synchronized void onTimeout(final AsyncEvent event) throws IOException {
        completeImpl();
    }

    @Override
    public void onError(final AsyncEvent event) throws IOException {
        logger.error("onError", event.getThrowable());
    }

    ...
}

2018-08-01 更新：实际上并没有完全修复警告，请参阅：“Closed while Pending/Unready” warnings from Jetty

【讨论】：

哇！这是一些很棒的细节。您可以将此作为 Jetty 问题提交吗？ github.com/eclipse/jetty.project/issues - 也许我们可以找到更好的方法。即使它只是让错误/问题发生时更加清晰。
谢谢！我想我可以创建这个问题，但可能只是在星期一。（如果有时间，请随意创建一个。）

【解决方案2】：

您正在使用 Servlet 规范中的异步 I/O 并手动关闭了响应流。

关键事实：close call 意味着写入操作。

在 Jetty 案例中，IOException 告诉您手动调用关闭响应流导致了不希望的副作用。

因此，在异步 I/O 模式下，应该在使用 ServletOutputStream.close() 验证 isReady() 为真之前调用 ServletOutputStream.isReady()。

请注意，任何时候都可能需要允许ServletOutputStream.close()，尤其是在此调度中调用了AsyncContext.complete() 时。

但是，如果先前的写入尚未完成和/或 ServletOutputStream.isReady() 尚未被调用，则允许这种裸 ServletOutputStream.close() 的特定情况，但行为是中止响应，丢弃任何未决/未写入的写在ServletOutputStream。

这就是你得到IOException("Closed while Pending/Unready")的原因

现在您应该问自己为什么要关闭响应流？根据 Servlet 规范，这不是必需的，如果您想使用 RequestDispatcher 将多个 Servlet 响应组合在一起，则不希望这样做。

【讨论】：

感谢您的回答！我想我找到了一些东西，请您检查问题中的更新吗？
感谢有人关闭流的有用提示。原来是Jetty把它关了。我能够调试、重现和修复错误。（详细信息在我下面的回答中。）