Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'main' into polish-project-diagnostics

Browse source 
Also fix false failure in ModelHandle::condition when parking is not forbidden.
This commit is contained in:
Max Brunsfeld 2022-01-05 10:53:18 -08:00
commit 8728d3292d
16 changed files with 676 additions and 202 deletions
Download patch file Download diff file Expand all files Collapse all files

197
crates/language/src/buffer.rs
View file

@ -65,8 +65,8 @@ pub struct Buffer {
syntax_tree: Mutex<Option<SyntaxTree>>,
parsing_in_background: bool,
parse_count: usize,
remote_selections: TreeMap<ReplicaId, Arc<[Selection<Anchor>]>>,
diagnostics: DiagnosticSet,
remote_selections: TreeMap<ReplicaId, SelectionSet>,
selections_update_count: usize,
diagnostics_update_count: usize,
language_server: Option<LanguageServerState>,
@ -79,14 +79,20 @@ pub struct BufferSnapshot {
text: text::BufferSnapshot,
tree: Option<Tree>,
diagnostics: DiagnosticSet,
remote_selections: TreeMap<ReplicaId, Arc<[Selection<Anchor>]>>,
diagnostics_update_count: usize,
remote_selections: TreeMap<ReplicaId, SelectionSet>,
selections_update_count: usize,
is_parsing: bool,
language: Option<Arc<Language>>,
parse_count: usize,
}
#[derive(Clone, Debug)]
struct SelectionSet {
selections: Arc<[Selection<Anchor>]>,
lamport_timestamp: clock::Lamport,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GroupId {
source: Arc<str>,
@ -131,10 +137,6 @@ pub enum Operation {
selections: Arc<[Selection<Anchor>]>,
lamport_timestamp: clock::Lamport,
},
RemoveSelections {
replica_id: ReplicaId,
lamport_timestamp: clock::Lamport,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
@ -286,18 +288,37 @@ impl Buffer {
file: Option<Box<dyn File>>,
cx: &mut ModelContext<Self>,
) -> Result<Self> {
let mut buffer =
text::Buffer::new(replica_id, message.id, History::new(message.content.into()));
let ops = message
.history
.into_iter()
.map(|op| text::Operation::Edit(proto::deserialize_edit_operation(op)));
buffer.apply_ops(ops)?;
let fragments_len = message.fragments.len();
let buffer = TextBuffer::from_parts(
replica_id,
message.id,
&message.visible_text,
&message.deleted_text,
message
.undo_map
.into_iter()
.map(proto::deserialize_undo_map_entry),
message
.fragments
.into_iter()
.enumerate()
.map(|(i, fragment)| {
proto::deserialize_buffer_fragment(fragment, i, fragments_len)
}),
message.lamport_timestamp,
From::from(message.version),
);
let mut this = Self::build(buffer, file);
for selection_set in message.selections {
this.remote_selections.insert(
selection_set.replica_id as ReplicaId,
proto::deserialize_selections(selection_set.selections),
SelectionSet {
selections: proto::deserialize_selections(selection_set.selections),
lamport_timestamp: clock::Lamport {
replica_id: selection_set.replica_id as ReplicaId,
value: selection_set.lamport_timestamp,
},
},
);
}
let snapshot = this.snapshot();
@ -307,27 +328,53 @@ impl Buffer {
cx,
);
let deferred_ops = message
.deferred_operations
.into_iter()
.map(proto::deserialize_operation)
.collect::<Result<Vec<_>>>()?;
this.apply_ops(deferred_ops, cx)?;
Ok(this)
}
pub fn to_proto(&self) -> proto::Buffer {
proto::Buffer {
id: self.remote_id(),
content: self.text.base_text().to_string(),
history: self
visible_text: self.text.text(),
deleted_text: self.text.deleted_text(),
undo_map: self
.text
.history()
.map(proto::serialize_edit_operation)
.undo_history()
.map(proto::serialize_undo_map_entry)
.collect(),
version: From::from(&self.version),
lamport_timestamp: self.lamport_clock.value,
fragments: self
.text
.fragments()
.map(proto::serialize_buffer_fragment)
.collect(),
selections: self
.remote_selections
.iter()
.map(|(replica_id, selections)| proto::SelectionSet {
.map(|(replica_id, set)| proto::SelectionSet {
replica_id: *replica_id as u32,
selections: proto::serialize_selections(selections),
selections: proto::serialize_selections(&set.selections),
lamport_timestamp: set.lamport_timestamp.value,
})
.collect(),
diagnostics: proto::serialize_diagnostics(self.diagnostics.iter()),
deferred_operations: self
.deferred_ops
.iter()
.map(proto::serialize_operation)
.chain(
self.text
.deferred_ops()
.map(|op| proto::serialize_operation(&Operation::Buffer(op.clone()))),
)
.collect(),
}
}
@ -1081,6 +1128,13 @@ impl Buffer {
cx: &mut ModelContext<Self>,
) {
let lamport_timestamp = self.text.lamport_clock.tick();
self.remote_selections.insert(
self.text.replica_id(),
SelectionSet {
selections: selections.clone(),
lamport_timestamp,
},
);
self.send_operation(
Operation::UpdateSelections {
replica_id: self.text.replica_id(),
@ -1092,14 +1146,7 @@ impl Buffer {
}
pub fn remove_active_selections(&mut self, cx: &mut ModelContext<Self>) {
let lamport_timestamp = self.text.lamport_clock.tick();
self.send_operation(
Operation::RemoveSelections {
replica_id: self.text.replica_id(),
lamport_timestamp,
},
cx,
);
self.set_active_selections(Arc::from([]), cx);
}
fn update_language_server(&mut self) {
@ -1287,6 +1334,7 @@ impl Buffer {
})
.collect::<Vec<_>>();
self.text.apply_ops(buffer_ops)?;
self.deferred_ops.insert(deferred_ops);
self.flush_deferred_ops(cx);
self.did_edit(&old_version, was_dirty, cx);
// Notify independently of whether the buffer was edited as the operations could include a
@ -1322,7 +1370,6 @@ impl Buffer {
Operation::UpdateSelections { selections, .. } => selections
.iter()
.all(|s| self.can_resolve(&s.start) && self.can_resolve(&s.end)),
Operation::RemoveSelections { .. } => true,
}
}
@ -1346,15 +1393,19 @@ impl Buffer {
selections,
lamport_timestamp,
} => {
self.remote_selections.insert(replica_id, selections);
self.text.lamport_clock.observe(lamport_timestamp);
self.selections_update_count += 1;
}
Operation::RemoveSelections {
replica_id,
lamport_timestamp,
} => {
self.remote_selections.remove(&replica_id);
if let Some(set) = self.remote_selections.get(&replica_id) {
if set.lamport_timestamp > lamport_timestamp {
return;
}
}
self.remote_selections.insert(
replica_id,
SelectionSet {
selections,
lamport_timestamp,
},
);
self.text.lamport_clock.observe(lamport_timestamp);
self.selections_update_count += 1;
}
@ -1448,6 +1499,10 @@ impl Buffer {
#[cfg(any(test, feature = "test-support"))]
impl Buffer {
pub fn set_group_interval(&mut self, group_interval: Duration) {
self.text.set_group_interval(group_interval);
}
pub fn randomly_edit<T>(
&mut self,
rng: &mut T,
@ -1456,9 +1511,38 @@ impl Buffer {
) where
T: rand::Rng,
{
self.start_transaction();
self.text.randomly_edit(rng, old_range_count);
self.end_transaction(cx);
let mut old_ranges: Vec<Range<usize>> = Vec::new();
for _ in 0..old_range_count {
let last_end = old_ranges.last().map_or(0, |last_range| last_range.end + 1);
if last_end > self.len() {
break;
}
old_ranges.push(self.text.random_byte_range(last_end, rng));
}
let new_text_len = rng.gen_range(0..10);
let new_text: String = crate::random_char_iter::RandomCharIter::new(&mut *rng)
.take(new_text_len)
.collect();
log::info!(
"mutating buffer {} at {:?}: {:?}",
self.replica_id(),
old_ranges,
new_text
);
self.edit(old_ranges.iter().cloned(), new_text.as_str(), cx);
}
pub fn randomly_undo_redo(&mut self, rng: &mut impl rand::Rng, cx: &mut ModelContext<Self>) {
let was_dirty = self.is_dirty();
let old_version = self.version.clone();
let ops = self.text.randomly_undo_redo(rng);
if !ops.is_empty() {
for op in ops {
self.send_operation(Operation::Buffer(op), cx);
self.did_edit(&old_version, was_dirty, cx);
}
}
}
}
@ -1711,20 +1795,30 @@ impl BufferSnapshot {
{
self.remote_selections
.iter()
.filter(|(replica_id, _)| **replica_id != self.text.replica_id())
.map(move |(replica_id, selections)| {
let start_ix = match selections
.binary_search_by(|probe| probe.end.cmp(&range.start, self).unwrap())
{
.filter(|(replica_id, set)| {
**replica_id != self.text.replica_id() && !set.selections.is_empty()
})
.map(move |(replica_id, set)| {
let start_ix = match set.selections.binary_search_by(|probe| {
probe
.end
.cmp(&range.start, self)
.unwrap()
.then(Ordering::Greater)
}) {
Ok(ix) | Err(ix) => ix,
};
let end_ix = match selections
.binary_search_by(|probe| probe.start.cmp(&range.end, self).unwrap())
{
let end_ix = match set.selections.binary_search_by(|probe| {
probe
.start
.cmp(&range.end, self)
.unwrap()
.then(Ordering::Less)
}) {
Ok(ix) | Err(ix) => ix,
};
(*replica_id, selections[start_ix..end_ix].iter())
(*replica_id, set.selections[start_ix..end_ix].iter())
})
}
@ -2007,9 +2101,6 @@ impl operation_queue::Operation for Operation {
}
| Operation::UpdateSelections {
lamport_timestamp, ..
}
| Operation::RemoveSelections {
lamport_timestamp, ..
} => *lamport_timestamp,
}
}

101
crates/language/src/proto.rs
View file

@ -1,6 +1,7 @@
use crate::{diagnostic_set::DiagnosticEntry, Diagnostic, Operation};
use anyhow::{anyhow, Result};
use clock::ReplicaId;
use collections::HashSet;
use lsp::DiagnosticSeverity;
use rpc::proto;
use std::sync::Arc;
@ -32,7 +33,7 @@ pub fn serialize_operation(operation: &Operation) -> proto::Operation {
counts: undo
.counts
.iter()
.map(|(edit_id, count)| proto::operation::UndoCount {
.map(|(edit_id, count)| proto::UndoCount {
replica_id: edit_id.replica_id as u32,
local_timestamp: edit_id.value,
count: *count,
@ -49,13 +50,6 @@ pub fn serialize_operation(operation: &Operation) -> proto::Operation {
lamport_timestamp: lamport_timestamp.value,
selections: serialize_selections(selections),
}),
Operation::RemoveSelections {
replica_id,
lamport_timestamp,
} => proto::operation::Variant::RemoveSelections(proto::operation::RemoveSelections {
replica_id: *replica_id as u32,
lamport_timestamp: lamport_timestamp.value,
}),
Operation::UpdateDiagnostics {
diagnostics,
lamport_timestamp,
@ -87,6 +81,43 @@ pub fn serialize_edit_operation(operation: &EditOperation) -> proto::operation::
}
}
pub fn serialize_undo_map_entry(
(edit_id, counts): (&clock::Local, &[(clock::Local, u32)]),
) -> proto::UndoMapEntry {
proto::UndoMapEntry {
replica_id: edit_id.replica_id as u32,
local_timestamp: edit_id.value,
counts: counts
.iter()
.map(|(undo_id, count)| proto::UndoCount {
replica_id: undo_id.replica_id as u32,
local_timestamp: undo_id.value,
count: *count,
})
.collect(),
}
}
pub fn serialize_buffer_fragment(fragment: &text::Fragment) -> proto::BufferFragment {
proto::BufferFragment {
replica_id: fragment.insertion_timestamp.replica_id as u32,
local_timestamp: fragment.insertion_timestamp.local,
lamport_timestamp: fragment.insertion_timestamp.lamport,
insertion_offset: fragment.insertion_offset as u32,
len: fragment.len as u32,
visible: fragment.visible,
deletions: fragment
.deletions
.iter()
.map(|clock| proto::VectorClockEntry {
replica_id: clock.replica_id as u32,
timestamp: clock.value,
})
.collect(),
max_undos: From::from(&fragment.max_undos),
}
}
pub fn serialize_selections(selections: &Arc<[Selection<Anchor>]>) -> Vec<proto::Selection> {
selections
.iter()
@ -200,13 +231,6 @@ pub fn deserialize_operation(message: proto::Operation) -> Result<Operation> {
selections: Arc::from(selections),
}
}
proto::operation::Variant::RemoveSelections(message) => Operation::RemoveSelections {
replica_id: message.replica_id as ReplicaId,
lamport_timestamp: clock::Lamport {
replica_id: message.replica_id as ReplicaId,
value: message.lamport_timestamp,
},
},
proto::operation::Variant::UpdateDiagnostics(message) => Operation::UpdateDiagnostics {
diagnostics: deserialize_diagnostics(message.diagnostics),
lamport_timestamp: clock::Lamport {
@ -236,6 +260,53 @@ pub fn deserialize_edit_operation(edit: proto::operation::Edit) -> EditOperation
}
}
pub fn deserialize_undo_map_entry(
entry: proto::UndoMapEntry,
) -> (clock::Local, Vec<(clock::Local, u32)>) {
(
clock::Local {
replica_id: entry.replica_id as u16,
value: entry.local_timestamp,
},
entry
.counts
.into_iter()
.map(|undo_count| {
(
clock::Local {
replica_id: undo_count.replica_id as u16,
value: undo_count.local_timestamp,
},
undo_count.count,
)
})
.collect(),
)
}
pub fn deserialize_buffer_fragment(
message: proto::BufferFragment,
ix: usize,
count: usize,
) -> Fragment {
Fragment {
id: locator::Locator::from_index(ix, count),
insertion_timestamp: InsertionTimestamp {
replica_id: message.replica_id as ReplicaId,
local: message.local_timestamp,
lamport: message.lamport_timestamp,
},
insertion_offset: message.insertion_offset as usize,
len: message.len as usize,
visible: message.visible,
deletions: HashSet::from_iter(message.deletions.into_iter().map(|entry| clock::Local {
replica_id: entry.replica_id as ReplicaId,
value: entry.timestamp,
})),
max_undos: From::from(message.max_undos),
}
}
pub fn deserialize_selections(selections: Vec<proto::Selection>) -> Arc<[Selection<Anchor>]> {
Arc::from(
selections

192
crates/language/src/tests.rs
View file

@ -1,13 +1,18 @@
use super::*;
use clock::ReplicaId;
use collections::BTreeMap;
use gpui::{ModelHandle, MutableAppContext};
use rand::prelude::*;
use std::{
cell::RefCell,
env,
iter::FromIterator,
ops::Range,
rc::Rc,
time::{Duration, Instant},
};
use unindent::Unindent as _;
use util::test::Network;
#[cfg(test)]
#[ctor::ctor]
@ -758,6 +763,193 @@ async fn test_empty_diagnostic_ranges(mut cx: gpui::TestAppContext) {
});
}
#[gpui::test]
fn test_serialization(cx: &mut gpui::MutableAppContext) {
let mut now = Instant::now();
let buffer1 = cx.add_model(|cx| {
let mut buffer = Buffer::new(0, "abc", cx);
buffer.edit([3..3], "D", cx);
now += Duration::from_secs(1);
buffer.start_transaction_at(now);
buffer.edit([4..4], "E", cx);
buffer.end_transaction_at(now, cx);
assert_eq!(buffer.text(), "abcDE");
buffer.undo(cx);
assert_eq!(buffer.text(), "abcD");
buffer.edit([4..4], "F", cx);
assert_eq!(buffer.text(), "abcDF");
buffer
});
assert_eq!(buffer1.read(cx).text(), "abcDF");
let message = buffer1.read(cx).to_proto();
let buffer2 = cx.add_model(|cx| Buffer::from_proto(1, message, None, cx).unwrap());
assert_eq!(buffer2.read(cx).text(), "abcDF");
}
#[gpui::test(iterations = 100)]
fn test_random_collaboration(cx: &mut MutableAppContext, mut rng: StdRng) {
let min_peers = env::var("MIN_PEERS")
.map(|i| i.parse().expect("invalid `MIN_PEERS` variable"))
.unwrap_or(1);
let max_peers = env::var("MAX_PEERS")
.map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
.unwrap_or(5);
let operations = env::var("OPERATIONS")
.map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
.unwrap_or(10);
let base_text_len = rng.gen_range(0..10);
let base_text = RandomCharIter::new(&mut rng)
.take(base_text_len)
.collect::<String>();
let mut replica_ids = Vec::new();
let mut buffers = Vec::new();
let mut network = Network::new(rng.clone());
for i in 0..rng.gen_range(min_peers..=max_peers) {
let buffer = cx.add_model(|cx| {
let mut buffer = Buffer::new(i as ReplicaId, base_text.as_str(), cx);
buffer.set_group_interval(Duration::from_millis(rng.gen_range(0..=200)));
buffer
});
buffers.push(buffer);
replica_ids.push(i as ReplicaId);
network.add_peer(i as ReplicaId);
log::info!("Adding initial peer with replica id {}", i);
}
log::info!("initial text: {:?}", base_text);
let mut now = Instant::now();
let mut mutation_count = operations;
let mut active_selections = BTreeMap::default();
loop {
let replica_index = rng.gen_range(0..replica_ids.len());
let replica_id = replica_ids[replica_index];
let buffer = &mut buffers[replica_index];
let mut new_buffer = None;
match rng.gen_range(0..100) {
0..=29 if mutation_count != 0 => {
buffer.update(cx, |buffer, cx| {
buffer.start_transaction_at(now);
buffer.randomly_edit(&mut rng, 5, cx);
buffer.end_transaction_at(now, cx);
log::info!("buffer {} text: {:?}", buffer.replica_id(), buffer.text());
});
mutation_count -= 1;
}
30..=39 if mutation_count != 0 => {
buffer.update(cx, |buffer, cx| {
let mut selections = Vec::new();
for id in 0..rng.gen_range(1..=5) {
let range = buffer.random_byte_range(0, &mut rng);
selections.push(Selection {
id,
start: buffer.anchor_before(range.start),
end: buffer.anchor_before(range.end),
reversed: false,
goal: SelectionGoal::None,
});
}
let selections: Arc<[Selection<Anchor>]> = selections.into();
log::info!(
"peer {} setting active selections: {:?}",
replica_id,
selections
);
active_selections.insert(replica_id, selections.clone());
buffer.set_active_selections(selections, cx);
});
mutation_count -= 1;
}
40..=49 if replica_ids.len() < max_peers => {
let old_buffer = buffer.read(cx).to_proto();
let new_replica_id = replica_ids.len() as ReplicaId;
log::info!(
"Adding new replica {} (replicating from {})",
new_replica_id,
replica_id
);
new_buffer = Some(cx.add_model(|cx| {
let mut new_buffer =
Buffer::from_proto(new_replica_id, old_buffer, None, cx).unwrap();
new_buffer.set_group_interval(Duration::from_millis(rng.gen_range(0..=200)));
new_buffer
}));
replica_ids.push(new_replica_id);
network.replicate(replica_id, new_replica_id);
}
50..=69 if mutation_count != 0 => {
buffer.update(cx, |buffer, cx| {
buffer.randomly_undo_redo(&mut rng, cx);
log::info!("buffer {} text: {:?}", buffer.replica_id(), buffer.text());
});
mutation_count -= 1;
}
70..=99 if network.has_unreceived(replica_id) => {
let ops = network
.receive(replica_id)
.into_iter()
.map(|op| proto::deserialize_operation(op).unwrap());
if ops.len() > 0 {
log::info!(
"peer {} applying {} ops from the network.",
replica_id,
ops.len()
);
buffer.update(cx, |buffer, cx| buffer.apply_ops(ops, cx).unwrap());
}
}
_ => {}
}
buffer.update(cx, |buffer, _| {
let ops = buffer
.operations
.drain(..)
.map(|op| proto::serialize_operation(&op))
.collect();
network.broadcast(buffer.replica_id(), ops);
});
now += Duration::from_millis(rng.gen_range(0..=200));
buffers.extend(new_buffer);
if mutation_count == 0 && network.is_idle() {
break;
}
}
let first_buffer = buffers[0].read(cx);
for buffer in &buffers[1..] {
let buffer = buffer.read(cx);
assert_eq!(
buffer.text(),
first_buffer.text(),
"Replica {} text != Replica 0 text",
buffer.replica_id()
);
}
for buffer in &buffers {
let buffer = buffer.read(cx).snapshot();
let expected_remote_selections = active_selections
.iter()
.filter(|(replica_id, _)| **replica_id != buffer.replica_id())
.map(|(replica_id, selections)| (*replica_id, selections.iter().collect::<Vec<_>>()))
.collect::<Vec<_>>();
let actual_remote_selections = buffer
.remote_selections_in_range(Anchor::min()..Anchor::max())
.map(|(replica_id, selections)| (replica_id, selections.collect::<Vec<_>>()))
.collect::<Vec<_>>();
assert_eq!(actual_remote_selections, expected_remote_selections);
}
}
fn chunks_with_diagnostics<T: ToOffset + ToPoint>(
buffer: &Buffer,
range: Range<T>,