[BUG] Search pipeline seen executing on transport_worker thread

[austintlee]

Describe the bug
I have a search processor that uses a org.opensearch.client.Client object to execute a TransportAction. This invocation returns a BaseFuture and the processor does .get(), blocking until the client returns a response. Occasionally, the get() call blocks indefinitely and brings the whole cluster into bad state.

A thread dump on the node that hung revealed that the search processor was executing on a transport_worker thread.

"opensearch[opensearch-node1][transport_worker][T#2]" #32 daemon prio=5 os_prio=0 cpu=61810.65ms elapsed=43062.56s allocated=2771M defined_classes=251      tid=0x0000fffef4009140 nid=0x13d waiting on condition  [0x0000ffff88cda000]
    java.lang.Thread.State: WAITING (parking)
     at jdk.internal.misc.Unsafe.park(java.base@17.0.8/Native Method)
     - parking to wait for  <0x00000000ee75b640> (a org.opensearch.common.util.concurrent.BaseFuture$Sync)
     at java.util.concurrent.locks.LockSupport.park(java.base@17.0.8/LockSupport.java:211)
     at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.base@17.0.8/AbstractQueuedSynchronizer.java:715)
     at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(java.base@17.0.8/AbstractQueuedSynchronizer.java:1047)
     at org.opensearch.common.util.concurrent.BaseFuture$Sync.get(BaseFuture.java:272)
     at org.opensearch.common.util.concurrent.BaseFuture.get(BaseFuture.java:104)
     at org.opensearch.common.util.concurrent.FutureUtils.get(FutureUtils.java:74)
     at org.opensearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:55)
     at org.opensearch.searchpipelines.questionanswering.generative.llm.DefaultLlmImpl.doChatCompletion(DefaultLlmImpl.java:84)
     at org.opensearch.searchpipelines.questionanswering.generative.GenerativeQAResponseProcessor.processResponse(GenerativeQAResponseProcessor.java:109)
     at org.opensearch.search.pipeline.Pipeline.transformResponse(Pipeline.java:177)
     at org.opensearch.search.pipeline.PipelinedRequest.transformResponse(PipelinedRequest.java:31)
     at org.opensearch.action.search.TransportSearchAction.lambda$executeRequest$0(TransportSearchAction.java:398)
     at org.opensearch.action.search.TransportSearchAction$$Lambda$4884/0x0000003001d5d850.accept(Unknown Source)
     at org.opensearch.core.action.ActionListener$1.onResponse(ActionListener.java:82)
     at org.opensearch.core.action.ActionListener$5.onResponse(ActionListener.java:268)
     at org.opensearch.action.search.AbstractSearchAsyncAction.sendSearchResponse(AbstractSearchAsyncAction.java:671)
     at org.opensearch.action.search.ExpandSearchPhase.run(ExpandSearchPhase.java:132)
     at org.opensearch.action.search.AbstractSearchAsyncAction.executePhase(AbstractSearchAsyncAction.java:428)
     at org.opensearch.action.search.AbstractSearchAsyncAction.executeNextPhase(AbstractSearchAsyncAction.java:422)

I do see in a few other thread dumps I captured the same processor running on a search thread.

From:

OpenSearch/server/src/main/java/org/opensearch/common/util/concurrent/BaseFuture.java

Line 109 in 8807d7a

return Transports.assertNotTransportThread(BLOCKING_OP_REASON)

And:

OpenSearch/modules/transport-netty4/src/main/java/org/opensearch/transport/netty4/Netty4MessageChannelHandler.java

Lines 86 to 89 in 8807d7a

	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
	assert Transports.assertDefaultThreadContext(transport.getThreadPool().getThreadContext());
	assert Transports.assertTransportThread();
	assert msg instanceof ByteBuf : "Expected message type ByteBuf, found: " + msg.getClass();

I can see that we don't expect a blocking call to happen on transport threads and there is code specifically in BaseFuture.get to disallow invocations on transport threads, although assert won't trigger unless the OpenSearch process is run with the ea JVM flag.

Can we ensure that search processors run on search threads? Or are they really allowed to run on transport threads which means that I should not have any blocking calls in my search processor?

To Reproduce
Steps to reproduce the behavior:

1. Go to '...'
2. Click on '....'
3. Scroll down to '....'
4. See error

Expected behavior
A clear and concise description of what you expected to happen.

Plugins
Please list all plugins currently enabled.

Screenshots
If applicable, add screenshots to help explain your problem.

Host/Environment (please complete the following information):

• OS: [e.g. iOS]
• Version [e.g. 22]

Additional context
Add any other context about the problem here.

[austintlee]

transport_worker thread example full trace:

"opensearch[opensearch-node1][transport_worker][T#2]" #32 daemon prio=5 os_prio=0 cpu=61810.65ms elapsed=43062.56s allocated=2771M defined_classes=251      tid=0x0000fffef4009140 nid=0x13d waiting on condition  [0x0000ffff88cda000]
    java.lang.Thread.State: WAITING (parking)
     at jdk.internal.misc.Unsafe.park(java.base@17.0.8/Native Method)
     - parking to wait for  <0x00000000ee75b640> (a org.opensearch.common.util.concurrent.BaseFuture$Sync)
     at java.util.concurrent.locks.LockSupport.park(java.base@17.0.8/LockSupport.java:211)
     at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(java.base@17.0.8/AbstractQueuedSynchronizer.java:715)
     at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(java.base@17.0.8/AbstractQueuedSynchronizer.java:1047)
     at org.opensearch.common.util.concurrent.BaseFuture$Sync.get(BaseFuture.java:272)
     at org.opensearch.common.util.concurrent.BaseFuture.get(BaseFuture.java:104)
     at org.opensearch.common.util.concurrent.FutureUtils.get(FutureUtils.java:74)
     at org.opensearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:55)
     at org.opensearch.searchpipelines.questionanswering.generative.llm.DefaultLlmImpl.doChatCompletion(DefaultLlmImpl.java:84)
     at org.opensearch.searchpipelines.questionanswering.generative.GenerativeQAResponseProcessor.processResponse(GenerativeQAResponseProcessor.java:109)
     at org.opensearch.search.pipeline.Pipeline.transformResponse(Pipeline.java:177)
     at org.opensearch.search.pipeline.PipelinedRequest.transformResponse(PipelinedRequest.java:31)
     at org.opensearch.action.search.TransportSearchAction.lambda$executeRequest$0(TransportSearchAction.java:398)
     at org.opensearch.action.search.TransportSearchAction$$Lambda$4884/0x0000003001d5d850.accept(Unknown Source)
     at org.opensearch.core.action.ActionListener$1.onResponse(ActionListener.java:82)
     at org.opensearch.core.action.ActionListener$5.onResponse(ActionListener.java:268)
     at org.opensearch.action.search.AbstractSearchAsyncAction.sendSearchResponse(AbstractSearchAsyncAction.java:671)
     at org.opensearch.action.search.ExpandSearchPhase.run(ExpandSearchPhase.java:132)
     at org.opensearch.action.search.AbstractSearchAsyncAction.executePhase(AbstractSearchAsyncAction.java:428)
     at org.opensearch.action.search.AbstractSearchAsyncAction.executeNextPhase(AbstractSearchAsyncAction.java:422)
     at org.opensearch.action.search.FetchSearchPhase.moveToNextPhase(FetchSearchPhase.java:298)
     at org.opensearch.action.search.FetchSearchPhase.lambda$innerRun$1(FetchSearchPhase.java:138)
     at org.opensearch.action.search.FetchSearchPhase$$Lambda$5645/0x0000003001e62400.run(Unknown Source)
     at org.opensearch.action.search.CountedCollector.countDown(CountedCollector.java:66)
     at org.opensearch.action.search.CountedCollector$$Lambda$5941/0x0000003001ee0cc0.run(Unknown Source)
     at org.opensearch.action.search.ArraySearchPhaseResults.consumeResult(ArraySearchPhaseResults.java:61)
     at org.opensearch.action.search.CountedCollector.onResult(CountedCollector.java:74)
     at org.opensearch.action.search.FetchSearchPhase$2.innerOnResponse(FetchSearchPhase.java:243)
     at org.opensearch.action.search.FetchSearchPhase$2.innerOnResponse(FetchSearchPhase.java:238)
     at org.opensearch.action.search.SearchActionListener.onResponse(SearchActionListener.java:59)
     at org.opensearch.action.search.SearchActionListener.onResponse(SearchActionListener.java:44)
     at org.opensearch.action.ActionListenerResponseHandler.handleResponse(ActionListenerResponseHandler.java:70)
     at org.opensearch.action.search.SearchTransportService$ConnectionCountingHandler.handleResponse(SearchTransportService.java:746)
     at org.opensearch.transport.TransportService$6.handleResponse(TransportService.java:880)
     at org.opensearch.transport.TransportService$ContextRestoreResponseHandler.handleResponse(TransportService.java:1496)
     at org.opensearch.transport.InboundHandler.doHandleResponse(InboundHandler.java:394)
     at org.opensearch.transport.InboundHandler.handleResponse(InboundHandler.java:386)
     at org.opensearch.transport.InboundHandler.messageReceived(InboundHandler.java:161)
     at org.opensearch.transport.InboundHandler.inboundMessage(InboundHandler.java:115)
     at org.opensearch.transport.TcpTransport.inboundMessage(TcpTransport.java:767)
     at org.opensearch.transport.netty4.Netty4MessageChannelHandler$$Lambda$4575/0x0000003001cd14b8.accept(Unknown Source)
     at org.opensearch.transport.InboundPipeline.forwardFragments(InboundPipeline.java:175)
     at org.opensearch.transport.InboundPipeline.doHandleBytes(InboundPipeline.java:150)
     at org.opensearch.transport.InboundPipeline.handleBytes(InboundPipeline.java:115)
     at org.opensearch.transport.netty4.Netty4MessageChannelHandler.channelRead(Netty4MessageChannelHandler.java:95)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
     at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:280)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
     at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
     at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
     at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
     at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
     at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:689)
     at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:652)
     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
     at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
     at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
     at java.lang.Thread.run(java.base@17.0.8/Thread.java:833)

    Locked ownable synchronizers:
     - None

[austintlee]

search thread example:

"opensearch[demo-node4][search][T#6]@22544" daemon prio=5 tid=0xb0 nid=NA waiting
  java.lang.Thread.State: WAITING
          at jdk.internal.misc.Unsafe.park(Unsafe.java:-1)
          at java.util.concurrent.locks.LockSupport.park(LockSupport.java:211)
          at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:715)
          at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1047)
          at org.opensearch.common.util.concurrent.BaseFuture$Sync.get(BaseFuture.java:272)
          at org.opensearch.common.util.concurrent.BaseFuture.get(BaseFuture.java:104)
          at org.opensearch.common.util.concurrent.FutureUtils.get(FutureUtils.java:74)
          at org.opensearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:55)
          at org.opensearch.searchpipelines.questionanswering.generative.llm.DefaultLlmImpl.doChatCompletion(DefaultLlmImpl.java:84)
          at org.opensearch.searchpipelines.questionanswering.generative.GenerativeQAResponseProcessor.processResponse(GenerativeQAResponseProcessor.java:102)
          at org.opensearch.search.pipeline.Pipeline.transformResponse(Pipeline.java:177)
          at org.opensearch.search.pipeline.PipelinedRequest.transformResponse(PipelinedRequest.java:31)
          at org.opensearch.action.search.TransportSearchAction.lambda$executeRequest$0(TransportSearchAction.java:398)
          at org.opensearch.action.search.TransportSearchAction$$Lambda$4889/0x00007fb484d4a940.accept(Unknown Source:-1)
          at org.opensearch.core.action.ActionListener$1.onResponse(ActionListener.java:82)
          at org.opensearch.core.action.ActionListener$5.onResponse(ActionListener.java:268)
          at org.opensearch.action.search.AbstractSearchAsyncAction.sendSearchResponse(AbstractSearchAsyncAction.java:671)
          at org.opensearch.action.search.ExpandSearchPhase.run(ExpandSearchPhase.java:132)
          at org.opensearch.action.search.AbstractSearchAsyncAction.executePhase(AbstractSearchAsyncAction.java:428)
          at org.opensearch.action.search.AbstractSearchAsyncAction.executeNextPhase(AbstractSearchAsyncAction.java:422)
          at org.opensearch.action.search.FetchSearchPhase.moveToNextPhase(FetchSearchPhase.java:298)
          at org.opensearch.action.search.FetchSearchPhase.lambda$innerRun$1(FetchSearchPhase.java:138)
          at org.opensearch.action.search.FetchSearchPhase$$Lambda$5039/0x00007fb484d7d8e0.run(Unknown Source:-1)
          at org.opensearch.action.search.CountedCollector.countDown(CountedCollector.java:66)
          at org.opensearch.action.search.CountedCollector$$Lambda$5893/0x00007fb484f172a8.run(Unknown Source:-1)
          at org.opensearch.action.search.ArraySearchPhaseResults.consumeResult(ArraySearchPhaseResults.java:61)
          at org.opensearch.action.search.CountedCollector.onResult(CountedCollector.java:74)
          at org.opensearch.action.search.FetchSearchPhase$2.innerOnResponse(FetchSearchPhase.java:243)
          at org.opensearch.action.search.FetchSearchPhase$2.innerOnResponse(FetchSearchPhase.java:238)
          at org.opensearch.action.search.SearchActionListener.onResponse(SearchActionListener.java:59)
          at org.opensearch.action.search.SearchActionListener.onResponse(SearchActionListener.java:44)
          at org.opensearch.action.ActionListenerResponseHandler.handleResponse(ActionListenerResponseHandler.java:70)
          at org.opensearch.action.search.SearchTransportService$ConnectionCountingHandler.handleResponse(SearchTransportService.java:746)
          at org.opensearch.transport.TransportService$6.handleResponse(TransportService.java:880)
          at org.opensearch.transport.TransportService$ContextRestoreResponseHandler.handleResponse(TransportService.java:1496)
          at org.opensearch.transport.TransportService$DirectResponseChannel.processResponse(TransportService.java:1579)
          at org.opensearch.transport.TransportService$DirectResponseChannel.sendResponse(TransportService.java:1559)
          at org.opensearch.transport.TaskTransportChannel.sendResponse(TaskTransportChannel.java:71)
          at org.opensearch.action.support.ChannelActionListener.onResponse(ChannelActionListener.java:62)
          at org.opensearch.action.support.ChannelActionListener.onResponse(ChannelActionListener.java:45)
          at org.opensearch.search.SearchService$3.onResponse(SearchService.java:1194)
          at org.opensearch.action.ActionRunnable.lambda$supply$0(ActionRunnable.java:74)
          at org.opensearch.action.ActionRunnable$$Lambda$4821/0x00007fb484d29ba8.accept(Unknown Source:-1)
          at org.opensearch.action.ActionRunnable$2.doRun(ActionRunnable.java:89)
          at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52)
          at org.opensearch.threadpool.TaskAwareRunnable.doRun(TaskAwareRunnable.java:78)
          at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52)
          at org.opensearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:59)
          at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:908)
          at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
          at java.lang.Thread.run(Thread.java:833)

[reta]

@austintlee the difference between these two flows is that the first one (#10248 (comment)) is run against remote node (by coordinator) but the latter one (#10248 (comment)) uses local transports (the coordinator node itself)

[msfroh]

Hmm.... I had been saying that we will eventually need to support async execution in search pipelines. It sounds like that time may be now. For response processors, it's arguably easier than for request processors (since they already get executed on a callback.)

Ingest pipelines assume async execution, but that also leads to #6338.

[heemin32]

Unless other processors in the search pipeline make async call to a transport action, the search processor should be run in search thread I guess? At least that is what I witnessed in ingest pipeline.

Do you have other search processors in the search pipeline which are called before your processor?

[austintlee]

@heemin32 not at the moment, but we plan to introduce a search request processor that is also going to invoke a transport action AND pass state (#9405) and this processor will be used in conjunction with the search response process (the one mentioned above).

[heemin32]

Okay. Seems like we need to see where the thread is converted from search thread to transport thread in the search pipeline implementation. I agree with @austintlee that search pipeline should run only in search thread which is similar to what ingest pipeline behave. Also, support of async won't do much good here because we lose search back pressure mechanism by releasing search thread earlier than search is completed.

[austintlee]

@heemin32 Does the ingest pipeline allow invocation of any transport actions? If so, can you provide an example that I can take a look at?

[heemin32]

https://github.com/opensearch-project/geospatial/blob/main/src/main/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessor.java#L155

https://github.com/opensearch-project/geospatial/blob/main/src/main/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessor.java#L161

The first code calls GET index API and the second code calls Query index API synchronously.

[msfroh]

I cobbled together an implementation of async response processors: msfroh@6b4279c

It's mostly untested (besides unit tests), but it might let you avoid the blocking get() call.