[ashie]

Which issue(s) this PR fixes:
None

What this PR does / why we need it:

(This PR depends on #5527 and includes its commits.)

This PR adds opt-in worker-thread parsing to in_tail. The
default remains synchronous (num_threads = 1), setting
num_threads to 2 or more enables parsing on worker threads
when the configured parser and hooks are supported.

Parsing uses worker-specific parser instances. Event emission
and position updates remain on the event loop, and processing
for each file stays serialized. Unsupported parser
configurations and overridden compatibility hooks continue
through the synchronous path. The worker pool is bounded and
handles asynchronous completion, queue saturation, rotation,
and shutdown.

The goal is to reduce delays to other files and watcher
callbacks while a large file is being processed. In an
exploratory local Test Driver measurement, while reading an
80 MiB JSONL file with a 1 MiB/s per-file read limit, median
probe-file latency was about 48 ms with synchronous parsing
and about 1 ms with four workers. The result was similar with
the StatWatcher enabled or disabled. This is a result for
that specific setup, not a general performance guarantee. If
all workers are occupied, tasks can still wait; this change
does not guarantee a latency bound or claim a general
throughput improvement.

Docs Changes:
Updated the num_threads configuration description.
fluent/fluentd-docs-gitbook#716

Release Note:
Add opt-in worker-thread parsing to in_tail via
num_threads. The default remains synchronous.

Assisted-by: GPT-6.1 Sol (OpenAI)

[ashie]

test/plugin/in_tail/test_worker_pool.rb:117:15: C: [Correctable] Performance/TimesMap: Use Array.new(2) with a block instead of .times.map.
    results = 2.times.map { result_from(pool) }
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
test/plugin/in_tail/test_worker_pool.rb:210:15: C: [Correctable] Performance/TimesMap: Use Array.new(3) with a block instead of .times.map.
    results = 3.times.map { result_from(pool) }
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

470 files inspected, 2 offenses detected, 2 offenses autocorrectable
Error: Process completed with exit code 1.

[ashie]

For the CI failure, it doesn't seem caused by this change.
I've opened another PR for it: #5565

[Watson1978]

Reviewed at 16c2ae6. I ran the worker path with extra scripts on Linux (Ruby 4.0.7) and reproduced each issue below. The inline comments have the details and the numbers. The scripts are here: https://gist.github.com/Watson1978/97704fc8a8c66f1d411d89c00aeeb5e4

Must fix before merge

• detach_watcher can submit a batch inside tw.detach after the deferral decision, and the completion then corrupts another file's position entry (in_tail.rb:554).
• A file that waited for queue capacity stops reading after its retried batch completes (io_handler.rb:288).
• An exception in the completion path kills the process, and the handler stays pending afterwards (worker_pool.rb:271, io_handler.rb:189).

Should fix

• A detached watcher whose batch failed to emit is never retried or closed, and the final close cannot close a pending watcher (tail_watcher.rb:134, tail_watcher.rb:77).
• The position file is never closed in worker mode (in_tail.rb:349).
• Shutdown waits for SHUTDOWN_TIMEOUT whenever a file is not at EOF (in_tail.rb:835).
• A result published during join is dropped when the completion watcher is detached (in_tail.rb:348).
• A worker that does not finish by the join timeout outlives terminate (in_tail.rb:747).

The last four should items are one shutdown sequence: stop the workers, process the remaining results, close the watched files, then close the position file. They may be easier to fix together. The orphaned watcher at tail_watcher.rb:134 is a runtime path and needs its own fix.

[Watson1978]

Re-reviewed at 9f91db5. I re-ran the reproduction scripts from the first round against this commit and all of the original reproduction cases now pass: the batch submitted by detach keeps its position entry through compaction, the retried batch resumes reading (1200 of 1200 records), the completion path survives a failing position write, the failed-emit watcher is closed right away, the position file is closed, shutdown no longer incurs the full drain timeout for a throttled backlog, no result was dropped during join in 15 shutdowns, and no worker thread survives terminate. Thanks for the thorough fix.

One case of the position corruption is still open with follow_inodes true, so I am keeping the request for changes for that alone. There is also one flaky test. Details inline. The new reproduction is in the same gist: https://gist.github.com/Watson1978/97704fc8a8c66f1d411d89c00aeeb5e4

[Watson1978]

Reviewed 8dfb08b.

b08383e closes the follow_inodes case. test_follow_inodes_refresh_unwatches_pending_detached.rb from the gist passes now. The new test_follow_inodes_refresh_preserves_worker_draining_position_entries fails on 9f91db5 and passes on this head for both data variants. 8dfb08b makes test_worker_parse_error_retries_synchronously deterministic, 5 of 5 runs here. test/plugin/test_in_tail.rb and the in_tail unit tests pass locally on Linux. The failing Windows job is test_without_event_time(TestFluentCat), which is unrelated to this PR.

One part of the earlier MUST 2 is still open, see the inline comment. The new commits did not introduce it. It is already on 9f91db5, and my earlier run of test_queue_saturation_stall.rb passed only by timing. With the watch timer off that test still fails about one run in three on both commits. I added test_parked_batch_loses_notification.rb to the gist. It reproduces the case deterministically without inotify.

https://gist.github.com/Watson1978/97704fc8a8c66f1d411d89c00aeeb5e4

[ashie]

Thanks for rechecking these cases. The remaining parked-batch issue
is fixed in a117775. I added a unit regression test and extended the
queue-saturation test to cover notifications during partial writes.
The local tests pass.

[ashie]

I'll fix the failed test:

https://github.com/fluent/fluentd/actions/runs/39745003992/job/119711197908?pr=5529

3) Failure: test_worker_position_write_error_does_not_stop_completion_processing(TailInputTest::singleline)
D:/a/fluentd/fluentd/test/plugin/test_in_tail.rb:1068:in `block in test_worker_position_write_error_does_not_stop_completion_processing'
     1065:         pool = plugin.instance_variable_get(:@worker_pool)
     1066:         waiting(5) { sleep 0.01 until driver.events.size == 2 && pool.statistics[:tasks].zero? && !watchers.first.pending? }
     1067:         assert_equal 0, watchers.first.pe.read_pos
  => 1068:         assert_equal 5, watchers.last.pe.read_pos
     1069:         watchers.first.on_notify
     1070:         waiting(5) { sleep 0.01 until watchers.first.pe.read_pos == 5 }
     1071:       end
D:/a/fluentd/fluentd/lib/fluent/test/driver/base.rb:204:in `block in run_actual'
C:/hostedtoolcache/windows/Ruby/3.3.12/x64/lib/ruby/gems/3.3.0/gems/timeout-0.6.1/lib/timeout.rb:296:in `block in timeout'
C:/hostedtoolcache/windows/Ruby/3.3.12/x64/lib/ruby/gems/3.3.0/gems/timeout-0.6.1/lib/timeout.rb:38:in `handle_timeout'
C:/hostedtoolcache/windows/Ruby/3.3.12/x64/lib/ruby/gems/3.3.0/gems/timeout-0.6.1/lib/timeout.rb:306:in `timeout'
D:/a/fluentd/fluentd/lib/fluent/test/driver/base.rb:203:in `run_actual'
D:/a/fluentd/fluentd/lib/fluent/test/driver/base.rb:96:in `run'
D:/a/fluentd/fluentd/lib/fluent/test/driver/base_owner.rb:130:in `run'
D:/a/fluentd/fluentd/test/plugin/test_in_tail.rb:1047:in `test_worker_position_write_error_does_not_stop_completion_processing'
<5> expected but was
<0>
Error: <5> expected but was
<0>.

[Watson1978]

Thanks

[ashie]

Thanks!