possibility of an indefinite while loop when https stream reader thread notifies and the other thread has not yet read the condVar

[Abhinav-Upadhya-ni]

Suppose a scenario, where we have 2 threads doing the read operation,

it is possible that when the thread has errored out, in say read_exact method, now when it tries to notify the other thread, it is possible that the other thread is somewhere between while loop and claiming the cache mutex, so the notification is missed,

and since it errored out in read_exact, it has not set the state.reader to the right state and is None, which can potentially cause missed notification and an indefinite while loop.

we should also add retries in case we face the issue when doing read_exact, especially if the error is Interrupted or UnexpectedEOF.

[Abhinav-Upadhya-ni]

maybe we add a read_failed_somewhere check before we take the state.reader

// Claim CACHE mutex
        let mut state = self.state.lock().unwrap();
        // Is there block in cache?
        // - If yes, release CACHE mutex, and return
        if let Some(bytes_read_from_cache) = state.read_from_cache(pos, buf) {
            log::debug!("Immediate cache success");
            return Ok(bytes_read_from_cache);
        }
        // check if read failed somewhere
        if state.read_failed_somewhere {
            return Err(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "another thread experienced a problem creating a reader",
            ));
        }
        // - If no, check if read in progress
        let mut reading_stuff = state.reader.take();

and similarly, for retry, we'd probably need to match the read_exact to Ok(_) and Err(e), then handle accordingly.
I'm using take instead of as_mut, to avoid issues when we want to create a new stream reader,
this has improvement scope but a retry mechanism seems right here.

// use take instead of as_mut() so that we can create new stream reader without any hassle
        let (mut reader, mut reader_pos) = reading_stuff.reader.take().unwrap();
        if pos > reader_pos {
            log::debug!(
                "Read: reading ahead from 0x{:x} to 0x{:x} without skipping",
                reader_pos,
                pos
            );
        }
        while pos >= reader_pos {
            // Fast forward beyond the desired position, recording any reads in the cache
            // for later.
            let to_read = min(max_block, self.len as usize - reader_pos as usize);
            let mut new_block = vec![0u8; to_read];
            let mut retries = 0;
            while retries < MAX_RETRIES {
                match reader.read_exact(&mut new_block) {
                    Ok(_) => break,
                    Err(e) => match e.kind() {
                        ErrorKind::UnexpectedEof | ErrorKind::Interrupted => {
                            retries += 1;
                            // This is a bit of a hack, but it seems to be the only way
                            // to get the underlying stream to retry.
                            log::error!(
                                "Error reading from stream: {:?} - retrying {}/{}",
                                e,
                                retries,
                                MAX_RETRIES
                            );
                            // Recreate the HTTP stream since the server errored out on the old stream
                            reading_stuff.reader = Some((
                                BufReader::new(
                                    reading_stuff
                                        .range_fetcher
                                        .fetch_range(reader_pos)
                                        .map_err(|e| {
                                            std::io::Error::new(
                                                ErrorKind::Unsupported,
                                                e.to_string(),
                                            )
                                        })?,
                                ),
                                reader_pos,
                            ));
                            let (new_reader, new_reader_pos) = reading_stuff.reader.take().unwrap();
                            // Update the reader and reader_pos variables
                            reader = new_reader;
                            reader_pos = new_reader_pos;
                        }
                        _ => {
                            log::error!("Error reading from stream: {:?}", e);
                            return Err(std::io::Error::new(
                                std::io::ErrorKind::BrokenPipe,
                                "error reading from stream",
                            ));
                        }
                    },
                }
            }
            if retries == MAX_RETRIES {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "reached max number of retries reading from stream",
                ));
            }
            //     claim STATE mutex
            let mut state = self.state.lock().unwrap();
            state.insert(reader_pos, new_block);
            // Tell any waiting threads they should re-check the cache
            self.read_completed.notify_all();
            reader_pos += to_read as u64;
        }
        // Because the above condition is >=, and because we know the request was not
        // to read at the very end of the file, we know we now have some data in the
        // cache which can satisfy the request.
        //     claim STATE mutex
        let mut state = self.state.lock().unwrap();
        let bytes_read = state
            .read_from_cache(pos, buf)
            .expect("Cache still couldn't satisfy request event after reading beyond read pos");
        log::debug!("Cache success after read");
        if reader_created {
            state.stats.num_http_streams += 1;
        }
        //     Update reading_stuff with right information
        reading_stuff.reader = Some((reader, reader_pos));

[Abhinav-Upadhya-ni]

this also results in incomplete download.

maybe there could be a better way of doing things here.
Please let me know on this,

Thanks

[Abhinav-Upadhya-ni]

hello, any update on this?
thanks!

[pgrace-google]

Hi! thanks for raising this concern!
an indefinite while loop is certainly not what we want - is this something you've run into? or is this just a theoretical concern?
Either way, happy to take a PR that makes this more robust!

[Abhinav-Upadhya-ni]

Hey, yes, we were able to reproduce the issue. We mocked the read_exact method to return an error after it had been called more than 10(random number) times, and we observed that the download api which we wrote as a wrapper to use the ripunzip module, gets stuck.

[Abhinav-Upadhya-ni]

I couldn't raise a pr due to some permission issues, also a little new to rust, so I might not be able to fully utilize the crates very effectively.

but the solution would look something like this mentioned here in the 2nd comment.

[Abhinav-Upadhya-ni]

Hello!

is there any update on this?

Thanks!

[pgrace-google]

hi! again, i am happy to review a pr!

[Abhinav-Upadhya-ni]

Hello!

can you please review the pr?

Thanks!

[pgrace-google]

Hi! thank you for the PR - I have been out sick for the past week and I am just catching up on things. This is a best effort maintained repo, and I will get to it probably next week!

[Abhinav-Upadhya-ni]

thanks @pgrace-google !

[Abhinav-Upadhya-ni]

hello! any update on the pr?