No particular reason, I was just reading code and confused about the bounds
and it turned out they weren't needed.

Signed-off-by: Ben Pfaff <blp@feldera.com>

The trait documentation says so:

```rust
    /// Returns `true` if `self` is an input operator.
    ///
    /// An input operator feeds new data into the circuit. Examples are
    /// the `Input` and `Generator` operators.
    fn is_input(&self) -> bool {
        false
    }
```

I don't know why they did not return `true` for these input operators
before.  I don't know what the consequences are for returning the wrong
value (this is not in response to some bug I discovered).

Signed-off-by: Ben Pfaff <blp@feldera.com>

I thought maybe it was possible to have an operator indicate that it was
already flushed immediately and then never get evaluated.  This is not the
case.

Plus a few other minor comment improvements.

Signed-off-by: Ben Pfaff <blp@feldera.com>

As far as I can tell, `eval` and `eval_owned` were the same except for
a clone() call, so this simplifies things.

Signed-off-by: Ben Pfaff <blp@feldera.com>

This will enable adding a field that can't be cloned, in an upcoming
commit.

Signed-off-by: Ben Pfaff <blp@feldera.com>

Before commit e60a8a3 ("[dbsp] Use new protocol for exchanging data
among hosts."), each remote exchange RPC waited for the data to be picked
up out of its mailboxes before returning.  That commit changed that, since
now sending data doesn't wait for a reply from the remote host (there is
in fact no reply).  In fact, it didn't do anything to prevent such races.

This fixes the problem.  Instead of introducing a wait on the RPC sender
side, it introduces it on the receiver side.  The receiver waits for the
mailboxes it is writing to be empty before writing to them.

I noticed that we have a 2-d array of notifiers for this purpose but we
were going to wait on all of the notifiers in the column for a given
sender in sequence.  It is more efficient in time and space to simply
have one per sender instead, so I made that change.  At the same time,
I noticed that Broadcast could simply use that same Notify instead of
adding its own via registering a callback, so I made that simplification
too.

sender_notifies is now redundant with sender_callbacks, but it would
involve changing a lot of code to switch from one to the other, so this
commit does not make that change.

Signed-off-by: Ben Pfaff <blp@feldera.com>

Signed-off-by: Ben Pfaff <blp@feldera.com>

Signed-off-by: Ben Pfaff <blp@feldera.com>

Signed-off-by: Ben Pfaff <blp@feldera.com>

In practice these were always supplied as Runtime::worker_index() and it
didn't make sense to supply any other value.

Signed-off-by: Ben Pfaff <blp@feldera.com>

Until now, the code for exchange has always bound its own socket to listen
for incoming connections.  This works fine in what is our common case,
where the listening socket is a well-known port in a container that doesn't
have anything else running.  But for tests, or for situations where the
host running the pipeline has other software running and doesn't reserve
specific ports for the pipeline, it's more reliable to bind all the
listening ports (probably letting the kernel choose a port number), then
start the pipelines with the port numbers known in advance.  This commit
allows for that.  The initial use case will be unit tests, in upcoming
commits.

Signed-off-by: Ben Pfaff <blp@feldera.com>

By moving this out of `Exchange` into an argument for try_receive_all(),
we move it closer to where it's actually needed, and it's no longer
necessary to box it.

Signed-off-by: Ben Pfaff <blp@feldera.com>

This separation was painful and I'm glad to get rid of it.

Signed-off-by: Ben Pfaff <blp@feldera.com>

The word `Service` wasn't doing anything here.

Signed-off-by: Ben Pfaff <blp@feldera.com>

Until now, exchange has proceeded in lockstep: every sender sends one
item per receiver, then every receiver receives one item per sender.  This
commit adds an alternative model, called "streaming" exchange.  In
streaming exchange, senders still send a single item to each receiver per
step, but receivers might get multiple items or no items in a given step.
This should allow for better performance, especially in multihost
environments.

Signed-off-by: Ben Pfaff <blp@feldera.com>

Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>

This causes a failure in the `proptest_max_retain_values_test` test that
I am trying to track down.  (The failure doesn't happen if I change
`ExchangeKind::Stream` to `ExchangeKind::Sync` in
dyn_shard_accumulate().)

Signed-off-by: Ben Pfaff <blp@feldera.com>