Skip to content

Archiver reader lock refactor#3807

Open
jacomago wants to merge 10 commits into
ControlSystemStudio:masterfrom
jacomago:archiver-reader-lock-refactor
Open

Archiver reader lock refactor#3807
jacomago wants to merge 10 commits into
ControlSystemStudio:masterfrom
jacomago:archiver-reader-lock-refactor

Conversation

@jacomago
Copy link
Copy Markdown
Contributor

@jacomago jacomago commented May 26, 2026

Refactors for ArchiveFetchJob, ApplianceValueIterator, ApplianceArchiveReader.

Refactors:

  • Remove IteratorListener (didn't seem necessary to me)
  • Add more logging
  • Make fetch data in ArchiveFetchJob asynchrounous

Fixes:

  • Make lock in ApplianceValueIterator per object rather than global

Feature:

  • Add a timeout archive_read_timeout_ms to databrowser_preferences, so down ArchiveSources don't block fetching for others.

Testing:

  • Add tests for the fix
  • Add more generic tests for the involved classes

Checklist

  • Testing:

    • [ x] The feature has automated tests
    • [ x] Tests were run
    • If not, explain how you tested your changes

  • Documentation:

    • [x ] The feature is documented
    • The documentation is up to date
    • Release notes:
      • Added an entry if the change is breaking or significant
      • Added an entry when adding a new feature

@jacomago jacomago self-assigned this May 26, 2026
@jacomago
Copy link
Copy Markdown
Contributor Author

jacomago commented May 26, 2026

Not sure whether to split this up into multiple MRs:

  • Lock changes
  • Timeout addition
  • Asynchronous fetching

@kasemir
Copy link
Copy Markdown
Collaborator

kasemir commented May 26, 2026

Make fetch data in ArchiveFetchJob asynchrounous

The archive fetch job is already a background thread, and concurrent_requests controls how many are started.
Why add one more layer of threading?

@kasemir
Copy link
Copy Markdown
Collaborator

kasemir commented May 26, 2026

Is the intent of the added level of threading only for the timeout?

Before:
One ArchiveFetchJob iterates over data sources, reads from one source, merges with data-fetched-so-far?

With this PR:
One ArchiveFetchJob iterates over data sources, starts fetch in background to read from one source, 'gets' the future to obtain the result, merges with data-fetched-so-far?

jacomago and others added 9 commits May 27, 2026 08:13
@jacomago @claude
Add failing regression tests for hasNext()/close() deadlock
6bd4de3 
With synchronized hasNext(), a worker blocked inside mainIterator.hasNext()
holds the 'this' monitor indefinitely; close() cannot acquire it and hangs.
closeCompletesWhileHasNextIsBlocking reproduces this with @timeout(5).

Includes FakeDataRetrieval/FakeApplianceArchiveReader/BlockingGenMsgIterator
test infrastructure and Mockito dependency.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Fix hasNext()/close() deadlock by removing synchronized from hasNext()
09169f2 
hasNext() no longer holds the 'this' monitor while blocking on a streaming
network read. close() can now acquire the lock concurrently, set closed=true,
and close the underlying stream — which in turn unblocks the reader.

next() still guards its mainIterator.next() call with synchronized(this) and
rechecks closed, so the close-between-hasNext-and-next race remains safe.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Add failing regression test for static lock serialising concurrent fe…
bded065 
…tches

twoIteratorFetchesProceedConcurrently shows that two iterators for
independent PVs block each other inside fetchDataInternal() because
they compete for a single static monitor.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Make fetchDataInternal lock instance-level, not JVM-global
c4b5efc 
Changing the lock from static to a per-instance final field means each
iterator only serialises its own concurrent fetchDataInternal calls.
Fetches for different PVs can now proceed concurrently.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Add iterator routing, type and lifecycle coverage tests
393b89b 
ApplianceArchiveReaderTest: verifies enum PVs route to non-numeric iterator,
numeric scalars use the optimized path, cancel() closes active iterators,
and the WeakHashMap releases references after GC.

ApplianceMeanValueIteratorTest: checks the mean_<interval>() operator URL,
and that determineDisplay rejects enum/waveform types.

ApplianceOptimizedValueIteratorTest: verifies the optimized_N() URL,
VStatistics output when useStatistics=true, and VNumber when false.

ApplianceNonNumericOptimizedValueIteratorTest: checks n=1 routes raw fetch,
n>1 uses nth_N() operator, and the 1.5x boundary gives n=2.

ApplianceStatisticsValueIteratorTest: verifies all five operator streams
(mean, std, min, max, count) are opened, and that close() closes all five.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Add failing test for per-source archive read timeout
ed506cb 
faultySourceTimesOutAndLoopContinues fails: the slow source blocks
WorkerThread.run() indefinitely because no timeout guards the fetch
loop; @timeout(5) fires after 5 s, proving the bug.

Scaffolding added so the tests compile:
- archive_read_timeout_ms preference key and 30 s default
- ArchiveFetchJob.openReader() hook so TestableFetchJob can inject
  fake readers without going through ArchiveReaders SPI
- Package-private test-only constructor that skips JobManager
- Mockito added to databrowser pom

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Add per-source read timeout to ArchiveFetchJob fetch loop
4845fdc 
WorkerThread.run() now submits each source fetch to Activator.thread_pool
and calls Future.get(archive_read_timeout_ms, MILLISECONDS). On timeout:
- Future.cancel(true) interrupts the carrier thread
- The active ArchiveReader.cancel() is called to close the connection
- archiveFetchFailed() is reported and the loop advances to the next source

fetchFromSource() extracts the reader-open + iterator-drain logic that
previously lived inline in run(), keeping the outer loop readable.

archive_read_timeout_ms=0 disables the timeout (Future.get() with no
deadline), preserving the previous behaviour when needed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Add generic functional tests alongside existing regression tests
47d1f6d 
Broadens test coverage beyond the three regression scenarios (static
lock, hasNext/close deadlock, per-source timeout) to cover the normal
code paths that a reader of these classes relies on.

- ApplianceValueIteratorExtractDataTest (new, 13 tests): extractData()
  across all supported payload types — scalar numerics, enum, string,
  waveform — plus alarm severity and display header extraction.

- ApplianceArchiveReaderTest (+6 tests): getOptimizedValues() routing
  (points≤count→raw, statistics, mean, old-appliance fallback on fetch
  failure), iterator map registration, and close()→cancel() delegation.

- ApplianceStatisticsValueIteratorTest (+2 tests): VStatistics assembly
  from five sub-iterators, and null-safe next() after close().

- ArchiveFetchJobTest (+7 tests): UnknownChannelException→channelNotFound
  routing (full and partial), two healthy sources completing together,
  IOException→archiveFetchFailed, cancel-before-start, and RAW vs
  OPTIMIZED request type dispatch.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago @claude
Improve error logging for archive fetch failures and fallback paths
5c82220 
- ApplianceValueIterator.close(): log IOException at WARNING instead of
  wrapping it in IllegalStateException; move closed=true into finally so
  it is always set even when the stream close throws
- ApplianceStatisticsValueIterator.closeStream(): add package-level logger
  and emit WARNING instead of silently swallowing the IOException
- ApplianceArchiveReader.getOptimizedValues(): log FINE for expected type
  fallbacks and WARNING when the optimized or count-based path fails
- ArchiveFetchJob.WorkerThread: log WARNING for ExecutionException and
  generic Exception fetch failures, not just TimeoutException

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago jacomago force-pushed the archiver-reader-lock-refactor branch 2 times, most recently from 0638701 to dac09a8 Compare May 27, 2026 12:47
@jacomago @claude
Move per-source read timeout to outer polling loop
4d19a4f 
Timeout is now checked in the existing run(JobMonitor) poll loop rather
than a per-source virtual-thread watchdog, eliminating the extra thread
per source. A stuck source triggers cancel() on the whole job.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@jacomago jacomago force-pushed the archiver-reader-lock-refactor branch from dac09a8 to 4d19a4f Compare May 27, 2026 13:16
@jacomago
Copy link
Copy Markdown
Contributor Author

jacomago commented May 27, 2026 •
edited
Loading

Refactored so the timeout handling is in ArchiveFetchJob.run rather than in ArchiveFetchJob.WorkerThread.run.

Also now uses the existing cancel rather than some explicit other way of stopping the job.

@jacomago
Copy link
Copy Markdown
Contributor Author

jacomago commented May 27, 2026

@georgweiss Note I didn't add a notification for the timeout in this PR. I'm not sure how to do it, but I thought would be better as a follow up.

@sonarqubecloud
Copy link
Copy Markdown

sonarqubecloud Bot commented May 27, 2026

Quality Gate Passed Quality Gate passed

Issues
29 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

kasemir
kasemir approved these changes May 27, 2026
View reviewed changes
Copy link
Copy Markdown
Collaborator

@kasemir kasemir left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the ArchiveFetchJob, this PR basically changes

if (monitor.isCanceled())
    worker.cancel();

into

if (monitor.isCanceled()  ||  we_have_exceeded_archive_read_timeout_ms))
    worker.cancel();

The ArchiveReader is otherwise created and used the same way,
so this should be perfectly safe for the different archive reader
implementations.

@kasemir
Copy link
Copy Markdown
Collaborator

kasemir commented May 27, 2026

Also now uses the existing cancel rather than some explicit other way of stopping the job.

Yes, that is good.
The different archive reader implementations all have their own way of cancelling.
RDB implementation tries to cancel the JDBC transaction, archive appliance might try to close the web service connection. The ArchiveFetchJob should simply call 'ArchiveReader.cancel' so the reader implementation can then do what's most appropriate. The ArchiveFetchJob should not interrupt the thread and potentially put the reader into a bad state.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@kasemir kasemir kasemir approved these changes

@shroffk shroffk Awaiting requested review from shroffk

@georgweiss georgweiss Awaiting requested review from georgweiss

@abrahamwolk abrahamwolk Awaiting requested review from abrahamwolk

Assignees

@jacomago jacomago

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@jacomago @kasemir