HDDS-15422. Stream read seek should not close stream

[szetszwo]

What changes were proposed in this pull request?

• Sequential read: close the block (gRPC) stream at block boundary
• Position read: do not close
• Seek: do not close

What is the link to the Apache JIRA

HDDS-15422

How was this patch tested?

By existing tests. Checked that the StreamReader won't be closed for seek and position read.

[yandrey321]

we need tests for validating that stream is not closed when:

1. seek inside the length of prefetched buffer
2. seek outside the length of prefetched buffer
3. read last N bytes from the input stream, seek to 0, read

[szetszwo]

we need tests for validating that stream is not closed when ...

@yandrey321 , We already have such a lot test for seek and position read; see TestStreamBlockInputStream.

... Checked that the StreamReader won't be closed for seek and position read.

The way I checked it is to turn on debug log and run org.apache.hadoop.ozone.client.rpc.read.TestStreamBlockInputStream before and after this fix. You may try it.

[ss77892]

When I was checking HBase behavior, I noticed that it's quite a common scenario where it does seek {current} -> 0 and seek 0 -> current + small offset without any reads in between. Usually, the last seek lands in the existing buffer/prefetch. Would it be more reasonable if we try to preserve the buffer if the actual read happen for the data we already have in the buffer?

[szetszwo]

That's a good point! Let's put it back to the queue for seek.

[yandrey321]

we need tests for validating that stream is not closed when ...

@yandrey321 , We already have such a lot test for seek and position read; see TestStreamBlockInputStream.

... Checked that the StreamReader won't be closed for seek and position read.

The way I checked it is to turn on debug log and run org.apache.hadoop.ozone.client.rpc.read.TestStreamBlockInputStream before and after this fix. You may try it.

we need to make sure that we'd not have regression when the next person would change the behavior.

[szetszwo]

we need to make sure that we'd not have regression when the next person would change the behavior.

@yandrey321 , this is a good suggestion only if such test is easy to be done in low cost. It would be great if you could contribute on it. I look forward to see your contribution!

[sodonnel]

I am not sure I understand what is happening here. Say we have pre-read enabled, 32MB. This means the queue will receive 32 x 1MB chunks from the first read in the file.

Then the read buffer will pull potentally one of those chunks off the queue into the buffer, not necessairly all of them - is that correct?

Do we need to ensure the queue is empty before pushing data read from the queue back onto it?

Then onces we seek, even if we push the read buffer back to the queue, or if the queue has some data, how do we know where to read from the queue? Eg, we read 1MB at offset zero. The queue gets 32MB of pre-read data.

Then we seek to offset 100MB - all that queued data needs to be dropped and a new read from the server.

However if we seek instread to 20MB, then on the queue is some data we need.

[szetszwo]

Then the read buffer will pull potentally one of those chunks off the queue into the buffer, not necessairly all of them - is that correct?

It is correct -- the protos are in the queue and they will be pulled one by one.

Do we need to ensure the queue is empty before pushing data read from the queue back onto it?

No, because we will drop it if the proto is outside the required range; see the change below:

     ByteBuffer readFromQueue() throws IOException {
       final long blockOffset = readBlock.getOffset();
       final long pos = getPos();
       if (pos < blockOffset) {
-        // This should not happen, and if it does, we have a bug.
-        setFailedAndThrow(new IllegalStateException(
-            this + ": out of order, position " + pos + " < block offset " + blockOffset));
+        // This can happen after seek, just drop the buffer
+        return null;
       }
       final long offset = pos - blockOffset;
       if (offset > 0) {
         dataBuffer.position(Math.toIntExact(Math.min(offset, dataBuffer.limit())));
       }
       LOG.debug("{}: return response positon {}, length {} (block offset {}, length {})",
           name, pos, dataBuffer.remaining(), blockOffset, data.size());
       return dataBuffer;
     }

... how do we know where to read from the queue? ...

The proto has the offset and length information. The readFromQueue() method will check and set the dataBuffer before returning it; see above.

Then we seek to offset 100MB - all that queued data needs to be dropped ...

That's correct.

... if we seek instread to 20MB, then on the queue is some data we need.

Continue with your example, the code will drop the first 20MB and use the remaining 12MB.

[sodonnel]

OK, I think I understand better now.

In this code we check if the seek is < 0, or > blocklength or == pos. Otherwise we offer the current buffer back to the queue.

Should we also check if the position is within the bounds of the current buffer?

Eg we are at position 1 so we have a 1MB buffer, 32MB of pre-read on the queue.

We seek to position 100. This, I think, would trigger putting the current buffer onto THE BACK of the queue.

Then we would read and discard all the 32MB pre-read, before hitting the buffer we already had?

I think we can check if pos > position && pos <= position + buffer.remaining() and if so, just set buffer.position accordingly?

[szetszwo]

That's a good idea. We could put back the buffer in the head (instead of the tail) of the queue. Let me see how to do it.

[sodonnel]

I think a seek within the current buffer doesn't even need to go back on the queue - just set the buffer position and it will continue reading as usual from the next read.

[szetszwo]

Sure, just pushed a change for reusing the buffer.

[sodonnel]

I agree with @yandrey321 that there needs to be some tests here. The Seek logic has various boundary cases, eg:

• seek to a position behind the current buffer
• seek within the current buffer
• seek within the pre-read chunks
• seek past all the pre-read chunks
• seek to the start
• seek out of range

The tests should validate data read correctness, and that unnecessary reads are not issued to the server when relevant.

To commit this code without tests is not going to help the next person that comes along, and this code is complex enough that its difficult (at least for me) to follow or review the correctness of.

[sodonnel]

Lets say we have 32MB pre-read in the queue and we are within the first 1MB. Then we seek to 10MB offset. reuseReadBuffer will return null.

Then later in read() we call dataAvailableToRead which triggers a call streamingReader.read() if there is no remaining in the buffer or the buffer is null.

  private synchronized boolean dataAvailableToRead(int length, boolean preRead) throws IOException {
    if (position >= blockLength) {
      return false;
    }
    initialize();

    if (!hasRemaining(readBuffer)) {
      readBuffer = streamingReader.read(length, preRead);
    }
    Preconditions.assertTrue(hasRemaining(readBuffer));
    return true;
  }

Inside streamingReadder.read:

 private ReadBuffer read(int length, boolean preRead) throws IOException {
      checkError();
      if (future.isDone()) {
        return null; // Stream ended
      }

      readBlock(length, preRead);  // !! Reads more data before polling

      while (true) {
        final ReadBlockResponseProto proto = poll();

It pre-reads more data duplicating data already on the queue and also pulling more data onto the queue that the pre-read limit.

This is the sort of thing the boundary tests for 'reading within the pre-read doesn't issue new read calls' would catch.

[sodonnel]

I've attached a diff that has a unit test to validate that we don't re-read data already on the queue when seeking forward. I tried to create a similar test for a backwards seek but its more difficult as we want to test that the queue is drained before requesting more data, which isn't really possible with the current structure.

But I think we need to change the reading behavior so it reads through the queue and only reads more data from the datanode if the queue has been drained.

In a dysfunctional case, we could have read 32MB onto the queue with pre-read. Seek from 0.1MB to 1.1MB. This would trigger another 32MB read onto the end of the queue, but the data we want is there are the head. Now there is 31 + 32MB on the queue. Seek now to 3.1MB, the same happens, as we have the next chunk we need available.

Even with the backwards seek, we could have 32MB on the queue, and then read 32MB more before throwing away the first 32MB. We should throw away the first 32MB first.

forward-seek-test.patch

[sodonnel]

Claude suggests there is a problem if we have pre-read to the end of the stream so the stream is done, and then seek backwards. This is what it says, although I am not sure how the stream can be done and we are still reading, but it sounds plausible:

"Infinite spin in the read() while-loop when a backward seek leaves stale ahead-of-position responses in the queue and then the stream ends."

Client at position 50 MB seeks backward to 10 MB; reuseReadBuffer returns null; streamingReader is alive with ~30 stale responses (for offsets 51–80 MB) queued. After readBlock() sends a new request at 10
MB, poll() returns each stale response in turn. For each, getByteBuffer(staleProto, 10 MB) finds blockOffset(10 MB) < protoOffset(51+ MB) and returns null; read is null; hasRemaining is false; the loop continues. If the future
completes (stream done) before new responses arrive — e.g. server finishes sending to the old read boundary — poll() returns null at line 474-475. Line 511 NPEs. The loop has no exit path for a null proto, so even without the NPE
it would spin forever: there is no break or return null when proto is null and the stream is done."

[szetszwo]

It is a bug -- it should return null when the proto returned by poll() is null.

[sodonnel]

This hasn't changed as part of this PR, but why do we only close the reader if preRead is true?

[szetszwo]

preRead:

• true for sequential read
• false for position read

For position read, we should keep the reader running.

[ss77892]

I let the AI validate the correctness, and it has found a couple of issues. Here is the patch that has tests to validate and fixes.
0001-HDDS-15422.-Fix-StreamBlockInputStream-seek-re-fetch.patch

[szetszwo]

I let the AI validate the correctness, ...

@ss77892 , thanks for checking it! I will fix the bugs in the PR. For the tests, let's add them in a new JIRA.

yandrey321 Requested changes with read-only permissions

@yandrey321 , are you still requesting changes?

[yandrey321]

@szetszwo I think that we need specific use-cases to validate what is happening in seek scenarios. And it would also nice to have a benchmark that can be run to validate improvements.

[ss77892]

LGTM. I have a set of patches to improve the logic of stream read, but they depends on this commit.

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[yandrey321]

@ss77892 is it ok to commit this fix as a foundation for future improvements suggested in comments? Could you please open JIRAs for these improvements?

[yandrey321]

lets submit these changes and open JIRA for improvements that we discussed in comments. I can pick up some of these work items, including unit tests.

@ss77892 what do you think?