indexedlog/log/
wait.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 *
4 * This source code is licensed under the MIT license found in the
5 * LICENSE file in the root directory of this source tree.
6 */
7
8use std::path::PathBuf;
9
10use super::meta::LogMetadata;
11use super::Log;
12use super::META_FILE;
13use crate::errors::IoResultExt;
14
15/// State to detect on-disk updates.
16pub struct Wait {
17    // The "meta" file to watch.
18    path: PathBuf,
19    // The "len" and "epoch" specified by the meta file.
20    state: (u64, u64),
21}
22
23impl Wait {
24    /// Construct `Wait` from a `Log`.
25    pub fn from_log(log: &Log) -> crate::Result<Self> {
26        let path = match log.dir.as_opt_path() {
27            None => {
28                return Err(crate::errors::Error::programming(
29                    "Wait does not support in-memory Log",
30                ));
31            }
32            Some(dir) => dir.join(META_FILE),
33        };
34        Ok(Self {
35            path,
36            state: state_from_meta(&log.meta),
37        })
38    }
39
40    /// Wait for on-disk changes that changes the backing `Log`.
41    pub fn wait_for_change(&mut self) -> crate::Result<()> {
42        // Initialize atomicfile Wait before read_meta() to avoid races.
43        let mut atomic_wait = atomicfile::Wait::from_path(&self.path)
44            .context(&self.path, "initialize file change detector")?;
45        tracing::debug!(" waiting meta change: {}", self.path.display());
46        let mut new_state;
47        loop {
48            let new_meta = LogMetadata::read_file(&self.path)?;
49            new_state = state_from_meta(&new_meta);
50            if new_state != self.state {
51                tracing::trace!(" state changed: {}", self.path.display());
52                break;
53            } else {
54                tracing::trace!(" state not changed: {}", self.path.display());
55                // Block.
56                atomic_wait
57                    .wait_for_change()
58                    .context(&self.path, "waiting file change")?;
59            }
60        }
61        self.state = new_state;
62        Ok(())
63    }
64}
65
66fn state_from_meta(meta: &LogMetadata) -> (u64, u64) {
67    (meta.primary_len, meta.epoch)
68}
indexedlog/log/wait.rs

indexedlog/log/
wait.rs
