Skip to content

Commit e69fcc6

Browse files
fulmicotonfulmicoton
committed
blop
1 parent e5ef168 commit e69fcc6

9 files changed

Lines changed: 224 additions & 180 deletions

File tree

‎src/lib.rs‎

Lines changed: 29 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
use std::borrow::Cow;
21
use std::sync::Arc;
32

43
mod block_read_write;
@@ -12,23 +11,46 @@ mod record;
1211
mod recordlog;
1312
mod rolling;
1413

14+
pub use mem::PagesBuf;
1515
pub use multi_record_log::{MultiRecordLog, SyncPolicy};
1616

17-
#[derive(Debug, PartialEq, Eq)]
17+
#[derive(Debug)]
1818
pub struct Record<'a> {
1919
pub position: u64,
20-
pub payload: Cow<'a, [u8]>,
20+
payload: PagesBuf<'a>,
2121
}
2222

23+
2324
impl<'a> Record<'a> {
24-
pub fn new(position: u64, payload: &'a [u8]) -> Self {
25-
Record {
26-
position,
27-
payload: Cow::Borrowed(payload),
25+
#[cfg(test)]
26+
pub fn payload_equal(&self, mut payload: &[u8]) -> bool {
27+
use bytes::Buf;
28+
let mut self_payload = self.payload;
29+
if self_payload.remaining() != payload.len() {
30+
return false;
2831
}
32+
while self_payload.has_remaining() {
33+
let chunk = self_payload.chunk();
34+
let chunk_len = chunk.len();
35+
if chunk != &payload[..chunk_len] {
36+
return false;
37+
}
38+
self_payload.advance(chunk_len);
39+
payload = &payload[chunk_len..];
40+
}
41+
true
2942
}
3043
}
3144

45+
// impl<'a> Record<'a> {
46+
// pub fn new(position: u64, payload: &'a [u8]) -> Self {
47+
// Record {
48+
// position,
49+
// payload: Cow::Borrowed(payload),
50+
// }
51+
// }
52+
// }
53+
3254
#[derive(Clone, Default, Debug, Ord, PartialOrd, Eq, PartialEq)]
3355
pub struct FileNumber {
3456
file_number: Arc<u64>,

‎src/mem/arena.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -194,7 +194,7 @@ mod tests {
194194
arena.release_page(PageId(1));
195195
assert_eq!(arena.num_allocated_pages(), 2);
196196
arena.gc();
197-
assert_eq!(arena.num_allocated_pages(), 1);
197+
assert_eq!(arena.num_allocated_pages(), 2);
198198
assert_eq!(arena.get_page_id(), PageId(1));
199199
assert_eq!(arena.num_allocated_pages(), 2);
200200
}

‎src/mem/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ use self::arena::{Arena, PAGE_SIZE};
77
pub(crate) use self::queue::MemQueue;
88
pub(crate) use self::queues::MemQueues;
99
use self::rolling_buffer::RollingBuffer;
10+
pub use self::rolling_buffer::PagesBuf;
1011

1112
#[cfg(test)]
1213
mod tests;

‎src/mem/queue.rs‎

Lines changed: 25 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,4 @@
1-
use std::borrow::Cow;
21
use std::ops::{Bound, RangeBounds};
3-
4-
use bytes::buf::Buf;
5-
62
use crate::error::AppendError;
73
use crate::mem::{Arena, RollingBuffer};
84
use crate::{FileNumber, Record};
@@ -24,21 +20,19 @@ pub(crate) struct MemQueue {
2420
record_metas: Vec<RecordMeta>,
2521
}
2622

27-
fn concatenate_buffers<'a>(mut buffers: impl Iterator<Item = &'a [u8]>) -> Cow<'a, [u8]> {
28-
let Some(first) = buffers.next() else {
29-
return Cow::Borrowed(&[]);
30-
};
31-
let Some(second) = buffers.next() else {
32-
return Cow::Borrowed(first);
33-
};
34-
let mut concatenated_buffer: Vec<u8> = Vec::with_capacity(first.len() + second.len());
35-
concatenated_buffer.extend_from_slice(first);
36-
concatenated_buffer.extend_from_slice(second);
37-
for buffer in buffers {
38-
concatenated_buffer.extend_from_slice(buffer);
39-
}
40-
Cow::Owned(concatenated_buffer)
41-
}
23+
// fn concatenate_buffers<'a>(mut buf: impl Buf + 'a) -> Cow<'a, [u8]> {
24+
// let first_chunk: &'a buf = buf.chunk();
25+
// if buf.remaining() == first_chunk.len() {
26+
// return Cow::Borrowed(first_chunk);
27+
// }
28+
// let mut concatenated_buffer: Vec<u8> = Vec::with_capacity(buf.remaining());
29+
// while buf.has_remaining() {
30+
// let chunk = buf.chunk();
31+
// concatenated_buffer.extend_from_slice(chunk);
32+
// buf.advance(chunk.len());
33+
// }
34+
// Cow::Owned(concatenated_buffer)
35+
// }
4236

4337
impl MemQueue {
4438
pub fn with_next_position(next_position: u64) -> Self {
@@ -59,15 +53,14 @@ impl MemQueue {
5953
}
6054

6155
/// Returns the last record stored in the queue.
62-
pub fn last_record<'a>(&self, arena: &'a Arena) -> Option<Record<'a>> {
56+
pub fn last_record<'a>(&'a self, arena: &'a Arena) -> Option<Record<'a>> {
6357
let record = self.record_metas.last()?;
64-
let buf_iter = self
58+
let record_payload = self
6559
.concatenated_records
66-
.get_range(record.start_offset.., arena);
67-
let payload = concatenate_buffers(buf_iter);
60+
.get_range_buf(record.start_offset.., arena);
6861
Some(Record {
6962
position: record.position,
70-
payload,
63+
payload: record_payload,
7164
})
7265
}
7366

@@ -156,10 +149,10 @@ impl MemQueue {
156149
} else {
157150
Bound::Unbounded
158151
};
159-
let payload_iter = self
152+
let payload= self
160153
.concatenated_records
161-
.get_range((start_bound, end_bound), arena);
162-
let payload = concatenate_buffers(payload_iter);
154+
.get_range_buf((start_bound, end_bound), arena);
155+
// let payload = concatenate_buffers(payload_buf);
163156
Record { position, payload }
164157
})
165158
}
@@ -168,7 +161,7 @@ impl MemQueue {
168161
///
169162
/// If truncating to a future position, make the queue go forward to that position.
170163
/// Return the number of record removed.
171-
pub fn truncate(&mut self, truncate_up_to_pos: u64, arena: &mut Arena) -> usize {
164+
pub fn truncate_up_to_included(&mut self, truncate_up_to_pos: u64, arena: &mut Arena) -> usize {
172165
if self.start_position > truncate_up_to_pos {
173166
return 0;
174167
}
@@ -185,11 +178,11 @@ impl MemQueue {
185178

186179
let start_offset_to_keep: usize = self.record_metas[first_record_to_keep].start_offset;
187180
self.record_metas.drain(..first_record_to_keep);
188-
for record_meta in &mut self.record_metas {
189-
record_meta.start_offset -= start_offset_to_keep;
190-
}
181+
// for record_meta in &mut self.record_metas {
182+
// record_meta.start_offset -= start_offset_to_keep;
183+
// }
191184
self.concatenated_records
192-
.truncate_to(start_offset_to_keep, arena);
185+
.truncate_up_to_included(start_offset_to_keep, arena);
193186
self.start_position = truncate_up_to_pos + 1;
194187
first_record_to_keep
195188
}

‎src/mem/queues.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ impl MemQueues {
157157
/// not do anything.
158158
pub fn truncate(&mut self, queue_id: &str, position: u64) -> Option<usize> {
159159
let queue = self.queues.get_mut(queue_id)?;
160-
Some(queue.truncate(position, &mut self.arena))
160+
Some(queue.truncate_up_to_included(position, &mut self.arena))
161161
}
162162

163163
/// Return a tuple of (size, capacity) of memory used by the memqueues

‎src/mem/rolling_buffer.rs‎

Lines changed: 73 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -44,17 +44,15 @@ impl RollingBuffer {
4444
}
4545

4646
/// Truncate the buffer, dropping the first `truncate_len` bytes.
47-
pub fn truncate_to(&mut self, truncate_len: usize, arena: &mut Arena) {
47+
pub fn truncate_up_to_included(&mut self, truncate_len: usize, arena: &mut Arena) {
4848
if truncate_len >= self.range.len() {
4949
self.clear(arena);
5050
return;
5151
}
5252
let num_pages = num_pages_required((self.range.start + truncate_len)..self.range.end);
5353
assert!(num_pages <= self.page_ids.len());
5454
let num_pages_to_drop = self.page_ids.len() - num_pages;
55-
let new_start = self.range.start + truncate_len - num_pages_to_drop * PAGE_SIZE;
56-
let new_end = self.range.end - num_pages_to_drop * PAGE_SIZE;
57-
self.range = new_start..new_end;
55+
self.range.start = self.range.start + truncate_len;
5856
if num_pages_to_drop > 0 {
5957
for page_id in self.page_ids.drain(..(self.page_ids.len() - num_pages)) {
6058
arena.release_page(page_id);
@@ -92,26 +90,35 @@ impl RollingBuffer {
9290
&'slf self,
9391
range: Range<usize>,
9492
arena: &'a Arena,
95-
) -> impl Buf + 'slf {
96-
let start = (self.range.start + range.start).min(self.range.end);
97-
let end = (self.range.start + range.end).clamp(start, self.range.end);
98-
let len = end - start;
99-
let skip_pages = start / PAGE_SIZE;
93+
) -> PagesBuf<'slf> {
94+
let Range {start, end} = range;
95+
assert!(start >= self.range.start);
96+
assert!(end <= self.range.end);
97+
if end <= start {
98+
return PagesBuf {
99+
arena,
100+
start_offset: 0,
101+
page_ids: &[],
102+
remaining_len: 0,
103+
};
104+
}
105+
let start_page_id = start / PAGE_SIZE;
106+
let start_inner_page_id = self.range.start / PAGE_SIZE;
107+
let skip_pages = start_page_id - start_inner_page_id;
100108
let start_offset = start % PAGE_SIZE;
101109
PagesBuf {
102110
arena,
103111
start_offset,
104112
page_ids: &self.page_ids[skip_pages..],
105-
remaining_len: len,
113+
remaining_len: end - start,
106114
}
107-
.take(len)
108115
}
109116

110117
pub fn get_range_buf<'slf, 'a: 'slf, R>(
111118
&'slf self,
112119
range_bounds: R,
113120
arena: &'a Arena,
114-
) -> impl Buf + 'slf
121+
) -> PagesBuf<'slf>
115122
where
116123
R: RangeBounds<usize> + 'static,
117124
{
@@ -123,69 +130,49 @@ impl RollingBuffer {
123130
let end = match range_bounds.end_bound() {
124131
Bound::Included(pos) => pos + 1,
125132
Bound::Excluded(pos) => *pos,
126-
Bound::Unbounded => self.len(),
133+
Bound::Unbounded => self.range.end
127134
};
128135
self.get_range_buf_aux(start..end, arena)
129136
}
130-
131-
pub fn get_range<'slf, 'a: 'slf, R>(
132-
&'slf self,
133-
range_bounds: R,
134-
arena: &'a Arena,
135-
) -> impl Iterator<Item = &'a [u8]> + 'slf
136-
where
137-
R: RangeBounds<usize> + 'static,
138-
{
139-
let start = match range_bounds.start_bound() {
140-
Bound::Included(pos) => *pos,
141-
Bound::Excluded(pos) => pos + 1,
142-
Bound::Unbounded => 0,
143-
};
144-
let end = match range_bounds.end_bound() {
145-
Bound::Included(pos) => pos + 1,
146-
Bound::Excluded(pos) => *pos,
147-
Bound::Unbounded => self.len(),
148-
};
149-
self.get_range_aux(start..end, arena)
150-
}
151-
152-
pub fn get_range_aux<'slf, 'a: 'slf>(
153-
&'slf self,
154-
range: Range<usize>,
155-
arena: &'a Arena,
156-
) -> impl Iterator<Item = &'a [u8]> + 'slf {
157-
let start = (self.range.start + range.start).min(self.range.end);
158-
let end = (self.range.start + range.end).clamp(start, self.range.end);
159-
let mut remaining_len = end - start;
160-
let skip_pages = start / PAGE_SIZE;
161-
let mut start_in_page = start % PAGE_SIZE;
162-
self.page_ids[skip_pages..]
163-
.iter()
164-
.copied()
165-
.map(move |page_id| {
166-
let page = &arena.page(page_id)[start_in_page..];
167-
start_in_page = 0;
168-
let page_len = page.len().min(remaining_len);
169-
remaining_len -= page_len;
170-
&page[..page_len]
171-
})
172-
.take_while(|page| !page.is_empty())
173-
}
174137
}
175138

176-
struct PagesBuf<'a> {
139+
#[derive(Clone, Copy)]
140+
pub struct PagesBuf<'a> {
177141
arena: &'a Arena,
178142
page_ids: &'a [PageId],
179143
start_offset: usize,
180144
remaining_len: usize,
181145
}
182146

183-
impl<'a> Buf for PagesBuf<'a> {
184-
fn remaining(&self) -> usize {
185-
self.remaining_len
147+
impl<'a> std::fmt::Debug for PagesBuf<'a> {
148+
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
149+
let mut buf = self.clone();
150+
let mut vec = Vec::with_capacity(self.remaining_len);
151+
while buf.has_remaining() {
152+
let chunk = buf.chunk();
153+
vec.extend_from_slice(chunk);
154+
buf.advance(chunk.len());
155+
}
156+
vec.as_slice().fmt(f)
186157
}
158+
}
187159

188-
fn chunk(&self) -> &[u8] {
160+
impl<'a> PagesBuf<'a> {
161+
pub fn to_cow(mut self) -> std::borrow::Cow<'a, [u8]> {
162+
if self.page_ids.len() <= 1 {
163+
let chunk = self.chunk_with_lifetime();
164+
return std::borrow::Cow::Borrowed(chunk);
165+
}
166+
let mut buf = Vec::with_capacity(self.remaining_len);
167+
while self.has_remaining() {
168+
let chunk = self.chunk_with_lifetime();
169+
buf.extend_from_slice(chunk);
170+
self.advance(chunk.len());
171+
}
172+
std::borrow::Cow::Owned(buf)
173+
}
174+
175+
fn chunk_with_lifetime(&self) -> &'a [u8] {
189176
let Some(first_page_id) = self.page_ids.first().copied() else {
190177
return &[];
191178
};
@@ -197,6 +184,26 @@ impl<'a> Buf for PagesBuf<'a> {
197184
}
198185
}
199186

187+
}
188+
189+
impl<'a> Buf for PagesBuf<'a> {
190+
fn remaining(&self) -> usize {
191+
self.remaining_len
192+
}
193+
194+
fn chunk(&self) -> &[u8] {
195+
// let Some(first_page_id) = self.page_ids.first().copied() else {
196+
// return &[];
197+
// };
198+
// let current_page = &self.arena.page(first_page_id)[self.start_offset..];
199+
// if current_page.len() > self.remaining_len {
200+
// &current_page[..self.remaining_len]
201+
// } else {
202+
// current_page
203+
// }
204+
self.chunk_with_lifetime()
205+
}
206+
200207
fn advance(&mut self, mut cnt: usize) {
201208
while cnt > 0 {
202209
if self.page_ids.is_empty() {
@@ -235,16 +242,15 @@ mod tests {
235242
let mut arena = Arena::default();
236243
let text = b"hello happy tax payer";
237244
for truncate_len in 0..text.len() {
238-
let expected = &text[truncate_len..];
239245
let mut rolling_buffer = RollingBuffer::new();
240246
rolling_buffer.extend_from_slice(&b"hello"[..], &mut arena);
241247
rolling_buffer.extend_from_slice(&b" happy"[..], &mut arena);
242248
rolling_buffer.extend_from_slice(&b" tax payer"[..], &mut arena);
243-
rolling_buffer.truncate_to(truncate_len, &mut arena);
244-
for start in 0..expected.len() {
245-
for end in start..expected.len() {
249+
rolling_buffer.truncate_up_to_included(truncate_len, &mut arena);
250+
for start in truncate_len..text.len() {
251+
for end in start..text.len() {
246252
let bytes: Vec<u8> = to_vec(rolling_buffer.get_range_buf(start..end, &arena));
247-
assert_eq!(&expected[start..end], &bytes[..]);
253+
assert_eq!(&text[start..end], &bytes[..]);
248254
}
249255
}
250256
}

0 commit comments

Comments
 (0)