mirror of
https://github.com/scottlamb/moonfire-nvr.git
synced 2024-12-24 06:05:55 -05:00
WIP: use jiff crate
TODO: * use a released Retina * look over `base/time.rs` changes
This commit is contained in:
Scott Lamb
parent
f9e3fb56b3
commit
bee5b387a5
@ -17,6 +17,7 @@ even on minor releases, e.g. `v0.7.5` -> `v0.7.6`.
|
||||
recording ended.
|
||||
* fix [#121](https://github.com/scottlamb/moonfire-nvr/issues/121):
|
||||
iPhone live view.
|
||||
* use `jiff` for time manipulations.
|
||||
|
||||
## v0.7.16 (2024-05-30)
|
||||
|
||||
|
||||
@ -105,8 +105,8 @@ services:
|
||||
# - seccomp:unconfined
|
||||
|
||||
environment:
|
||||
# Edit zone below to taste. The `:` is functional.
|
||||
TZ: ":America/Los_Angeles"
|
||||
# Edit zone below to taste.
|
||||
TZ: "America/Los_Angeles"
|
||||
RUST_BACKTRACE: 1
|
||||
|
||||
# docker's default log driver won't rotate logs properly, and will throw
|
||||
@ -323,7 +323,6 @@ After=network-online.target
|
||||
|
||||
[Service]
|
||||
ExecStart=/usr/local/bin/moonfire-nvr run
|
||||
Environment=TZ=:/etc/localtime
|
||||
Environment=MOONFIRE_FORMAT=systemd
|
||||
Environment=MOONFIRE_LOG=info
|
||||
Environment=RUST_BACKTRACE=1
|
||||
|
||||
@ -13,9 +13,10 @@ need more help.
|
||||
* [Docker setup](#docker-setup)
|
||||
* [`"/etc/moonfire-nvr.toml" is a directory`](#etcmoonfire-nvrtoml-is-a-directory)
|
||||
* [`Error response from daemon: unable to find user UID: no matching entries in passwd file`](#error-response-from-daemon-unable-to-find-user-uid-no-matching-entries-in-passwd-file)
|
||||
* [`clock_gettime failed: EPERM: Operation not permitted`](#clock_gettime-failed-eperm-operation-not-permitted)
|
||||
* [`clock_gettime(CLOCK_MONOTONIC) failed: EPERM: Operation not permitted`](#clock_gettimeclock_monotonic-failed-eperm-operation-not-permitted)
|
||||
* [`VFS is unable to determine a suitable directory for temporary files`](#vfs-is-unable-to-determine-a-suitable-directory-for-temporary-files)
|
||||
* [Server errors](#server-errors)
|
||||
* [`unable to get IANA time zone name; check your $TZ and /etc/localtime`](#unable-to-get-iana-time-zone-name-check-your-tz-and-etclocaltime)
|
||||
* [`Error: pts not monotonically increasing; got 26615520 then 26539470`](#error-pts-not-monotonically-increasing-got-26615520-then-26539470)
|
||||
* [Out of disk space](#out-of-disk-space)
|
||||
* [Database or filesystem corruption errors](#database-or-filesystem-corruption-errors)
|
||||
@ -217,7 +218,7 @@ If Docker produces this error, look at this section of the docker compose setup:
|
||||
user: UID:GID
|
||||
```
|
||||
|
||||
#### `clock_gettime failed: EPERM: Operation not permitted`
|
||||
#### `clock_gettime(CLOCK_MONOTONIC) failed: EPERM: Operation not permitted`
|
||||
|
||||
If commands fail with an error like the following, you're likely running
|
||||
Docker with an overly restrictive `seccomp` setup. [This stackoverflow
|
||||
@ -227,7 +228,7 @@ the `- seccomp: unconfined` line in your Docker compose file.
|
||||
|
||||
```console
|
||||
$ sudo docker compose run --rm moonfire-nvr --version
|
||||
clock_gettime failed: EPERM: Operation not permitted
|
||||
clock_gettime(CLOCK_MONOTONIC) failed: EPERM: Operation not permitted
|
||||
|
||||
This indicates a broken environment. See the troubleshooting guide.
|
||||
```
|
||||
@ -250,6 +251,12 @@ container in your Docker compose file.
|
||||
|
||||
### Server errors
|
||||
|
||||
#### `unable to get IANA time zone name; check your $TZ and /etc/localtime`
|
||||
|
||||
Moonfire NVR loads the system time zone via the logic described at
|
||||
[`jiff::tz::TimeZone::system`](https://docs.rs/jiff/0.1.8/jiff/tz/struct.TimeZone.html#method.system)
|
||||
and expects to be able to get the IANA zone name.
|
||||
|
||||
#### `Error: pts not monotonically increasing; got 26615520 then 26539470`
|
||||
|
||||
If your streams cut out and you see error messages like this one in Moonfire
|
||||
|
||||
121
server/Cargo.lock
generated
121
server/Cargo.lock
generated
@ -45,21 +45,6 @@ dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "android-tzdata"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0"
|
||||
|
||||
[[package]]
|
||||
name = "android_system_properties"
|
||||
version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anyhow"
|
||||
version = "1.0.86"
|
||||
@ -215,20 +200,6 @@ version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
|
||||
|
||||
[[package]]
|
||||
name = "chrono"
|
||||
version = "0.4.38"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401"
|
||||
dependencies = [
|
||||
"android-tzdata",
|
||||
"iana-time-zone",
|
||||
"js-sys",
|
||||
"num-traits",
|
||||
"wasm-bindgen",
|
||||
"windows-targets 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cipher"
|
||||
version = "0.4.4"
|
||||
@ -361,7 +332,7 @@ dependencies = [
|
||||
"log",
|
||||
"num",
|
||||
"owning_ref",
|
||||
"time 0.3.36",
|
||||
"time",
|
||||
"unicode-segmentation",
|
||||
"unicode-width",
|
||||
"xi-unicode",
|
||||
@ -662,7 +633,7 @@ dependencies = [
|
||||
"cfg-if",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"wasi 0.11.0+wasi-snapshot-preview1",
|
||||
"wasi",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
@ -874,29 +845,6 @@ dependencies = [
|
||||
"want",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "iana-time-zone"
|
||||
version = "0.1.60"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e7ffbb5a1b541ea2561f8c41c087286cc091e21e556a4f09a8f6cbf17b69b141"
|
||||
dependencies = [
|
||||
"android_system_properties",
|
||||
"core-foundation-sys",
|
||||
"iana-time-zone-haiku",
|
||||
"js-sys",
|
||||
"wasm-bindgen",
|
||||
"windows-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "iana-time-zone-haiku"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
|
||||
dependencies = [
|
||||
"cc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ident_case"
|
||||
version = "1.0.1"
|
||||
@ -959,6 +907,31 @@ version = "1.0.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b"
|
||||
|
||||
[[package]]
|
||||
name = "jiff"
|
||||
version = "0.1.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ef8bc400f8312944a9f879db116fed372c4f0859af672eba2a80f79c767dd19"
|
||||
dependencies = [
|
||||
"jiff-tzdb-platform",
|
||||
"windows-sys 0.59.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jiff-tzdb"
|
||||
version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "05fac328b3df1c0f18a3c2ab6cb7e06e4e549f366017d796e3e66b6d6889abe6"
|
||||
|
||||
[[package]]
|
||||
name = "jiff-tzdb-platform"
|
||||
version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8da387d5feaf355954c2c122c194d6df9c57d865125a67984bb453db5336940"
|
||||
dependencies = [
|
||||
"jiff-tzdb",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "js-sys"
|
||||
version = "0.3.70"
|
||||
@ -1093,7 +1066,7 @@ checksum = "80e04d1dcff3aae0704555fe5fee3bcfaf3d1fdf8a7e521d5b9d2b42acb52cec"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
"wasi 0.11.0+wasi-snapshot-preview1",
|
||||
"wasi",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
@ -1102,9 +1075,9 @@ name = "moonfire-base"
|
||||
version = "0.0.0"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"chrono",
|
||||
"coded",
|
||||
"futures",
|
||||
"jiff",
|
||||
"libc",
|
||||
"nix",
|
||||
"nom",
|
||||
@ -1112,7 +1085,6 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"slab",
|
||||
"time 0.1.45",
|
||||
"tracing",
|
||||
"tracing-core",
|
||||
"tracing-log",
|
||||
@ -1132,6 +1104,7 @@ dependencies = [
|
||||
"h264-reader",
|
||||
"hashlink",
|
||||
"itertools",
|
||||
"jiff",
|
||||
"libc",
|
||||
"moonfire-base",
|
||||
"nix",
|
||||
@ -1147,7 +1120,6 @@ dependencies = [
|
||||
"serde_json",
|
||||
"smallvec",
|
||||
"tempfile",
|
||||
"time 0.1.45",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"ulid",
|
||||
@ -1174,6 +1146,7 @@ dependencies = [
|
||||
"http-serve",
|
||||
"hyper",
|
||||
"itertools",
|
||||
"jiff",
|
||||
"libc",
|
||||
"libsystemd",
|
||||
"log",
|
||||
@ -1197,7 +1170,6 @@ dependencies = [
|
||||
"smallvec",
|
||||
"sync_wrapper",
|
||||
"tempfile",
|
||||
"time 0.1.45",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"toml",
|
||||
@ -1706,8 +1678,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "retina"
|
||||
version = "0.4.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ef9828fb04b8b2bd763887cf4be07aa85aecaa7fce3ee3c7f57bf61e804e9e5c"
|
||||
source = "git+https://github.com/scottlamb/retina.git?rev=6d5ccd156a503b7f83427e4871f1ada887e485f1#6d5ccd156a503b7f83427e4871f1ada887e485f1"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"bitstream-io",
|
||||
@ -1716,6 +1687,7 @@ dependencies = [
|
||||
"h264-reader",
|
||||
"hex",
|
||||
"http-auth",
|
||||
"jiff",
|
||||
"log",
|
||||
"once_cell",
|
||||
"pin-project",
|
||||
@ -1725,7 +1697,6 @@ dependencies = [
|
||||
"sdp-types",
|
||||
"smallvec",
|
||||
"thiserror",
|
||||
"time 0.1.45",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"url",
|
||||
@ -2102,17 +2073,6 @@ dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.1.45"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1b797afad3f312d1c66a56d11d0316f916356d11bd158fbc6ca6389ff6bf805a"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"wasi 0.10.0+wasi-snapshot-preview1",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.3.36"
|
||||
@ -2508,12 +2468,6 @@ dependencies = [
|
||||
"try-lock",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wasi"
|
||||
version = "0.10.0+wasi-snapshot-preview1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1a143597ca7c7793eff794def352d41792a93c481eb1042423ff7ff72ba2c31f"
|
||||
|
||||
[[package]]
|
||||
name = "wasi"
|
||||
version = "0.11.0+wasi-snapshot-preview1"
|
||||
@ -2650,15 +2604,6 @@ version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
|
||||
|
||||
[[package]]
|
||||
name = "windows-core"
|
||||
version = "0.52.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "33ab640c8d7e35bf8ba19b884ba838ceb4fba93a4e8c65a9059d08afcfc683d9"
|
||||
dependencies = [
|
||||
"windows-targets 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.48.0"
|
||||
|
||||
@ -26,6 +26,7 @@ members = ["base", "db"]
|
||||
base64 = "0.21.0"
|
||||
h264-reader = "0.7.0"
|
||||
itertools = "0.12.0"
|
||||
jiff = "0.1.8"
|
||||
nix = "0.27.0"
|
||||
pretty-hex = "0.4.0"
|
||||
ring = "0.17.0"
|
||||
@ -51,6 +52,7 @@ http = "0.2.3"
|
||||
http-serve = { version = "0.3.1", features = ["dir"] }
|
||||
hyper = { version = "0.14.2", features = ["http1", "server", "stream", "tcp"] }
|
||||
itertools = { workspace = true }
|
||||
jiff = { workspace = true, features = ["tz-system"] }
|
||||
libc = "0.2"
|
||||
log = { version = "0.4" }
|
||||
memchr = "2.0.2"
|
||||
@ -60,14 +62,14 @@ password-hash = "0.5.0"
|
||||
pretty-hex = { workspace = true }
|
||||
protobuf = "3.0"
|
||||
reffers = "0.7.0"
|
||||
retina = "0.4.9"
|
||||
#retina = "0.4.9"
|
||||
retina = { git = "https://github.com/scottlamb/retina.git", rev = "6d5ccd156a503b7f83427e4871f1ada887e485f1" }
|
||||
ring = { workspace = true }
|
||||
rusqlite = { workspace = true }
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
smallvec = { version = "1.7", features = ["union"] }
|
||||
sync_wrapper = "0.1.0"
|
||||
time = "0.1"
|
||||
tokio = { version = "1.24", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
|
||||
tokio-tungstenite = "0.20.0"
|
||||
toml = "0.8"
|
||||
@ -122,4 +124,4 @@ debug = 1
|
||||
tracing = { git = "https://github.com/scottlamb/tracing", rev = "861b443d7b2da400ca7b09111957f33c80135908" }
|
||||
tracing-core = { git = "https://github.com/scottlamb/tracing", rev = "861b443d7b2da400ca7b09111957f33c80135908" }
|
||||
tracing-log = { git = "https://github.com/scottlamb/tracing", rev = "861b443d7b2da400ca7b09111957f33c80135908" }
|
||||
tracing-subscriber = { git = "https://github.com/scottlamb/tracing", rev = "861b443d7b2da400ca7b09111957f33c80135908" }
|
||||
tracing-subscriber = { git = "https://github.com/scottlamb/tracing", rev = "861b443d7b2da400ca7b09111957f33c80135908" }
|
||||
|
||||
@ -15,17 +15,16 @@ path = "lib.rs"
|
||||
|
||||
[dependencies]
|
||||
ahash = "0.8"
|
||||
chrono = "0.4.23"
|
||||
coded = { git = "https://github.com/scottlamb/coded", rev = "2c97994974a73243d5dd12134831814f42cdb0e8"}
|
||||
futures = "0.3"
|
||||
jiff = { workspace = true }
|
||||
libc = "0.2"
|
||||
nix = { workspace = true }
|
||||
nix = { workspace = true, features = ["time"] }
|
||||
nom = "7.0.0"
|
||||
rusqlite = { workspace = true }
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
slab = "0.4"
|
||||
time = "0.1"
|
||||
tracing = { workspace = true }
|
||||
tracing-core = { workspace = true }
|
||||
tracing-log = { workspace = true }
|
||||
|
||||
@ -3,28 +3,91 @@
|
||||
// SPDX-License-Identifier: GPL-v3.0-or-later WITH GPL-3.0-linking-exception.
|
||||
|
||||
//! Clock interface and implementations for testability.
|
||||
//!
|
||||
//! Note these types are in a more standard nanosecond-based format, where
|
||||
//! [`crate::time`] uses Moonfire's 90 kHz time base.
|
||||
|
||||
use std::mem;
|
||||
use nix::sys::time::{TimeSpec, TimeValLike as _};
|
||||
use std::sync::Mutex;
|
||||
use std::sync::{mpsc, Arc};
|
||||
use std::thread;
|
||||
use std::time::Duration as StdDuration;
|
||||
use time::{Duration, Timespec};
|
||||
pub use std::time::Duration;
|
||||
use tracing::warn;
|
||||
|
||||
use crate::error::Error;
|
||||
use crate::shutdown::ShutdownError;
|
||||
|
||||
#[derive(Copy, Clone, PartialEq, Eq)]
|
||||
pub struct SystemTime(pub TimeSpec);
|
||||
|
||||
impl SystemTime {
|
||||
pub fn new(sec: nix::sys::time::time_t, nsec: i64) -> Self {
|
||||
SystemTime(TimeSpec::new(sec, nsec))
|
||||
}
|
||||
|
||||
pub fn as_secs(&self) -> i64 {
|
||||
self.0.num_seconds()
|
||||
}
|
||||
}
|
||||
|
||||
impl std::ops::Add<Duration> for SystemTime {
|
||||
type Output = SystemTime;
|
||||
|
||||
fn add(self, rhs: Duration) -> SystemTime {
|
||||
SystemTime(self.0 + TimeSpec::from(rhs))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct Instant(pub TimeSpec);
|
||||
|
||||
impl Instant {
|
||||
pub fn from_secs(secs: i64) -> Self {
|
||||
Instant(TimeSpec::seconds(secs))
|
||||
}
|
||||
|
||||
pub fn saturating_sub(&self, o: &Instant) -> Duration {
|
||||
if o > self {
|
||||
Duration::default()
|
||||
} else {
|
||||
Duration::from(self.0 - o.0)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for Instant {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
self.0.fmt(f)
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: should use saturating always?
|
||||
impl std::ops::Sub<Instant> for Instant {
|
||||
type Output = Duration;
|
||||
|
||||
fn sub(self, rhs: Instant) -> Duration {
|
||||
Duration::from(self.0 - rhs.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::ops::Add<Duration> for Instant {
|
||||
type Output = Instant;
|
||||
|
||||
fn add(self, rhs: Duration) -> Instant {
|
||||
Instant(self.0 + TimeSpec::from(rhs))
|
||||
}
|
||||
}
|
||||
|
||||
/// Abstract interface to the system clocks. This is for testability.
|
||||
pub trait Clocks: Send + Sync + 'static {
|
||||
/// Gets the current time from `CLOCK_REALTIME`.
|
||||
fn realtime(&self) -> Timespec;
|
||||
fn realtime(&self) -> SystemTime;
|
||||
|
||||
/// Gets the current time from a monotonic clock.
|
||||
///
|
||||
/// On Linux, this uses `CLOCK_BOOTTIME`, which includes suspended time.
|
||||
/// On other systems, it uses `CLOCK_MONOTONIC`.
|
||||
fn monotonic(&self) -> Timespec;
|
||||
fn monotonic(&self) -> Instant;
|
||||
|
||||
/// Causes the current thread to sleep for the specified time.
|
||||
fn sleep(&self, how_long: Duration);
|
||||
@ -33,7 +96,7 @@ pub trait Clocks: Send + Sync + 'static {
|
||||
fn recv_timeout<T>(
|
||||
&self,
|
||||
rcv: &mpsc::Receiver<T>,
|
||||
timeout: StdDuration,
|
||||
timeout: Duration,
|
||||
) -> Result<T, mpsc::RecvTimeoutError>;
|
||||
}
|
||||
|
||||
@ -52,7 +115,7 @@ where
|
||||
Err(e) => e.into(),
|
||||
};
|
||||
shutdown_rx.check()?;
|
||||
let sleep_time = Duration::seconds(1);
|
||||
let sleep_time = Duration::from_secs(1);
|
||||
warn!(
|
||||
exception = %e.chain(),
|
||||
"sleeping for 1 s after error"
|
||||
@ -64,49 +127,38 @@ where
|
||||
#[derive(Copy, Clone)]
|
||||
pub struct RealClocks {}
|
||||
|
||||
impl RealClocks {
|
||||
fn get(&self, clock: libc::clockid_t) -> Timespec {
|
||||
unsafe {
|
||||
let mut ts = mem::MaybeUninit::uninit();
|
||||
assert_eq!(0, libc::clock_gettime(clock, ts.as_mut_ptr()));
|
||||
let ts = ts.assume_init();
|
||||
Timespec::new(
|
||||
// On 32-bit arm builds, `tv_sec` is an `i32` and requires conversion.
|
||||
// On other platforms, the `.into()` is a no-op.
|
||||
#[allow(clippy::useless_conversion)]
|
||||
ts.tv_sec.into(),
|
||||
ts.tv_nsec as i32,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Clocks for RealClocks {
|
||||
fn realtime(&self) -> Timespec {
|
||||
self.get(libc::CLOCK_REALTIME)
|
||||
fn realtime(&self) -> SystemTime {
|
||||
SystemTime(
|
||||
nix::time::clock_gettime(nix::time::ClockId::CLOCK_REALTIME)
|
||||
.expect("clock_gettime(REALTIME) should succeed"),
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn monotonic(&self) -> Timespec {
|
||||
self.get(libc::CLOCK_BOOTTIME)
|
||||
fn monotonic(&self) -> Instant {
|
||||
Instant(
|
||||
nix::time::clock_gettime(nix::time::ClockId::CLOCK_BOOTTIME)
|
||||
.expect("clock_gettime(BOOTTIME) should succeed"),
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
fn monotonic(&self) -> Timespec {
|
||||
self.get(libc::CLOCK_MONOTONIC)
|
||||
fn monotonic(&self) -> Instant {
|
||||
Instant(
|
||||
nix::time::clock_gettime(nix::time::ClockId::CLOCK_MONOTONIC)
|
||||
.expect("clock_gettime(MONOTONIC) should succeed"),
|
||||
)
|
||||
}
|
||||
|
||||
fn sleep(&self, how_long: Duration) {
|
||||
match how_long.to_std() {
|
||||
Ok(d) => thread::sleep(d),
|
||||
Err(err) => warn!(%err, "invalid duration {:?}", how_long),
|
||||
};
|
||||
thread::sleep(how_long)
|
||||
}
|
||||
|
||||
fn recv_timeout<T>(
|
||||
&self,
|
||||
rcv: &mpsc::Receiver<T>,
|
||||
timeout: StdDuration,
|
||||
timeout: Duration,
|
||||
) -> Result<T, mpsc::RecvTimeoutError> {
|
||||
rcv.recv_timeout(timeout)
|
||||
}
|
||||
@ -117,7 +169,7 @@ impl Clocks for RealClocks {
|
||||
pub struct TimerGuard<'a, C: Clocks + ?Sized, S: AsRef<str>, F: FnOnce() -> S + 'a> {
|
||||
clocks: &'a C,
|
||||
label_f: Option<F>,
|
||||
start: Timespec,
|
||||
start: Instant,
|
||||
}
|
||||
|
||||
impl<'a, C: Clocks + ?Sized, S: AsRef<str>, F: FnOnce() -> S + 'a> TimerGuard<'a, C, S, F> {
|
||||
@ -138,9 +190,9 @@ where
|
||||
{
|
||||
fn drop(&mut self) {
|
||||
let elapsed = self.clocks.monotonic() - self.start;
|
||||
if elapsed.num_seconds() >= 1 {
|
||||
if elapsed.as_secs() >= 1 {
|
||||
let label_f = self.label_f.take().unwrap();
|
||||
warn!("{} took {}!", label_f().as_ref(), elapsed);
|
||||
warn!("{} took {:?}!", label_f().as_ref(), elapsed);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -150,42 +202,42 @@ where
|
||||
pub struct SimulatedClocks(Arc<SimulatedClocksInner>);
|
||||
|
||||
struct SimulatedClocksInner {
|
||||
boot: Timespec,
|
||||
boot: SystemTime,
|
||||
uptime: Mutex<Duration>,
|
||||
}
|
||||
|
||||
impl SimulatedClocks {
|
||||
pub fn new(boot: Timespec) -> Self {
|
||||
pub fn new(boot: SystemTime) -> Self {
|
||||
SimulatedClocks(Arc::new(SimulatedClocksInner {
|
||||
boot,
|
||||
uptime: Mutex::new(Duration::seconds(0)),
|
||||
uptime: Mutex::new(Duration::from_secs(0)),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
impl Clocks for SimulatedClocks {
|
||||
fn realtime(&self) -> Timespec {
|
||||
fn realtime(&self) -> SystemTime {
|
||||
self.0.boot + *self.0.uptime.lock().unwrap()
|
||||
}
|
||||
fn monotonic(&self) -> Timespec {
|
||||
Timespec::new(0, 0) + *self.0.uptime.lock().unwrap()
|
||||
fn monotonic(&self) -> Instant {
|
||||
Instant(TimeSpec::from(*self.0.uptime.lock().unwrap()))
|
||||
}
|
||||
|
||||
/// Advances the clock by the specified amount without actually sleeping.
|
||||
fn sleep(&self, how_long: Duration) {
|
||||
let mut l = self.0.uptime.lock().unwrap();
|
||||
*l = *l + how_long;
|
||||
*l += how_long;
|
||||
}
|
||||
|
||||
/// Advances the clock by the specified amount if data is not immediately available.
|
||||
fn recv_timeout<T>(
|
||||
&self,
|
||||
rcv: &mpsc::Receiver<T>,
|
||||
timeout: StdDuration,
|
||||
timeout: Duration,
|
||||
) -> Result<T, mpsc::RecvTimeoutError> {
|
||||
let r = rcv.recv_timeout(StdDuration::new(0, 0));
|
||||
let r = rcv.recv_timeout(Duration::new(0, 0));
|
||||
if r.is_err() {
|
||||
self.sleep(Duration::from_std(timeout).unwrap());
|
||||
self.sleep(timeout);
|
||||
}
|
||||
r
|
||||
}
|
||||
|
||||
@ -14,24 +14,48 @@ use std::fmt;
|
||||
use std::ops;
|
||||
use std::str::FromStr;
|
||||
|
||||
use super::clock::SystemTime;
|
||||
|
||||
type IResult<'a, I, O> = nom::IResult<I, O, nom::error::VerboseError<&'a str>>;
|
||||
|
||||
pub const TIME_UNITS_PER_SEC: i64 = 90_000;
|
||||
|
||||
/// The zone to use for all time handling.
|
||||
///
|
||||
/// In normal operation this is assigned from `jiff::tz::TimeZone::system()` at
|
||||
/// startup, but tests set it to a known political time zone instead.
|
||||
///
|
||||
/// Note that while fresh calls to `jiff::tz::TimeZone::system()` might return
|
||||
/// new values, this time zone is fixed for the entire run. This is important
|
||||
/// for `moonfire_db::days::Map`, where it's expected that adding values and
|
||||
/// then later subtracting them will cancel out.
|
||||
static GLOBAL_ZONE: std::sync::OnceLock<jiff::tz::TimeZone> = std::sync::OnceLock::new();
|
||||
|
||||
pub fn init_zone<F: FnOnce() -> jiff::tz::TimeZone>(f: F) {
|
||||
GLOBAL_ZONE.get_or_init(f);
|
||||
}
|
||||
|
||||
pub fn global_zone() -> jiff::tz::TimeZone {
|
||||
GLOBAL_ZONE
|
||||
.get()
|
||||
.expect("global zone should be initialized")
|
||||
.clone()
|
||||
}
|
||||
|
||||
/// A time specified as 90,000ths of a second since 1970-01-01 00:00:00 UTC.
|
||||
#[derive(Clone, Copy, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
pub struct Time(pub i64);
|
||||
|
||||
/// Returns a parser for a `len`-digit non-negative number which fits into an i32.
|
||||
fn fixed_len_num<'a>(len: usize) -> impl FnMut(&'a str) -> IResult<'a, &'a str, i32> {
|
||||
/// Returns a parser for a `len`-digit non-negative number which fits into `T`.
|
||||
fn fixed_len_num<'a, T: FromStr>(len: usize) -> impl FnMut(&'a str) -> IResult<'a, &'a str, T> {
|
||||
map_res(
|
||||
take_while_m_n(len, len, |c: char| c.is_ascii_digit()),
|
||||
|input: &str| input.parse::<i32>(),
|
||||
|input: &str| input.parse(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Parses `YYYY-mm-dd` into pieces.
|
||||
fn parse_datepart(input: &str) -> IResult<&str, (i32, i32, i32)> {
|
||||
fn parse_datepart(input: &str) -> IResult<&str, (i16, i8, i8)> {
|
||||
tuple((
|
||||
fixed_len_num(4),
|
||||
preceded(tag("-"), fixed_len_num(2)),
|
||||
@ -40,7 +64,7 @@ fn parse_datepart(input: &str) -> IResult<&str, (i32, i32, i32)> {
|
||||
}
|
||||
|
||||
/// Parses `HH:MM[:SS[:FFFFF]]` into pieces.
|
||||
fn parse_timepart(input: &str) -> IResult<&str, (i32, i32, i32, i32)> {
|
||||
fn parse_timepart(input: &str) -> IResult<&str, (i8, i8, i8, i32)> {
|
||||
let (input, (hr, _, min)) = tuple((fixed_len_num(2), tag(":"), fixed_len_num(2)))(input)?;
|
||||
let (input, stuff) = opt(tuple((
|
||||
preceded(tag(":"), fixed_len_num(2)),
|
||||
@ -57,16 +81,16 @@ fn parse_zone(input: &str) -> IResult<&str, i32> {
|
||||
map(
|
||||
tuple((
|
||||
opt(nom::character::complete::one_of(&b"+-"[..])),
|
||||
fixed_len_num(2),
|
||||
fixed_len_num::<i32>(2),
|
||||
tag(":"),
|
||||
fixed_len_num(2),
|
||||
fixed_len_num::<i32>(2),
|
||||
)),
|
||||
|(sign, hr, _, min)| {
|
||||
let off = hr * 3600 + min * 60;
|
||||
if sign == Some('-') {
|
||||
off
|
||||
} else {
|
||||
-off
|
||||
} else {
|
||||
off
|
||||
}
|
||||
},
|
||||
),
|
||||
@ -77,10 +101,6 @@ impl Time {
|
||||
pub const MIN: Self = Time(i64::MIN);
|
||||
pub const MAX: Self = Time(i64::MAX);
|
||||
|
||||
pub fn new(tm: time::Timespec) -> Self {
|
||||
Time(tm.sec * TIME_UNITS_PER_SEC + tm.nsec as i64 * TIME_UNITS_PER_SEC / 1_000_000_000)
|
||||
}
|
||||
|
||||
/// Parses a time as either 90,000ths of a second since epoch or a RFC 3339-like string.
|
||||
///
|
||||
/// The former is 90,000ths of a second since 1970-01-01T00:00:00 UTC, excluding leap seconds.
|
||||
@ -114,38 +134,27 @@ impl Time {
|
||||
);
|
||||
}
|
||||
let (tm_hour, tm_min, tm_sec, subsec) = opt_time.unwrap_or((0, 0, 0, 0));
|
||||
let mut tm = time::Tm {
|
||||
tm_sec,
|
||||
tm_min,
|
||||
tm_hour,
|
||||
tm_mday,
|
||||
tm_mon,
|
||||
tm_year,
|
||||
tm_wday: 0,
|
||||
tm_yday: 0,
|
||||
tm_isdst: -1,
|
||||
tm_utcoff: 0,
|
||||
tm_nsec: 0,
|
||||
};
|
||||
if tm.tm_mon == 0 {
|
||||
bail!(InvalidArgument, msg("time {input:?} has month 0"));
|
||||
}
|
||||
tm.tm_mon -= 1;
|
||||
if tm.tm_year < 1900 {
|
||||
bail!(InvalidArgument, msg("time {input:?} has year before 1900"));
|
||||
}
|
||||
tm.tm_year -= 1900;
|
||||
let dt = jiff::civil::DateTime::new(tm_year, tm_mon, tm_mday, tm_hour, tm_min, tm_sec, 0)
|
||||
.map_err(|e| err!(InvalidArgument, source(e)))?;
|
||||
|
||||
// The time crate doesn't use tm_utcoff properly; it just calls timegm() if tm_utcoff == 0,
|
||||
// mktime() otherwise. If a zone is specified, use the timegm path and a manual offset.
|
||||
// If no zone is specified, use the tm_utcoff path. This is pretty lame, but follow the
|
||||
// chrono crate's lead and just use 0 or 1 to choose between these functions.
|
||||
let sec = if let Some(off) = opt_zone {
|
||||
tm.to_timespec().sec + i64::from(off)
|
||||
} else {
|
||||
tm.tm_utcoff = 1;
|
||||
tm.to_timespec().sec
|
||||
};
|
||||
let tz =
|
||||
if let Some(off) = opt_zone {
|
||||
jiff::tz::TimeZone::fixed(jiff::tz::Offset::from_seconds(off).map_err(|e| {
|
||||
err!(InvalidArgument, msg("invalid time zone offset"), source(e))
|
||||
})?)
|
||||
} else {
|
||||
global_zone()
|
||||
};
|
||||
let sec = tz
|
||||
.into_ambiguous_zoned(dt)
|
||||
.compatible()
|
||||
.map_err(|e| err!(InvalidArgument, source(e)))?
|
||||
.timestamp()
|
||||
.as_second();
|
||||
Ok(Time(sec * TIME_UNITS_PER_SEC + i64::from(subsec)))
|
||||
}
|
||||
|
||||
@ -155,6 +164,18 @@ impl Time {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<SystemTime> for Time {
|
||||
fn from(tm: SystemTime) -> Self {
|
||||
Time(tm.0.tv_sec() * TIME_UNITS_PER_SEC + tm.0.tv_nsec() * 9 / 100_000)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<jiff::Timestamp> for Time {
|
||||
fn from(tm: jiff::Timestamp) -> Self {
|
||||
Time((tm.as_nanosecond() * 9 / 100_000) as i64)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::str::FromStr for Time {
|
||||
type Err = Error;
|
||||
|
||||
@ -199,32 +220,39 @@ impl fmt::Debug for Time {
|
||||
|
||||
impl fmt::Display for Time {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
let tm = time::at(time::Timespec {
|
||||
sec: self.0 / TIME_UNITS_PER_SEC,
|
||||
nsec: 0,
|
||||
});
|
||||
let zone_minutes = tm.tm_utcoff.abs() / 60;
|
||||
let tm = jiff::Zoned::new(
|
||||
jiff::Timestamp::from_second(self.0 / TIME_UNITS_PER_SEC).map_err(|_| fmt::Error)?,
|
||||
global_zone(),
|
||||
);
|
||||
write!(
|
||||
f,
|
||||
"{}:{:05}{}{:02}:{:02}",
|
||||
tm.strftime("%FT%T").map_err(|_| fmt::Error)?,
|
||||
"{}:{:05}{}",
|
||||
tm.strftime("%FT%T"),
|
||||
self.0 % TIME_UNITS_PER_SEC,
|
||||
if tm.tm_utcoff > 0 { '+' } else { '-' },
|
||||
zone_minutes / 60,
|
||||
zone_minutes % 60
|
||||
tm.strftime("%:z"),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// A duration specified in 1/90,000ths of a second.
|
||||
/// Durations are typically non-negative, but a `moonfire_db::db::CameraDayValue::duration` may be
|
||||
/// negative.
|
||||
/// Durations are typically non-negative, but a `moonfire_db::db::StreamDayValue::duration` may be
|
||||
/// negative when used as a `<StreamDayValue as Value>::Change`.
|
||||
#[derive(Clone, Copy, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
pub struct Duration(pub i64);
|
||||
|
||||
impl Duration {
|
||||
pub fn to_tm_duration(&self) -> time::Duration {
|
||||
time::Duration::nanoseconds(self.0 * 100000 / 9)
|
||||
impl From<Duration> for jiff::SignedDuration {
|
||||
fn from(d: Duration) -> Self {
|
||||
jiff::SignedDuration::from_nanos(d.0 * 100_000 / 9)
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<Duration> for std::time::Duration {
|
||||
type Error = std::num::TryFromIntError;
|
||||
|
||||
fn try_from(value: Duration) -> Result<Self, Self::Error> {
|
||||
Ok(std::time::Duration::from_nanos(
|
||||
u64::try_from(value.0)? * 100_000 / 9,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
@ -327,6 +355,15 @@ impl ops::SubAssign for Duration {
|
||||
}
|
||||
}
|
||||
|
||||
pub mod testutil {
|
||||
pub fn init_zone() {
|
||||
super::init_zone(|| {
|
||||
jiff::tz::TimeZone::get("America/Los_Angeles")
|
||||
.expect("America/Los_Angeles should exist")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{Duration, Time, TIME_UNITS_PER_SEC};
|
||||
@ -334,8 +371,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_parse_time() {
|
||||
std::env::set_var("TZ", "America/Los_Angeles");
|
||||
time::tzset();
|
||||
super::testutil::init_zone();
|
||||
#[rustfmt::skip]
|
||||
let tests = &[
|
||||
("2006-01-02T15:04:05-07:00", 102261550050000),
|
||||
@ -358,8 +394,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_format_time() {
|
||||
std::env::set_var("TZ", "America/Los_Angeles");
|
||||
time::tzset();
|
||||
super::testutil::init_zone();
|
||||
assert_eq!(
|
||||
"2006-01-02T15:04:05:00000-08:00",
|
||||
format!("{}", Time(102261874050000))
|
||||
|
||||
@ -17,12 +17,18 @@ use tracing_subscriber::{
|
||||
|
||||
struct FormatSystemd;
|
||||
|
||||
struct ChronoTimer;
|
||||
struct JiffTimer;
|
||||
|
||||
impl FormatTime for ChronoTimer {
|
||||
impl FormatTime for JiffTimer {
|
||||
fn format_time(&self, w: &mut Writer<'_>) -> std::fmt::Result {
|
||||
const TIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.6f";
|
||||
write!(w, "{}", chrono::Local::now().format(TIME_FORMAT))
|
||||
|
||||
// Always use the system time zone here, not `base::time::GLOBAL_ZONE`,
|
||||
// to resolve a chicken-and-egg problem. `jiff::tz::TimeZone::system()`
|
||||
// may log an error that is worth seeing. Therefore, we install the
|
||||
// tracing subscriber before initializing `GLOBAL_ZONE`. The latter
|
||||
// nly exists to override the zone for tests anyway.
|
||||
write!(w, "{}", jiff::Zoned::now().strftime(TIME_FORMAT))
|
||||
}
|
||||
}
|
||||
|
||||
@ -139,7 +145,7 @@ pub fn install() {
|
||||
let sub = tracing_subscriber::registry().with(
|
||||
tracing_subscriber::fmt::Layer::new()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_timer(ChronoTimer)
|
||||
.with_timer(JiffTimer)
|
||||
.with_thread_names(true)
|
||||
.with_filter(filter),
|
||||
);
|
||||
@ -164,7 +170,7 @@ pub fn install_for_tests() {
|
||||
let sub = tracing_subscriber::registry().with(
|
||||
tracing_subscriber::fmt::Layer::new()
|
||||
.with_test_writer()
|
||||
.with_timer(ChronoTimer)
|
||||
.with_timer(JiffTimer)
|
||||
.with_thread_names(true)
|
||||
.with_filter(filter),
|
||||
);
|
||||
|
||||
@ -25,6 +25,7 @@ futures = "0.3"
|
||||
h264-reader = { workspace = true }
|
||||
hashlink = "0.9.1"
|
||||
itertools = { workspace = true }
|
||||
jiff = { workspace = true }
|
||||
libc = "0.2"
|
||||
nix = { workspace = true, features = ["dir", "feature", "fs", "mman"] }
|
||||
num-rational = { version = "0.4.0", default-features = false, features = ["std"] }
|
||||
@ -38,7 +39,6 @@ serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
smallvec = "1.0"
|
||||
tempfile = "3.2.0"
|
||||
time = "0.1"
|
||||
tokio = { version = "1.24", features = ["macros", "rt-multi-thread", "sync"] }
|
||||
tracing = { workspace = true }
|
||||
ulid = "1.0.0"
|
||||
|
||||
@ -5,7 +5,7 @@
|
||||
//! In-memory indexes by calendar day.
|
||||
|
||||
use base::time::{Duration, Time, TIME_UNITS_PER_SEC};
|
||||
use base::{err, Error};
|
||||
use base::Error;
|
||||
use smallvec::SmallVec;
|
||||
use std::cmp;
|
||||
use std::collections::BTreeMap;
|
||||
@ -20,28 +20,22 @@ use tracing::{error, trace};
|
||||
pub struct Key(pub(crate) [u8; 10]);
|
||||
|
||||
impl Key {
|
||||
fn new(tm: time::Tm) -> Result<Self, Error> {
|
||||
fn new(tm: &jiff::Zoned) -> Result<Self, Error> {
|
||||
let mut s = Key([0u8; 10]);
|
||||
write!(
|
||||
&mut s.0[..],
|
||||
"{}",
|
||||
tm.strftime("%Y-%m-%d")
|
||||
.map_err(|e| err!(Internal, source(e)))?
|
||||
)?;
|
||||
write!(&mut s.0[..], "{}", tm.strftime("%Y-%m-%d"))?;
|
||||
Ok(s)
|
||||
}
|
||||
|
||||
pub fn bounds(&self) -> Range<Time> {
|
||||
let mut my_tm = time::strptime(self.as_ref(), "%Y-%m-%d").expect("days must be parseable");
|
||||
my_tm.tm_utcoff = 1; // to the time crate, values != 0 mean local time.
|
||||
my_tm.tm_isdst = -1;
|
||||
let start = Time(my_tm.to_timespec().sec * TIME_UNITS_PER_SEC);
|
||||
my_tm.tm_hour = 0;
|
||||
my_tm.tm_min = 0;
|
||||
my_tm.tm_sec = 0;
|
||||
my_tm.tm_mday += 1;
|
||||
let end = Time(my_tm.to_timespec().sec * TIME_UNITS_PER_SEC);
|
||||
start..end
|
||||
let date: jiff::civil::Date = self.as_ref().parse().expect("Key should be valid date");
|
||||
let start = date
|
||||
.to_zoned(base::time::global_zone())
|
||||
.expect("Key should be valid date");
|
||||
let end = start.tomorrow().expect("Key should have valid tomorrow");
|
||||
|
||||
// Note day boundaries are expected to always be whole numbers of seconds.
|
||||
Time(start.timestamp().as_second() * TIME_UNITS_PER_SEC)
|
||||
..Time(end.timestamp().as_second() * TIME_UNITS_PER_SEC)
|
||||
}
|
||||
}
|
||||
|
||||
@ -60,13 +54,14 @@ impl std::fmt::Debug for Key {
|
||||
pub trait Value: std::fmt::Debug + Default {
|
||||
type Change: std::fmt::Debug;
|
||||
|
||||
/// Applies the given change to this value.
|
||||
/// Applies the given change to this value; `c` may be positive or negative.
|
||||
fn apply(&mut self, c: &Self::Change);
|
||||
|
||||
fn is_empty(&self) -> bool;
|
||||
}
|
||||
|
||||
/// In-memory state about a particular stream on a particular day.
|
||||
/// In-memory state about a particular stream on a particular day, or a change
|
||||
/// to make via `<StreamValue as Value::apply>`.
|
||||
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub struct StreamValue {
|
||||
/// The number of recordings that overlap with this day.
|
||||
@ -81,6 +76,7 @@ pub struct StreamValue {
|
||||
impl Value for StreamValue {
|
||||
type Change = Self;
|
||||
|
||||
/// Applies the given change, which may have positive or negative recordings and duration.
|
||||
fn apply(&mut self, c: &StreamValue) {
|
||||
self.recordings += c.recordings;
|
||||
self.duration += c.duration;
|
||||
@ -198,42 +194,33 @@ impl<'a, V: Value> IntoIterator for &'a Map<V> {
|
||||
|
||||
impl Map<StreamValue> {
|
||||
/// Adjusts `self` to reflect the range of the given recording.
|
||||
///
|
||||
/// Note that the specified range may span two days. It will never span more because the maximum
|
||||
/// length of a recording entry is less than a day (even a 23-hour "spring forward" day).
|
||||
///
|
||||
/// This function swallows/logs date formatting errors because they shouldn't happen and there's
|
||||
/// not much that can be done about them. (The database operation has already gone through.)
|
||||
pub(crate) fn adjust(&mut self, r: Range<Time>, sign: i64) {
|
||||
// Find first day key.
|
||||
let sec = r.start.unix_seconds();
|
||||
let mut my_tm = time::at(time::Timespec { sec, nsec: 0 });
|
||||
let day = match Key::new(my_tm) {
|
||||
Ok(d) => d,
|
||||
Err(ref e) => {
|
||||
error!(
|
||||
"Unable to fill first day key from {:?}->{:?}: {}; will ignore.",
|
||||
r, my_tm, e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let start = jiff::Zoned::new(
|
||||
jiff::Timestamp::from_second(sec).expect("valid timestamp"),
|
||||
base::time::global_zone(),
|
||||
);
|
||||
let start_day = Key::new(&start).expect("valid key");
|
||||
|
||||
// Determine the start of the next day.
|
||||
// Use mytm to hold a non-normalized representation of the boundary.
|
||||
my_tm.tm_isdst = -1;
|
||||
my_tm.tm_hour = 0;
|
||||
my_tm.tm_min = 0;
|
||||
my_tm.tm_sec = 0;
|
||||
my_tm.tm_mday += 1;
|
||||
let boundary = my_tm.to_timespec();
|
||||
let boundary_90k = boundary.sec * TIME_UNITS_PER_SEC;
|
||||
let boundary = start
|
||||
.date()
|
||||
.tomorrow()
|
||||
.expect("valid tomorrow")
|
||||
.to_zoned(start.time_zone().clone())
|
||||
.expect("valid tomorrow");
|
||||
let boundary_90k = boundary.timestamp().as_second() * TIME_UNITS_PER_SEC;
|
||||
|
||||
// Adjust the first day.
|
||||
let first_day_delta = StreamValue {
|
||||
recordings: sign,
|
||||
duration: Duration(sign * (cmp::min(r.end.0, boundary_90k) - r.start.0)),
|
||||
};
|
||||
self.adjust_day(day, first_day_delta);
|
||||
self.adjust_day(start_day, first_day_delta);
|
||||
|
||||
if r.end.0 <= boundary_90k {
|
||||
return;
|
||||
@ -242,13 +229,12 @@ impl Map<StreamValue> {
|
||||
// Fill day with the second day. This requires a normalized representation so recalculate.
|
||||
// (The C mktime(3) already normalized for us once, but .to_timespec() discarded that
|
||||
// result.)
|
||||
let my_tm = time::at(boundary);
|
||||
let day = match Key::new(my_tm) {
|
||||
let day = match Key::new(&boundary) {
|
||||
Ok(d) => d,
|
||||
Err(ref e) => {
|
||||
error!(
|
||||
"Unable to fill second day key from {:?}: {}; will ignore.",
|
||||
my_tm, e
|
||||
boundary, e
|
||||
);
|
||||
return;
|
||||
}
|
||||
@ -263,35 +249,29 @@ impl Map<StreamValue> {
|
||||
|
||||
impl Map<SignalValue> {
|
||||
/// Adjusts `self` to reflect the range of the given recording.
|
||||
/// Note that the specified range may span several days (unlike StreamValue).
|
||||
///
|
||||
/// This function swallows/logs date formatting errors because they shouldn't happen and there's
|
||||
/// not much that can be done about them. (The database operation has already gone through.)
|
||||
/// Note that the specified range may span several days (unlike `StreamValue`).
|
||||
pub(crate) fn adjust(&mut self, mut r: Range<Time>, old_state: u16, new_state: u16) {
|
||||
// Find first day key.
|
||||
let sec = r.start.unix_seconds();
|
||||
let mut my_tm = time::at(time::Timespec { sec, nsec: 0 });
|
||||
let mut day = match Key::new(my_tm) {
|
||||
Ok(d) => d,
|
||||
Err(ref e) => {
|
||||
error!(
|
||||
"Unable to fill first day key from {:?}->{:?}: {}; will ignore.",
|
||||
r, my_tm, e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let mut tm = jiff::Zoned::new(
|
||||
jiff::Timestamp::from_second(sec).expect("valid timestamp"),
|
||||
base::time::global_zone(),
|
||||
);
|
||||
let mut day = Key::new(&tm).expect("valid date");
|
||||
|
||||
// Determine the start of the next day.
|
||||
// Use mytm to hold a non-normalized representation of the boundary.
|
||||
my_tm.tm_isdst = -1;
|
||||
my_tm.tm_hour = 0;
|
||||
my_tm.tm_min = 0;
|
||||
my_tm.tm_sec = 0;
|
||||
// Determine the starts of subsequent days.
|
||||
tm = tm
|
||||
.with()
|
||||
.hour(0)
|
||||
.minute(0)
|
||||
.second(0)
|
||||
.build()
|
||||
.expect("midnight is valid");
|
||||
|
||||
loop {
|
||||
my_tm.tm_mday += 1;
|
||||
let boundary_90k = my_tm.to_timespec().sec * TIME_UNITS_PER_SEC;
|
||||
tm = tm.tomorrow().expect("valid tomorrow");
|
||||
let boundary_90k = tm.timestamp().as_second() * TIME_UNITS_PER_SEC;
|
||||
|
||||
// Adjust this day.
|
||||
let duration = Duration(cmp::min(r.end.0, boundary_90k) - r.start.0);
|
||||
@ -308,23 +288,8 @@ impl Map<SignalValue> {
|
||||
return;
|
||||
}
|
||||
|
||||
// Fill day with the next day. This requires a normalized representation so
|
||||
// recalculate. (The C mktime(3) already normalized for us once, but .to_timespec()
|
||||
// discarded that result.)
|
||||
let my_tm = time::at(time::Timespec {
|
||||
sec: Time(boundary_90k).unix_seconds(),
|
||||
nsec: 0,
|
||||
});
|
||||
day = match Key::new(my_tm) {
|
||||
Ok(d) => d,
|
||||
Err(ref e) => {
|
||||
error!(
|
||||
"Unable to fill day key from {:?}: {}; will ignore.",
|
||||
my_tm, e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
// Fill day with the next day.
|
||||
day = Key::new(&tm).expect("valid date");
|
||||
r.start.0 = boundary_90k;
|
||||
}
|
||||
}
|
||||
|
||||
@ -621,7 +621,7 @@ pub struct LockedDatabase {
|
||||
|
||||
/// The monotonic time when the database was opened (whether in read-write mode or read-only
|
||||
/// mode).
|
||||
open_monotonic: recording::Time,
|
||||
open_monotonic: base::clock::Instant,
|
||||
|
||||
auth: auth::State,
|
||||
signal: signal::State,
|
||||
@ -1076,8 +1076,10 @@ impl LockedDatabase {
|
||||
r"update open set duration_90k = ?, end_time_90k = ? where id = ?",
|
||||
)?;
|
||||
let rows = stmt.execute(params![
|
||||
(recording::Time::new(clocks.monotonic()) - self.open_monotonic).0,
|
||||
recording::Time::new(clocks.realtime()).0,
|
||||
recording::Duration::try_from(clocks.monotonic() - self.open_monotonic)
|
||||
.expect("valid duration")
|
||||
.0,
|
||||
recording::Time::from(clocks.realtime()).0,
|
||||
o.id,
|
||||
])?;
|
||||
if rows != 1 {
|
||||
@ -2346,9 +2348,9 @@ impl<C: Clocks + Clone> Database<C> {
|
||||
// Note: the meta check comes after the version check to improve the error message when
|
||||
// trying to open a version 0 or version 1 database (which lacked the meta table).
|
||||
let (db_uuid, config) = raw::read_meta(&conn)?;
|
||||
let open_monotonic = recording::Time::new(clocks.monotonic());
|
||||
let open_monotonic = clocks.monotonic();
|
||||
let open = if read_write {
|
||||
let real = recording::Time::new(clocks.realtime());
|
||||
let real = recording::Time::from(clocks.realtime());
|
||||
let mut stmt = conn
|
||||
.prepare(" insert into open (uuid, start_time_90k, boot_uuid) values (?, ?, ?)")?;
|
||||
let open_uuid = SqlUuid(Uuid::new_v4());
|
||||
|
||||
@ -10,7 +10,6 @@ use crate::dir;
|
||||
use crate::writer;
|
||||
use base::clock::Clocks;
|
||||
use base::FastHashMap;
|
||||
use std::env;
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use tempfile::TempDir;
|
||||
@ -33,14 +32,13 @@ pub const TEST_VIDEO_SAMPLE_ENTRY_DATA: &[u8] =
|
||||
/// Performs global initialization for tests.
|
||||
/// * set up logging. (Note the output can be confusing unless `RUST_TEST_THREADS=1` is set in
|
||||
/// the program's environment prior to running.)
|
||||
/// * set `TZ=America/Los_Angeles` so that tests that care about calendar time get the expected
|
||||
/// results regardless of machine setup.)
|
||||
/// * set time zone `America/Los_Angeles` so that tests that care about
|
||||
/// calendar time get the expected results regardless of machine setup.)
|
||||
/// * use a fast but insecure password hashing format.
|
||||
pub fn init() {
|
||||
INIT.call_once(|| {
|
||||
base::tracing_setup::install_for_tests();
|
||||
env::set_var("TZ", "America/Los_Angeles");
|
||||
time::tzset();
|
||||
base::time::testutil::init_zone();
|
||||
crate::auth::set_test_config();
|
||||
});
|
||||
}
|
||||
|
||||
@ -19,8 +19,8 @@ use std::path::PathBuf;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::{mpsc, Arc};
|
||||
use std::thread;
|
||||
use std::time::Duration as StdDuration;
|
||||
use time::{Duration, Timespec};
|
||||
// use std::time::Duration as StdDuration;
|
||||
// use time::{Duration, Timespec};
|
||||
use tracing::{debug, trace, warn};
|
||||
|
||||
/// Trait to allow mocking out [crate::dir::SampleFileDir] in syncer tests.
|
||||
@ -103,7 +103,7 @@ struct Syncer<C: Clocks + Clone, D: DirWriter> {
|
||||
/// A plan to flush at a given instant due to a recently-saved recording's `flush_if_sec` parameter.
|
||||
struct PlannedFlush {
|
||||
/// Monotonic time at which this flush should happen.
|
||||
when: Timespec,
|
||||
when: base::clock::Instant,
|
||||
|
||||
/// Recording which prompts this flush. If this recording is already flushed at the planned
|
||||
/// time, it can be skipped.
|
||||
@ -440,9 +440,7 @@ impl<C: Clocks + Clone, D: DirWriter> Syncer<C, D> {
|
||||
let now = self.db.clocks().monotonic();
|
||||
|
||||
// Calculate the timeout to use, mapping negative durations to 0.
|
||||
let timeout = (t - now)
|
||||
.to_std()
|
||||
.unwrap_or_else(|_| StdDuration::new(0, 0));
|
||||
let timeout = t.saturating_sub(&now);
|
||||
match self.db.clocks().recv_timeout(cmds, timeout) {
|
||||
Err(mpsc::RecvTimeoutError::Disconnected) => return false, // cmd senders gone.
|
||||
Err(mpsc::RecvTimeoutError::Timeout) => {
|
||||
@ -534,8 +532,11 @@ impl<C: Clocks + Clone, D: DirWriter> Syncer<C, D> {
|
||||
let c = db.cameras_by_id().get(&s.camera_id).unwrap();
|
||||
|
||||
// Schedule a flush.
|
||||
let how_soon =
|
||||
Duration::seconds(i64::from(s.config.flush_if_sec)) - wall_duration.to_tm_duration();
|
||||
let how_soon = base::clock::Duration::from_secs(u64::from(s.config.flush_if_sec))
|
||||
.saturating_sub(
|
||||
base::clock::Duration::try_from(wall_duration)
|
||||
.expect("wall_duration is non-negative"),
|
||||
);
|
||||
let now = self.db.clocks().monotonic();
|
||||
let when = now + how_soon;
|
||||
let reason = format!(
|
||||
@ -546,7 +547,7 @@ impl<C: Clocks + Clone, D: DirWriter> Syncer<C, D> {
|
||||
s.type_.as_str(),
|
||||
id
|
||||
);
|
||||
trace!("scheduling flush in {} because {}", how_soon, &reason);
|
||||
trace!("scheduling flush in {:?} because {}", how_soon, &reason);
|
||||
self.planned_flushes.push(PlannedFlush {
|
||||
when,
|
||||
reason,
|
||||
@ -600,15 +601,15 @@ impl<C: Clocks + Clone, D: DirWriter> Syncer<C, D> {
|
||||
return;
|
||||
}
|
||||
if let Err(e) = l.flush(&f.reason) {
|
||||
let d = Duration::minutes(1);
|
||||
let d = base::clock::Duration::from_secs(60);
|
||||
warn!(
|
||||
"flush failure on save for reason {}; will retry after {}: {:?}",
|
||||
"flush failure on save for reason {}; will retry after {:?}: {:?}",
|
||||
f.reason, d, e
|
||||
);
|
||||
self.planned_flushes
|
||||
.peek_mut()
|
||||
.expect("planned_flushes is non-empty")
|
||||
.when = self.db.clocks().monotonic() + Duration::minutes(1);
|
||||
.when = self.db.clocks().monotonic() + base::clock::Duration::from_secs(60);
|
||||
return;
|
||||
}
|
||||
|
||||
@ -1162,7 +1163,7 @@ mod tests {
|
||||
}
|
||||
|
||||
fn new_harness(flush_if_sec: u32) -> Harness {
|
||||
let clocks = SimulatedClocks::new(::time::Timespec::new(0, 0));
|
||||
let clocks = SimulatedClocks::new(base::clock::SystemTime::new(0, 0));
|
||||
let tdb = testutil::TestDb::new_with_flush_if_sec(clocks, flush_if_sec);
|
||||
let dir_id = *tdb
|
||||
.db
|
||||
@ -1653,7 +1654,7 @@ mod tests {
|
||||
let mut h = new_harness(60); // flush_if_sec=60
|
||||
|
||||
// There's a database constraint forbidding a recording starting at t=0, so advance.
|
||||
h.db.clocks().sleep(time::Duration::seconds(1));
|
||||
h.db.clocks().sleep(base::clock::Duration::from_secs(1));
|
||||
|
||||
// Setup: add a 3-byte recording.
|
||||
let video_sample_entry_id =
|
||||
@ -1700,7 +1701,7 @@ mod tests {
|
||||
h.db.lock().flush("forced").unwrap();
|
||||
assert!(h.syncer.iter(&h.syncer_rx)); // DatabaseFlushed
|
||||
assert_eq!(h.syncer.planned_flushes.len(), 1);
|
||||
h.db.clocks().sleep(time::Duration::seconds(30));
|
||||
h.db.clocks().sleep(base::clock::Duration::from_secs(30));
|
||||
|
||||
// Then, a 1-byte recording.
|
||||
let mut w = Writer::new(&h.dir, &h.db, &h.channel, testutil::TEST_STREAM_ID);
|
||||
@ -1735,13 +1736,22 @@ mod tests {
|
||||
|
||||
assert_eq!(h.syncer.planned_flushes.len(), 2);
|
||||
let db_flush_count_before = h.db.lock().flushes();
|
||||
assert_eq!(h.db.clocks().monotonic(), time::Timespec::new(31, 0));
|
||||
assert_eq!(
|
||||
h.db.clocks().monotonic(),
|
||||
base::clock::Instant::from_secs(31)
|
||||
);
|
||||
assert!(h.syncer.iter(&h.syncer_rx)); // planned flush (no-op)
|
||||
assert_eq!(h.db.clocks().monotonic(), time::Timespec::new(61, 0));
|
||||
assert_eq!(
|
||||
h.db.clocks().monotonic(),
|
||||
base::clock::Instant::from_secs(61)
|
||||
);
|
||||
assert_eq!(h.db.lock().flushes(), db_flush_count_before);
|
||||
assert_eq!(h.syncer.planned_flushes.len(), 1);
|
||||
assert!(h.syncer.iter(&h.syncer_rx)); // planned flush
|
||||
assert_eq!(h.db.clocks().monotonic(), time::Timespec::new(91, 0));
|
||||
assert_eq!(
|
||||
h.db.clocks().monotonic(),
|
||||
base::clock::Instant::from_secs(91)
|
||||
);
|
||||
assert_eq!(h.db.lock().flushes(), db_flush_count_before + 1);
|
||||
assert_eq!(h.syncer.planned_flushes.len(), 0);
|
||||
assert!(h.syncer.iter(&h.syncer_rx)); // DatabaseFlushed
|
||||
|
||||
@ -78,7 +78,7 @@ pub fn run(args: Args) -> Result<i32, Error> {
|
||||
.map(db::Permissions::from)
|
||||
.unwrap_or_else(|| u.permissions.clone());
|
||||
let creation = db::auth::Request {
|
||||
when_sec: Some(db.clocks().realtime().sec),
|
||||
when_sec: Some(db.clocks().realtime().as_secs()),
|
||||
user_agent: None,
|
||||
addr: None,
|
||||
};
|
||||
|
||||
@ -44,89 +44,6 @@ pub struct Args {
|
||||
read_only: bool,
|
||||
}
|
||||
|
||||
// These are used in a hack to get the name of the current time zone (e.g. America/Los_Angeles).
|
||||
// They seem to be correct for Linux and macOS at least.
|
||||
const LOCALTIME_PATH: &str = "/etc/localtime";
|
||||
const TIMEZONE_PATH: &str = "/etc/timezone";
|
||||
|
||||
// Some well-known zone paths looks like the following:
|
||||
// /usr/share/zoneinfo/* for Linux and macOS < High Sierra
|
||||
// /var/db/timezone/zoneinfo/* for macOS High Sierra
|
||||
// /etc/zoneinfo/* for NixOS
|
||||
fn zoneinfo_name(path: &str) -> Option<&str> {
|
||||
path.rsplit_once("/zoneinfo/").map(|(_, name)| name)
|
||||
}
|
||||
|
||||
/// Attempt to resolve the timezone of the server.
|
||||
/// The Javascript running in the browser needs this to match the server's timezone calculations.
|
||||
fn resolve_zone() -> Result<String, Error> {
|
||||
// If the environmental variable `TZ` exists, is valid UTF-8, and doesn't just reference
|
||||
// `/etc/localtime/`, use that.
|
||||
if let Ok(tz) = ::std::env::var("TZ") {
|
||||
let mut p: &str = &tz;
|
||||
|
||||
// Strip of an initial `:` if present. Having `TZ` set in this way is a trick to avoid
|
||||
// repeated `tzset` calls:
|
||||
// https://blog.packagecloud.io/eng/2017/02/21/set-environment-variable-save-thousands-of-system-calls/
|
||||
if p.starts_with(':') {
|
||||
p = &p[1..];
|
||||
}
|
||||
|
||||
if let Some(p) = zoneinfo_name(p) {
|
||||
return Ok(p.to_owned());
|
||||
}
|
||||
|
||||
if !p.starts_with('/') {
|
||||
return Ok(p.to_owned());
|
||||
}
|
||||
|
||||
if p != LOCALTIME_PATH {
|
||||
bail!(
|
||||
FailedPrecondition,
|
||||
msg("unable to resolve env TZ={tz} to a timezone")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// If `LOCALTIME_PATH` is a symlink, use that. On some systems, it's instead a copy of the
|
||||
// desired timezone, which unfortunately doesn't contain its own name.
|
||||
match ::std::fs::read_link(LOCALTIME_PATH) {
|
||||
Ok(localtime_dest) => {
|
||||
let localtime_dest = match localtime_dest.to_str() {
|
||||
Some(d) => d,
|
||||
None => bail!(
|
||||
FailedPrecondition,
|
||||
msg("{LOCALTIME_PATH} symlink destination is invalid UTF-8")
|
||||
),
|
||||
};
|
||||
if let Some(p) = zoneinfo_name(localtime_dest) {
|
||||
return Ok(p.to_owned());
|
||||
}
|
||||
bail!(
|
||||
FailedPrecondition,
|
||||
msg("unable to resolve {LOCALTIME_PATH} symlink destination {localtime_dest} to a timezone"),
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
use ::std::io::ErrorKind;
|
||||
if e.kind() != ErrorKind::NotFound && e.kind() != ErrorKind::InvalidInput {
|
||||
bail!(e, msg("unable to read {LOCALTIME_PATH} symlink"));
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// If `TIMEZONE_PATH` is a file, use its contents as the zone name, trimming whitespace.
|
||||
match ::std::fs::read_to_string(TIMEZONE_PATH) {
|
||||
Ok(z) => Ok(z.trim().to_owned()),
|
||||
Err(e) => {
|
||||
bail!(
|
||||
e,
|
||||
msg("unable to resolve timezone from TZ env, {LOCALTIME_PATH}, or {TIMEZONE_PATH}"),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Syncer {
|
||||
dir: Arc<dir::SampleFileDir>,
|
||||
channel: writer::SyncerChannel<::std::fs::File>,
|
||||
@ -337,7 +254,13 @@ async fn inner(
|
||||
}
|
||||
info!("Directories are opened.");
|
||||
|
||||
let time_zone_name = resolve_zone()?;
|
||||
let zone = base::time::global_zone();
|
||||
let Some(time_zone_name) = zone.iana_name() else {
|
||||
bail!(
|
||||
Unknown,
|
||||
msg("unable to get IANA time zone name; check your $TZ and /etc/localtime")
|
||||
);
|
||||
};
|
||||
info!("Resolved timezone: {}", &time_zone_name);
|
||||
|
||||
// Start a streamer for each stream.
|
||||
@ -460,7 +383,7 @@ async fn inner(
|
||||
.clone()
|
||||
.map(db::Permissions::from),
|
||||
trust_forward_hdrs: b.trust_forward_headers,
|
||||
time_zone_name: time_zone_name.clone(),
|
||||
time_zone_name: time_zone_name.to_owned(),
|
||||
privileged_unix_uid: b.own_uid_is_privileged.then_some(own_euid),
|
||||
})?);
|
||||
let make_svc = make_service_fn(move |conn: &crate::web::accept::Conn| {
|
||||
|
||||
@ -70,13 +70,13 @@ fn main() {
|
||||
// anything (with timestamps...) so we can print a helpful error.
|
||||
if let Err(e) = nix::time::clock_gettime(nix::time::ClockId::CLOCK_MONOTONIC) {
|
||||
eprintln!(
|
||||
"clock_gettime failed: {e}\n\n\
|
||||
"clock_gettime(CLOCK_MONOTONIC) failed: {e}\n\n\
|
||||
This indicates a broken environment. See the troubleshooting guide."
|
||||
);
|
||||
std::process::exit(1);
|
||||
}
|
||||
|
||||
base::tracing_setup::install();
|
||||
base::time::init_zone(jiff::tz::TimeZone::system);
|
||||
|
||||
// Get the program name from the OS (e.g. if invoked as `target/debug/nvr`: `nvr`),
|
||||
// falling back to the crate name if conversion to a path/UTF-8 string fails.
|
||||
|
||||
@ -82,7 +82,7 @@ use tracing::{debug, error, trace, warn};
|
||||
/// This value should be incremented any time a change is made to this file that causes different
|
||||
/// bytes or headers to be output for a particular set of `FileBuilder` options. Incrementing this
|
||||
/// value will cause the etag to change as well.
|
||||
const FORMAT_VERSION: [u8; 1] = [0x09];
|
||||
const FORMAT_VERSION: [u8; 1] = [0x0a];
|
||||
|
||||
/// An `ftyp` (ISO/IEC 14496-12 section 4.3 `FileType`) box.
|
||||
const NORMAL_FTYP_BOX: &[u8] = &[
|
||||
@ -1028,7 +1028,7 @@ impl FileBuilder {
|
||||
let start_sec = wall.start.unix_seconds();
|
||||
let end_sec =
|
||||
(wall.end + recording::Duration(TIME_UNITS_PER_SEC - 1)).unix_seconds();
|
||||
s.num_subtitle_samples = (end_sec - start_sec) as u16;
|
||||
s.num_subtitle_samples = (end_sec - start_sec + 1) as u16;
|
||||
self.num_subtitle_samples += s.num_subtitle_samples as u32;
|
||||
}
|
||||
|
||||
@ -1844,19 +1844,21 @@ impl FileInner {
|
||||
.unix_seconds();
|
||||
let len = usize::try_from(len).unwrap();
|
||||
let mut v = Vec::with_capacity(len);
|
||||
// TODO(slamb): is this right?!? might have an off-by-one here.
|
||||
for ts in start_sec..end_sec {
|
||||
let mut tm = jiff::Zoned::new(
|
||||
jiff::Timestamp::from_second(start_sec).expect("timestamp is valid"),
|
||||
base::time::global_zone(),
|
||||
);
|
||||
let mut cur_sec = start_sec;
|
||||
loop {
|
||||
v.write_u16::<BigEndian>(SUBTITLE_LENGTH as u16)
|
||||
.expect("Vec write shouldn't fail");
|
||||
let tm = time::at(time::Timespec { sec: ts, nsec: 0 });
|
||||
use std::io::Write;
|
||||
write!(
|
||||
v,
|
||||
"{}",
|
||||
tm.strftime(SUBTITLE_TEMPLATE)
|
||||
.err_kind(ErrorKind::Internal)?
|
||||
)
|
||||
.expect("Vec write shouldn't fail");
|
||||
use std::io::Write as _;
|
||||
write!(v, "{}", tm.strftime(SUBTITLE_TEMPLATE)).expect("Vec write shouldn't fail");
|
||||
if cur_sec == end_sec {
|
||||
break;
|
||||
}
|
||||
tm += std::time::Duration::from_secs(1);
|
||||
cur_sec += 1;
|
||||
}
|
||||
assert_eq!(len, v.len());
|
||||
Ok(ARefss::new(v)
|
||||
@ -2848,7 +2850,7 @@ mod tests {
|
||||
hash.to_hex().as_str()
|
||||
);
|
||||
const EXPECTED_ETAG: &str =
|
||||
"\"791114c469130970608dd999b0ecf5861d077ec33fad2f0b040996e4aae4e30f\"";
|
||||
"\"37c89bda9f0513acdc2ab95f48f03b3f797dfa3fb30bbefa6549fdc7296afed2\"";
|
||||
assert_eq!(
|
||||
Some(HeaderValue::from_str(EXPECTED_ETAG).unwrap()),
|
||||
mp4.etag()
|
||||
@ -2873,11 +2875,11 @@ mod tests {
|
||||
// combine ranges from the new format with ranges from the old format.
|
||||
let hash = digest(&mp4).await;
|
||||
assert_eq!(
|
||||
"1f85ec7ea7f061b7d8f696c337a3258abc2bf830e81ac23c1342131669d7bb14",
|
||||
"8f17df9b43dc55654a1e4e00126e7477f43234693d4f1fae72185798a09479d7",
|
||||
hash.to_hex().as_str()
|
||||
);
|
||||
const EXPECTED_ETAG: &str =
|
||||
"\"85703b9abadd4292e119f2f7b0d6a16e99acf8b3ba98fcb6498e60ac5cb0b0b7\"";
|
||||
"\"d4af0554a50f6dfff2f7f95a14e16f720bd5fe36a9570cd4fd32f6664f1487c4\"";
|
||||
assert_eq!(
|
||||
Some(HeaderValue::from_str(EXPECTED_ETAG).unwrap()),
|
||||
mp4.etag()
|
||||
@ -2906,7 +2908,7 @@ mod tests {
|
||||
hash.to_hex().as_str()
|
||||
);
|
||||
const EXPECTED_ETAG: &str =
|
||||
"\"3d2031124fb995bf2fc4930e7affdcd51add396e062cfab97e1001224c5ee42c\"";
|
||||
"\"7165c1a866451b7e714a8ad47f4a0022a3749212e945321b35b2f8aaee8aea5c\"";
|
||||
assert_eq!(
|
||||
Some(HeaderValue::from_str(EXPECTED_ETAG).unwrap()),
|
||||
mp4.etag()
|
||||
@ -2932,11 +2934,11 @@ mod tests {
|
||||
// combine ranges from the new format with ranges from the old format.
|
||||
let hash = digest(&mp4).await;
|
||||
assert_eq!(
|
||||
"9c0302294f8f34d14fc8069fea1a65c1593a4c01134c07ab994b7398004f2b63",
|
||||
"caf8b23f3b6ee959981687ff0bcbf8d6b01db9daef35695b2600ffb9f8b54fe1",
|
||||
hash.to_hex().as_str()
|
||||
);
|
||||
const EXPECTED_ETAG: &str =
|
||||
"\"aa9bb2f63787a7d21227981135326c948db3e0b3dae5d0d39c77df69d0baf504\"";
|
||||
"\"167ad6b44502cb09eb15d08fdd2c360e4e54e521251eceeebddf74c4041b0b38\"";
|
||||
assert_eq!(
|
||||
Some(HeaderValue::from_str(EXPECTED_ETAG).unwrap()),
|
||||
mp4.etag()
|
||||
@ -2965,7 +2967,7 @@ mod tests {
|
||||
hash.to_hex().as_str()
|
||||
);
|
||||
const EXPECTED_ETAG: &str =
|
||||
"\"0a6accaa7b583c94209eba58b00b39a804a5c4a8c99043e58e72fed7acd8dfc6\"";
|
||||
"\"2c591788cf06f09b55450cd98cb07c670d580413359260f2d18b9595bd0b430d\"";
|
||||
assert_eq!(
|
||||
Some(HeaderValue::from_str(EXPECTED_ETAG).unwrap()),
|
||||
mp4.etag()
|
||||
|
||||
@ -120,7 +120,7 @@ where
|
||||
pub fn run(&mut self) {
|
||||
while self.shutdown_rx.check().is_ok() {
|
||||
if let Err(err) = self.run_once() {
|
||||
let sleep_time = time::Duration::seconds(1);
|
||||
let sleep_time = base::clock::Duration::from_secs(1);
|
||||
warn!(
|
||||
err = %err.chain(),
|
||||
"sleeping for 1 s after error"
|
||||
@ -181,7 +181,7 @@ where
|
||||
self.opener
|
||||
.open(self.short_name.clone(), self.url.clone(), options)?
|
||||
};
|
||||
let realtime_offset = self.db.clocks().realtime() - clocks.monotonic();
|
||||
let realtime_offset = self.db.clocks().realtime().0 - clocks.monotonic().0;
|
||||
let mut video_sample_entry_id = {
|
||||
let _t = TimerGuard::new(&clocks, || "inserting video sample entry");
|
||||
self.db
|
||||
@ -214,10 +214,10 @@ where
|
||||
debug!("have first key frame");
|
||||
seen_key_frame = true;
|
||||
}
|
||||
let frame_realtime = clocks.monotonic() + realtime_offset;
|
||||
let local_time = recording::Time::new(frame_realtime);
|
||||
let frame_realtime = base::clock::SystemTime(realtime_offset + clocks.monotonic().0);
|
||||
let local_time = recording::Time::from(frame_realtime);
|
||||
rotate = if let Some(r) = rotate {
|
||||
if frame_realtime.sec > r && frame.is_key {
|
||||
if frame_realtime.as_secs() > r && frame.is_key {
|
||||
trace!("close on normal rotation");
|
||||
let _t = TimerGuard::new(&clocks, || "closing writer");
|
||||
w.close(Some(frame.pts), None)?;
|
||||
@ -245,7 +245,7 @@ where
|
||||
let r = match rotate {
|
||||
Some(r) => r,
|
||||
None => {
|
||||
let sec = frame_realtime.sec;
|
||||
let sec = frame_realtime.as_secs();
|
||||
let r = sec - (sec % self.rotate_interval_sec) + self.rotate_offset_sec;
|
||||
let r = r + if r <= sec {
|
||||
self.rotate_interval_sec
|
||||
@ -300,8 +300,8 @@ mod tests {
|
||||
struct ProxyingStream {
|
||||
clocks: clock::SimulatedClocks,
|
||||
inner: Box<dyn stream::Stream>,
|
||||
buffered: time::Duration,
|
||||
slept: time::Duration,
|
||||
buffered: base::clock::Duration,
|
||||
slept: base::clock::Duration,
|
||||
ts_offset: i64,
|
||||
ts_offset_pkts_left: u32,
|
||||
pkts_left: u32,
|
||||
@ -310,7 +310,7 @@ mod tests {
|
||||
impl ProxyingStream {
|
||||
fn new(
|
||||
clocks: clock::SimulatedClocks,
|
||||
buffered: time::Duration,
|
||||
buffered: base::clock::Duration,
|
||||
inner: Box<dyn stream::Stream>,
|
||||
) -> ProxyingStream {
|
||||
clocks.sleep(buffered);
|
||||
@ -318,7 +318,7 @@ mod tests {
|
||||
clocks,
|
||||
inner,
|
||||
buffered,
|
||||
slept: time::Duration::seconds(0),
|
||||
slept: base::clock::Duration::default(),
|
||||
ts_offset: 0,
|
||||
ts_offset_pkts_left: 0,
|
||||
pkts_left: 0,
|
||||
@ -349,9 +349,10 @@ mod tests {
|
||||
// Avoid accumulating conversion error by tracking the total amount to sleep and how
|
||||
// much we've already slept, rather than considering each frame in isolation.
|
||||
{
|
||||
let goal = frame.pts + i64::from(frame.duration);
|
||||
let goal = time::Duration::nanoseconds(
|
||||
goal * 1_000_000_000 / recording::TIME_UNITS_PER_SEC,
|
||||
let goal =
|
||||
u64::try_from(frame.pts).unwrap() + u64::try_from(frame.duration).unwrap();
|
||||
let goal = base::clock::Duration::from_nanos(
|
||||
goal * 1_000_000_000 / u64::try_from(recording::TIME_UNITS_PER_SEC).unwrap(),
|
||||
);
|
||||
let duration = goal - self.slept;
|
||||
let buf_part = cmp::min(self.buffered, duration);
|
||||
@ -430,12 +431,15 @@ mod tests {
|
||||
async fn basic() {
|
||||
testutil::init();
|
||||
// 2015-04-25 00:00:00 UTC
|
||||
let clocks = clock::SimulatedClocks::new(time::Timespec::new(1429920000, 0));
|
||||
clocks.sleep(time::Duration::seconds(86400)); // to 2015-04-26 00:00:00 UTC
|
||||
let clocks = clock::SimulatedClocks::new(clock::SystemTime::new(1429920000, 0));
|
||||
clocks.sleep(clock::Duration::from_secs(86400)); // to 2015-04-26 00:00:00 UTC
|
||||
|
||||
let stream = stream::testutil::Mp4Stream::open("src/testdata/clip.mp4").unwrap();
|
||||
let mut stream =
|
||||
ProxyingStream::new(clocks.clone(), time::Duration::seconds(2), Box::new(stream));
|
||||
let mut stream = ProxyingStream::new(
|
||||
clocks.clone(),
|
||||
clock::Duration::from_secs(2),
|
||||
Box::new(stream),
|
||||
);
|
||||
stream.ts_offset = 123456; // starting pts of the input should be irrelevant
|
||||
stream.ts_offset_pkts_left = u32::max_value();
|
||||
stream.pkts_left = u32::max_value();
|
||||
|
||||
@ -315,7 +315,7 @@ impl Service {
|
||||
) -> Result<Response<Body>, std::convert::Infallible> {
|
||||
let request_id = ulid::Ulid::new();
|
||||
let authreq = auth::Request {
|
||||
when_sec: Some(self.db.clocks().realtime().sec),
|
||||
when_sec: Some(self.db.clocks().realtime().as_secs()),
|
||||
addr: if self.trust_forward_hdrs {
|
||||
req.headers()
|
||||
.get("X-Real-IP")
|
||||
@ -532,22 +532,22 @@ impl Service {
|
||||
.user_agent
|
||||
.as_ref()
|
||||
.map(|u| String::from_utf8_lossy(&u[..]));
|
||||
let when = authreq.when_sec.map(|sec| {
|
||||
jiff::Timestamp::from_second(sec)
|
||||
.expect("valid time")
|
||||
.to_zoned(base::time::global_zone())
|
||||
.strftime("%FT%T%:z")
|
||||
});
|
||||
Ok(plain_response(
|
||||
StatusCode::OK,
|
||||
format!(
|
||||
"when: {}\n\
|
||||
"when: {:?}\n\
|
||||
host: {:?}\n\
|
||||
addr: {:?}\n\
|
||||
user_agent: {:?}\n\
|
||||
secure: {:?}\n\
|
||||
caller: {:?}\n",
|
||||
time::at(time::Timespec {
|
||||
sec: authreq.when_sec.unwrap(),
|
||||
nsec: 0
|
||||
})
|
||||
.strftime("%FT%T")
|
||||
.map(|f| f.to_string())
|
||||
.unwrap_or_else(|e| e.to_string()),
|
||||
when,
|
||||
host.as_deref(),
|
||||
&authreq.addr,
|
||||
agent.as_deref(),
|
||||
|
||||
@ -41,7 +41,7 @@ impl Service {
|
||||
let r = extract_json_body(&mut req).await?;
|
||||
let r: json::PostSignalsRequest = parse_json_body(&r)?;
|
||||
require_csrf_if_session(&caller, r.csrf)?;
|
||||
let now = recording::Time::new(self.db.clocks().realtime());
|
||||
let now = recording::Time::from(self.db.clocks().realtime());
|
||||
let mut l = self.db.lock();
|
||||
let start = match r.start {
|
||||
json::PostSignalsTimeBase::Epoch(t) => t,
|
||||
|
||||
@ -180,10 +180,10 @@ impl Service {
|
||||
}
|
||||
}
|
||||
if let Some(start) = start_time_for_filename {
|
||||
let tm = time::at(time::Timespec {
|
||||
sec: start.unix_seconds(),
|
||||
nsec: 0,
|
||||
});
|
||||
let zone = base::time::global_zone();
|
||||
let tm = jiff::Timestamp::from_second(start.unix_seconds())
|
||||
.expect("valid start")
|
||||
.to_zoned(zone);
|
||||
let stream_abbrev = if stream_type == db::StreamType::Main {
|
||||
"main"
|
||||
} else {
|
||||
@ -196,7 +196,7 @@ impl Service {
|
||||
};
|
||||
builder.set_filename(&format!(
|
||||
"{}-{}-{}.{}",
|
||||
tm.strftime("%Y%m%d%H%M%S").unwrap(),
|
||||
tm.strftime("%Y%m%d%H%M%S"),
|
||||
camera_name,
|
||||
stream_abbrev,
|
||||
suffix
|
||||
|
||||
Loading…
Reference in New Issue
Block a user