Improve pipeline backoff

[Mallets]

This PR improves the pipeline backoff implementation and reduces the possibility of spinning tasks.

Without this patch, it may happen before that in high-load scenario we could observe the following logs:

2024-06-07T07:56:30.783760Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)
2024-06-07T07:56:30.783771Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)
2024-06-07T07:56:30.783772Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)
2024-06-07T07:56:30.783774Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)
2024-06-07T07:56:30.783775Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)
2024-06-07T07:56:30.783830Z  WARN tx-0 ThreadId(16) zenoh_transport::common::pipeline: Pipeline pull backoff overflow detected! Retrying in 4294967295ns. (tcp/127.0.0.1:7447 => tcp/127.0.0.1:51696)

With this patch, in my tests I didn't observe those logs anymore.
What I expect is that this patch heavily mitigates the problem although it might not solve it in 100% of the cases.

[Mallets]

@yellowhatter can you please help in reviewing this PR?

[yellowhatter]

@Mallets looks nice!

I've been in parallel working on a little bit different design. Our queue implementation uses mutex to achieve synchronization around current batch. This is a weak point, because it makes a producer thread to go to sleep when current batch is being processed by a consumer task:

pipeline.rs: 138:
// Lock the current serialization batch.
let mut c_guard = self.mutex.current();

My idea is that the producer thread should take care about always having "one batch behind": after each serialization check atomic count of batches in s_out and move current batch (potentially not full) to s_out if s_out is empty. In this case, there won't be any producer/consumer contention around current batch. Regarding the performance in terms of batch size utilization, this approach will essentially regulate batch utilization, because misperforming (due to sending non-full batches) consumer will cause producer to manage to completely fill current, restoring consumer performance....
The only problem of this approach is how to efficiently consume non-empty current when there is no new serializations