| 1 | use crate::block::{ClientID, ID}; |
| 2 | use crate::block_store::BlockStore; |
| 3 | use crate::encoding::read::Error; |
| 4 | use crate::iter::TxnIterator; |
| 5 | use crate::slice::BlockSlice; |
| 6 | use crate::store::Store; |
| 7 | use crate::updates::decoder::{Decode, Decoder}; |
| 8 | use crate::updates::encoder::{Encode, Encoder}; |
| 9 | use crate::utils::client_hasher::ClientHasher; |
| 10 | use crate::ReadTxn; |
| 11 | use std::collections::hash_map::Entry; |
| 12 | use std::collections::HashMap; |
| 13 | use std::hash::{BuildHasherDefault, Hash, Hasher}; |
| 14 | use std::ops::Range; |
| 15 | // Note: use native Rust [Range](https://doc.rust-lang.org/std/ops/struct.Range.html) |
| 16 | // as it's left-inclusive/right-exclusive and defines the exact capabilities we care about here. |
| 17 | |
| 18 | impl Encode for Range<u32> { |
| 19 | fn encode<E: Encoder>(&self, encoder: &mut E) { |
| 20 | encoder.write_ds_clock(self.start); |
| 21 | encoder.write_ds_len(self.end - self.start) |
| 22 | } |
| 23 | } |
| 24 | |
| 25 | impl Decode for Range<u32> { |
| 26 | fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, Error> { |
| 27 | let clock = decoder.read_ds_clock()?; |
| 28 | let len = decoder.read_ds_len()?; |
| 29 | Ok(clock..(clock + len)) |
| 30 | } |
| 31 | } |
| 32 | |
| 33 | /// [IdRange] describes a single space of an [ID] clock values, belonging to the same client. |
| 34 | /// It can contain from a single continuous space, or multiple ones having "holes" between them. |
| 35 | #[derive(Clone, PartialEq, Eq, Hash)] |
| 36 | pub enum IdRange { |
| 37 | /// A single continuous range of clocks. |
| 38 | Continuous(Range<u32>), |
| 39 | /// A multiple ranges containing clock values, separated from each other by other clock ranges |
| 40 | /// not included in this [IdRange]. |
| 41 | Fragmented(Vec<Range<u32>>), |
| 42 | } |
| 43 | |
| 44 | impl IdRange { |
| 45 | pub fn with_capacity(capacity: usize) -> Self { |
| 46 | IdRange::Fragmented(Vec::with_capacity(capacity)) |
| 47 | } |
| 48 | |
| 49 | /// Check if range is empty (doesn't cover any clock space). |
| 50 | pub fn is_empty(&self) -> bool { |
| 51 | match self { |
| 52 | IdRange::Continuous(r) => r.start == r.end, |
| 53 | IdRange::Fragmented(rs) => rs.is_empty(), |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | /// Inverts current [IdRange], returning another [IdRange] that contains all |
| 58 | /// "holes" (ranges not included in current range). If current range is a continuous space |
| 59 | /// starting from the initial clock (eg. [0..5)), then returned range will be empty. |
| 60 | pub fn invert(&self) -> IdRange { |
| 61 | match self { |
| 62 | IdRange::Continuous(range) => IdRange::Continuous(0..range.start), |
| 63 | IdRange::Fragmented(ranges) => { |
| 64 | let mut inv = Vec::new(); |
| 65 | let mut start = 0; |
| 66 | for range in ranges.iter() { |
| 67 | if range.start > start { |
| 68 | inv.push(start..range.start); |
| 69 | } |
| 70 | start = range.end; |
| 71 | } |
| 72 | match inv.len() { |
| 73 | 0 => IdRange::Continuous(0..0), |
| 74 | 1 => IdRange::Continuous(inv[0].clone()), |
| 75 | _ => IdRange::Fragmented(inv), |
| 76 | } |
| 77 | } |
| 78 | } |
| 79 | } |
| 80 | |
| 81 | /// Check if given clock exists within current [IdRange]. |
| 82 | pub fn contains(&self, clock: u32) -> bool { |
| 83 | match self { |
| 84 | IdRange::Continuous(range) => range.contains(&clock), |
| 85 | IdRange::Fragmented(ranges) => ranges.iter().any(|r| r.contains(&clock)), |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | /// Iterate over ranges described by current [IdRange]. |
| 90 | pub fn iter(&self) -> IdRangeIter<'_> { |
| 91 | let (range, inner) = match self { |
| 92 | IdRange::Continuous(range) => (Some(range), None), |
| 93 | IdRange::Fragmented(ranges) => (None, Some(ranges.iter())), |
| 94 | }; |
| 95 | IdRangeIter { range, inner } |
| 96 | } |
| 97 | |
| 98 | fn push(&mut self, range: Range<u32>) { |
| 99 | match self { |
| 100 | IdRange::Continuous(r) => { |
| 101 | if r.end >= range.start { |
| 102 | if r.start > range.end { |
| 103 | *self = IdRange::Fragmented(vec![range, r.clone()]) |
| 104 | } else { |
| 105 | // two ranges overlap - merge them |
| 106 | r.end = range.end.max(r.end); |
| 107 | r.start = range.start.min(r.start); |
| 108 | } |
| 109 | } else { |
| 110 | *self = IdRange::Fragmented(vec![r.clone(), range]) |
| 111 | } |
| 112 | } |
| 113 | IdRange::Fragmented(ranges) => { |
| 114 | if ranges.is_empty() { |
| 115 | *self = IdRange::Continuous(range); |
| 116 | } else { |
| 117 | let last_idx = ranges.len() - 1; |
| 118 | let last = &mut ranges[last_idx]; |
| 119 | if !Self::try_join(last, &range) { |
| 120 | ranges.push(range); |
| 121 | } |
| 122 | } |
| 123 | } |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | /// Alters current [IdRange] by compacting its internal implementation (in fragmented case). |
| 128 | /// Example: fragmented space of [0,3), [3,5), [6,7) will be compacted into [0,5), [6,7). |
| 129 | fn squash(&mut self) { |
| 130 | if let IdRange::Fragmented(ranges) = self { |
| 131 | if !ranges.is_empty() { |
| 132 | ranges.sort_by(|a, b| a.start.cmp(&b.start)); |
| 133 | let mut new_len = 1; |
| 134 | |
| 135 | let len = ranges.len() as isize; |
| 136 | let head = ranges.as_mut_ptr(); |
| 137 | let mut current = unsafe { head.as_mut().unwrap() }; |
| 138 | let mut i = 1; |
| 139 | while i < len { |
| 140 | let next = unsafe { head.offset(i).as_ref().unwrap() }; |
| 141 | if !Self::try_join(current, next) { |
| 142 | // current and next are disjoined eg. [0,5) & [6,9) |
| 143 | |
| 144 | // move current pointer one index to the left: by using new_len we |
| 145 | // squash ranges possibly already merged to current |
| 146 | current = unsafe { head.offset(new_len).as_mut().unwrap() }; |
| 147 | |
| 148 | // make next a new current |
| 149 | current.start = next.start; |
| 150 | current.end = next.end; |
| 151 | new_len += 1; |
| 152 | } |
| 153 | |
| 154 | i += 1; |
| 155 | } |
| 156 | |
| 157 | if new_len == 1 { |
| 158 | *self = IdRange::Continuous(ranges[0].clone()) |
| 159 | } else if ranges.len() != new_len as usize { |
| 160 | ranges.truncate(new_len as usize); |
| 161 | } |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | fn is_squashed(&self) -> bool { |
| 167 | match self { |
| 168 | IdRange::Continuous(_) => true, |
| 169 | IdRange::Fragmented(ranges) => { |
| 170 | let mut i = ranges.iter(); |
| 171 | if let Some(r) = i.next() { |
| 172 | let mut prev_end = r.end; |
| 173 | while let Some(r) = i.next() { |
| 174 | if r.start < prev_end { |
| 175 | return false; |
| 176 | } |
| 177 | prev_end = r.end; |
| 178 | } |
| 179 | true |
| 180 | } else { |
| 181 | true |
| 182 | } |
| 183 | } |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | fn merge(&mut self, other: IdRange) { |
| 188 | let raw = std::mem::take(self); |
| 189 | *self = match (raw, other) { |
| 190 | (IdRange::Continuous(mut a), IdRange::Continuous(b)) => { |
| 191 | let never_intersect = a.end < b.start || b.end < a.start; |
| 192 | if never_intersect { |
| 193 | IdRange::Fragmented(vec![a, b]) |
| 194 | } else { |
| 195 | a.start = a.start.min(b.start); |
| 196 | a.end = a.end.max(b.end); |
| 197 | IdRange::Continuous(a) |
| 198 | } |
| 199 | } |
| 200 | (IdRange::Fragmented(mut a), IdRange::Continuous(b)) => { |
| 201 | a.push(b); |
| 202 | IdRange::Fragmented(a) |
| 203 | } |
| 204 | (IdRange::Continuous(a), IdRange::Fragmented(b)) => { |
| 205 | let mut v = b; |
| 206 | v.push(a); |
| 207 | IdRange::Fragmented(v) |
| 208 | } |
| 209 | (IdRange::Fragmented(mut a), IdRange::Fragmented(mut b)) => { |
| 210 | a.append(&mut b); |
| 211 | IdRange::Fragmented(a) |
| 212 | } |
| 213 | }; |
| 214 | } |
| 215 | |
| 216 | fn encode_raw<E: Encoder>(&self, encoder: &mut E) { |
| 217 | match self { |
| 218 | IdRange::Continuous(range) => { |
| 219 | encoder.write_var(1u32); |
| 220 | range.encode(encoder) |
| 221 | } |
| 222 | IdRange::Fragmented(ranges) => { |
| 223 | encoder.write_var(ranges.len() as u32); |
| 224 | for range in ranges.iter() { |
| 225 | range.encode(encoder); |
| 226 | } |
| 227 | } |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | #[inline] |
| 232 | fn try_join(a: &mut Range<u32>, b: &Range<u32>) -> bool { |
| 233 | if Self::disjoint(a, b) { |
| 234 | false |
| 235 | } else { |
| 236 | a.start = a.start.min(b.start); |
| 237 | a.end = a.end.max(b.end); |
| 238 | true |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | #[inline] |
| 243 | fn disjoint(a: &Range<u32>, b: &Range<u32>) -> bool { |
| 244 | a.start > b.end || b.start > a.end |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | impl Default for IdRange { |
| 249 | fn default() -> Self { |
| 250 | IdRange::Continuous(0..0) |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | impl Encode for IdRange { |
| 255 | fn encode<E: Encoder>(&self, encoder: &mut E) { |
| 256 | if self.is_squashed() { |
| 257 | self.encode_raw(encoder) |
| 258 | } else { |
| 259 | let mut clone = self.clone(); |
| 260 | clone.squash(); |
| 261 | clone.encode_raw(encoder); |
| 262 | } |
| 263 | } |
| 264 | } |
| 265 | |
| 266 | impl Decode for IdRange { |
| 267 | fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, Error> { |
| 268 | match decoder.read_var::<u32>()? { |
| 269 | 1 => { |
| 270 | let range = Range::decode(decoder)?; |
| 271 | Ok(IdRange::Continuous(range)) |
| 272 | } |
| 273 | len => { |
| 274 | let mut ranges = Vec::with_capacity(len as usize); |
| 275 | let mut i = 0; |
| 276 | while i < len { |
| 277 | ranges.push(Range::decode(decoder)?); |
| 278 | i += 1; |
| 279 | } |
| 280 | Ok(IdRange::Fragmented(ranges)) |
| 281 | } |
| 282 | } |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | pub struct IdRangeIter<'a> { |
| 287 | inner: Option<std::slice::Iter<'a, Range<u32>>>, |
| 288 | range: Option<&'a Range<u32>>, |
| 289 | } |
| 290 | |
| 291 | impl<'a> Iterator for IdRangeIter<'a> { |
| 292 | type Item = &'a Range<u32>; |
| 293 | |
| 294 | fn next(&mut self) -> Option<Self::Item> { |
| 295 | if let Some(inner) = &mut self.inner { |
| 296 | inner.next() |
| 297 | } else { |
| 298 | self.range.take() |
| 299 | } |
| 300 | } |
| 301 | } |
| 302 | |
| 303 | /// Implement this to efficiently let IdRange iterator work in descending order |
| 304 | impl<'a> DoubleEndedIterator for IdRangeIter<'a> { |
| 305 | fn next_back(&mut self) -> Option<Self::Item> { |
| 306 | if let Some(inner) = &mut self.inner { |
| 307 | inner.next_back() |
| 308 | } else { |
| 309 | self.range.take() |
| 310 | } |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | /// DeleteSet is a temporary object that is created when needed. |
| 315 | /// - When created in a transaction, it must only be accessed after sorting and merging. |
| 316 | /// - This DeleteSet is sent to other clients. |
| 317 | /// - We do not create a DeleteSet when we send a sync message. The DeleteSet message is created |
| 318 | /// directly from StructStore. |
| 319 | /// - We read a DeleteSet as a apart of sync/update message. In this case the DeleteSet is already |
| 320 | /// sorted and merged. |
| 321 | #[derive(Default, Clone, PartialEq, Eq)] |
| 322 | pub struct IdSet(HashMap<ClientID, IdRange, BuildHasherDefault<ClientHasher>>); |
| 323 | |
| 324 | pub(crate) type Iter<'a> = std::collections::hash_map::Iter<'a, ClientID, IdRange>; |
| 325 | |
| 326 | //TODO: I'd say we should split IdSet and DeleteSet into two structures. While DeleteSet can be |
| 327 | // implemented in terms of IdSet, it has more specific methods (related to deletion process), while |
| 328 | // IdSet could contain wider area of use cases. |
| 329 | impl IdSet { |
| 330 | pub fn new() -> Self { |
| 331 | Self::default() |
| 332 | } |
| 333 | |
| 334 | /// Returns number of clients stored; |
| 335 | pub fn len(&self) -> usize { |
| 336 | self.0.len() |
| 337 | } |
| 338 | |
| 339 | pub(crate) fn iter(&self) -> Iter<'_> { |
| 340 | self.0.iter() |
| 341 | } |
| 342 | |
| 343 | /// Check if current [IdSet] contains given `id`. |
| 344 | pub fn contains(&self, id: &ID) -> bool { |
| 345 | if let Some(ranges) = self.0.get(&id.client) { |
| 346 | ranges.contains(id.clock) |
| 347 | } else { |
| 348 | false |
| 349 | } |
| 350 | } |
| 351 | |
| 352 | /// Checks if current ID set contains any data. |
| 353 | pub fn is_empty(&self) -> bool { |
| 354 | self.0.is_empty() || self.0.values().all(|r| r.is_empty()) |
| 355 | } |
| 356 | |
| 357 | /// Compacts an internal ranges representation. |
| 358 | pub fn squash(&mut self) { |
| 359 | for block in self.0.values_mut() { |
| 360 | block.squash(); |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | pub fn insert(&mut self, id: ID, len: u32) { |
| 365 | let range = id.clock..(id.clock + len); |
| 366 | match self.0.entry(id.client) { |
| 367 | Entry::Occupied(r) => { |
| 368 | r.into_mut().push(range); |
| 369 | } |
| 370 | Entry::Vacant(e) => { |
| 371 | e.insert(IdRange::Continuous(range)); |
| 372 | } |
| 373 | } |
| 374 | } |
| 375 | |
| 376 | /// Inserts a new ID `range` corresponding with a given `client`. |
| 377 | pub fn insert_range(&mut self, client: ClientID, range: IdRange) { |
| 378 | self.0.insert(client, range); |
| 379 | } |
| 380 | |
| 381 | /// Merges another ID set into a current one, combining their information about observed ID |
| 382 | /// ranges and squashing them if necessary. |
| 383 | pub fn merge(&mut self, other: Self) { |
| 384 | other.0.into_iter().for_each(|(client, range)| { |
| 385 | if let Some(r) = self.0.get_mut(&client) { |
| 386 | r.merge(range) |
| 387 | } else { |
| 388 | self.0.insert(client, range); |
| 389 | } |
| 390 | }); |
| 391 | self.squash() |
| 392 | } |
| 393 | |
| 394 | pub fn get(&self, client_id: &ClientID) -> Option<&IdRange> { |
| 395 | self.0.get(client_id) |
| 396 | } |
| 397 | } |
| 398 | |
| 399 | impl Encode for IdSet { |
| 400 | fn encode<E: Encoder>(&self, encoder: &mut E) { |
| 401 | encoder.write_var(self.0.len() as u32); |
| 402 | for (&client_id, block) in self.0.iter() { |
| 403 | encoder.reset_ds_cur_val(); |
| 404 | encoder.write_var(client_id); |
| 405 | block.encode(encoder); |
| 406 | } |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | impl Decode for IdSet { |
| 411 | fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, Error> { |
| 412 | let mut set = Self::new(); |
| 413 | let client_len: u32 = decoder.read_var()?; |
| 414 | let mut i = 0; |
| 415 | while i < client_len { |
| 416 | decoder.reset_ds_cur_val(); |
| 417 | let client: u32 = decoder.read_var()?; |
| 418 | let range = IdRange::decode(decoder)?; |
| 419 | set.0.insert(client as ClientID, range); |
| 420 | i += 1; |
| 421 | } |
| 422 | Ok(set) |
| 423 | } |
| 424 | } |
| 425 | |
| 426 | impl Hash for IdSet { |
| 427 | fn hash<H: Hasher>(&self, state: &mut H) { |
| 428 | for (client, range) in self.0.iter() { |
| 429 | client.hash(state); |
| 430 | range.hash(state); |
| 431 | } |
| 432 | } |
| 433 | } |
| 434 | |
| 435 | /// [DeleteSet] contains information about all blocks (described by clock ranges) that have been |
| 436 | /// subjected to delete process. |
| 437 | #[derive(Clone, Eq, PartialEq, Hash)] |
| 438 | pub struct DeleteSet(IdSet); |
| 439 | |
| 440 | impl From<IdSet> for DeleteSet { |
| 441 | fn from(id_set: IdSet) -> Self { |
| 442 | DeleteSet(id_set) |
| 443 | } |
| 444 | } |
| 445 | |
| 446 | impl<'a> From<&'a BlockStore> for DeleteSet { |
| 447 | /// Creates a [DeleteSet] by reading all deleted blocks and including their clock ranges into |
| 448 | /// the delete set itself. |
| 449 | fn from(store: &'a BlockStore) -> Self { |
| 450 | let mut set = DeleteSet(IdSet::new()); |
| 451 | for (&client, blocks) in store.iter() { |
| 452 | let mut deletes = IdRange::with_capacity(blocks.len()); |
| 453 | for block in blocks.iter() { |
| 454 | if block.is_deleted() { |
| 455 | let (start, end) = block.clock_range(); |
| 456 | deletes.push(start..(end + 1)); |
| 457 | } |
| 458 | } |
| 459 | |
| 460 | if !deletes.is_empty() { |
| 461 | set.0.insert_range(client, deletes); |
| 462 | } |
| 463 | } |
| 464 | set |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | impl Default for DeleteSet { |
| 469 | fn default() -> Self { |
| 470 | Self::new() |
| 471 | } |
| 472 | } |
| 473 | |
| 474 | impl std::fmt::Debug for DeleteSet { |
| 475 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 476 | std::fmt::Display::fmt(self, f) |
| 477 | } |
| 478 | } |
| 479 | impl std::fmt::Display for DeleteSet { |
| 480 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 481 | std::fmt::Display::fmt(&self.0, f) |
| 482 | } |
| 483 | } |
| 484 | |
| 485 | impl std::fmt::Debug for IdSet { |
| 486 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 487 | std::fmt::Display::fmt(self, f) |
| 488 | } |
| 489 | } |
| 490 | impl std::fmt::Display for IdSet { |
| 491 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 492 | let mut s = f.debug_struct(""); |
| 493 | for (k, v) in self.iter() { |
| 494 | s.field(&k.to_string(), v); |
| 495 | } |
| 496 | s.finish() |
| 497 | } |
| 498 | } |
| 499 | |
| 500 | impl std::fmt::Debug for IdRange { |
| 501 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 502 | std::fmt::Display::fmt(self, f) |
| 503 | } |
| 504 | } |
| 505 | impl std::fmt::Display for IdRange { |
| 506 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 507 | match self { |
| 508 | IdRange::Continuous(r) => write!(f, "[{}..{})", r.start, r.end), |
| 509 | IdRange::Fragmented(r) => { |
| 510 | write!(f, "[")?; |
| 511 | for r in r.iter() { |
| 512 | write!(f, " [{}..{})", r.start, r.end)?; |
| 513 | } |
| 514 | write!(f, " ]") |
| 515 | } |
| 516 | } |
| 517 | } |
| 518 | } |
| 519 | |
| 520 | impl DeleteSet { |
| 521 | /// Creates a new empty delete set instance. |
| 522 | pub fn new() -> Self { |
| 523 | DeleteSet(IdSet::new()) |
| 524 | } |
| 525 | |
| 526 | /// Inserts an information about delete block (identified by `id` and having a specified length) |
| 527 | /// inside of a current delete set. |
| 528 | pub fn insert(&mut self, id: ID, len: u32) { |
| 529 | self.0.insert(id, len) |
| 530 | } |
| 531 | |
| 532 | /// Returns number of clients stored; |
| 533 | pub fn len(&self) -> usize { |
| 534 | self.0.len() |
| 535 | } |
| 536 | |
| 537 | /// Checks if delete set contains any clock ranges. |
| 538 | pub fn is_empty(&self) -> bool { |
| 539 | self.0.is_empty() |
| 540 | } |
| 541 | |
| 542 | /// Checks if given block `id` is considered deleted from the perspective of current delete set. |
| 543 | pub fn is_deleted(&self, id: &ID) -> bool { |
| 544 | self.0.contains(id) |
| 545 | } |
| 546 | |
| 547 | /// Returns an iterator over all client-range pairs registered in this delete set. |
| 548 | pub fn iter(&self) -> Iter<'_> { |
| 549 | self.0.iter() |
| 550 | } |
| 551 | |
| 552 | /// Merges another delete set into a current one, combining their information about deleted |
| 553 | /// clock ranges. |
| 554 | pub fn merge(&mut self, other: Self) { |
| 555 | self.0.merge(other.0) |
| 556 | } |
| 557 | |
| 558 | /// Squashes the contents of a current delete set. This operation means, that in case when |
| 559 | /// delete set contains any overlapping ranges within, they will be squashed together to |
| 560 | /// optimize the space and make future encoding more compact. |
| 561 | pub fn squash(&mut self) { |
| 562 | self.0.squash() |
| 563 | } |
| 564 | |
| 565 | pub fn range(&self, client_id: &ClientID) -> Option<&IdRange> { |
| 566 | self.0.get(client_id) |
| 567 | } |
| 568 | |
| 569 | pub(crate) fn try_squash_with(&mut self, store: &mut Store) { |
| 570 | // try to merge deleted / gc'd items |
| 571 | for (&client, range) in self.iter() { |
| 572 | let blocks = store.blocks.get_client_blocks_mut(client); |
| 573 | for r in range.iter().rev() { |
| 574 | // start with merging the item next to the last deleted item |
| 575 | let mut si = |
| 576 | (blocks.len() - 1).min(1 + blocks.find_pivot(r.end - 1).unwrap_or_default()); |
| 577 | let mut block = &blocks[si]; |
| 578 | |
| 579 | let mut valid_range = usize::MAX..usize::MIN; |
| 580 | |
| 581 | while si > 0 && block.clock_start() >= r.start { |
| 582 | valid_range.start = valid_range.start.min(si); |
| 583 | valid_range.end = valid_range.end.max(si); |
| 584 | si -= 1; |
| 585 | block = &blocks[si]; |
| 586 | } |
| 587 | |
| 588 | if valid_range.start != usize::MAX && valid_range.end != usize::MIN { |
| 589 | blocks.squash_left_range_compaction(valid_range.start..=valid_range.end); |
| 590 | } |
| 591 | } |
| 592 | } |
| 593 | } |
| 594 | |
| 595 | pub(crate) fn deleted_blocks(&self) -> DeletedBlocks { |
| 596 | DeletedBlocks::new(self) |
| 597 | } |
| 598 | } |
| 599 | |
| 600 | impl Decode for DeleteSet { |
| 601 | fn decode<D: Decoder>(decoder: &mut D) -> Result<Self, Error> { |
| 602 | Ok(DeleteSet(IdSet::decode(decoder)?)) |
| 603 | } |
| 604 | } |
| 605 | |
| 606 | impl Encode for DeleteSet { |
| 607 | #[inline] |
| 608 | fn encode<E: Encoder>(&self, encoder: &mut E) { |
| 609 | self.0.encode(encoder) |
| 610 | } |
| 611 | } |
| 612 | |
| 613 | pub(crate) struct DeletedBlocks<'ds> { |
| 614 | ds_iter: Iter<'ds>, |
| 615 | current_range: Option<&'ds Range<u32>>, |
| 616 | current_client_id: Option<ClientID>, |
| 617 | range_iter: Option<IdRangeIter<'ds>>, |
| 618 | current_index: Option<usize>, |
| 619 | } |
| 620 | |
| 621 | impl<'ds> DeletedBlocks<'ds> { |
| 622 | pub(crate) fn new(ds: &'ds DeleteSet) -> Self { |
| 623 | let ds_iter = ds.iter(); |
| 624 | DeletedBlocks { |
| 625 | ds_iter, |
| 626 | current_client_id: None, |
| 627 | current_range: None, |
| 628 | range_iter: None, |
| 629 | current_index: None, |
| 630 | } |
| 631 | } |
| 632 | } |
| 633 | |
| 634 | impl<'ds> TxnIterator for DeletedBlocks<'ds> { |
| 635 | type Item = BlockSlice; |
| 636 | |
| 637 | fn next<T: ReadTxn>(&mut self, txn: &T) -> Option<Self::Item> { |
| 638 | if let Some(r) = self.current_range { |
| 639 | let mut block = if let Some(idx) = self.current_index.as_mut() { |
| 640 | if let Some(block) = txn |
| 641 | .store() |
| 642 | .blocks |
| 643 | .get_client(&self.current_client_id?) |
| 644 | .unwrap() |
| 645 | .get(*idx) |
| 646 | { |
| 647 | *idx += 1; |
| 648 | block.as_slice() |
| 649 | } else { |
| 650 | self.current_range = None; |
| 651 | self.current_index = None; |
| 652 | return self.next(txn); |
| 653 | } |
| 654 | } else { |
| 655 | // first block for a particular client |
| 656 | let list = txn |
| 657 | .store() |
| 658 | .blocks |
| 659 | .get_client(&self.current_client_id?) |
| 660 | .unwrap(); |
| 661 | if let Some(idx) = list.find_pivot(r.start) { |
| 662 | let mut block = list[idx].as_slice(); |
| 663 | let clock = block.clock_start(); |
| 664 | |
| 665 | // check if we don't need to cut first block |
| 666 | if clock < r.start { |
| 667 | block.trim_start(r.start - clock); |
| 668 | } |
| 669 | self.current_index = Some(idx + 1); |
| 670 | block |
| 671 | } else { |
| 672 | self.current_range = None; |
| 673 | self.current_index = None; |
| 674 | return self.next(txn); |
| 675 | } |
| 676 | }; |
| 677 | |
| 678 | // check if this is the last block for a current client |
| 679 | let clock = block.clock_start(); |
| 680 | let block_len = block.len(); |
| 681 | if clock > r.end { |
| 682 | // move to the next range |
| 683 | self.current_range = None; |
| 684 | self.current_index = None; |
| 685 | return self.next(txn); |
| 686 | } else if clock < r.end && clock + block_len > r.end { |
| 687 | // we need to cut the last block |
| 688 | block.trim_end(clock + block_len - r.end); |
| 689 | self.current_range = None; |
| 690 | self.current_index = None; |
| 691 | } |
| 692 | |
| 693 | if clock + block_len >= r.end { |
| 694 | self.current_range = None; |
| 695 | self.current_index = None; |
| 696 | } |
| 697 | |
| 698 | Some(block) |
| 699 | } else { |
| 700 | let range_iter = if let Some(iter) = self.range_iter.as_mut() { |
| 701 | iter |
| 702 | } else { |
| 703 | let (client_id, range) = self.ds_iter.next()?; |
| 704 | self.current_client_id = Some(client_id.clone()); |
| 705 | self.current_index = None; |
| 706 | self.range_iter = Some(range.iter()); |
| 707 | self.range_iter.as_mut().unwrap() |
| 708 | }; |
| 709 | self.current_range = match range_iter.next() { |
| 710 | None => { |
| 711 | let (client_id, range) = self.ds_iter.next()?; |
| 712 | self.current_client_id = Some(client_id.clone()); |
| 713 | self.current_index = None; |
| 714 | let mut iter = range.iter(); |
| 715 | let range = iter.next(); |
| 716 | self.range_iter = Some(iter); |
| 717 | range |
| 718 | } |
| 719 | other => other, |
| 720 | }; |
| 721 | return self.next(txn); |
| 722 | } |
| 723 | } |
| 724 | } |
| 725 | |
| 726 | #[cfg(test)] |
| 727 | mod test { |
| 728 | use crate::block::ItemContent; |
| 729 | use crate::id_set::{IdRange, IdSet}; |
| 730 | use crate::iter::TxnIterator; |
| 731 | use crate::slice::BlockSlice; |
| 732 | use crate::test_utils::exchange_updates; |
| 733 | use crate::updates::decoder::{Decode, DecoderV1}; |
| 734 | use crate::updates::encoder::{Encode, Encoder, EncoderV1}; |
| 735 | use crate::{DeleteSet, Doc, Options, ReadTxn, Text, Transact, ID}; |
| 736 | use std::collections::HashSet; |
| 737 | use std::fmt::Debug; |
| 738 | |
| 739 | #[test] |
| 740 | fn id_range_merge_continous() { |
| 741 | // `b` entirely within `a` |
| 742 | let mut a = IdRange::Continuous(0..5); |
| 743 | a.merge(IdRange::Continuous(2..4)); |
| 744 | assert_eq!(a, IdRange::Continuous(0..5)); |
| 745 | |
| 746 | // the tail of `a` crosses the head of `b` |
| 747 | let mut a = IdRange::Continuous(0..5); |
| 748 | a.merge(IdRange::Continuous(4..9)); |
| 749 | assert_eq!(a, IdRange::Continuous(0..9)); |
| 750 | |
| 751 | // `b` is immediately adjacent to the end of `a` |
| 752 | let mut a = IdRange::Continuous(0..5); |
| 753 | a.merge(IdRange::Continuous(5..9)); |
| 754 | assert_eq!(a, IdRange::Continuous(0..9)); |
| 755 | |
| 756 | // `a` does not intersect with `b` |
| 757 | let mut a = IdRange::Continuous(0..4); |
| 758 | a.merge(IdRange::Continuous(6..9)); |
| 759 | assert_eq!(a, IdRange::Fragmented(vec![0..4, 6..9])); |
| 760 | } |
| 761 | |
| 762 | #[test] |
| 763 | fn id_range_compact() { |
| 764 | let mut r = IdRange::Fragmented(vec![(0..3), (3..5), (6..7)]); |
| 765 | r.squash(); |
| 766 | assert_eq!(r, IdRange::Fragmented(vec![(0..5), (6..7)])); |
| 767 | } |
| 768 | |
| 769 | #[test] |
| 770 | fn id_range_invert() { |
| 771 | assert!(IdRange::Continuous(0..3).invert().is_empty()); |
| 772 | |
| 773 | assert_eq!( |
| 774 | IdRange::Continuous(3..5).invert(), |
| 775 | IdRange::Continuous(0..3) |
| 776 | ); |
| 777 | |
| 778 | assert_eq!( |
| 779 | IdRange::Fragmented(vec![0..3, 4..5]).invert(), |
| 780 | IdRange::Continuous(3..4) |
| 781 | ); |
| 782 | |
| 783 | assert_eq!( |
| 784 | IdRange::Fragmented(vec![3..4, 7..9]).invert(), |
| 785 | IdRange::Fragmented(vec![0..3, 4..7]) |
| 786 | ); |
| 787 | } |
| 788 | |
| 789 | #[test] |
| 790 | fn id_range_contains() { |
| 791 | assert!(!IdRange::Continuous(1..3).contains(0)); |
| 792 | assert!(IdRange::Continuous(1..3).contains(1)); |
| 793 | assert!(IdRange::Continuous(1..3).contains(2)); |
| 794 | assert!(!IdRange::Continuous(1..3).contains(3)); |
| 795 | |
| 796 | assert!(!IdRange::Fragmented(vec![1..3, 4..5]).contains(0)); |
| 797 | assert!(IdRange::Fragmented(vec![1..3, 4..5]).contains(1)); |
| 798 | assert!(IdRange::Fragmented(vec![1..3, 4..5]).contains(2)); |
| 799 | assert!(!IdRange::Fragmented(vec![1..3, 4..5]).contains(3)); |
| 800 | assert!(IdRange::Fragmented(vec![1..3, 4..5]).contains(4)); |
| 801 | assert!(!IdRange::Fragmented(vec![1..3, 4..5]).contains(5)); |
| 802 | assert!(!IdRange::Fragmented(vec![1..3, 4..5]).contains(6)); |
| 803 | } |
| 804 | |
| 805 | #[test] |
| 806 | fn id_range_push() { |
| 807 | let mut range = IdRange::Continuous(0..0); |
| 808 | |
| 809 | range.push(0..4); |
| 810 | assert_eq!(range, IdRange::Continuous(0..4)); |
| 811 | |
| 812 | range.push(4..6); |
| 813 | assert_eq!(range, IdRange::Continuous(0..6)); |
| 814 | |
| 815 | range.push(7..9); |
| 816 | assert_eq!(range, IdRange::Fragmented(vec![0..6, 7..9])); |
| 817 | } |
| 818 | |
| 819 | #[test] |
| 820 | fn id_range_encode_decode() { |
| 821 | roundtrip(&IdRange::Continuous(0..4)); |
| 822 | roundtrip(&IdRange::Fragmented(vec![1..4, 5..8])); |
| 823 | } |
| 824 | |
| 825 | #[test] |
| 826 | fn id_set_encode_decode() { |
| 827 | let mut set = IdSet::new(); |
| 828 | set.insert(ID::new(124, 0), 1); |
| 829 | set.insert(ID::new(1337, 0), 12); |
| 830 | set.insert(ID::new(124, 1), 3); |
| 831 | |
| 832 | roundtrip(&set); |
| 833 | } |
| 834 | |
| 835 | fn roundtrip<T>(value: &T) |
| 836 | where |
| 837 | T: Encode + Decode + PartialEq + Debug, |
| 838 | { |
| 839 | let mut encoder = EncoderV1::new(); |
| 840 | value.encode(&mut encoder); |
| 841 | let buf = encoder.to_vec(); |
| 842 | let mut decoder = DecoderV1::from(buf.as_slice()); |
| 843 | let decoded = T::decode(&mut decoder).unwrap(); |
| 844 | |
| 845 | assert_eq!(value, &decoded); |
| 846 | } |
| 847 | |
| 848 | #[test] |
| 849 | fn deleted_blocks() { |
| 850 | let mut o = Options::default(); |
| 851 | o.client_id = 1; |
| 852 | o.skip_gc = true; |
| 853 | let d1 = Doc::with_options(o.clone()); |
| 854 | let t1 = d1.get_or_insert_text("test"); |
| 855 | |
| 856 | o.client_id = 2; |
| 857 | let d2 = Doc::with_options(o); |
| 858 | let t2 = d2.get_or_insert_text("test"); |
| 859 | |
| 860 | t1.insert(&mut d1.transact_mut(), 0, "aaaaa"); |
| 861 | t1.insert(&mut d1.transact_mut(), 0, "bbb"); |
| 862 | |
| 863 | exchange_updates(&[&d1, &d2]); |
| 864 | |
| 865 | t2.insert(&mut d2.transact_mut(), 4, "cccc"); |
| 866 | |
| 867 | exchange_updates(&[&d1, &d2]); |
| 868 | |
| 869 | // t1: 'bbbaccccaaaa' |
| 870 | t1.remove_range(&mut d1.transact_mut(), 2, 2); // => 'bbccccaaaa' |
| 871 | t1.remove_range(&mut d1.transact_mut(), 3, 1); // => 'bbcccaaaa' |
| 872 | t1.remove_range(&mut d1.transact_mut(), 3, 1); // => 'bbccaaaa' |
| 873 | t1.remove_range(&mut d1.transact_mut(), 7, 1); // => 'bbccaaa' |
| 874 | |
| 875 | let blocks = { |
| 876 | let mut txn = d1.transact_mut(); |
| 877 | let s = txn.snapshot(); |
| 878 | |
| 879 | let mut blocks = HashSet::new(); |
| 880 | |
| 881 | let mut i = 0; |
| 882 | let mut deleted = s.delete_set.deleted_blocks(); |
| 883 | while let Some(BlockSlice::Item(b)) = deleted.next(&txn) { |
| 884 | let item = txn.store.materialize(b); |
| 885 | if let ItemContent::String(str) = &item.content { |
| 886 | let t = ( |
| 887 | item.is_deleted(), |
| 888 | item.id, |
| 889 | item.len(), |
| 890 | str.as_str().to_string(), |
| 891 | ); |
| 892 | blocks.insert(t); |
| 893 | } |
| 894 | i += 1; |
| 895 | if i == 5 { |
| 896 | break; |
| 897 | } |
| 898 | } |
| 899 | blocks |
| 900 | }; |
| 901 | |
| 902 | let expected = HashSet::from([ |
| 903 | (true, ID::new(1, 0), 1, "a".to_owned()), |
| 904 | (true, ID::new(1, 4), 1, "a".to_owned()), |
| 905 | (true, ID::new(1, 7), 1, "b".to_owned()), |
| 906 | (true, ID::new(2, 1), 2, "cc".to_owned()), |
| 907 | ]); |
| 908 | |
| 909 | assert_eq!(blocks, expected); |
| 910 | } |
| 911 | |
| 912 | #[test] |
| 913 | fn deleted_blocks2() { |
| 914 | let mut ds = DeleteSet::new(); |
| 915 | let doc = Doc::with_client_id(1); |
| 916 | let txt = doc.get_or_insert_text("test"); |
| 917 | txt.push(&mut doc.transact_mut(), "testab"); |
| 918 | ds.insert(ID::new(1, 5), 1); |
| 919 | let txn = doc.transact_mut(); |
| 920 | let mut i = ds.deleted_blocks(); |
| 921 | let ptr = i.next(&txn).unwrap(); |
| 922 | let start = ptr.clock_start(); |
| 923 | let end = ptr.clock_end(); |
| 924 | assert_eq!(start, 5); |
| 925 | assert_eq!(end, 5); |
| 926 | assert!(i.next(&txn).is_none()); |
| 927 | } |
| 928 | } |
| 929 | |
amadan