| 183 | } |
| 184 | |
| 185 | void ProgressiveAttachment::MarkRPCAsDone(bool rpc_failed) { |
| 186 | // Notes: |
| 187 | // * Writing here is more timely than being flushed in next Write(), in |
| 188 | // some extreme situations, the delay before next Write() may be |
| 189 | // significant. |
| 190 | // * Write() should be outside lock because a failed write triggers |
| 191 | // SetFailed() which runs the closure to NotifyOnStopped() which may |
| 192 | // call methods requesting for the lock again. Another solution is to |
| 193 | // use recursive lock. |
| 194 | // * _saved_buf can't be much longer than FLAGS_socket_max_unwritten_bytes, |
| 195 | // ignoring EOVERCROWDED simplifies the error handling. |
| 196 | // * If the iteration repeats too many times, _pause_from_mark_rpc_as_done |
| 197 | // is set to true to make Write() fails with EOVERCROWDED temporarily to |
| 198 | // stop _saved_buf from growing. |
| 199 | const int MAX_TRY = 3; |
| 200 | int ntry = 0; |
| 201 | bool permanent_error = false; |
| 202 | do { |
| 203 | std::unique_lock<butil::Mutex> mu(_mutex); |
| 204 | if (_saved_buf.empty() || permanent_error || rpc_failed) { |
| 205 | butil::IOBuf tmp; |
| 206 | tmp.swap(_saved_buf); // Clear _saved_buf outside lock. |
| 207 | _pause_from_mark_rpc_as_done = false; |
| 208 | _rpc_state.store((rpc_failed? RPC_FAILED: RPC_SUCCEED), |
| 209 | butil::memory_order_release); |
| 210 | mu.unlock(); |
| 211 | return; |
| 212 | } |
| 213 | if (++ntry > MAX_TRY) { |
| 214 | _pause_from_mark_rpc_as_done = true; |
| 215 | } |
| 216 | butil::IOBuf copied; |
| 217 | copied.swap(_saved_buf); |
| 218 | mu.unlock(); |
| 219 | Socket::WriteOptions wopt; |
| 220 | wopt.ignore_eovercrowded = true; |
| 221 | if (_httpsock->Write(&copied, &wopt) != 0) { |
| 222 | permanent_error = true; |
| 223 | } |
| 224 | } while (true); |
| 225 | } |
| 226 | |
| 227 | butil::EndPoint ProgressiveAttachment::remote_side() const { |
| 228 | return _httpsock ? _httpsock->remote_side() : butil::EndPoint(); |