Unsafe access to state in multi-threaded operators

[pschafhalter]

Currently, access to state isn't locked. When using the TimeVersionedState, we could introduce unsafe accesses due to concurrency if a watermark callback for time t and a data callback for time t + 1 run simultaneously.

Currently, this shouldn't cause problems because only 1 callback can access state at a time. In the future, we may want to increase parallelism in multi-threaded operators and then this will cause problems.

Lines causing the problem (also in scripts/make_callback_builder.py):

erdos/src/dataflow/stream/internal_stateful_read_stream.rs

Lines 91 to 95 in fd590d0

	let state_ref_mut =
	unsafe { Arc::get_mut_unchecked(&mut state_arc) };
	state_ref_mut.set_access_context(AccessContext::Callback);
	state_ref_mut.set_current_time(msg.timestamp.clone());
	(callback)(msg_copy.timestamp, msg_copy.data, state_ref_mut)

Potential solutions:

1. Locking.
2. Cloning.
3. Providing a view on a subsection of the state.