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

Until now the node identifiers for operator invocations have been formatted
as `[123.456]`.  This is fine except that the profiles show them as
`n123_456` and search only works if they're in the latter form, so this
commit changes them to use that form.

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

…cation.

With this change, one can use `GlobalNodeId::node_identifier()` in a format
string without an extra allocation to make a String and then copy it.

This seems like a good idea because each operator evaluation calls this
(if profile capture is running).

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

Until now, each iteration of the spine merge loop has done three steps per
level like this:

1. Run a merge step, if there's a merger for this level.
2. If the merger is done, update the spine and drop the merger.
3. Start a new merger, if there's no merger for this level.

This meant that there was always a time gap between creating a merger (in
step 3) and running a merge step for it (in step 1), because all the other
mergers and levels would get a chance to run in between.  This made it hard
to interpret the elapsed time of a merge for level-0 merges, which always
complete in a single merge step.

This commit changes the pattern to:

1. Start a new merger, if there's no merger for this level.
2. Run a merge step, if there's a merger for this level.
3. If the merger is done, update the spine and drop the merger.

This means that elapsed time for level-0 merges is now accurate.

In addition, the previous implementation took the shared state lock on
every iteration, even though it was technically needed only to start a new
merge.  This commit fixes that.  It also fixes how the previous
implementation sometimes took the lock twice successively.  I don't know
whether lock contention was a serious issue here.

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

ShardedAccumulatorReceiver did not wait for any kind of backpressure in
the spine that it created.  This meant that in some cases the spine could
end up with thousands of loose batches, which could start exhausting
memory and lead to memory pressure elsewhere, causing an overall slowdown.
This commit fixes the problem by making the receiver operator wait until
the number of batches in the spine reaches an acceptable level.

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

This made the problem with too many batches in the receiver's spine
evident from a circuit profile.

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

ShardedExchangeSender was sending data to one remote worker, then waiting
for the channel to no longer be backed up, then sending it to another
remote worker, and so on.  This commit switches to sending all the data
to remote workers, then waiting for them all together, reducing the amount
of serialization.

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

Until now, exchange profile spans have included exchange IDs, which allows
one to distinguish them from one another in the profiles but it doesn't
allow for associating them with particular operators.  This commit enables
the latter by adding operators' global node ids for spans associated with
exchanges and sharded accumulators.

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

The sharded accumulator adds larger number of smaller batches to its spine
than the exchange operator.  We can get faster merges by merging more of
these batches at a time.

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

The other names of this kind were prefixed by the operator name.

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