Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refactor StoreTailer.readingDocument #1414

Open
wants to merge 2 commits into
base: develop
Choose a base branch
from
Open

Refactor StoreTailer.readingDocument #1414

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2f50ef6
Refactor StoreTailer.readDocument
tgd Aug 10, 2023
d44dd6e
Rollback switch re-ordering
tgd Aug 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 40 additions & 15 deletions src/main/java/net/openhft/chronicle/queue/impl/single/StoreTailer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package net.openhft.chronicle.queue.impl.single;

import net.openhft.chronicle.assertions.AssertUtil;
import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.MappedBytes;
import net.openhft.chronicle.bytes.MappedBytesStore;
Expand All @@ -42,6 +43,7 @@
import java.io.StreamCorruptedException;
import java.text.ParseException;

import static net.openhft.chronicle.assertions.AssertUtil.SKIP_ASSERTIONS;
import static net.openhft.chronicle.queue.TailerDirection.*;
import static net.openhft.chronicle.queue.TailerState.*;
import static net.openhft.chronicle.queue.impl.single.ScanResult.*;
Expand Down Expand Up @@ -202,11 +204,17 @@ public String toString() {
@Override
public DocumentContext readingDocument(final boolean includeMetaData) {
DocumentContext documentContext = readingDocument0(includeMetaData);
// this check was added after a strange behaviour seen by one client. It should be impossible.
checkReadRemainingPrecondition(documentContext);
return documentContext;
}

/**
* Added in response to client bug please see <a href="https://github.com/OpenHFT/Chronicle-Queue/issues/801">801</a>.
*/
private void checkReadRemainingPrecondition(DocumentContext documentContext) {
if (documentContext.wire() != null)
if (documentContext.wire().bytes().readRemaining() >= 1 << 30)
throw new AssertionError("readRemaining " + documentContext.wire().bytes().readRemaining());
return documentContext;
}

DocumentContext readingDocument0(final boolean includeMetaData) {
Expand All @@ -228,16 +236,14 @@ DocumentContext readingDocument0(final boolean includeMetaData) {
if (tryAgain)
next = next0(includeMetaData);

Wire wire = context.wire();
if (wire != null && context.present(next)) {
Bytes<?> bytes = wire.bytes();
context.setStart(bytes.readPosition() - 4);
readingDocumentFound = true;
this.lastReadIndex = this.index();
return context;
// An entry has been found, prepare the context and return it.
if (next && (context.wire() != null)) {
Copy link
Contributor

@nicktindall nicktindall Mar 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

I notice we went from if (wire != null && context.present(next)) to if (next && context.wire() != null) I can see that context.present(...) just returns the value passed in but it also has the side-effect 🤮 of setting the present field in the context.

Is that change significant?

Copy link
Contributor

@nicktindall nicktindall Mar 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Oh I see it happens in init now. Much nicer.

return prepareStoreTailerContext();
}

readingDocumentCycleNotFound(next);
if (state == CYCLE_NOT_FOUND) {
readingDocumentCycleNotFound(next);
}

} catch (StreamCorruptedException e) {
throw new IllegalStateException(e);
Expand All @@ -251,6 +257,16 @@ DocumentContext readingDocument0(final boolean includeMetaData) {
return INSTANCE;
}

/**
* A document has been found - prepare the tailer context for return to the caller.
*/
private StoreTailerContext prepareStoreTailerContext() {
context.init();
readingDocumentFound = true;
this.lastReadIndex = this.index();
return context;
}

private void readingDocumentDBUE(DecoratedBufferUnderflowException e) {
if (queue.isReadOnly()) {
Jvm.warn().on(StoreTailer.class,
Expand All @@ -262,12 +278,12 @@ private void readingDocumentDBUE(DecoratedBufferUnderflowException e) {
}

private void readingDocumentCycleNotFound(boolean next) {
RollCycle rollCycle = queue.rollCycle();
if (state == CYCLE_NOT_FOUND && direction == FORWARD) {
if (direction == FORWARD) {
RollCycle rollCycle = queue.rollCycle();
int firstCycle = queue.firstCycle();
if (rollCycle.toCycle(index()) < firstCycle)
toStart();
} else if (!next && state == CYCLE_NOT_FOUND && cycle != queue.cycle()) {
} else if (!next && cycle != queue.cycle()) {
// appenders have moved on, it's possible that linearScan is hitting EOF, which is ignored
// since we can't find an entry at current index, indicate that we're at the end of a cycle
state = TailerState.END_OF_CYCLE;
Expand Down Expand Up @@ -1420,6 +1436,9 @@ protected void finalize() throws Throwable {
}
}

/**
* Supports a view to a document somewhere in the queue.
*/
class StoreTailerContext extends BinaryReadDocumentContext {
StoreTailerContext() {
super(null);
Expand All @@ -1446,8 +1465,14 @@ public void close() {
super.close();
}

boolean present(final boolean present) {
return this.present = present;
/**
* Initialise the context with a new document ensuring that the start of this context points to the start of
* this document in the underlying buffer.
*/
public void init() {
present = true;
long bytesReadPosition = wire.bytes().readPosition();
setStart(bytesReadPosition - Wires.SPB_HEADER_SIZE);
}

public void wire(@Nullable final AbstractWire wire) {
Expand Down