About parallelism

[liyiheng]

If process of an operator could be called in parallel, a lock for state.get and state.set may required.
Such as:

fluxus/crates/fluxus-api/src/operators/window_aggregator.rs

Lines 53 to 58 in 7828dec

	let current = self
	.state
	.get(&window_key)
	.unwrap_or_else(|| self.init.clone());
	let new_value = (self.f)(current, record.data.clone());
	self.state.set(window_key, new_value.clone());

fluxus/crates/fluxus-api/src/operators/window_sorter.rs

Lines 46 to 52 in 7828dec

	let mut current = self.state.get(&window_key).unwrap_or_default();
	let index = current
	.binary_search_by(|prob| (self.f)(prob, &record.data))
	.unwrap_or_else(|i| i);
	current.insert(index, record.data.clone());
	self.state.set(window_key, current.clone());

fluxus/crates/fluxus-api/src/operators/window_sorter.rs

Lines 103 to 112 in 7828dec

	let mut current = self.state.get(&window_key).unwrap_or_default();
	let index = current
	.binary_search_by(|prob| match self.method {
	SortOrder::Asc => prob.timestamp.cmp(&record.timestamp),
	SortOrder::Desc => record.timestamp.cmp(&prob.timestamp),
	})
	.unwrap_or_else(|i| i);
	current.insert(index, record.clone());
	self.state.set(window_key, current.clone());