pseudotcp_adapter.cc 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507
  1. // Copyright 2015 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "remoting/protocol/pseudotcp_adapter.h"
  5. #include <stddef.h>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/compiler_specific.h"
  9. #include "base/logging.h"
  10. #include "base/time/time.h"
  11. #include "base/timer/timer.h"
  12. #include "net/base/address_list.h"
  13. #include "net/base/completion_once_callback.h"
  14. #include "net/base/io_buffer.h"
  15. #include "net/base/net_errors.h"
  16. #include "net/traffic_annotation/network_traffic_annotation.h"
  17. #include "remoting/protocol/p2p_datagram_socket.h"
  18. using cricket::PseudoTcp;
  19. namespace {
  20. const int kReadBufferSize = 65536; // Maximum size of a packet.
  21. const uint16_t kDefaultMtu = 1280;
  22. } // namespace
  23. namespace remoting {
  24. namespace protocol {
  25. class PseudoTcpAdapter::Core : public cricket::IPseudoTcpNotify,
  26. public base::RefCounted<Core> {
  27. public:
  28. explicit Core(std::unique_ptr<P2PDatagramSocket> socket);
  29. Core(const Core&) = delete;
  30. Core& operator=(const Core&) = delete;
  31. // Functions used to implement net::StreamSocket.
  32. int Read(const scoped_refptr<net::IOBuffer>& buffer,
  33. int buffer_size,
  34. net::CompletionOnceCallback callback);
  35. int Write(const scoped_refptr<net::IOBuffer>& buffer,
  36. int buffer_size,
  37. net::CompletionOnceCallback callback,
  38. const net::NetworkTrafficAnnotationTag& traffic_annotation);
  39. net::CompletionOnceCallback Connect(net::CompletionOnceCallback callback);
  40. // cricket::IPseudoTcpNotify interface.
  41. // These notifications are triggered from NotifyPacket.
  42. void OnTcpOpen(cricket::PseudoTcp* tcp) override;
  43. void OnTcpReadable(cricket::PseudoTcp* tcp) override;
  44. void OnTcpWriteable(cricket::PseudoTcp* tcp) override;
  45. // This is triggered by NotifyClock or NotifyPacket.
  46. void OnTcpClosed(cricket::PseudoTcp* tcp, uint32_t error) override;
  47. // This is triggered by NotifyClock, NotifyPacket, Recv and Send.
  48. WriteResult TcpWritePacket(cricket::PseudoTcp* tcp,
  49. const char* buffer,
  50. size_t len) override;
  51. void SetAckDelay(int delay_ms);
  52. void SetNoDelay(bool no_delay);
  53. void SetReceiveBufferSize(int32_t size);
  54. void SetSendBufferSize(int32_t size);
  55. void SetWriteWaitsForSend(bool write_waits_for_send);
  56. void DeleteSocket();
  57. private:
  58. friend class base::RefCounted<Core>;
  59. ~Core() override;
  60. // These are invoked by the underlying Socket, and may trigger callbacks.
  61. // They hold a reference to |this| while running, to protect from deletion.
  62. void OnRead(int result);
  63. void OnWritten(int result);
  64. // These may trigger callbacks, so the holder must hold a reference on
  65. // the stack while calling them.
  66. void DoReadFromSocket();
  67. void HandleReadResults(int result);
  68. void HandleTcpClock();
  69. // Checks if current write has completed in the write-waits-for-send
  70. // mode.
  71. void CheckWriteComplete();
  72. // This re-sets |timer| without triggering callbacks.
  73. void AdjustClock();
  74. net::CompletionOnceCallback connect_callback_;
  75. net::CompletionOnceCallback read_callback_;
  76. net::CompletionOnceCallback write_callback_;
  77. cricket::PseudoTcp pseudo_tcp_;
  78. std::unique_ptr<P2PDatagramSocket> socket_;
  79. scoped_refptr<net::IOBuffer> read_buffer_;
  80. int read_buffer_size_;
  81. scoped_refptr<net::IOBuffer> write_buffer_;
  82. int write_buffer_size_;
  83. // Whether we need to wait for data to be sent before completing write.
  84. bool write_waits_for_send_;
  85. // Set to true in the write-waits-for-send mode when we've
  86. // successfully writtend data to the send buffer and waiting for the
  87. // data to be sent to the remote end.
  88. bool waiting_write_position_;
  89. // Number of the bytes written by the last write stored while we wait
  90. // for the data to be sent (i.e. when waiting_write_position_ = true).
  91. int last_write_result_;
  92. bool socket_write_pending_;
  93. scoped_refptr<net::IOBuffer> socket_read_buffer_;
  94. base::OneShotTimer timer_;
  95. };
  96. PseudoTcpAdapter::Core::Core(std::unique_ptr<P2PDatagramSocket> socket)
  97. : pseudo_tcp_(this, 0),
  98. socket_(std::move(socket)),
  99. write_waits_for_send_(false),
  100. waiting_write_position_(false),
  101. socket_write_pending_(false) {
  102. // Doesn't trigger callbacks.
  103. pseudo_tcp_.NotifyMTU(kDefaultMtu);
  104. }
  105. PseudoTcpAdapter::Core::~Core() = default;
  106. int PseudoTcpAdapter::Core::Read(const scoped_refptr<net::IOBuffer>& buffer,
  107. int buffer_size,
  108. net::CompletionOnceCallback callback) {
  109. DCHECK(read_callback_.is_null());
  110. // Reference the Core in case a callback deletes the adapter.
  111. scoped_refptr<Core> core(this);
  112. int result = pseudo_tcp_.Recv(buffer->data(), buffer_size);
  113. if (result < 0) {
  114. result = net::MapSystemError(pseudo_tcp_.GetError());
  115. DCHECK(result < 0);
  116. }
  117. if (result == net::ERR_IO_PENDING) {
  118. read_buffer_ = buffer;
  119. read_buffer_size_ = buffer_size;
  120. read_callback_ = std::move(callback);
  121. }
  122. AdjustClock();
  123. return result;
  124. }
  125. int PseudoTcpAdapter::Core::Write(
  126. const scoped_refptr<net::IOBuffer>& buffer,
  127. int buffer_size,
  128. net::CompletionOnceCallback callback,
  129. const net::NetworkTrafficAnnotationTag& /*traffic_annotation*/) {
  130. DCHECK(write_callback_.is_null());
  131. // Reference the Core in case a callback deletes the adapter.
  132. scoped_refptr<Core> core(this);
  133. int result = pseudo_tcp_.Send(buffer->data(), buffer_size);
  134. if (result < 0) {
  135. result = net::MapSystemError(pseudo_tcp_.GetError());
  136. DCHECK(result < 0);
  137. }
  138. AdjustClock();
  139. if (result == net::ERR_IO_PENDING) {
  140. write_buffer_ = buffer;
  141. write_buffer_size_ = buffer_size;
  142. write_callback_ = std::move(callback);
  143. return result;
  144. }
  145. if (result < 0)
  146. return result;
  147. // Need to wait until the data is sent to the peer when
  148. // send-confirmation mode is enabled.
  149. if (write_waits_for_send_ && pseudo_tcp_.GetBytesBufferedNotSent() > 0) {
  150. DCHECK(!waiting_write_position_);
  151. waiting_write_position_ = true;
  152. last_write_result_ = result;
  153. write_buffer_ = buffer;
  154. write_buffer_size_ = buffer_size;
  155. write_callback_ = std::move(callback);
  156. return net::ERR_IO_PENDING;
  157. }
  158. return result;
  159. }
  160. net::CompletionOnceCallback PseudoTcpAdapter::Core::Connect(
  161. net::CompletionOnceCallback callback) {
  162. DCHECK_EQ(pseudo_tcp_.State(), cricket::PseudoTcp::TCP_LISTEN);
  163. // Reference the Core in case a callback deletes the adapter.
  164. scoped_refptr<Core> core(this);
  165. // Start the connection attempt.
  166. int result = pseudo_tcp_.Connect();
  167. if (result < 0)
  168. return callback;
  169. AdjustClock();
  170. connect_callback_ = std::move(callback);
  171. DoReadFromSocket();
  172. return {};
  173. }
  174. void PseudoTcpAdapter::Core::OnTcpOpen(PseudoTcp* tcp) {
  175. DCHECK(tcp == &pseudo_tcp_);
  176. if (connect_callback_)
  177. std::move(connect_callback_).Run(net::OK);
  178. OnTcpReadable(tcp);
  179. OnTcpWriteable(tcp);
  180. }
  181. void PseudoTcpAdapter::Core::OnTcpReadable(PseudoTcp* tcp) {
  182. DCHECK_EQ(tcp, &pseudo_tcp_);
  183. if (read_callback_.is_null())
  184. return;
  185. int result = pseudo_tcp_.Recv(read_buffer_->data(), read_buffer_size_);
  186. if (result < 0) {
  187. result = net::MapSystemError(pseudo_tcp_.GetError());
  188. DCHECK(result < 0);
  189. if (result == net::ERR_IO_PENDING)
  190. return;
  191. }
  192. AdjustClock();
  193. read_buffer_.reset();
  194. std::move(read_callback_).Run(result);
  195. }
  196. void PseudoTcpAdapter::Core::OnTcpWriteable(PseudoTcp* tcp) {
  197. DCHECK_EQ(tcp, &pseudo_tcp_);
  198. if (write_callback_.is_null())
  199. return;
  200. if (waiting_write_position_) {
  201. CheckWriteComplete();
  202. return;
  203. }
  204. int result = pseudo_tcp_.Send(write_buffer_->data(), write_buffer_size_);
  205. if (result < 0) {
  206. result = net::MapSystemError(pseudo_tcp_.GetError());
  207. DCHECK(result < 0);
  208. if (result == net::ERR_IO_PENDING)
  209. return;
  210. }
  211. AdjustClock();
  212. if (write_waits_for_send_ && pseudo_tcp_.GetBytesBufferedNotSent() > 0) {
  213. DCHECK(!waiting_write_position_);
  214. waiting_write_position_ = true;
  215. last_write_result_ = result;
  216. return;
  217. }
  218. write_buffer_.reset();
  219. std::move(write_callback_).Run(result);
  220. }
  221. void PseudoTcpAdapter::Core::OnTcpClosed(PseudoTcp* tcp, uint32_t error) {
  222. DCHECK_EQ(tcp, &pseudo_tcp_);
  223. if (!connect_callback_.is_null()) {
  224. std::move(connect_callback_).Run(net::MapSystemError(error));
  225. }
  226. if (!read_callback_.is_null()) {
  227. std::move(read_callback_).Run(net::MapSystemError(error));
  228. }
  229. if (!write_callback_.is_null()) {
  230. std::move(write_callback_).Run(net::MapSystemError(error));
  231. }
  232. }
  233. void PseudoTcpAdapter::Core::SetAckDelay(int delay_ms) {
  234. pseudo_tcp_.SetOption(cricket::PseudoTcp::OPT_ACKDELAY, delay_ms);
  235. }
  236. void PseudoTcpAdapter::Core::SetNoDelay(bool no_delay) {
  237. pseudo_tcp_.SetOption(cricket::PseudoTcp::OPT_NODELAY, no_delay ? 1 : 0);
  238. }
  239. void PseudoTcpAdapter::Core::SetReceiveBufferSize(int32_t size) {
  240. pseudo_tcp_.SetOption(cricket::PseudoTcp::OPT_RCVBUF, size);
  241. }
  242. void PseudoTcpAdapter::Core::SetSendBufferSize(int32_t size) {
  243. pseudo_tcp_.SetOption(cricket::PseudoTcp::OPT_SNDBUF, size);
  244. }
  245. void PseudoTcpAdapter::Core::SetWriteWaitsForSend(bool write_waits_for_send) {
  246. write_waits_for_send_ = write_waits_for_send;
  247. }
  248. void PseudoTcpAdapter::Core::DeleteSocket() {
  249. // Don't dispatch outstanding callbacks when the socket is deleted.
  250. read_callback_.Reset();
  251. read_buffer_.reset();
  252. write_callback_.Reset();
  253. write_buffer_.reset();
  254. connect_callback_.Reset();
  255. socket_.reset();
  256. }
  257. cricket::IPseudoTcpNotify::WriteResult PseudoTcpAdapter::Core::TcpWritePacket(
  258. PseudoTcp* tcp,
  259. const char* buffer,
  260. size_t len) {
  261. DCHECK_EQ(tcp, &pseudo_tcp_);
  262. // If we already have a write pending, we behave like a congested network,
  263. // returning success for the write, but dropping the packet. PseudoTcp will
  264. // back-off and retransmit, adjusting for the perceived congestion.
  265. if (socket_write_pending_)
  266. return IPseudoTcpNotify::WR_SUCCESS;
  267. scoped_refptr<net::IOBuffer> write_buffer =
  268. base::MakeRefCounted<net::IOBuffer>(len);
  269. memcpy(write_buffer->data(), buffer, len);
  270. // Our underlying socket is datagram-oriented, which means it should either
  271. // send exactly as many bytes as we requested, or fail.
  272. int result;
  273. if (socket_) {
  274. result =
  275. socket_->Send(write_buffer.get(), len,
  276. base::BindRepeating(&PseudoTcpAdapter::Core::OnWritten,
  277. base::Unretained(this)));
  278. } else {
  279. result = net::ERR_CONNECTION_CLOSED;
  280. }
  281. if (result == net::ERR_IO_PENDING) {
  282. socket_write_pending_ = true;
  283. return IPseudoTcpNotify::WR_SUCCESS;
  284. } else if (result == net::ERR_MSG_TOO_BIG) {
  285. return IPseudoTcpNotify::WR_TOO_LARGE;
  286. } else if (result < 0) {
  287. return IPseudoTcpNotify::WR_FAIL;
  288. } else {
  289. return IPseudoTcpNotify::WR_SUCCESS;
  290. }
  291. }
  292. void PseudoTcpAdapter::Core::DoReadFromSocket() {
  293. if (!socket_read_buffer_.get())
  294. socket_read_buffer_ = base::MakeRefCounted<net::IOBuffer>(kReadBufferSize);
  295. int result = 1;
  296. while (socket_ && result > 0) {
  297. result = socket_->Recv(socket_read_buffer_.get(), kReadBufferSize,
  298. base::BindRepeating(&PseudoTcpAdapter::Core::OnRead,
  299. base::Unretained(this)));
  300. if (result != net::ERR_IO_PENDING)
  301. HandleReadResults(result);
  302. }
  303. }
  304. void PseudoTcpAdapter::Core::HandleReadResults(int result) {
  305. if (result <= 0) {
  306. LOG(ERROR) << "Read returned " << result;
  307. return;
  308. }
  309. // TODO(wez): Disconnect on failure of NotifyPacket?
  310. pseudo_tcp_.NotifyPacket(socket_read_buffer_->data(), result);
  311. AdjustClock();
  312. CheckWriteComplete();
  313. }
  314. void PseudoTcpAdapter::Core::OnRead(int result) {
  315. // Reference the Core in case a callback deletes the adapter.
  316. scoped_refptr<Core> core(this);
  317. HandleReadResults(result);
  318. if (result >= 0)
  319. DoReadFromSocket();
  320. }
  321. void PseudoTcpAdapter::Core::OnWritten(int result) {
  322. // Reference the Core in case a callback deletes the adapter.
  323. scoped_refptr<Core> core(this);
  324. socket_write_pending_ = false;
  325. if (result < 0) {
  326. LOG(WARNING) << "Write failed. Error code: " << result;
  327. }
  328. }
  329. void PseudoTcpAdapter::Core::AdjustClock() {
  330. long timeout = 0;
  331. if (pseudo_tcp_.GetNextClock(PseudoTcp::Now(), timeout)) {
  332. timer_.Stop();
  333. timer_.Start(FROM_HERE, base::Milliseconds(std::max(timeout, 0L)), this,
  334. &PseudoTcpAdapter::Core::HandleTcpClock);
  335. }
  336. }
  337. void PseudoTcpAdapter::Core::HandleTcpClock() {
  338. // Reference the Core in case a callback deletes the adapter.
  339. scoped_refptr<Core> core(this);
  340. pseudo_tcp_.NotifyClock(PseudoTcp::Now());
  341. AdjustClock();
  342. CheckWriteComplete();
  343. }
  344. void PseudoTcpAdapter::Core::CheckWriteComplete() {
  345. if (!write_callback_.is_null() && waiting_write_position_) {
  346. if (pseudo_tcp_.GetBytesBufferedNotSent() == 0) {
  347. waiting_write_position_ = false;
  348. write_buffer_.reset();
  349. std::move(write_callback_).Run(last_write_result_);
  350. }
  351. }
  352. }
  353. // Public interface implementation.
  354. PseudoTcpAdapter::PseudoTcpAdapter(std::unique_ptr<P2PDatagramSocket> socket)
  355. : core_(new Core(std::move(socket))) {}
  356. PseudoTcpAdapter::~PseudoTcpAdapter() {
  357. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  358. // Make sure that the underlying socket is destroyed before PseudoTcp.
  359. core_->DeleteSocket();
  360. }
  361. int PseudoTcpAdapter::Read(const scoped_refptr<net::IOBuffer>& buffer,
  362. int buffer_size,
  363. net::CompletionOnceCallback callback) {
  364. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  365. return core_->Read(buffer, buffer_size, std::move(callback));
  366. }
  367. int PseudoTcpAdapter::Write(
  368. const scoped_refptr<net::IOBuffer>& buffer,
  369. int buffer_size,
  370. net::CompletionOnceCallback callback,
  371. const net::NetworkTrafficAnnotationTag& traffic_annotation) {
  372. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  373. return core_->Write(buffer, buffer_size, std::move(callback),
  374. traffic_annotation);
  375. }
  376. int PseudoTcpAdapter::SetReceiveBufferSize(int32_t size) {
  377. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  378. core_->SetReceiveBufferSize(size);
  379. return net::OK;
  380. }
  381. int PseudoTcpAdapter::SetSendBufferSize(int32_t size) {
  382. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  383. core_->SetSendBufferSize(size);
  384. return net::OK;
  385. }
  386. net::CompletionOnceCallback PseudoTcpAdapter::Connect(
  387. net::CompletionOnceCallback callback) {
  388. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  389. return core_->Connect(std::move(callback));
  390. }
  391. void PseudoTcpAdapter::SetAckDelay(int delay_ms) {
  392. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  393. core_->SetAckDelay(delay_ms);
  394. }
  395. void PseudoTcpAdapter::SetNoDelay(bool no_delay) {
  396. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  397. core_->SetNoDelay(no_delay);
  398. }
  399. void PseudoTcpAdapter::SetWriteWaitsForSend(bool write_waits_for_send) {
  400. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  401. core_->SetWriteWaitsForSend(write_waits_for_send);
  402. }
  403. } // namespace protocol
  404. } // namespace remoting