go/store/datas/pull: pull_chunk_fetcher: Move chunk fetching to a streaming interface instead of batch. #7644
Conversation
…eaming interface instead of batch. We want to better pipeline I/O when pulling from a remote, and moving a streaming interface where storage can see more of the addresses that are needed at once will allow us to achieve it. For now, we implement the streaming interface just by calling the existing batch get interface.
| mu.Lock() | ||
| missing.Remove(chk.H) | ||
| mu.Unlock() | ||
| select { |
There was a problem hiding this comment.
This select block is a little odd to me. Why aren't you just sticking chk into f.resCh and calling it a day? These are unlimited channels, so they aren't going to block, right? Does this avoid getting a panic when the channel has been closed prematurely?
There was a problem hiding this comment.
The channels are blocking and unbuffered...it's a synchronous exchange. So here we don't have a way to bail early but the reader might have already shut down, so we need to get through the batch of responses by ignoring them.
| if err != nil { | ||
| return err | ||
| } | ||
| if len(cChk.FullCompressedChunk) == 0 { |
There was a problem hiding this comment.
The only way this would happen would be if a CompressedChunk{} was returned, which should always be accompanied by an error. Is this out of an abundance of caution, or is there another case where this will come up?
We want to better pipeline I/O when pulling from a remote, and moving a streaming interface where storage can see more of the addresses that are needed at once will allow us to achieve it.
For now, we implement the streaming interface just by calling the existing batch get interface.