From 5563eb806dd153d22475a2dac342b4a2c5a5a76a Mon Sep 17 00:00:00 2001 From: Paul Masurel Date: Sat, 26 Nov 2016 23:56:08 +0900 Subject: [PATCH] NOBUG simdcompression --- Cargo.toml | 2 +- src/compression/composite.rs | 3 +- src/compression/compression.rs | 408 -------------------------- src/compression/compression_nosimd.rs | 127 ++++++++ src/compression/compression_simd.rs | 112 +++++++ src/compression/mod.rs | 256 +++++++++++++++- src/core/index.rs | 13 +- src/indexer/mod.rs | 2 +- src/indexer/segment_updater.rs | 19 ++ src/postings/freq_handler.rs | 2 +- src/postings/segment_postings.rs | 2 +- src/postings/serializer.rs | 1 + 12 files changed, 520 insertions(+), 427 deletions(-) delete mode 100644 src/compression/compression.rs create mode 100644 src/compression/compression_nosimd.rs create mode 100644 src/compression/compression_simd.rs diff --git a/Cargo.toml b/Cargo.toml index 72d74d233..1f7482fbc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,5 +50,5 @@ debug-assertions = false [features] -default = [] +default = ["simdcompression"] simdcompression = ["libc", "gcc"] diff --git a/src/compression/composite.rs b/src/compression/composite.rs index b4cec5285..915ab9f42 100644 --- a/src/compression/composite.rs +++ b/src/compression/composite.rs @@ -1,5 +1,6 @@ -use compression::{BlockEncoder, BlockDecoder}; +use super::{BlockEncoder, BlockDecoder}; use super::NUM_DOCS_PER_BLOCK; +use compression::{VIntEncoder, VIntDecoder}; pub struct CompositeEncoder { block_encoder: BlockEncoder, diff --git a/src/compression/compression.rs b/src/compression/compression.rs deleted file mode 100644 index 1ab68f98c..000000000 --- a/src/compression/compression.rs +++ /dev/null @@ -1,408 +0,0 @@ - -use super::NUM_DOCS_PER_BLOCK; - -const COMPRESSED_BLOCK_MAX_SIZE: usize = NUM_DOCS_PER_BLOCK * 4 + 1; - - -#[cfg(feature="simdcompression")] -mod compression { - use libc::size_t; - extern { - fn compress_sorted_cpp( - data: *const u32, - output: *mut u8, - offset: u32) -> size_t; - - fn uncompress_sorted_cpp( - compressed_data: *const u8, - output: *mut u32, - offset: u32) -> size_t; - - fn compress_unsorted_cpp( - data: *const u32, - output: *mut u8) -> size_t; - - fn uncompress_unsorted_cpp( - compressed_data: *const u8, - output: *mut u32) -> size_t; - } - - pub fn compress_sorted(vals: &[u32], output: &mut [u8], offset: u32) -> usize { - unsafe { compress_sorted_cpp(vals.as_ptr(), output.as_mut_ptr(), offset) } - } - - pub fn uncompress_sorted(compressed_data: &[u8], output: &mut [u32], offset: u32) -> usize { - unsafe { uncompress_sorted_cpp(compressed_data.as_ptr(), output.as_mut_ptr(), offset) } - } - - pub fn compress_unsorted(vals: &[u32], output: &mut [u8]) -> usize { - unsafe { compress_unsorted_cpp(vals.as_ptr(), output.as_mut_ptr()) } - } - - pub fn uncompress_unsorted(compressed_data: &[u8], output: &mut [u32]) -> usize { - unsafe { uncompress_unsorted_cpp(compressed_data.as_ptr(), output.as_mut_ptr()) } - } -} - - -#[cfg(not(feature="simdcompression"))] -mod compression { - - use compression::NUM_DOCS_PER_BLOCK; - use common::bitpacker::compute_num_bits; - use common::bitpacker::{BitPacker, BitUnpacker}; - use std::cmp; - use std::io::Write; - - pub fn compress_sorted(vals: &[u32], mut output: &mut [u8], offset: u32) -> usize { - // TODO remove the alloc - let mut deltas = Vec::with_capacity(NUM_DOCS_PER_BLOCK); - unsafe { deltas.set_len(NUM_DOCS_PER_BLOCK ); } - let mut max_delta = 0; - { - let mut local_offset = offset; - for i in 0..NUM_DOCS_PER_BLOCK { - let val = vals[i]; - let delta = val - local_offset; - max_delta = cmp::max(max_delta, delta); - deltas[i] = delta; - local_offset = val; - } - } - let num_bits = compute_num_bits(max_delta); - output.write_all(&[num_bits]).unwrap(); - let mut bit_packer = BitPacker::new(num_bits as usize); - for val in &deltas { - bit_packer.write(*val, &mut output).unwrap(); - } - 1 + bit_packer.close(&mut output).expect("packing in memory should never fail") - } - - pub fn uncompress_sorted(compressed_data: &[u8], output: &mut [u32], mut offset: u32) -> usize { - let num_bits = compressed_data[0]; - let bit_unpacker = BitUnpacker::new(&compressed_data[1..], num_bits as usize); - for i in 0..NUM_DOCS_PER_BLOCK { - let delta = bit_unpacker.get(i); - let val = offset + delta; - output[i] = val; - offset = val; - } - 1 + (num_bits as usize * NUM_DOCS_PER_BLOCK + 7) / 8 - } - - pub fn compress_unsorted(vals: &[u32], mut output: &mut [u8]) -> usize { - let max = vals.iter().cloned().max().expect("compress unsorted called with an empty array"); - let num_bits = compute_num_bits(max); - output.write_all(&[num_bits]).unwrap(); - let mut bit_packer = BitPacker::new(num_bits as usize); - for val in vals { - bit_packer.write(*val, &mut output).unwrap(); - } - 1 + bit_packer.close(&mut output).expect("packing in memory should never fail") - } - - pub fn uncompress_unsorted(compressed_data: &[u8], output: &mut [u32]) -> usize { - let num_bits = compressed_data[0]; - let bit_unpacker = BitUnpacker::new(&compressed_data[1..], num_bits as usize); - for i in 0..NUM_DOCS_PER_BLOCK { - output[i] = bit_unpacker.get(i); - } - 1 + (num_bits as usize * NUM_DOCS_PER_BLOCK + 7) / 8 - } -} - - -pub struct BlockEncoder { - output: [u8; COMPRESSED_BLOCK_MAX_SIZE], - output_len: usize, -} - -impl BlockEncoder { - - pub fn new() -> BlockEncoder { - BlockEncoder { - output: [0u8; COMPRESSED_BLOCK_MAX_SIZE], - output_len: 0, - } - } - - pub fn compress_block_sorted(&mut self, vals: &[u32], offset: u32) -> &[u8] { - let compressed_size = compression::compress_sorted(vals, &mut self.output, offset); - &self.output[..compressed_size] - } - - pub fn compress_block_unsorted(&mut self, vals: &[u32]) -> &[u8] { - let compressed_size = compression::compress_unsorted(vals, &mut self.output); - &self.output[..compressed_size] - } - - pub fn compress_vint_sorted(&mut self, input: &[u32], mut offset: u32) -> &[u8] { - let mut byte_written = 0; - for &v in input { - let mut to_encode: u32 = v - offset; - offset = v; - loop { - let next_byte: u8 = (to_encode % 128u32) as u8; - to_encode /= 128u32; - if to_encode == 0u32 { - self.output[byte_written] = next_byte | 128u8; - byte_written += 1; - break; - } - else { - self.output[byte_written] = next_byte; - byte_written += 1; - } - } - } - &self.output[..byte_written] - } - - pub fn compress_vint_unsorted(&mut self, input: &[u32]) -> &[u8] { - let mut byte_written = 0; - for &v in input { - let mut to_encode: u32 = v; - loop { - let next_byte: u8 = (to_encode % 128u32) as u8; - to_encode /= 128u32; - if to_encode == 0u32 { - self.output[byte_written] = next_byte | 128u8; - byte_written += 1; - break; - } - else { - self.output[byte_written] = next_byte; - byte_written += 1; - } - } - } - &self.output[..byte_written] - } - -} - -pub struct BlockDecoder { - output: [u32; COMPRESSED_BLOCK_MAX_SIZE], - output_len: usize, -} - - -impl BlockDecoder { - pub fn new() -> BlockDecoder { - BlockDecoder::with_val(0u32) - } - - pub fn with_val(val: u32) -> BlockDecoder { - BlockDecoder { - output: [val; COMPRESSED_BLOCK_MAX_SIZE], - output_len: 0, - } - } - - pub fn uncompress_block_sorted<'a>(&mut self, compressed_data: &'a [u8], offset: u32) -> &'a[u8] { - let consumed_size = compression::uncompress_sorted(compressed_data, &mut self.output, offset); - self.output_len = NUM_DOCS_PER_BLOCK; - &compressed_data[consumed_size..] - } - - pub fn uncompress_block_unsorted<'a>(&mut self, compressed_data: &'a [u8]) -> &'a[u8] { - let consumed_size = compression::uncompress_unsorted(compressed_data, &mut self.output); - self.output_len = NUM_DOCS_PER_BLOCK; - &compressed_data[consumed_size..] - } - - pub fn uncompress_vint_sorted<'a>( - &mut self, - compressed_data: &'a [u8], - offset: u32, - num_els: usize) -> &'a [u8] { - let mut read_byte = 0; - let mut result = offset; - for i in 0..num_els { - let mut shift = 0u32; - loop { - let cur_byte = compressed_data[read_byte]; - read_byte += 1; - result += ((cur_byte % 128u8) as u32) << shift; - if cur_byte & 128u8 != 0u8 { - break; - } - shift += 7; - } - self.output[i] = result; - } - self.output_len = num_els; - &compressed_data[read_byte..] - } - - pub fn uncompress_vint_unsorted<'a>( - &mut self, - compressed_data: &'a [u8], - num_els: usize) -> &'a [u8] { - let mut read_byte = 0; - for i in 0..num_els { - let mut result = 0u32; - let mut shift = 0u32; - loop { - let cur_byte = compressed_data[read_byte]; - read_byte += 1; - result += ((cur_byte % 128u8) as u32) << shift; - if cur_byte & 128u8 != 0u8 { - break; - } - shift += 7; - } - self.output[i] = result; - } - self.output_len = num_els; - &compressed_data[read_byte..] - } - - #[inline] - pub fn output_array(&self,) -> &[u32] { - &self.output[..self.output_len] - } - - #[inline] - pub fn output(&self, idx: usize) -> u32 { - self.output[idx] - } -} - - - -#[cfg(test)] -mod tests { - - use super::*; - use test::Bencher; - use compression::NUM_DOCS_PER_BLOCK; - use compression::tests::generate_array; - - #[test] - fn test_encode_sorted_block() { - let vals: Vec = (0u32..128u32).map(|i| i*7).collect(); - let mut encoder = BlockEncoder::new(); - let compressed_data = encoder.compress_block_sorted(&vals, 0); - let mut decoder = BlockDecoder::new(); - { - let remaining_data = decoder.uncompress_block_sorted(compressed_data, 0); - assert_eq!(remaining_data.len(), 0); - } - for i in 0..128 { - assert_eq!(vals[i], decoder.output(i)); - } - } - - #[test] - fn test_encode_sorted_block_with_offset() { - let vals: Vec = (0u32..128u32).map(|i| 11 + i*7).collect(); - let mut encoder = BlockEncoder::new(); - let compressed_data = encoder.compress_block_sorted(&vals, 10); - let mut decoder = BlockDecoder::new(); - { - let remaining_data = decoder.uncompress_block_sorted(compressed_data, 10); - assert_eq!(remaining_data.len(), 0); - } - for i in 0..128 { - assert_eq!(vals[i], decoder.output(i)); - } - } - - #[test] - fn test_encode_sorted_block_with_junk() { - let mut compressed: Vec = Vec::new(); - let n = 128; - let vals: Vec = (0..n).map(|i| 11u32 + (i as u32)*7u32).collect(); - let mut encoder = BlockEncoder::new(); - let compressed_data = encoder.compress_block_sorted(&vals, 10); - compressed.extend_from_slice(compressed_data); - compressed.push(173u8); - let mut decoder = BlockDecoder::new(); - { - let remaining_data = decoder.uncompress_block_sorted(&compressed, 10); - assert_eq!(remaining_data.len(), 1); - assert_eq!(remaining_data[0], 173u8); - } - for i in 0..n { - assert_eq!(vals[i], decoder.output(i)); - } - } - - #[test] - fn test_encode_unsorted_block_with_junk() { - let mut compressed: Vec = Vec::new(); - let n = 128; - let vals: Vec = (0..n).map(|i| 11u32 + (i as u32)*7u32 % 12).collect(); - let mut encoder = BlockEncoder::new(); - let compressed_data = encoder.compress_block_unsorted(&vals); - compressed.extend_from_slice(compressed_data); - compressed.push(173u8); - let mut decoder = BlockDecoder::new(); - { - let remaining_data = decoder.uncompress_block_unsorted(&compressed); - assert_eq!(remaining_data.len(), 1); - assert_eq!(remaining_data[0], 173u8); - } - for i in 0..n { - assert_eq!(vals[i], decoder.output(i)); - } - } - - - #[test] - fn test_encode_vint() { - { - let expected_length = 123; - let mut encoder = BlockEncoder::new(); - let input: Vec = (0u32..123u32) - .map(|i| 4 + i * 7 / 2) - .into_iter() - .collect(); - for offset in &[0u32, 1u32, 2u32] { - let encoded_data = encoder.compress_vint_sorted(&input, *offset); - assert_eq!(encoded_data.len(), expected_length); - let mut decoder = BlockDecoder::new(); - let remaining_data = decoder.uncompress_vint_sorted(&encoded_data, *offset, input.len()); - assert_eq!(0, remaining_data.len()); - assert_eq!(input, decoder.output_array()); - } - } - { - let mut encoder = BlockEncoder::new(); - let input = vec!(3u32, 17u32, 187u32); - let encoded_data = encoder.compress_vint_sorted(&input, 0); - assert_eq!(encoded_data.len(), 4); - assert_eq!(encoded_data[0], 3u8 + 128u8); - assert_eq!(encoded_data[1], (17u8 - 3u8) + 128u8); - assert_eq!(encoded_data[2], (187u8 - 17u8 - 128u8)); - assert_eq!(encoded_data[3], (1u8 + 128u8)); - } - } - - - #[bench] - fn bench_compress(b: &mut Bencher) { - let mut encoder = BlockEncoder::new(); - let data = generate_array(NUM_DOCS_PER_BLOCK, 0.1); - b.iter(|| { - encoder.compress_block_sorted(&data, 0u32); - }); - } - - #[bench] - fn bench_uncompress(b: &mut Bencher) { - let mut encoder = BlockEncoder::new(); - let data = generate_array(NUM_DOCS_PER_BLOCK, 0.1); - let compressed = encoder.compress_block_sorted(&data, 0u32); - let mut decoder = BlockDecoder::new(); - b.iter(|| { - decoder.uncompress_block_sorted(compressed, 0u32); - }); - } - -} - - - - \ No newline at end of file diff --git a/src/compression/compression_nosimd.rs b/src/compression/compression_nosimd.rs new file mode 100644 index 000000000..501a4e1fe --- /dev/null +++ b/src/compression/compression_nosimd.rs @@ -0,0 +1,127 @@ +use common::bitpacker::compute_num_bits; +use common::bitpacker::{BitPacker, BitUnpacker}; +use std::cmp; +use std::io::Write; +use super::NUM_DOCS_PER_BLOCK; + +const COMPRESSED_BLOCK_MAX_SIZE: usize = NUM_DOCS_PER_BLOCK * 4 + 1; + +pub fn compress_sorted(vals: &mut [u32], mut output: &mut [u8], offset: u32) -> usize { + let mut max_delta = 0; + { + let mut local_offset = offset; + for i in 0..NUM_DOCS_PER_BLOCK { + let val = vals[i]; + let delta = val - local_offset; + max_delta = cmp::max(max_delta, delta); + vals[i] = delta; + local_offset = val; + } + } + let num_bits = compute_num_bits(max_delta); + output.write_all(&[num_bits]).unwrap(); + let mut bit_packer = BitPacker::new(num_bits as usize); + for val in vals { + bit_packer.write(*val, &mut output).unwrap(); + } + 1 + bit_packer.close(&mut output).expect("packing in memory should never fail") +} + + + +pub struct BlockEncoder { + pub output: [u8; COMPRESSED_BLOCK_MAX_SIZE], + pub output_len: usize, + input_buffer: [u32; NUM_DOCS_PER_BLOCK], +} + +impl BlockEncoder { + + pub fn new() -> BlockEncoder { + BlockEncoder { + output: [0u8; COMPRESSED_BLOCK_MAX_SIZE], + output_len: 0, + input_buffer: [0u32; NUM_DOCS_PER_BLOCK], + } + } + + pub fn compress_block_sorted(&mut self, vals: &[u32], offset: u32) -> &[u8] { + self.input_buffer.clone_from_slice(vals); + let compressed_size = compress_sorted(&mut self.input_buffer, &mut self.output, offset); + &self.output[..compressed_size] + } + + pub fn compress_block_unsorted(&mut self, vals: &[u32]) -> &[u8] { + let compressed_size: usize = { + let mut output: &mut [u8] = &mut self.output; + let max = vals.iter().cloned().max().expect("compress unsorted called with an empty array"); + let num_bits = compute_num_bits(max); + output.write_all(&[num_bits]).unwrap(); + let mut bit_packer = BitPacker::new(num_bits as usize); + for val in vals { + bit_packer.write(*val, &mut output).unwrap(); + } + 1 + bit_packer.close(&mut output).expect("packing in memory should never fail") + }; + &self.output[..compressed_size] + } + +} + +pub struct BlockDecoder { + pub output: [u32; COMPRESSED_BLOCK_MAX_SIZE], + pub output_len: usize, +} + + +impl BlockDecoder { + pub fn new() -> BlockDecoder { + BlockDecoder::with_val(0u32) + } + + pub fn with_val(val: u32) -> BlockDecoder { + BlockDecoder { + output: [val; COMPRESSED_BLOCK_MAX_SIZE], + output_len: 0, + } + } + + pub fn uncompress_block_sorted<'a>(&mut self, compressed_data: &'a [u8], mut offset: u32) -> &'a[u8] { + let consumed_size = { + let num_bits = compressed_data[0]; + let bit_unpacker = BitUnpacker::new(&compressed_data[1..], num_bits as usize); + for i in 0..NUM_DOCS_PER_BLOCK { + let delta = bit_unpacker.get(i); + let val = offset + delta; + self.output[i] = val; + offset = val; + } + 1 + (num_bits as usize * NUM_DOCS_PER_BLOCK + 7) / 8 + }; + self.output_len = NUM_DOCS_PER_BLOCK; + &compressed_data[consumed_size..] + } + + pub fn uncompress_block_unsorted<'a>(&mut self, compressed_data: &'a [u8]) -> &'a[u8] { + let num_bits = compressed_data[0]; + let bit_unpacker = BitUnpacker::new(&compressed_data[1..], num_bits as usize); + for i in 0..NUM_DOCS_PER_BLOCK { + self.output[i] = bit_unpacker.get(i); + } + let consumed_size = 1 + (num_bits as usize * NUM_DOCS_PER_BLOCK + 7) / 8; + self.output_len = NUM_DOCS_PER_BLOCK; + &compressed_data[consumed_size..] + } + + #[inline] + pub fn output_array(&self,) -> &[u32] { + &self.output[..self.output_len] + } + + #[inline] + pub fn output(&self, idx: usize) -> u32 { + self.output[idx] + } +} + + diff --git a/src/compression/compression_simd.rs b/src/compression/compression_simd.rs new file mode 100644 index 000000000..605d0ec03 --- /dev/null +++ b/src/compression/compression_simd.rs @@ -0,0 +1,112 @@ + +use super::NUM_DOCS_PER_BLOCK; + +use libc::size_t; + +const COMPRESSED_BLOCK_MAX_SIZE: usize = NUM_DOCS_PER_BLOCK * 4 + 1; + +extern { + fn compress_sorted_cpp( + data: *const u32, + output: *mut u8, + offset: u32) -> size_t; + + fn uncompress_sorted_cpp( + compressed_data: *const u8, + output: *mut u32, + offset: u32) -> size_t; + + fn compress_unsorted_cpp( + data: *const u32, + output: *mut u8) -> size_t; + + fn uncompress_unsorted_cpp( + compressed_data: *const u8, + output: *mut u32) -> size_t; +} + +fn compress_sorted(vals: &[u32], output: &mut [u8], offset: u32) -> usize { + unsafe { compress_sorted_cpp(vals.as_ptr(), output.as_mut_ptr(), offset) } +} + +fn uncompress_sorted(compressed_data: &[u8], output: &mut [u32], offset: u32) -> usize { + unsafe { uncompress_sorted_cpp(compressed_data.as_ptr(), output.as_mut_ptr(), offset) } +} + +fn compress_unsorted(vals: &[u32], output: &mut [u8]) -> usize { + unsafe { compress_unsorted_cpp(vals.as_ptr(), output.as_mut_ptr()) } +} + +fn uncompress_unsorted(compressed_data: &[u8], output: &mut [u32]) -> usize { + unsafe { uncompress_unsorted_cpp(compressed_data.as_ptr(), output.as_mut_ptr()) } +} + + +pub struct BlockEncoder { + pub output: [u8; COMPRESSED_BLOCK_MAX_SIZE], + pub output_len: usize, +} + +impl BlockEncoder { + + pub fn new() -> BlockEncoder { + BlockEncoder { + output: [0u8; COMPRESSED_BLOCK_MAX_SIZE], + output_len: 0, + } + } + + pub fn compress_block_sorted(&mut self, vals: &[u32], offset: u32) -> &[u8] { + let compressed_size = compress_sorted(vals, &mut self.output, offset); + &self.output[..compressed_size] + } + + pub fn compress_block_unsorted(&mut self, vals: &[u32]) -> &[u8] { + let compressed_size = compress_unsorted(vals, &mut self.output); + &self.output[..compressed_size] + } + +} + +pub struct BlockDecoder { + pub output: [u32; COMPRESSED_BLOCK_MAX_SIZE], + pub output_len: usize, +} + + +impl BlockDecoder { + pub fn new() -> BlockDecoder { + BlockDecoder::with_val(0u32) + } + + pub fn with_val(val: u32) -> BlockDecoder { + BlockDecoder { + output: [val; COMPRESSED_BLOCK_MAX_SIZE], + output_len: 0, + } + } + + pub fn uncompress_block_sorted<'a>(&mut self, compressed_data: &'a [u8], offset: u32) -> &'a[u8] { + let consumed_size = uncompress_sorted(compressed_data, &mut self.output, offset); + self.output_len = NUM_DOCS_PER_BLOCK; + &compressed_data[consumed_size..] + } + + pub fn uncompress_block_unsorted<'a>(&mut self, compressed_data: &'a [u8]) -> &'a[u8] { + let consumed_size = uncompress_unsorted(compressed_data, &mut self.output); + self.output_len = NUM_DOCS_PER_BLOCK; + &compressed_data[consumed_size..] + } + + #[inline] + pub fn output_array(&self,) -> &[u32] { + &self.output[..self.output_len] + } + + #[inline] + pub fn output(&self, idx: usize) -> u32 { + self.output[idx] + } +} + + diff --git a/src/compression/mod.rs b/src/compression/mod.rs index cc4dd31c8..7de8ef558 100644 --- a/src/compression/mod.rs +++ b/src/compression/mod.rs @@ -1,11 +1,130 @@ #![allow(dead_code)] -mod compression; -pub use self::compression::{BlockEncoder, BlockDecoder}; mod composite; -pub use self::composite::CompositeEncoder; -pub use self::composite::CompositeDecoder; +pub use self::composite::{CompositeEncoder, CompositeDecoder}; + +#[cfg(feature="simdcompression")] +mod compression_simd; +#[cfg(feature="simdcompression")] +pub use self::compression_simd::{BlockEncoder, BlockDecoder}; + + +#[cfg(not(feature="simdcompression"))] +mod compression_nosimd; +#[cfg(not(feature="simdcompression"))] +pub use self::compression_nosimd::{BlockEncoder, BlockDecoder}; + + +pub trait VIntEncoder { + fn compress_vint_sorted(&mut self, input: &[u32], offset: u32) -> &[u8]; + fn compress_vint_unsorted(&mut self, input: &[u32]) -> &[u8]; +} + +pub trait VIntDecoder { + fn uncompress_vint_sorted<'a>(&mut self, compressed_data: &'a [u8], offset: u32, num_els: usize) -> &'a [u8]; + fn uncompress_vint_unsorted<'a>(&mut self, compressed_data: &'a [u8], num_els: usize) -> &'a [u8]; +} + +impl VIntEncoder for BlockEncoder{ + + fn compress_vint_sorted(&mut self, input: &[u32], mut offset: u32) -> &[u8] { + let mut byte_written = 0; + for &v in input { + let mut to_encode: u32 = v - offset; + offset = v; + loop { + let next_byte: u8 = (to_encode % 128u32) as u8; + to_encode /= 128u32; + if to_encode == 0u32 { + self.output[byte_written] = next_byte | 128u8; + byte_written += 1; + break; + } + else { + self.output[byte_written] = next_byte; + byte_written += 1; + } + } + } + &self.output[..byte_written] + } + + fn compress_vint_unsorted(&mut self, input: &[u32]) -> &[u8] { + let mut byte_written = 0; + for &v in input { + let mut to_encode: u32 = v; + loop { + let next_byte: u8 = (to_encode % 128u32) as u8; + to_encode /= 128u32; + if to_encode == 0u32 { + self.output[byte_written] = next_byte | 128u8; + byte_written += 1; + break; + } + else { + self.output[byte_written] = next_byte; + byte_written += 1; + } + } + } + &self.output[..byte_written] + } +} + +impl VIntDecoder for BlockDecoder { + + fn uncompress_vint_sorted<'a>( + &mut self, + compressed_data: &'a [u8], + offset: u32, + num_els: usize) -> &'a [u8] { + let mut read_byte = 0; + let mut result = offset; + for i in 0..num_els { + let mut shift = 0u32; + loop { + let cur_byte = compressed_data[read_byte]; + read_byte += 1; + result += ((cur_byte % 128u8) as u32) << shift; + if cur_byte & 128u8 != 0u8 { + break; + } + shift += 7; + } + self.output[i] = result; + } + self.output_len = num_els; + &compressed_data[read_byte..] + } + + fn uncompress_vint_unsorted<'a>( + &mut self, + compressed_data: &'a [u8], + num_els: usize) -> &'a [u8] { + let mut read_byte = 0; + for i in 0..num_els { + let mut result = 0u32; + let mut shift = 0u32; + loop { + let cur_byte = compressed_data[read_byte]; + read_byte += 1; + result += ((cur_byte % 128u8) as u32) << shift; + if cur_byte & 128u8 != 0u8 { + break; + } + shift += 7; + } + self.output[i] = result; + } + self.output_len = num_els; + &compressed_data[read_byte..] + } + +} + + + pub const NUM_DOCS_PER_BLOCK: usize = 128; //< should be a power of 2 to let the compiler optimize. @@ -15,7 +134,9 @@ pub mod tests { use rand::Rng; use rand::SeedableRng; use rand::XorShiftRng; - + use super::*; + use test::Bencher; + fn generate_array_with_seed(n: usize, ratio: f32, seed_val: u32) -> Vec { let seed: &[u32; 4] = &[1, 2, 3, seed_val]; let mut rng: XorShiftRng = XorShiftRng::from_seed(*seed); @@ -28,6 +149,127 @@ pub mod tests { pub fn generate_array(n: usize, ratio: f32) -> Vec { generate_array_with_seed(n, ratio, 4) } + + #[test] + fn test_encode_sorted_block() { + let vals: Vec = (0u32..128u32).map(|i| i*7).collect(); + let mut encoder = BlockEncoder::new(); + let compressed_data = encoder.compress_block_sorted(&vals, 0); + let mut decoder = BlockDecoder::new(); + { + let remaining_data = decoder.uncompress_block_sorted(compressed_data, 0); + assert_eq!(remaining_data.len(), 0); + } + for i in 0..128 { + assert_eq!(vals[i], decoder.output(i)); + } + } + + #[test] + fn test_encode_sorted_block_with_offset() { + let vals: Vec = (0u32..128u32).map(|i| 11 + i*7).collect(); + let mut encoder = BlockEncoder::new(); + let compressed_data = encoder.compress_block_sorted(&vals, 10); + let mut decoder = BlockDecoder::new(); + { + let remaining_data = decoder.uncompress_block_sorted(compressed_data, 10); + assert_eq!(remaining_data.len(), 0); + } + for i in 0..128 { + assert_eq!(vals[i], decoder.output(i)); + } + } + + #[test] + fn test_encode_sorted_block_with_junk() { + let mut compressed: Vec = Vec::new(); + let n = 128; + let vals: Vec = (0..n).map(|i| 11u32 + (i as u32)*7u32).collect(); + let mut encoder = BlockEncoder::new(); + let compressed_data = encoder.compress_block_sorted(&vals, 10); + compressed.extend_from_slice(compressed_data); + compressed.push(173u8); + let mut decoder = BlockDecoder::new(); + { + let remaining_data = decoder.uncompress_block_sorted(&compressed, 10); + assert_eq!(remaining_data.len(), 1); + assert_eq!(remaining_data[0], 173u8); + } + for i in 0..n { + assert_eq!(vals[i], decoder.output(i)); + } + } + + #[test] + fn test_encode_unsorted_block_with_junk() { + let mut compressed: Vec = Vec::new(); + let n = 128; + let vals: Vec = (0..n).map(|i| 11u32 + (i as u32)*7u32 % 12).collect(); + let mut encoder = BlockEncoder::new(); + let compressed_data = encoder.compress_block_unsorted(&vals); + compressed.extend_from_slice(compressed_data); + compressed.push(173u8); + let mut decoder = BlockDecoder::new(); + { + let remaining_data = decoder.uncompress_block_unsorted(&compressed); + assert_eq!(remaining_data.len(), 1); + assert_eq!(remaining_data[0], 173u8); + } + for i in 0..n { + assert_eq!(vals[i], decoder.output(i)); + } + } + + + #[test] + fn test_encode_vint() { + { + let expected_length = 123; + let mut encoder = BlockEncoder::new(); + let input: Vec = (0u32..123u32) + .map(|i| 4 + i * 7 / 2) + .into_iter() + .collect(); + for offset in &[0u32, 1u32, 2u32] { + let encoded_data = encoder.compress_vint_sorted(&input, *offset); + assert_eq!(encoded_data.len(), expected_length); + let mut decoder = BlockDecoder::new(); + let remaining_data = decoder.uncompress_vint_sorted(&encoded_data, *offset, input.len()); + assert_eq!(0, remaining_data.len()); + assert_eq!(input, decoder.output_array()); + } + } + { + let mut encoder = BlockEncoder::new(); + let input = vec!(3u32, 17u32, 187u32); + let encoded_data = encoder.compress_vint_sorted(&input, 0); + assert_eq!(encoded_data.len(), 4); + assert_eq!(encoded_data[0], 3u8 + 128u8); + assert_eq!(encoded_data[1], (17u8 - 3u8) + 128u8); + assert_eq!(encoded_data[2], (187u8 - 17u8 - 128u8)); + assert_eq!(encoded_data[3], (1u8 + 128u8)); + } + } + + + #[bench] + fn bench_compress(b: &mut Bencher) { + let mut encoder = BlockEncoder::new(); + let data = generate_array(NUM_DOCS_PER_BLOCK, 0.1); + b.iter(|| { + encoder.compress_block_sorted(&data, 0u32); + }); + } + + #[bench] + fn bench_uncompress(b: &mut Bencher) { + let mut encoder = BlockEncoder::new(); + let data = generate_array(NUM_DOCS_PER_BLOCK, 0.1); + let compressed = encoder.compress_block_sorted(&data, 0u32); + let mut decoder = BlockDecoder::new(); + b.iter(|| { + decoder.uncompress_block_sorted(compressed, 0u32); + }); + } + } - - diff --git a/src/core/index.rs b/src/core/index.rs index 46b8a5c75..d29cc1fc3 100644 --- a/src/core/index.rs +++ b/src/core/index.rs @@ -19,6 +19,7 @@ use indexer::SegmentManager; use core::IndexMeta; use core::META_FILEPATH; use super::segment::create_segment; +use indexer::segment_updater::save_new_metas; const NUM_SEARCHERS: usize = 12; @@ -37,10 +38,6 @@ fn load_metas(directory: &Directory) -> Result { .map_err(|e| Error::CorruptedFile(META_FILEPATH.clone(), Box::new(e))) } -// pub fn set_metas(index: &mut Index, docstamp: u64) { -// index.docstamp = docstamp; -// } - /// Tantivy's Search Index pub struct Index { segment_manager: Arc, @@ -62,11 +59,13 @@ impl Index { } /// Creates a new index in a given filepath. - /// /// The index will use the `MMapDirectory`. + /// + /// If a previous index was in this directory, then its meta file will be destroyed. pub fn create(directory_path: &Path, schema: Schema) -> Result { - let directory = Box::new(try!(MmapDirectory::open(directory_path))); - Index::from_directory(directory, schema) + let mut directory = MmapDirectory::open(directory_path)?; + save_new_metas(schema.clone(), 0, &mut directory)?; + Index::from_directory(box directory, schema) } /// Creates a new index in a temp directory. diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 81d8105f1..804e4d8a5 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -7,7 +7,7 @@ mod log_merge_policy; mod segment_register; mod segment_writer; mod segment_manager; -mod segment_updater; +pub mod segment_updater; mod directory_lock; pub use self::segment_serializer::SegmentSerializer; diff --git a/src/indexer/segment_updater.rs b/src/indexer/segment_updater.rs index 2e8a190c2..89b66f5e7 100644 --- a/src/indexer/segment_updater.rs +++ b/src/indexer/segment_updater.rs @@ -40,6 +40,25 @@ fn create_metas(segment_manager: &SegmentManager, schema: Schema, docstamp: u64) } +/// Save the index meta file. +/// This operation is atomic : +/// Either +// - it fails, in which case an error is returned, +/// and the `meta.json` remains untouched, +/// - it success, and `meta.json` is written +/// and flushed. +/// +/// This method is not part of tantivy's public API +pub fn save_new_metas(schema: Schema, + docstamp: u64, + directory: &mut Directory) + -> Result<()> { + let segment_manager = SegmentManager::from_segments(Vec::new()); + save_metas(&segment_manager, schema, docstamp, directory) +} + + + /// Save the index meta file. /// This operation is atomic : /// Either diff --git a/src/postings/freq_handler.rs b/src/postings/freq_handler.rs index 78d15bf20..30efa08ac 100644 --- a/src/postings/freq_handler.rs +++ b/src/postings/freq_handler.rs @@ -1,7 +1,7 @@ use compression::BlockDecoder; use common::VInt; use common::BinarySerializable; -use compression::CompositeDecoder; +use compression::{CompositeDecoder, VIntDecoder}; use postings::SegmentPostingsOption; use compression::NUM_DOCS_PER_BLOCK; diff --git a/src/postings/segment_postings.rs b/src/postings/segment_postings.rs index da81fb0b1..fb313e76e 100644 --- a/src/postings/segment_postings.rs +++ b/src/postings/segment_postings.rs @@ -1,4 +1,4 @@ -use compression::{NUM_DOCS_PER_BLOCK, BlockDecoder}; +use compression::{NUM_DOCS_PER_BLOCK, BlockDecoder, VIntDecoder}; use DocId; use postings::{Postings, FreqHandler, DocSet, HasLen}; use std::num::Wrapping; diff --git a/src/postings/serializer.rs b/src/postings/serializer.rs index 9ba558509..afbfbe948 100644 --- a/src/postings/serializer.rs +++ b/src/postings/serializer.rs @@ -14,6 +14,7 @@ use core::Segment; use std::io; use core::SegmentComponent; use std::io::Write; +use compression::VIntEncoder; use common::VInt; use common::BinarySerializable;