Demonstrate FrontierInterest::IfCapability

[frankmcsherry]

This PR explores an alternate form of arrange_core that produces a trace handle that .. 1. cannot be shared across dataflows, but 2. only needs to be activated on frontier transitions if it holds a capability. At the moment, this is presented as a new method that needs to be opted in to, so .. potentially not great risk to merge.

Both TraceAgent and TraceWriter are refactored to extract the shared queue that communicates batches and progress statements, wrapped around Inner types that do everything else. Without the shared queue, you have an Arranged<G,Tr> that can be used in the local dataflow, just without an import method to bring it in to other dataflows.

There is a new example event_driven.rs borrowed from timely dataflow, in which we construct d dataflows each of one input, a region containing an arrange operator and k filter operators, with a probe at the end outside the region. The experiment repeatedly adds an update to one input, and ticks all inputs.

Using the existing arrange_core which has the Always frontier interest:

Running `target/release/examples/event_driven 1000 1000 no`
Local: false
2.39241975s     dataflows built (1000 x 1000)
2.392448125s    round 0 complete in 0 steps
6.588212375s    round 10 complete in 5 steps
10.337022125s   round 20 complete in 5 steps
13.826779125s   round 30 complete in 5 steps
17.38620175s    round 40 complete in 5 steps
20.938500333s   round 50 complete in 5 steps
24.418297583s   round 60 complete in 5 steps
27.9850445s     round 70 complete in 5 steps
31.47804925s    round 80 complete in 5 steps
35.038579041s   round 90 complete in 5 steps
38.634787708s   round 100 complete in 5 steps
^C

Using the dataflow-local operator with the same arguments:

Running `target/release/examples/event_driven 1000 1000 local`
Local: true
2.401872292s    dataflows built (1000 x 1000)
2.401896292s    round 0 complete in 0 steps
3.567310292s    round 100 complete in 5 steps
3.88450825s     round 200 complete in 5 steps
4.220108834s    round 300 complete in 5 steps
4.557004375s    round 400 complete in 5 steps
4.87334s        round 500 complete in 5 steps
5.19050575s     round 600 complete in 5 steps
5.504117375s    round 700 complete in 5 steps
5.824875125s    round 800 complete in 5 steps
6.133310125s    round 900 complete in 5 steps
6.440401209s    round 1000 complete in 5 steps
^C

The "speed-up" takes the iteration time from ~360ms to 4ms.

[frankmcsherry]

One thought while reading through the diff is that it should be possible, and is probably worth it, to factor the core operator logic into something parameterized by a &mut Option<Queues>, which is then optionally set to something and the trace elevated from Inner to shared. It's a bit of a weird dance to get the queues in there, but it's also awkward writing this logic twice, as it is .. subtle and load-bearing.

[frankmcsherry]

The big thing here is the commit Reduce must consult input frontier, which fixes an oversight in reduce where it would ignore its input frontier and treat the input arrangement as the source of truth about frontier information. Works great, until you have nested scopes and a held capability at an interesting time that the arrangement never sees. Then you hang forever.