[BACKPORT 2.18] Track and automatically recover from stuck outbound c…

…alls

Summary:
Track all outbound calls submitted to a Reactor in a multi-index container in that reactor. Remove them from the container as soon as we learn the callback has been called, using a multiple-producer single-consumer (MPSC) queue where the consumer is the reactor thread. Report stuck calls with detailed debug information. A call is considered stuck if it is a certain time past its expiration (specified by reactor_based_outbound_call_expiration_delay_ms -- set it to 0 to disable this feature). If the call does not have a timeout, the default value stuck_outbound_call_default_timeout_sec is used for reporting only, but not for timeout enforcement.

OutboundCall now keeps track of its expiration deadline, and an additional field active_call_state_, which indicates whether the call was added to the connection's active_calls_ in the corresponding connection, or removed from it, and the removal reason.

If in the OutboundCall destructor the state of the call is not final, or the callback has not been called, we will log that even in release build.

In OutboundCall::SetState, if the current state is already finished, treat this the same as any other invalid state transition. Logically, there should never be a race between e.g. processing a call response and a call timeout, because both events happen on the reactor thread.

Fixing error handling in DoQueueOutboundData. If there is an error sending the call, we destroy the connection.

Introduce a typedef CallHandle and a special value kUnknownCallHandle instead of just using the max value of size_t in case the call handle is unknown or not being used. Change the error handling logic in Connection::DoQueueOutboundData to avoid returning kUnknownCallHandle in case of errors, and make sure the callback is called on the call in case of those errors. The connection was already being shut down.

Update the YB_STRUCT_TO_STRING macro implementation to always add parentheses around an explicitly specified field value.

We introduce multiple ways to simulate stuck outbound calls for testing.

TEST_simulated_sent_stuck_call_probability specifies the probability of pretending a call is sent to the remote server in Connection::QueueOutboundCall, but instead just transitioning it to a SENT state. This is similar to the situation that we have observed.
TEST_simulated_failure_to_send_call_probability specifies the probability of a network error in stream_->Send() called from Connection::DoQueueOutboundData. This will cause the connection to be closed. Prior to this revision, this kind of an error would cause Connection::QueueOutboundCall to attempt to schedule a timer in a connection that has already been shut down.
TEST_outbound_call_skip_callback_probability specifies the probability of skipping calling the callback on a particular remote RPC call. We don't do this for local calls.
Also replacing a DFATAL with an WARNING log in InboundCall::QueueResponse to avoid crashing tests during shutdown. Failures to queue a response on the server side should just result in timeouts on the client side in the worst case.

We are combining multiple original revisions from the master branch to avoid introducing and then fixing a memory leak bug, and to keep the resulting patch cleaner.

Original revisions:

- D27735 ( 1fbff49 ) [yugabyte#18685] Track stuck OutboundCalls and enforce timeouts at Reactor level
- D27919 ( ddb817b ) [yugabyte#18808] A write-once lock-free weak pointer class: it is frequently necessary to have a weak pointer equivalent that can only be written to once, but can be read without locking for performance.
- D28085 ( a5ba08f ) [yugabyte#18855] Add functions for formatting a CoarseTimePoint relative to the current time. Useful for logging deadlines.
- D28138 ( d0a44db ) [yugabyte#18877] Log invalid OutboundCall state transitions and other OutboundCall improvements. OutboundCall will track invalid state transitions and report them in the DebugString and in the destructor, in addition to reporting them as they happen. Do not allow OutboundCall callback to be invoked more than once and log occurrences of that. Use a compare-and-swap loop in RpcCall::Transferred and log invalid transfer_state transitions. invoke_callback_time_ and sent_time_ in OutboundCall were being accessed without synchronization, so making them atomic. OutboundCall::DebugString should not queue a task in the reactor to print the connection details in relation to the call anymore. That is now done by a separate method, QueueDumpConnectionState, called explicitly.
- D28531 ( 0648341 ) [yugabyte#19090] Fix a memory leak in tracked outbound calls in Reactor. When an OutboundCall callback is invoked, notify the reactor through a multi-producer single-consumer queue that is accessed from the OutboundCall via a weak pointer. Also switching OutboundCall allocation away from using make_shared to prevent long-lived weak OutboundCall pointers from consuming memory.

Test Plan:
Jenkins

---

Manual testing details below.

First test mode
===============

```
bin/yb-ctl wipe_restart  --tserver_flags="TEST_simulated_sent_stuck_call_probability=0.0001,reactor_based_outbound_call_expiration_delay_ms=1000" --rf=3

java -jar yb-sample-apps.jar  --workload CassandraBatchKeyValue --nodes 127.0.0.1:9042,127.0.0.2:9042,127.0.0.3:9042 --num_threads_read 16 --num_threads_write 16 --num_reads 1000000000000 --num_writes 1000000000000
```

Look for output like this in the logs:

```
W1017 21:23:13.309160 2187266 connection.cc:300] Connection (0x0000556728cd2158) client 127.0.0.1:38331 => 127.0.0.2:9100: Simulating a call stuck in SENT state: RPC call 0x000055672aea9080: yb.consensus.ConsensusService.UpdateConsensus -> { remote: 127.0.0.2:9100 idx: 4 protocol: 0x00007f03547e75d0 -> tcpc }, id: 70811, state: SENT, transfer_state: PENDING, start_time: 257388.919s (0.004s ago), sent_time: -inf, trigger_callback_time: -inf, invoke_callback_time: -inf, expiration_time: +inf, now: 257388.923s, connection: Connection (0x0000556728cd2158) client 127.0.0.1:38331 => 127.0.0.2:9100, active_call_state: kNotAdded
...
W1017 21:23:17.377820 2187266 reactor.cc:682] TabletServer_R000: Stuck OutboundCall: RPC call 0x000055672aea9080: yb.consensus.ConsensusService.UpdateConsensus -> { remote: 127.0.0.2:9100 idx: 4 protocol: 0x00007f03547e75d0 -> tcpc }, id: 70811, state: SENT, transfer_state: PENDING, start_time: 257388.919s (4.073s ago), sent_time: -inf, trigger_callback_time: -inf, invoke_callback_time: -inf, expiration_time: 257391.923s (1.069s ago), now: 257392.992s, connection: Connection (0x0000556728cd2158) client 127.0.0.1:38331 => 127.0.0.2:9100, active_call_state: kNotAdded (forcing a timeout)
W1017 21:23:17.378927 2187522 consensus_peers.cc:605] T 775284184b9241a89206fc0cdaa59302 P a5cddad5f52643e183e281d890cea703 -> Peer 1b86c67a8a50457794972712d3f0a860 ([host: "127.0.0.2" port: 9100], []): Couldn't send request.  Status: Timed out (yb/rpc/outbound_call.cc:614): UpdateConsensus RPC (request call id 70811) to 127.0.0.2:9100 timed out after 3.000s. Retrying in the next heartbeat period. Already tried 1 times. State: 2
```

Ensure that all stuck outbound calls are eventually detected. After stopping the workload, the output of the command below should gradually decrease to zero after a couple of minutes:

```
log_path="$HOME/yugabyte-data/node-1/disk-1/yb-data/tserver/logs/yb-tserver.INFO"; grep "Simulating a call stuck in SENT state" "$log_path" | egrep -Eo 'id: [0-9]+' | awk '{print $NF}' | sort -n | while read c; do if ! grep "id: $c," "$log_path" | grep -q "Stuck OutboundCall"; then echo "Undetected stuck OutboundCall: id=$c"; fi; done | wc -l
```

Also, these stuck calls should stop appearing in the log because they would be forcibly timed out. If they were not, they would be logged every minute.

Second test mode
================

```
bin/yb-ctl wipe_restart  --tserver_flags="TEST_simulated_failure_to_send_call_probability=0.00001,reactor_based_outbound_call_expiration_delay_ms=1000" --rf=3
```

Same workload as above.

Look for "Simulated failure to send outbound data" in the tablet server log:

```
W1017 21:37:57.068120 2190494 connection.cc:389] Simulated network failure: Network error (yb/rpc/connection.cc:388): Simulated failure to send outbound data for 0x00005623a3e23b80 -> Call yb.tserver.TabletServerService.Write 127.0.0.3:36207 => 127.0.0.1:9100 (request call id 907743)
...
W1017 21:37:57.068182 2190494 inbound_call.cc:99] 0x00005623a3e23b80 -> Call yb.tserver.TabletServerService.Write 127.0.0.3:36207 => 127.0.0.1:9100 (request call id 907743): Connection torn down before Call yb.tserver.TabletServerService.Write 127.0.0.3:36207 => 127.0.0.1:9100 (request call id 907743) could send its response: Network error (yb/rpc/connection.cc:388): Simulated failure to send outbound data for 0x00005623a3e23b80 -> Call yb.tserver.TabletServerService.Write 127.0.0.3:36207 => 127.0.0.1:9100 (request call id 907743)
```

There should be no stuck outbound calls in the log, because these simulated failures will just result in the connection being closed. Depending on the value TEST_simulated_failure_to_send_call_probability, the workload may succeed or fail. With the value 0.00001, it manages to make progress in my experience.

Third test mode
===============

```
bin/yb-ctl wipe_restart  --tserver_flags="TEST_outbound_call_skip_callback_probability=0.0001,reactor_based_outbound_call_expiration_delay_ms=1000" --rf=3
```

Same workload as above.

Look for "Skipping OutboundCall callback as a test" and "Stuck OutboundCall" in the log.

Output:
```
W1017 21:40:48.326967 2193494 outbound_call.cc:421] OutboundCall@0x00005578719ec2c0: Skipping OutboundCall callback as a test: RPC call 0x00005578719ec2c0: yb.consensus.ConsensusService.UpdateConsensus -> { remote: 127.0.0.2:9100 idx: 2 protocol: 0x00007f11979b25d0 -> tcpc }, id: 3497, state: FINISHED_SUCCESS, transfer_state: FINISHED, start_time: 258443.926s (0.016s ago), sent_time: 258443.926s (0.016s ago), trigger_callback_time: 258443.941s (0.001s ago), invoke_callback_time: -inf, expiration_time: 258446.926s (2.983s from now), now: 258443.942s, connection: Connection (0x000055786dd06158) client 127.0.0.1:52911 => 127.0.0.2:9100, active_call_state: kErasedOnResponse
. . .
W1017 21:40:52.362062 2193263 reactor.cc:682] TabletServer_R002: Stuck OutboundCall: RPC call 0x00005578719ec2c0: yb.consensus.ConsensusService.UpdateConsensus -> { remote: 127.0.0.2:9100 idx: 2 protocol: 0x00007f11979b25d0 -> tcpc }, id: 3497, state: FINISHED_SUCCESS, transfer_state: FINISHED, start_time: 258443.926s (4.050s ago), sent_time: 258443.926s (4.050s ago), trigger_callback_time: 258443.941s (4.035s ago), invoke_callback_time: -inf, expiration_time: 258446.926s (1.050s ago), now: 258447.976s, connection: Connection (0x000055786dd06158) client 127.0.0.1:52911 => 127.0.0.2:9100, active_call_state: kErasedOnResponse
. . .
W1017 21:41:22.461496 2193263 reactor.cc:682] TabletServer_R002: Stuck OutboundCall: RPC call 0x00005578719ec2c0: yb.consensus.ConsensusService.UpdateConsensus -> { remote: 127.0.0.2:9100 idx: 2 protocol: 0x00007f11979b25d0 -> tcpc }, id: 3497, state: FINISHED_SUCCESS, transfer_state: FINISHED, start_time: 258443.926s (34.149s ago), sent_time: 258443.926s (34.149s ago), trigger_callback_time: 258443.941s (34.134s ago), invoke_callback_time: -inf, expiration_time: 258446.926s (31.149s ago), now: 258478.075s, connection: Connection (0x000055786dd06158) client 127.0.0.1:52911 => 127.0.0.2:9100, active_call_state: kErasedOnResponse
```

Similarly to the first test, wait for all stuck outbound calls to get reported:
```
log_path="$HOME/yugabyte-data/node-1/disk-1/yb-data/tserver/logs/yb-tserver.INFO"; grep "Skipping OutboundCall callback as a test" "$log_path" | egrep -Eo 'id: [0-9]+' | awk '{print $NF}' | sort -n | while read c; do if ! grep "id: $c," "$log_path" | grep -q "Stuck OutboundCall"; then echo "Undetected stuck OutboundCall: id=$c"; fi; done | wc -l
```

In this mode, because we are skipping callbacks but allowing state transitions to the finished state, forced expiration at reactor level does not work.

Reviewers: bogdan, rthallam, sergei

Reviewed By: sergei

Subscribers: ybase

Differential Revision: https://phorge.dev.yugabyte.com/D29215

src/yb/docdb/deadline_info.cc

@@ -43,8 +43,9 @@ bool DeadlineInfo::CheckAndSetDeadlinePassed() {
 }
 
 std::string DeadlineInfo::ToString() const {
+  auto now = CoarseMonoClock::now();
   return Format("{ now: $0 deadline: $1 counter: $2 }",
-                CoarseMonoClock::now(), deadline_, counter_);
+                now, ToStringRelativeToNow(deadline_, now), counter_);
 }
 
 void SimulateTimeoutIfTesting(CoarseTimePoint* deadline) {