MyFavMirrors/moonfire-nvr
1
0
Fork 0
mirror of https://github.com/scottlamb/moonfire-nvr.git synced 2025-02-03 09:55:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

upgrade to async hyper

Browse Source 
serve_generated_bytes is >3X faster. One caveat is that the reactor thread may
stall when reading from the memory-mapped slice. Moonfire NVR is basically a
single-user program, so that may not be so bad, but we'll see.
This commit is contained in:
Scott Lamb 2017-03-02 19:29:28 -08:00
parent 618709734a
commit 1cf27c189f
11 changed files with 1067 additions and 774 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

752
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

20
Cargo.toml
View File

@ -4,17 +4,18 @@ version = "0.1.0"
authors = ["Scott Lamb <slamb@slamb.org>"]
[features]
nightly = []
# The nightly feature is used within moonfire-nvr itself to gate the
# benchmarks. Also pass it along to crates that can benefit from it.
nightly = ["parking_lot/nightly"]
[dependencies]
byteorder = "1.0"
chan = "0.1"
chan-signal = "0.2"
docopt = "0.7"
futures = "0.1"
fnv = "1.0"
http-entity = { git = "https://github.com/scottlamb/http-entity" }
hyper = { git = "https://github.com/scottlamb/hyper", branch = "moonfire-on-0.10.x" }
lazycell = "0.5"
http-entity = { git = "https://github.com/scottlamb/http-entity", branch = "hyper-0.11.x" }
hyper = { git = "https://github.com/scottlamb/hyper", branch = "moonfire-on-0.11.x" }
lazy_static = "0.2"
libc = "0.2"
log = { version = "0.3", features = ["release_max_level_info"] }
@ -22,6 +23,8 @@ lru-cache = "0.1"
memmap = "0.5"
mime = "0.2"
openssl = "0.9"
parking_lot = { version = "0.3.8", features = [] }
reffers = { git = "https://github.com/diwic/reffers-rs" }
regex = "0.2"
rusqlite = "0.9"
rustc-serialize = "0.3"
@ -34,10 +37,13 @@ slog-stdlog = "1.1"
slog-term = "1.3"
smallvec = "0.3"
time = "0.1"
tokio-core = "0.1"
tokio-signal = "0.1"
url = "1.4"
uuid = { version = "0.4", features = ["serde", "v4"] }
[dev-dependencies]
reqwest = "0.3"
tempdir = "0.3"
[dependencies.cursive]
@ -65,4 +71,4 @@ lto = true
debug = true
[replace]
"hyper:0.10.4" = { git = "https://github.com/scottlamb/hyper", branch = "moonfire-on-0.10.x" }
"https://github.com/hyperium/hyper#hyper:0.11.0-a.0" = { git = "https://github.com/scottlamb/hyper", branch = "moonfire-on-0.11.x" }

44
src/cmds/run.rs
View File

@ -28,17 +28,18 @@
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use chan_signal;
use clock;
use db;
use dir;
use error::Error;
use hyper::server::Server;
use futures::{BoxFuture, Future, Stream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use stream;
use streamer;
use tokio_core::reactor;
use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
use web;
const USAGE: &'static str = r#"
@ -65,13 +66,18 @@ struct Args {
flag_read_only: bool,
}
fn setup_shutdown_future(h: &reactor::Handle) -> BoxFuture<(), ()> {
let int = Signal::new(SIGINT, h).flatten_stream().into_future();
let term = Signal::new(SIGTERM, h).flatten_stream().into_future();
int.select(term)
.map(|_| ())
.map_err(|_| ())
.boxed()
}
pub fn run() -> Result<(), Error> {
let args: Args = super::parse_args(USAGE)?;
// Watch for termination signals.
// This must be started before any threads are spawned (such as the async logger thread) so
// that signals will be blocked in all threads.
let signal = chan_signal::notify(&[chan_signal::Signal::INT, chan_signal::Signal::TERM]);
super::install_logger(true);
let (_db_dir, conn) = super::open_conn(
&args.flag_db_dir,
@ -81,7 +87,7 @@ pub fn run() -> Result<(), Error> {
info!("Database is loaded.");
// Start a streamer for each camera.
let shutdown = Arc::new(AtomicBool::new(false));
let shutdown_streamers = Arc::new(AtomicBool::new(false));
let mut streamers = Vec::new();
let syncer = if !args.flag_read_only {
let (syncer_channel, syncer_join) = dir::start_syncer(dir.clone()).unwrap();
@ -92,7 +98,7 @@ pub fn run() -> Result<(), Error> {
dir: &dir,
clocks: &clock::REAL,
opener: &*stream::FFMPEG,
shutdown: &shutdown,
shutdown: &shutdown_streamers,
};
for (i, (id, camera)) in l.cameras_by_id().iter().enumerate() {
let rotate_offset_sec = streamer::ROTATE_INTERVAL_SEC * i as i64 / cameras as i64;
@ -108,24 +114,28 @@ pub fn run() -> Result<(), Error> {
} else { None };
// Start the web interface.
let server = Server::http(args.flag_http_addr.as_str()).unwrap();
let h = web::Handler::new(db.clone(), dir.clone());
let _guard = server.handle(h);
info!("Ready to serve HTTP requests");
let addr = args.flag_http_addr.parse().unwrap();
let server = ::hyper::server::Http::new()
.bind(&addr, move || Ok(web::Service::new(db.clone(), dir.clone())))
.unwrap();
// Wait for a signal and shut down.
chan_select! {
signal.recv() -> signal => info!("Received signal {:?}; shutting down streamers.", signal),
}
shutdown.store(true, Ordering::SeqCst);
let shutdown = setup_shutdown_future(&server.handle());
info!("Ready to serve HTTP requests");
server.run_until(shutdown).unwrap();
info!("Shutting down streamers.");
shutdown_streamers.store(true, Ordering::SeqCst);
for streamer in streamers.drain(..) {
streamer.join().unwrap();
}
if let Some((syncer_channel, syncer_join)) = syncer {
info!("Shutting down syncer.");
drop(syncer_channel);
syncer_join.join().unwrap();
}
info!("Exiting.");
::std::process::exit(0);
}

9
src/db.rs
View File

@ -55,6 +55,7 @@ use error::{Error, ResultExt};
use fnv;
use lru_cache::LruCache;
use openssl::hash;
use parking_lot::{Mutex,MutexGuard};
use recording::{self, TIME_UNITS_PER_SEC};
use rusqlite;
use std::collections::BTreeMap;
@ -64,7 +65,7 @@ use std::io::Write;
use std::ops::Range;
use std::str;
use std::string::String;
use std::sync::{Arc,Mutex,MutexGuard};
use std::sync::Arc;
use std::vec::Vec;
use time;
use uuid::Uuid;
@ -1347,7 +1348,7 @@ impl Database {
},
}));
{
let mut l = &mut *db.0.lock().unwrap();
let mut l = &mut *db.lock();
l.init_video_sample_entries().annotate_err("init_video_sample_entries")?;
l.init_cameras().annotate_err("init_cameras")?;
for (&camera_id, ref mut camera) in &mut l.state.cameras_by_id {
@ -1361,13 +1362,13 @@ impl Database {
/// Locks the database; the returned reference is the only way to perform (read or write)
/// operations.
pub fn lock(&self) -> MutexGuard<LockedDatabase> { self.0.lock().unwrap() }
pub fn lock(&self) -> MutexGuard<LockedDatabase> { self.0.lock() }
/// For testing. Closes the database and return the connection. This allows verification that
/// a newly opened database is in an acceptable state.
#[cfg(test)]
fn close(self) -> rusqlite::Connection {
self.0.into_inner().unwrap().conn
self.0.into_inner().conn
}
}

2
src/error.rs
View File

@ -85,7 +85,7 @@ impl error::Error for Error {
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> result::Result<(), fmt::Error> {
write!(f, "Error: {}", self.description)
write!(f, "Error: {}\ncause: {:?}", self.description, self.cause)
}
}

9
src/main.rs
View File

@ -32,23 +32,23 @@
extern crate byteorder;
extern crate core;
#[macro_use] extern crate chan;
extern crate chan_signal;
extern crate docopt;
#[macro_use] extern crate ffmpeg;
extern crate ffmpeg_sys;
extern crate futures;
extern crate fnv;
extern crate http_entity;
extern crate hyper;
#[macro_use] extern crate lazy_static;
extern crate lazycell;
extern crate libc;
#[macro_use] extern crate log;
extern crate lru_cache;
extern crate reffers;
extern crate rusqlite;
extern crate memmap;
#[macro_use] extern crate mime;
extern crate openssl;
extern crate parking_lot;
extern crate regex;
extern crate rustc_serialize;
extern crate serde;
@ -60,6 +60,8 @@ extern crate slog_stdlog;
extern crate slog_term;
extern crate smallvec;
extern crate time;
extern crate tokio_core;
extern crate tokio_signal;
extern crate url;
extern crate uuid;
@ -71,7 +73,6 @@ mod dir;
mod error;
mod h264;
mod json;
mod mmapfile;
mod mp4;
mod recording;
mod slices;

71
src/mmapfile.rs
View File

@ -1,71 +0,0 @@
// This file is part of Moonfire NVR, a security camera digital video recorder.
// Copyright (C) 2016 Scott Lamb <slamb@slamb.org>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// In addition, as a special exception, the copyright holders give
// permission to link the code of portions of this program with the
// OpenSSL library under certain conditions as described in each
// individual source file, and distribute linked combinations including
// the two.
//
// You must obey the GNU General Public License in all respects for all
// of the code used other than OpenSSL. If you modify file(s) with this
// exception, you may extend this exception to your version of the
// file(s), but you are not obligated to do so. If you do not wish to do
// so, delete this exception statement from your version. If you delete
// this exception statement from all source files in the program, then
// also delete it here.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Memory-mapped file serving.
extern crate memmap;
use error::Result;
use std::fs::File;
use std::io;
use std::ops::Range;
/// Memory-mapped file slice.
/// This struct is meant to be used in constructing an implementation of the `http_entity::Entity`
/// or `pieces::ContextWriter` traits. The file in question should be immutable, as files shrinking
/// during `mmap` will cause the process to fail with `SIGBUS`. Moonfire NVR sample files satisfy
/// this requirement:
///
/// * They should only be modified by Moonfire NVR itself. Installation instructions encourage
/// creating a dedicated user/group for Moonfire NVR and ensuring only this group has
/// permissions to Moonfire NVR's directories.
/// * Moonfire NVR never modifies sample files after inserting their matching recording entries
/// into the database. They are kept as-is until they are deleted.
pub struct MmapFileSlice {
f: File,
range: Range<u64>,
}
impl MmapFileSlice {
pub fn new(f: File, range: Range<u64>) -> MmapFileSlice {
MmapFileSlice{f: f, range: range}
}
pub fn write_to(&self, range: Range<u64>, out: &mut io::Write) -> Result<()> {
// TODO: overflow check (in case u64 is larger than usize).
let r = self.range.start + range.start .. self.range.start + range.end;
assert!(r.end <= self.range.end,
"requested={:?} within={:?}", range, self.range);
let mmap = memmap::Mmap::open_with_offset(
&self.f, memmap::Protection::Read, r.start as usize, (r.end - r.start) as usize)?;
unsafe { out.write_all(mmap.as_slice())?; }
Ok(())
}
}

327
src/mp4.rs
View File

@ -83,14 +83,18 @@ use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
use db;
use dir;
use error::Error;
use futures::{Future, Stream};
use futures::stream;
use http_entity;
use hyper::header;
use mmapfile;
use memmap;
use openssl::hash;
use parking_lot::{Once, ONCE_INIT};
use recording::{self, TIME_UNITS_PER_SEC};
use slices::{self, Slices};
use reffers::ARefs;
use slices::{self, Body, Chunk, Slices};
use smallvec::SmallVec;
use lazycell::LazyCell;
use std::cell::UnsafeCell;
use std::cmp;
use std::io;
use std::ops::Range;
@ -100,7 +104,7 @@ use strutil;
use time::Timespec;
/// This value should be incremented any time a change is made to this file that causes different
/// bytes to be output for a particular set of `Builder` options. Incrementing this value will
/// bytes to be output for a particular set of `Mp4Builder` options. Incrementing this value will
/// cause the etag to change as well.
const FORMAT_VERSION: [u8; 1] = [0x03];
@ -329,42 +333,46 @@ struct Segment {
/// 1. stts: `slice[.. stsz_start]`
/// 2. stsz: `slice[stsz_start .. stss_start]`
/// 3. stss: `slice[stss_start ..]`
/// Access only through `write_index`.
index: LazyCell<Result<Box<[u8]>, ()>>,
/// Access only through `get_index`.
index: UnsafeCell<Result<Box<[u8]>, ()>>,
/// The 1-indexed frame number in the `File` of the first frame in this segment.
first_frame_num: u32,
num_subtitle_samples: u32,
num_subtitle_samples: u16,
index_once: Once,
}
unsafe impl Sync for Segment {}
impl Segment {
fn new(db: &db::LockedDatabase, row: &db::ListRecordingsRow, rel_range_90k: Range<i32>,
first_frame_num: u32) -> Result<Self, Error> {
Ok(Segment{
s: recording::Segment::new(db, row, rel_range_90k)?,
index: LazyCell::new(),
index: UnsafeCell::new(Err(())),
index_once: ONCE_INIT,
first_frame_num: first_frame_num,
num_subtitle_samples: 0,
})
}
fn write_index<F>(&self, r: Range<u64>, db: &db::Database, out: &mut io::Write, f: F)
-> Result<(), Error>
fn get_index<'a, F>(&'a self, db: &db::Database, f: F) -> Result<&'a [u8], Error>
where F: FnOnce(&[u8], SegmentLengths) -> &[u8] {
let index = self.index.borrow_with(|| {
db.lock()
self.index_once.call_once(|| {
let index = unsafe { &mut *self.index.get() };
*index = db.lock()
.with_recording_playback(self.s.camera_id, self.s.recording_id,
|playback| self.build_index(playback))
.map_err(|e| { error!("Unable to build index for segment: {:?}", e); })
.map_err(|e| { error!("Unable to build index for segment: {:?}", e); });
});
let index = match *index {
Ok(ref b) => &b[..],
let index: &'a _ = unsafe { &*self.index.get() };
match *index {
Ok(ref b) => return Ok(f(&b[..], self.lens())),
Err(()) => {
return Err(Error::new("Unable to build index; see previous error.".to_owned()))
},
};
out.write_all(&f(&index, self.lens())[r.start as usize .. r.end as usize])?;
Ok(())
}
}
fn lens(&self) -> SegmentLengths {
@ -466,8 +474,8 @@ enum SliceType {
VideoSampleEntry = 2, // param is index into m.video_sample_entries
Stts = 3, // param is index into m.segments
Stsz = 4, // param is index into m.segments
Co64 = 5, // param is unused
Stss = 6, // param is index into m.segments
Stss = 5, // param is index into m.segments
Co64 = 6, // param is unused
VideoSampleData = 7, // param is index into m.segments
SubtitleSampleData = 8, // param is index into m.segments
@ -482,49 +490,69 @@ impl Slice {
Ok(Slice(end | ((t as u64) << 40) | ((p as u64) << 44)))
}
}
impl Slice {
fn t(&self) -> SliceType {
// This value is guaranteed to be a valid SliceType because it was copied from a SliceType
// in Slice::new.
unsafe { ::std::mem::transmute(((self.0 >> 40) & 0xF) as u8) }
}
fn p(&self) -> usize { (self.0 >> 44) as usize }
fn wrap_index<F>(&self, mp4: &File, r: Range<u64>, f: &F) -> Result<Chunk, Error>
where F: Fn(&[u8], SegmentLengths) -> &[u8] {
let mp4 = ARefs::new(mp4.0.clone());
let r = r.start as usize .. r.end as usize;
let p = self.p();
mp4.try_map(|mp4| Ok(&mp4.segments[p].get_index(&mp4.db, f)?[r]))
}
}
impl slices::Slice for Slice {
type Ctx = File;
type Chunk = slices::Chunk;
fn end(&self) -> u64 { return self.0 & 0xFF_FF_FF_FF_FF }
fn write_to(&self, f: &File, r: Range<u64>, l: u64, out: &mut io::Write)
-> Result<(), Error> {
let t = self.t();
fn get_range(&self, f: &File, range: Range<u64>, len: u64) -> Body {
trace!("getting mp4 slice {:?}'s range {:?} / {}", self, range, len);
let p = self.p();
trace!("write {:?}, range {:?} out of len {}", self, r, l);
match t {
let res = match self.t() {
SliceType::Static => {
let s = STATIC_BYTESTRINGS[p];
let part = &s[r.start as usize .. r.end as usize];
out.write_all(part)?;
Ok(())
let part = &s[range.start as usize .. range.end as usize];
Ok(part.into())
},
SliceType::Buf => {
out.write_all(&f.buf[p+r.start as usize .. p+r.end as usize])?;
Ok(())
let r = ARefs::new(f.0.clone());
Ok(r.map(|f| &f.buf[p+range.start as usize .. p+range.end as usize]))
},
SliceType::VideoSampleEntry => {
out.write_all(&f.video_sample_entries[p].data[r.start as usize .. r.end as usize])?;
Ok(())
let r = ARefs::new(f.0.clone());
Ok(r.map(|f| &f.video_sample_entries[p]
.data[range.start as usize .. range.end as usize]))
},
SliceType::Stts => f.segments[p].write_index(r, &f.db, out, Segment::stts),
SliceType::Stsz => f.segments[p].write_index(r, &f.db, out, Segment::stsz),
SliceType::Stss => f.segments[p].write_index(r, &f.db, out, Segment::stss),
SliceType::Co64 => f.write_co64(r, l, out),
SliceType::VideoSampleData => f.write_video_sample_data(p, r, out),
SliceType::SubtitleSampleData => f.write_subtitle_sample_data(p, r, l, out),
SliceType::Stts => self.wrap_index(f, range.clone(), &Segment::stts),
SliceType::Stsz => self.wrap_index(f, range.clone(), &Segment::stsz),
SliceType::Stss => self.wrap_index(f, range.clone(), &Segment::stss),
SliceType::Co64 => f.0.get_co64(range.clone(), len),
SliceType::VideoSampleData => f.0.get_video_sample_data(p, range.clone()),
SliceType::SubtitleSampleData => f.0.get_subtitle_sample_data(p, range.clone(), len),
};
stream::once(res
.map_err(|e| {
error!("Error producing {:?}: {:?}", self, e);
::hyper::Error::Incomplete
})
.and_then(move |c| {
if c.len() != (range.end - range.start) as usize {
error!("Error producing {:?}: range {:?} produced incorrect len {}.",
self, range, c.len());
return Err(::hyper::Error::Incomplete);
}
Ok(c)
})).boxed()
}
fn get_slices(ctx: &File) -> &Slices<Self> { &ctx.0.slices }
}
impl ::std::fmt::Debug for Slice {
@ -626,8 +654,8 @@ impl FileBuilder {
let end_sec = (s.s.start +
recording::Duration(d.end as i64 + TIME_UNITS_PER_SEC - 1))
.unix_seconds();
s.num_subtitle_samples = (end_sec - start_sec) as u32;
self.num_subtitle_samples += s.num_subtitle_samples;
s.num_subtitle_samples = (end_sec - start_sec) as u16;
self.num_subtitle_samples += s.num_subtitle_samples as u32;
}
// Update the etag to reflect this segment.
@ -692,7 +720,7 @@ impl FileBuilder {
debug!("Estimated {} buf bytes; actually were {}", EST_BUF_LEN, self.body.buf.len());
}
debug!("slices: {:?}", self.body.slices);
Ok(File{
Ok(File(Arc::new(FileInner{
db: db,
dir: dir,
segments: self.segments,
@ -702,7 +730,7 @@ impl FileBuilder {
initial_sample_byte_pos: initial_sample_byte_pos,
last_modified: header::HttpDate(time::at(Timespec::new(max_end.unix_seconds(), 0))),
etag: header::EntityTag::strong(strutil::hex(&etag.finish()?)),
})
})))
}
/// Appends a `MovieBox` (ISO/IEC 14496-12 section 8.2.1).
@ -1015,7 +1043,7 @@ impl FileBuilder {
write_length!(self, {
self.body.buf.extend_from_slice(
b"stsc\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x01");
self.body.append_u32(self.num_subtitle_samples);
self.body.append_u32(self.num_subtitle_samples as u32);
self.body.append_u32(1);
})
}
@ -1042,7 +1070,7 @@ impl FileBuilder {
write_length!(self, {
self.body.buf.extend_from_slice(b"stsz\x00\x00\x00\x00");
self.body.append_u32((mem::size_of::<u16>() + SUBTITLE_LENGTH) as u32);
self.body.append_u32(self.num_subtitle_samples);
self.body.append_u32(self.num_subtitle_samples as u32);
})
}
@ -1123,7 +1151,7 @@ impl BodyState {
}
}
pub struct File {
struct FileInner {
db: Arc<db::Database>,
dir: Arc<dir::SampleFileDir>,
segments: Vec<Segment>,
@ -1135,60 +1163,67 @@ pub struct File {
etag: header::EntityTag,
}
impl File {
fn write_co64(&self, r: Range<u64>, l: u64, out: &mut io::Write) -> Result<(), Error> {
slices::clip_to_range(r, l, out, |w| {
impl FileInner {
fn get_co64(&self, r: Range<u64>, l: u64) -> Result<Chunk, Error> {
let mut v = Vec::with_capacity(l as usize);
let mut pos = self.initial_sample_byte_pos;
for s in &self.segments {
w.write_u64::<BigEndian>(pos)?;
v.write_u64::<BigEndian>(pos)?;
let r = s.s.sample_file_range();
pos += r.end - r.start;
}
Ok(())
})
Ok(ARefs::new(v).map(|v| &v[r.start as usize .. r.end as usize]))
}
fn write_video_sample_data(&self, i: usize, r: Range<u64>, out: &mut io::Write)
-> Result<(), Error> {
/// Gets a `Chunk` of video sample data from disk.
/// This works by `mmap()`ing in the data. There are a couple caveats:
///
/// * The thread which reads the resulting slice is likely to experience major page faults.
/// Eventually this will likely be rewritten to `mmap()` the memory in another thread, and
/// `mlock()` and send chunks of it to be read and `munlock()`ed to avoid this problem.
///
/// * If the backing file is truncated, the program will crash with `SIGBUS`. This shouldn't
/// happen because nothing should be touching Moonfire NVR's files but itself.
fn get_video_sample_data(&self, i: usize, r: Range<u64>) -> Result<Chunk, Error> {
let s = &self.segments[i];
let uuid = {
self.db.lock().with_recording_playback(s.s.camera_id, s.s.recording_id,
|p| Ok(p.sample_file_uuid))?
};
let f = self.dir.open_sample_file(uuid)?;
mmapfile::MmapFileSlice::new(f, s.s.sample_file_range()).write_to(r, out)
let mmap = Box::new(memmap::Mmap::open_with_offset(
&f, memmap::Protection::Read, r.start as usize, (r.end - r.start) as usize)?);
Ok(ARefs::new(mmap).map(|m| unsafe { m.as_slice() }))
}
fn write_subtitle_sample_data(&self, i: usize, r: Range<u64>, l: u64, out: &mut io::Write)
-> Result<(), Error> {
fn get_subtitle_sample_data(&self, i: usize, r: Range<u64>, l: u64) -> Result<Chunk, Error> {
let s = &self.segments[i];
let d = &s.s.desired_range_90k;
let start_sec = (s.s.start + recording::Duration(d.start as i64)).unix_seconds();
let end_sec = (s.s.start + recording::Duration(d.end as i64 + TIME_UNITS_PER_SEC - 1))
.unix_seconds();
slices::clip_to_range(r, l, out, |w| {
let mut v = Vec::with_capacity(l as usize);
for ts in start_sec .. end_sec {
w.write_u16::<BigEndian>(SUBTITLE_LENGTH as u16)?;
v.write_u16::<BigEndian>(SUBTITLE_LENGTH as u16)?;
let tm = time::at(time::Timespec{sec: ts, nsec: 0});
use std::io::Write;
write!(w, "{}", tm.strftime(SUBTITLE_TEMPLATE)?)?;
write!(v, "{}", tm.strftime(SUBTITLE_TEMPLATE)?)?;
}
Ok(())
})?;
Ok(())
Ok(ARefs::new(v).map(|v| &v[r.start as usize .. r.end as usize]))
}
}
impl http_entity::Entity<Error> for File {
#[derive(Clone)]
pub struct File(Arc<FileInner>);
impl http_entity::Entity<Body> for File {
fn add_headers(&self, hdrs: &mut header::Headers) {
hdrs.set(header::ContentType("video/mp4".parse().unwrap()));
}
fn last_modified(&self) -> Option<header::HttpDate> { Some(self.last_modified) }
fn etag(&self) -> Option<header::EntityTag> { Some(self.etag.clone()) }
fn len(&self) -> u64 { self.slices.len() }
fn write_to(&self, range: Range<u64>, out: &mut io::Write) -> Result<(), Error> {
self.slices.write_to(self, range, out)
}
fn last_modified(&self) -> Option<header::HttpDate> { Some(self.0.last_modified) }
fn etag(&self) -> Option<header::EntityTag> { Some(self.0.etag.clone()) }
fn len(&self) -> u64 { self.0.slices.len() }
fn get_range(&self, range: Range<u64>) -> Body { self.0.slices.get_range(self, range) }
}
/// Tests. There are two general strategies used to validate the resulting files:
@ -1206,14 +1241,13 @@ mod tests {
use byteorder::{BigEndian, ByteOrder};
use db;
use dir;
use error::Error;
use futures::Stream as FuturesStream;
use ffmpeg;
use hyper::header;
use openssl::hash;
use recording::{self, TIME_UNITS_PER_SEC};
use http_entity::{self, Entity};
use std::fs;
use std::io;
use std::ops::Range;
use std::path::Path;
use std::sync::Arc;
@ -1223,27 +1257,29 @@ mod tests {
use stream::{self, Opener, Stream};
use testutil::{self, TestDb, TEST_CAMERA_ID};
/// A wrapper around openssl's SHA-1 hashing that implements the `Write` trait.
struct Sha1(hash::Hasher);
impl Sha1 {
fn new() -> Sha1 { Sha1(hash::Hasher::new(hash::MessageDigest::sha1()).unwrap()) }
fn finish(mut self) -> Vec<u8> { self.0.finish().unwrap() }
}
impl io::Write for Sha1 {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.0.update(buf).unwrap();
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> { Ok(()) }
fn fill_slice(slice: &mut [u8], e: &http_entity::Entity<Body>, start: u64) {
let mut p = 0;
e.get_range(start .. start + slice.len() as u64)
.for_each(|chunk| {
slice[p .. p + chunk.len()].copy_from_slice(&chunk);
p += chunk.len();
Ok::<_, ::hyper::Error>(())
})
.wait()
.unwrap();
}
/// Returns the SHA-1 digest of the given `Entity`.
fn digest(e: &http_entity::Entity<Error>) -> Vec<u8> {
let mut sha1 = Sha1::new();
e.write_to(0 .. e.len(), &mut sha1).unwrap();
sha1.finish()
fn digest(e: &http_entity::Entity<Body>) -> Vec<u8> {
e.get_range(0 .. e.len())
.fold(hash::Hasher::new(hash::MessageDigest::sha1()).unwrap(), |mut sha1, chunk| {
sha1.update(&chunk).unwrap();
Ok::<_, ::hyper::Error>(sha1)
})
.wait()
.unwrap()
.finish()
.unwrap()
}
/// Information used within `BoxCursor` to describe a box on the stack.
@ -1256,13 +1292,13 @@ mod tests {
/// A cursor over the boxes in a `.mp4` file. Supports moving forward and up/down the box
/// stack, not backward. Panics on error.
#[derive(Clone)]
struct BoxCursor<'a> {
mp4: &'a http_entity::Entity<Error>,
struct BoxCursor {
mp4: File,
stack: Vec<Mp4Box>,
}
impl<'a> BoxCursor<'a> {
pub fn new(mp4: &'a http_entity::Entity<Error>) -> BoxCursor<'a> {
impl BoxCursor {
pub fn new(mp4: File) -> BoxCursor {
BoxCursor{
mp4: mp4,
stack: Vec::new(),
@ -1274,11 +1310,11 @@ mod tests {
fn internal_push(&mut self, pos: u64, max: u64) -> bool {
if pos == max { return false; }
let mut hdr = [0u8; 16];
self.mp4.write_to(pos .. pos+8, &mut &mut hdr[..]).unwrap();
fill_slice(&mut hdr[..8], &self.mp4, pos);
let (len, hdr_len, boxtype_slice) = match BigEndian::read_u32(&hdr[..4]) {
0 => (self.mp4.len() - pos, 8, &hdr[4..8]),
1 => {
self.mp4.write_to(pos+8 .. pos+12, &mut &mut hdr[..]).unwrap();
fill_slice(&mut hdr[8..], &self.mp4, pos + 8);
(BigEndian::read_u64(&hdr[4..12]), 16, &hdr[12..])
},
l => (l as u64, 8, &hdr[4..8]),
@ -1306,17 +1342,19 @@ mod tests {
/// Gets the specified byte range within the current box, starting after the box type.
/// Must not be at EOF.
pub fn get(&self, r: Range<u64>, mut buf: &mut [u8]) {
pub fn get(&self, start: u64, buf: &mut [u8]) {
let interior = &self.stack.last().expect("at root").interior;
assert!(r.end < interior.end - interior.start);
self.mp4.write_to(r.start+interior.start .. r.end+interior.start, &mut buf).unwrap();
assert!(start + (buf.len() as u64) < interior.end - interior.start);
fill_slice(buf, &self.mp4, start+interior.start);
}
pub fn get_all(&self) -> Vec<u8> {
let interior = self.stack.last().expect("at root").interior.clone();
let len = (interior.end - interior.start) as usize;
trace!("get_all: start={}, len={}", interior.start, len);
let mut out = Vec::with_capacity(len);
self.mp4.write_to(interior, &mut out).unwrap();
unsafe { out.set_len(len) };
fill_slice(&mut out[..], &self.mp4, interior.start);
out
}
@ -1324,7 +1362,7 @@ mod tests {
/// Must not be at EOF.
pub fn get_u32(&self, p: u64) -> u32 {
let mut buf = [0u8; 4];
self.get(p .. p+4, &mut buf);
self.get(p, &mut buf);
BigEndian::read_u32(&buf[..])
}
@ -1360,14 +1398,14 @@ mod tests {
}
/// Information returned by `find_track`.
struct Track<'a> {
edts_cursor: Option<BoxCursor<'a>>,
stbl_cursor: BoxCursor<'a>,
struct Track {
edts_cursor: Option<BoxCursor>,
stbl_cursor: BoxCursor,
}
/// Finds the `moov/trak` that has a `tkhd` associated with the given `track_id`, which must
/// exist.
fn find_track(mp4: &http_entity::Entity<Error>, track_id: u32) -> Track {
fn find_track(mp4: File, track_id: u32) -> Track {
let mut cursor = BoxCursor::new(mp4);
cursor.down();
assert!(cursor.find(b"moov"));
@ -1377,7 +1415,7 @@ mod tests {
cursor.down();
assert!(cursor.find(b"tkhd"));
let mut version = [0u8; 1];
cursor.get(0 .. 1, &mut version);
cursor.get(0, &mut version);
// Let id_pos be the offset after the FullBox section of the track_id.
let id_pos = match version[0] {
@ -1459,7 +1497,7 @@ mod tests {
db.list_recordings_by_time(TEST_CAMERA_ID, all_time, |r| {
let d = r.duration_90k;
assert!(skip_90k + shorten_90k < d);
builder.append(&db, r, skip_90k .. d - shorten_90k).unwrap();
builder.append(&*db, r, skip_90k .. d - shorten_90k).unwrap();
Ok(())
}).unwrap();
}
@ -1470,7 +1508,14 @@ mod tests {
let mut filename = dir.to_path_buf();
filename.push("clip.new.mp4");
let mut out = fs::OpenOptions::new().write(true).create_new(true).open(&filename).unwrap();
mp4.write_to(0 .. mp4.len(), &mut out).unwrap();
use ::std::io::Write;
mp4.get_range(0 .. mp4.len())
.for_each(|chunk| {
out.write_all(&chunk)?;
Ok(())
})
.wait()
.unwrap();
filename.to_str().unwrap().to_string()
}
@ -1536,7 +1581,7 @@ mod tests {
// Time range [2, 2+4+6+8) means the 2nd, 3rd, and 4th samples should be included.
let mp4 = make_mp4_from_encoder(&db, encoder, 2 .. 2+4+6+8);
let track = find_track(&mp4, 1);
let track = find_track(mp4, 1);
assert!(track.edts_cursor.is_none());
let mut cursor = track.stbl_cursor;
cursor.down();
@ -1590,7 +1635,7 @@ mod tests {
// Time range [2+4+6, 2+4+6+8) means the 4th sample should be included.
// The 3rd gets pulled in also because it's a sync frame and the 4th isn't.
let mp4 = make_mp4_from_encoder(&db, encoder, 2+4+6 .. 2+4+6+8);
let track = find_track(&mp4, 1);
let track = find_track(mp4, 1);
// Examine edts. It should skip the 3rd frame.
let mut cursor = track.edts_cursor.unwrap();
@ -1721,26 +1766,23 @@ mod tests {
#[cfg(all(test, feature="nightly"))]
mod bench {
extern crate reqwest;
extern crate test;
use futures::future;
use futures::stream::BoxStream;
use hyper;
use hyper::header;
use http_entity;
use recording;
use reffers::ARefs;
use self::test::Bencher;
use std::str;
use super::tests::create_mp4_from_db;
use testutil::{self, TestDb};
/// An HTTP server for benchmarking.
/// It's used as a singleton via `lazy_static!` for two reasons:
///
/// * to avoid running out of file descriptors. `#[bench]` functions apparently get called
/// many times as the number of iterations is tuned, and hyper servers
/// [can't be shut down](https://github.com/hyperium/hyper/issues/338), so
/// otherwise the default Ubuntu 16.04.1 ulimit of 1024 files is quickly exhausted.
/// * so that when getting a CPU profile of the benchmark, more of the profile focuses
/// on the HTTP serving rather than the setup.
/// It's used as a singleton via `lazy_static!` so that when getting a CPU profile of the
/// benchmark, more of the profile focuses on the HTTP serving rather than the setup.
///
/// Currently this only serves a single `.mp4` file but we could set up variations to benchmark
/// different scenarios: with/without subtitles and edit lists, different lengths, serving
@ -1752,32 +1794,40 @@ mod bench {
impl BenchServer {
fn new() -> BenchServer {
let mut listener = hyper::net::HttpListener::new("127.0.0.1:0").unwrap();
use hyper::net::NetworkListener;
let addr = listener.local_addr().unwrap();
let server = hyper::Server::new(listener);
let url = hyper::Url::parse(
format!("http://{}:{}/", addr.ip(), addr.port()).as_str()).unwrap();
let db = TestDb::new();
testutil::add_dummy_recordings_to_db(&db.db, 60);
let mp4 = create_mp4_from_db(db.db.clone(), db.dir.clone(), 0, 0, false);
let p = mp4.initial_sample_byte_pos;
use std::thread::spawn;
spawn(move || {
use hyper::server::{Request, Response, Fresh};
let (db, dir) = (db.db.clone(), db.dir.clone());
let _ = server.handle(move |req: Request, res: Response<Fresh>| {
let mp4 = create_mp4_from_db(db.clone(), dir.clone(), 0, 0, false);
http_entity::serve(&mp4, &req, res).unwrap();
});
let p = mp4.0.initial_sample_byte_pos;
let (tx, rx) = ::std::sync::mpsc::channel();
::std::thread::spawn(move || {
let addr = "127.0.0.1:0".parse().unwrap();
let server = hyper::server::Http::new()
.bind(&addr, move || Ok(MyService(mp4.clone())))
.unwrap();
tx.send(server.local_addr().unwrap()).unwrap();
server.run().unwrap();
});
let addr = rx.recv().unwrap();
BenchServer{
url: url,
url: hyper::Url::parse(&format!("http://{}:{}/", addr.ip(), addr.port())).unwrap(),
generated_len: p,
}
}
}
struct MyService(super::File);
impl hyper::server::Service for MyService {
type Request = hyper::server::Request;
type Response = hyper::server::Response<BoxStream<ARefs<'static, [u8]>, hyper::Error>>;
type Error = hyper::Error;
type Future = future::FutureResult<Self::Response, Self::Error>;
fn call(&self, req: hyper::server::Request) -> Self::Future {
future::ok(http_entity::serve(self.0.clone(), &req))
}
}
lazy_static! {
static ref SERVER: BenchServer = { BenchServer::new() };
}
@ -1816,11 +1866,12 @@ mod bench {
let p = server.generated_len;
let mut buf = Vec::with_capacity(p as usize);
b.bytes = p;
let client = hyper::Client::new();
let client = reqwest::Client::new().unwrap();
let mut run = || {
use self::reqwest::header::{Range, ByteRangeSpec};
let mut resp =
client.get(server.url.clone())
.header(header::Range::Bytes(vec![header::ByteRangeSpec::FromTo(0, p - 1)]))
.header(Range::Bytes(vec![ByteRangeSpec::FromTo(0, p - 1)]))
.send()
.unwrap();
buf.clear();

3
src/recording.rs
View File

@ -375,7 +375,8 @@ impl Segment {
/// desired start time. (The caller is responsible for creating an edit list to skip the
/// undesired portion.) It will end at the first frame after the desired range (unless the
/// desired range extends beyond the recording).
pub fn new(db: &db::LockedDatabase, recording: &db::ListRecordingsRow,
pub fn new(db: &db::LockedDatabase,
recording: &db::ListRecordingsRow,
desired_range_90k: Range<i32>) -> Result<Segment, Error> {
let mut self_ = Segment{
camera_id: recording.camera_id,

224
src/slices.rs
View File

@ -30,15 +30,20 @@
//! Tools for implementing a `http_entity::Entity` body composed from many "slices".
use error::{Error, Result};
use reffers::ARefs;
use futures::stream::{self, BoxStream};
use futures::Stream;
use std::fmt;
use std::io;
use std::ops::Range;
pub type Chunk = ARefs<'static, [u8]>;
pub type Body = stream::BoxStream<Chunk, ::hyper::Error>;
/// Writes a byte range to the given `io::Write` given a context argument; meant for use with
/// `Slices`.
pub trait Slice {
type Ctx;
pub trait Slice : Sync + Sized + 'static {
type Ctx: Send + Clone;
type Chunk: Send;
/// The byte position (relative to the start of the `Slices`) beyond the end of this slice.
/// Note the starting position (and thus length) are inferred from the previous slice.
@ -47,22 +52,10 @@ pub trait Slice {
/// Writes `r` to `out`, as in `http_entity::Entity::write_to`.
/// The additional argument `ctx` is as supplied to the `Slices`.
/// The additional argument `l` is the length of this slice, as determined by the `Slices`.
fn write_to(&self, ctx: &Self::Ctx, r: Range<u64>, l: u64, out: &mut io::Write) -> Result<()>;
}
fn get_range(&self, ctx: &Self::Ctx, r: Range<u64>, len: u64)
-> stream::BoxStream<Self::Chunk, ::hyper::Error>;
/// Calls `f` with an `io::Write` which delegates to `inner` only for the section defined by `r`.
/// This is useful for easily implementing the `ContextWriter` interface for pieces that generate
/// data on-the-fly rather than simply copying a buffer.
pub fn clip_to_range<F>(r: Range<u64>, l: u64, inner: &mut io::Write, mut f: F) -> Result<()>
where F: FnMut(&mut Vec<u8>) -> Result<()> {
// Just create a buffer for the whole slice and copy out the relevant portion.
// One might expect it to be faster to avoid this memory allocation and extra copying, but
// benchmarks show when making many 4-byte writes it's better to be able to inline many
// Vec::write_all calls then make one call through traits to hyper's write logic.
let mut buf = Vec::with_capacity(l as usize);
f(&mut buf)?;
inner.write_all(&buf[r.start as usize .. r.end as usize])?;
Ok(())
fn get_slices(ctx: &Self::Ctx) -> &Slices<Self>;
}
/// Helper to serve byte ranges from a body which is broken down into many "slices".
@ -114,56 +107,53 @@ impl<S> Slices<S> where S: Slice {
/// Writes `range` to `out`.
/// This interface mirrors `http_entity::Entity::write_to`, with the additional `ctx` argument.
pub fn write_to(&self, ctx: &S::Ctx, range: Range<u64>, out: &mut io::Write) -> Result<()> {
pub fn get_range(&self, ctx: &S::Ctx, range: Range<u64>)
-> BoxStream<S::Chunk, ::hyper::Error> {
if range.start > range.end || range.end > self.len {
return Err(Error{
description: format!("Bad range {:?} for slice of length {}", range, self.len),
cause: None});
error!("Bad range {:?} for slice of length {}", range, self.len);
return stream::once(Err(::hyper::Error::Incomplete)).boxed();
}
// Binary search for the first slice of the range to write, determining its index and
// (from the preceding slice) the start of its range.
let (mut i, mut slice_start) = match self.slices.binary_search_by_key(&range.start,
|s| s.end()) {
Ok(i) if i == self.slices.len() - 1 => return Ok(()), // at end.
let (i, slice_start) = match self.slices.binary_search_by_key(&range.start, |s| s.end()) {
Ok(i) => (i+1, self.slices[i].end()), // desired start == slice i's end; first is i+1!
Err(i) if i == 0 => (0, 0), // desired start < slice 0's end; first is 0.
Err(i) => (i, self.slices[i-1].end()), // desired start < slice i's end; first is i.
};
// There is at least one slice to write.
// Iterate through and write each slice until the end.
let mut start_pos = range.start - slice_start;
loop {
let s = &self.slices[i];
let end = s.end();
let l = end - slice_start;
if range.end <= end { // last slice.
return s.write_to(ctx, start_pos .. range.end - slice_start, l, out);
}
s.write_to(ctx, start_pos .. end - slice_start, l, out)?;
// Setup next iteration.
start_pos = 0;
slice_start = end;
i += 1;
}
let start_pos = range.start - slice_start;
let bodies = stream::unfold(
(ctx.clone(), i, start_pos, slice_start), move |(c, i, start_pos, slice_start)| {
let (body, end);
{
let self_ = S::get_slices(&c);
if i == self_.slices.len() { return None }
let s = &self_.slices[i];
if range.end == slice_start + start_pos { return None }
end = ::std::cmp::min(range.end, s.end());
let l = end - slice_start;
body = s.get_range(&c, start_pos .. end - slice_start, l);
};
Some(Ok::<_, ::hyper::Error>((body, (c, i+1, 0, end))))
});
bodies.flatten().boxed()
}
}
#[cfg(test)]
mod tests {
use error::{Error, Result};
use std::cell::RefCell;
use std::error::Error as E;
use std::io::Write;
use futures::{Future, Stream};
use futures::stream::{self, BoxStream};
use std::ops::Range;
use std::vec::Vec;
use super::{Slice, Slices, clip_to_range};
use super::{Slice, Slices};
use testutil;
#[derive(Debug, Eq, PartialEq)]
pub struct FakeWrite {
writer: &'static str,
pub struct FakeChunk {
slice: &'static str,
range: Range<u64>,
}
@ -173,18 +163,21 @@ mod tests {
}
impl Slice for FakeSlice {
type Ctx = RefCell<Vec<FakeWrite>>;
type Ctx = &'static Slices<FakeSlice>;
type Chunk = FakeChunk;
fn end(&self) -> u64 { self.end }
fn write_to(&self, ctx: &RefCell<Vec<FakeWrite>>, r: Range<u64>, _l: u64, _out: &mut Write)
-> Result<()> {
ctx.borrow_mut().push(FakeWrite{writer: self.name, range: r});
Ok(())
}
fn get_range(&self, _ctx: &&'static Slices<FakeSlice>, r: Range<u64>, _l: u64)
-> BoxStream<FakeChunk, ::hyper::Error> {
stream::once(Ok(FakeChunk{slice: self.name, range: r})).boxed()
}
pub fn new_slices() -> Slices<FakeSlice> {
fn get_slices(ctx: &&'static Slices<FakeSlice>) -> &'static Slices<Self> { *ctx }
}
lazy_static! {
static ref SLICES: Slices<FakeSlice> = {
let mut s = Slices::new();
s.append(FakeSlice{end: 5, name: "a"});
s.append(FakeSlice{end: 5+13, name: "b"});
@ -192,127 +185,62 @@ mod tests {
s.append(FakeSlice{end: 5+13+7+17, name: "d"});
s.append(FakeSlice{end: 5+13+7+17+19, name: "e"});
s
};
}
#[test]
pub fn size() {
assert_eq!(5 + 13 + 7 + 17 + 19, new_slices().len());
testutil::init();
assert_eq!(5 + 13 + 7 + 17 + 19, SLICES.len());
}
#[test]
pub fn exact_slice() {
// Test writing exactly slice b.
let s = new_slices();
let w = RefCell::new(Vec::new());
let mut dummy = Vec::new();
s.write_to(&w, 5 .. 18, &mut dummy).unwrap();
assert_eq!(&[FakeWrite{writer: "b", range: 0 .. 13}], &w.borrow()[..]);
testutil::init();
let out = SLICES.get_range(&&*SLICES, 5 .. 18).collect().wait().unwrap();
assert_eq!(&[FakeChunk{slice: "b", range: 0 .. 13}], &out[..]);
}
#[test]
pub fn offset_first() {
// Test writing part of slice a.
let s = new_slices();
let w = RefCell::new(Vec::new());
let mut dummy = Vec::new();
s.write_to(&w, 1 .. 3, &mut dummy).unwrap();
assert_eq!(&[FakeWrite{writer: "a", range: 1 .. 3}], &w.borrow()[..]);
testutil::init();
let out = SLICES.get_range(&&*SLICES, 1 .. 3).collect().wait().unwrap();
assert_eq!(&[FakeChunk{slice: "a", range: 1 .. 3}], &out[..]);
}
#[test]
pub fn offset_mid() {
// Test writing part of slice b, all of slice c, and part of slice d.
let s = new_slices();
let w = RefCell::new(Vec::new());
let mut dummy = Vec::new();
s.write_to(&w, 17 .. 26, &mut dummy).unwrap();
testutil::init();
let out = SLICES.get_range(&&*SLICES, 17 .. 26).collect().wait().unwrap();
assert_eq!(&[
FakeWrite{writer: "b", range: 12 .. 13},
FakeWrite{writer: "c", range: 0 .. 7},
FakeWrite{writer: "d", range: 0 .. 1},
], &w.borrow()[..]);
FakeChunk{slice: "b", range: 12 .. 13},
FakeChunk{slice: "c", range: 0 .. 7},
FakeChunk{slice: "d", range: 0 .. 1},
], &out[..]);
}
#[test]
pub fn everything() {
// Test writing the whole Slices.
let s = new_slices();
let w = RefCell::new(Vec::new());
let mut dummy = Vec::new();
s.write_to(&w, 0 .. 61, &mut dummy).unwrap();
testutil::init();
let out = SLICES.get_range(&&*SLICES, 0 .. 61).collect().wait().unwrap();
assert_eq!(&[
FakeWrite{writer: "a", range: 0 .. 5},
FakeWrite{writer: "b", range: 0 .. 13},
FakeWrite{writer: "c", range: 0 .. 7},
FakeWrite{writer: "d", range: 0 .. 17},
FakeWrite{writer: "e", range: 0 .. 19},
], &w.borrow()[..]);
FakeChunk{slice: "a", range: 0 .. 5},
FakeChunk{slice: "b", range: 0 .. 13},
FakeChunk{slice: "c", range: 0 .. 7},
FakeChunk{slice: "d", range: 0 .. 17},
FakeChunk{slice: "e", range: 0 .. 19},
], &out[..]);
}
#[test]
pub fn at_end() {
let s = new_slices();
let w = RefCell::new(Vec::new());
let mut dummy = Vec::new();
s.write_to(&w, 61 .. 61, &mut dummy).unwrap();
let empty: &[FakeWrite] = &[];
assert_eq!(empty, &w.borrow()[..]);
}
#[test]
pub fn test_clip_to_range() {
let mut out = Vec::new();
// Simple case: one write with everything.
clip_to_range(0 .. 5, 5, &mut out, |w| {
w.write_all(b"01234").unwrap();
Ok(())
}).unwrap();
assert_eq!(b"01234", &out[..]);
// Same in a few writes.
out.clear();
clip_to_range(0 .. 5, 5, &mut out, |w| {
w.write_all(b"0").unwrap();
w.write_all(b"123").unwrap();
w.write_all(b"4").unwrap();
Ok(())
}).unwrap();
assert_eq!(b"01234", &out[..]);
// Limiting to a prefix.
out.clear();
clip_to_range(0 .. 2, 5, &mut out, |w| {
w.write_all(b"0").unwrap(); // all of this write
w.write_all(b"123").unwrap(); // some of this write
w.write_all(b"4").unwrap(); // none of this write
Ok(())
}).unwrap();
assert_eq!(b"01", &out[..]);
// Limiting to part in the middle.
out.clear();
clip_to_range(2 .. 4, 5, &mut out, |w| {
w.write_all(b"0").unwrap(); // none of this write
w.write_all(b"1234").unwrap(); // middle of this write
w.write_all(b"5678").unwrap(); // none of this write
Ok(())
}).unwrap();
assert_eq!(b"23", &out[..]);
// If the callback returns an error, it should be propagated (fast path or not).
out.clear();
assert_eq!(
clip_to_range(0 .. 4, 4, &mut out, |_| Err(Error::new("some error".to_owned())))
.unwrap_err().description(),
"some error");
out.clear();
assert_eq!(
clip_to_range(0 .. 1, 4, &mut out, |_| Err(Error::new("some error".to_owned())))
.unwrap_err().description(),
"some error");
// TODO: if inner.write does a partial write, the next try should start at the correct
// position.
testutil::init();
let out = SLICES.get_range(&&*SLICES, 61 .. 61).collect().wait().unwrap();
let empty: &[FakeChunk] = &[];
assert_eq!(empty, &out[..]);
}
}

182
src/web.rs
View File

@ -35,15 +35,20 @@ use core::str::FromStr;
use db;
use dir::SampleFileDir;
use error::Error;
use futures::Stream;
use futures::{future, stream};
use json;
use http_entity;
use hyper::{header,server,status};
use hyper::uri::RequestUri;
use hyper::{header, status};
use hyper::server::{self, Request, Response};
use mime;
use mp4;
use parking_lot::MutexGuard;
use recording;
use reffers::ARefs;
use regex::Regex;
use serde_json;
use slices;
use std::cmp;
use std::fmt;
use std::io::Write;
@ -75,17 +80,6 @@ enum Path {
NotFound,
}
fn get_path_and_query(uri: &RequestUri) -> (&str, &str) {
match *uri {
RequestUri::AbsolutePath(ref both) => match both.find('?') {
Some(split) => (&both[..split], &both[split+1..]),
None => (both, ""),
},
RequestUri::AbsoluteUri(ref u) => (u.path(), u.query().unwrap_or("")),
_ => ("", ""),
}
}
fn decode_path(path: &str) -> Path {
if path == "/" {
return Path::CamerasList;
@ -116,8 +110,8 @@ fn decode_path(path: &str) -> Path {
}
}
fn is_json(req: &server::Request) -> bool {
if let Some(accept) = req.headers.get::<header::Accept>() {
fn is_json(req: &Request) -> bool {
if let Some(accept) = req.headers().get::<header::Accept>() {
return accept.len() == 1 && accept[0].item == *JSON &&
accept[0].quality == header::Quality(1000);
}
@ -182,11 +176,6 @@ impl fmt::Display for HumanizedTimestamp {
}
}
pub struct Handler {
db: Arc<db::Database>,
dir: Arc<SampleFileDir>,
}
#[derive(Debug, Eq, PartialEq)]
struct Segments {
ids: Range<i32>,
@ -227,33 +216,40 @@ impl Segments {
}
}
impl Handler {
pub struct Service {
db: Arc<db::Database>,
dir: Arc<SampleFileDir>,
}
impl Service {
pub fn new(db: Arc<db::Database>, dir: Arc<SampleFileDir>) -> Self {
Handler{db: db, dir: dir}
Service{db: db, dir: dir}
}
fn not_found(&self, mut res: server::Response) -> Result<(), Error> {
*res.status_mut() = status::StatusCode::NotFound;
res.send(b"not found")?;
Ok(())
fn not_found(&self) -> Result<Response<slices::Body>, Error> {
Ok(Response::new()
.with_status(status::StatusCode::NotFound)
.with_header(header::ContentType(mime!(Text/Plain)))
.with_body(stream::once(Ok(ARefs::new(&b"not found"[..]))).boxed()))
}
fn list_cameras(&self, req: &server::Request, mut res: server::Response) -> Result<(), Error> {
fn list_cameras(&self, req: &Request) -> Result<Response<slices::Body>, Error> {
let json = is_json(req);
let buf = {
let db = self.db.lock();
if json {
serde_json::to_vec(&json::ListCameras{cameras: db.cameras_by_id()})?
} else {
self.list_cameras_html(&db)?
self.list_cameras_html(db)?
}
};
res.headers_mut().set(header::ContentType(if json { JSON.clone() } else { HTML.clone() }));
res.send(&buf)?;
Ok(())
Ok(Response::new()
.with_header(header::ContentType(if json { JSON.clone() } else { HTML.clone() }))
.with_header(header::ContentLength(buf.len() as u64))
.with_body(stream::once(Ok(ARefs::new(buf))).boxed()))
}
fn list_cameras_html(&self, db: &db::LockedDatabase) -> Result<Vec<u8>, Error> {
fn list_cameras_html(&self, db: MutexGuard<db::LockedDatabase>) -> Result<Vec<u8>, Error> {
let mut buf = Vec::new();
buf.extend_from_slice(b"\
<!DOCTYPE html>\n\
@ -287,8 +283,8 @@ impl Handler {
Ok(buf)
}
fn camera(&self, uuid: Uuid, query: &str, req: &server::Request, mut res: server::Response)
-> Result<(), Error> {
fn camera(&self, uuid: Uuid, query: Option<&str>, req: &Request)
-> Result<Response<slices::Body>, Error> {
let json = is_json(req);
let buf = {
let db = self.db.lock();
@ -297,20 +293,22 @@ impl Handler {
.ok_or_else(|| Error::new("no such camera".to_owned()))?;
serde_json::to_vec(&json::Camera::new(camera, true))?
} else {
self.camera_html(&db, query, uuid)?
self.camera_html(db, query, uuid)?
}
};
res.headers_mut().set(header::ContentType(if json { JSON.clone() } else { HTML.clone() }));
res.send(&buf)?;
Ok(())
Ok(Response::new()
.with_header(header::ContentType(if json { JSON.clone() } else { HTML.clone() }))
.with_header(header::ContentLength(buf.len() as u64))
.with_body(stream::once(Ok(ARefs::new(buf))).boxed()))
}
fn camera_html(&self, db: &db::LockedDatabase, query: &str, uuid: Uuid)
-> Result<Vec<u8>, Error> {
fn camera_html(&self, db: MutexGuard<db::LockedDatabase>, query: Option<&str>,
uuid: Uuid) -> Result<Vec<u8>, Error> {
let (r, trim) = {
let mut time = recording::Time(i64::min_value()) .. recording::Time(i64::max_value());
let mut trim = false;
for (key, value) in form_urlencoded::parse(query.as_bytes()) {
if let Some(q) = query {
for (key, value) in form_urlencoded::parse(q.as_bytes()) {
let (key, value) = (key.borrow(), value.borrow());
match key {
"start_time" => time.start = recording::Time::parse(value)?,
@ -319,6 +317,7 @@ impl Handler {
_ => {},
}
};
}
(time, trim)
};
let camera = db.get_camera(uuid)
@ -396,13 +395,14 @@ impl Handler {
Ok(buf)
}
fn camera_recordings(&self, uuid: Uuid, query: &str, req: &server::Request,
mut res: server::Response) -> Result<(), Error> {
let r = Handler::get_optional_range(query)?;
fn camera_recordings(&self, uuid: Uuid, query: Option<&str>, req: &Request)
-> Result<Response<slices::Body>, Error> {
let r = Service::get_optional_range(query)?;
if !is_json(req) {
*res.status_mut() = status::StatusCode::NotAcceptable;
res.send(b"only available for JSON requests")?;
return Ok(());
return Ok(Response::new()
.with_status(status::StatusCode::NotAcceptable)
.with_body(stream::once(
Ok(ARefs::new(&b"only available for JSON requests"[..]))).boxed()));
}
let mut out = json::ListRecordings{recordings: Vec::new()};
{
@ -424,13 +424,14 @@ impl Handler {
})?;
}
let buf = serde_json::to_vec(&out)?;
res.headers_mut().set(header::ContentType(JSON.clone()));
res.send(&buf)?;
Ok(())
Ok(Response::new()
.with_header(header::ContentType(JSON.clone()))
.with_header(header::ContentLength(buf.len() as u64))
.with_body(stream::once(Ok(ARefs::new(buf))).boxed()))
}
fn camera_view_mp4(&self, uuid: Uuid, query: &str, req: &server::Request,
res: server::Response) -> Result<(), Error> {
fn camera_view_mp4(&self, uuid: Uuid, query: Option<&str>, req: &Request)
-> Result<Response<slices::Body>, Error> {
let camera_id = {
let db = self.db.lock();
let camera = db.get_camera(uuid)
@ -438,7 +439,8 @@ impl Handler {
camera.id
};
let mut builder = mp4::FileBuilder::new();
for (key, value) in form_urlencoded::parse(query.as_bytes()) {
if let Some(q) = query {
for (key, value) in form_urlencoded::parse(q.as_bytes()) {
let (key, value) = (key.borrow(), value.borrow());
match key {
"s" => {
@ -447,10 +449,11 @@ impl Handler {
debug!("camera_view_mp4: appending s={:?}", s);
let mut est_segments = (s.ids.end - s.ids.start) as usize;
if let Some(end) = s.end_time {
// There should be roughly ceil((end - start) / desired_recording_duration)
// recordings in the desired timespan if there are no gaps or overlap,
// possibly another for misalignment of the requested timespan with the
// rotate offset and another because rotation only happens at key frames.
// There should be roughly ceil((end - start) /
// desired_recording_duration) recordings in the desired timespan if
// there are no gaps or overlap, possibly another for misalignment of
// the requested timespan with the rotate offset and another because
// rotation only happens at key frames.
let ceil_durations = (end - s.start_time +
recording::DESIRED_RECORDING_DURATION - 1) /
recording::DESIRED_RECORDING_DURATION;
@ -514,17 +517,18 @@ impl Handler {
_ => return Err(Error::new(format!("parameter {} not understood", key))),
}
};
}
let mp4 = builder.build(self.db.clone(), self.dir.clone())?;
http_entity::serve(&mp4, req, res)?;
Ok(())
Ok(http_entity::serve(mp4, req))
}
/// Parses optional `start_time_90k` and `end_time_90k` query parameters, defaulting to the
/// full range of possible values.
fn get_optional_range(query: &str) -> Result<Range<recording::Time>, Error> {
fn get_optional_range(query: Option<&str>) -> Result<Range<recording::Time>, Error> {
let mut start = i64::min_value();
let mut end = i64::max_value();
for (key, value) in form_urlencoded::parse(query.as_bytes()) {
if let Some(q) = query {
for (key, value) in form_urlencoded::parse(q.as_bytes()) {
let (key, value) = (key.borrow(), value.borrow());
match key {
"start_time_90k" => start = i64::from_str(value)?,
@ -532,23 +536,30 @@ impl Handler {
_ => {},
}
};
}
Ok(recording::Time(start) .. recording::Time(end))
}
}
impl server::Handler for Handler {
fn handle(&self, req: server::Request, res: server::Response) {
let (path, query) = get_path_and_query(&req.uri);
let res = match decode_path(path) {
Path::CamerasList => self.list_cameras(&req, res),
Path::Camera(uuid) => self.camera(uuid, query, &req, res),
Path::CameraRecordings(uuid) => self.camera_recordings(uuid, query, &req, res),
Path::CameraViewMp4(uuid) => self.camera_view_mp4(uuid, query, &req, res),
Path::NotFound => self.not_found(res),
impl server::Service for Service {
type Request = Request;
type Response = Response<slices::Body>;
type Error = hyper::Error;
type Future = future::FutureResult<Self::Response, Self::Error>;
fn call(&self, req: Request) -> Self::Future {
debug!("request on: {}", req.uri());
let res = match decode_path(req.uri().path()) {
Path::CamerasList => self.list_cameras(&req),
Path::Camera(uuid) => self.camera(uuid, req.uri().query(), &req),
Path::CameraRecordings(uuid) => self.camera_recordings(uuid, req.uri().query(), &req),
Path::CameraViewMp4(uuid) => self.camera_view_mp4(uuid, req.uri().query(), &req),
Path::NotFound => self.not_found(),
};
if let Err(ref e) = res {
warn!("Error handling request: {}", e);
}
future::result(res.map_err(|e| {
error!("error: {}", e);
hyper::Error::Incomplete
}))
}
}
@ -599,6 +610,7 @@ mod tests {
#[cfg(all(test, feature="nightly"))]
mod bench {
extern crate reqwest;
extern crate test;
use hyper;
@ -612,18 +624,20 @@ mod bench {
impl Server {
fn new() -> Server {
let mut listener = hyper::net::HttpListener::new("127.0.0.1:0").unwrap();
use hyper::net::NetworkListener;
let addr = listener.local_addr().unwrap();
let server = hyper::Server::new(listener);
let url = format!("http://{}:{}", addr.ip(), addr.port());
let db = TestDb::new();
testutil::add_dummy_recordings_to_db(&db.db, 1440);
let (tx, rx) = ::std::sync::mpsc::channel();
::std::thread::spawn(move || {
let h = super::Handler::new(db.db.clone(), db.dir.clone());
let _ = server.handle(h);
let addr = "127.0.0.1:0".parse().unwrap();
let (db, dir) = (db.db.clone(), db.dir.clone());
let server = hyper::server::Http::new()
.bind(&addr, move || Ok(super::Service::new(db.clone(), dir.clone())))
.unwrap();
tx.send(server.local_addr().unwrap()).unwrap();
server.run().unwrap();
});
Server{base_url: url}
let addr = rx.recv().unwrap();
Server{base_url: format!("http://{}:{}", addr.ip(), addr.port())}
}
}
@ -635,13 +649,13 @@ mod bench {
fn serve_camera_html(b: &mut Bencher) {
testutil::init();
let server = &*SERVER;
let url = hyper::Url::parse(&format!("{}/cameras/{}/", server.base_url,
let url = reqwest::Url::parse(&format!("{}/cameras/{}/", server.base_url,
*testutil::TEST_CAMERA_UUID)).unwrap();
let mut buf = Vec::new();
b.iter(|| {
let client = hyper::Client::new();
let client = reqwest::Client::new().unwrap();
let mut resp = client.get(url.clone()).send().unwrap();
assert_eq!(resp.status, hyper::status::StatusCode::Ok);
assert_eq!(*resp.status(), reqwest::StatusCode::Ok);
buf.clear();
use std::io::Read;
resp.read_to_end(&mut buf).unwrap();