1 /*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include "rtc_base/async_tcp_socket.h"
12
13 #include <stdint.h>
14 #include <string.h>
15
16 #include <algorithm>
17 #include <memory>
18
19 #include "api/array_view.h"
20 #include "rtc_base/byte_order.h"
21 #include "rtc_base/checks.h"
22 #include "rtc_base/logging.h"
23 #include "rtc_base/network/sent_packet.h"
24 #include "rtc_base/third_party/sigslot/sigslot.h"
25 #include "rtc_base/time_utils.h" // for TimeMillis
26
27 #if defined(WEBRTC_POSIX)
28 #include <errno.h>
29 #endif // WEBRTC_POSIX
30
31 namespace rtc {
32
33 static const size_t kMaxPacketSize = 64 * 1024;
34
35 typedef uint16_t PacketLength;
36 static const size_t kPacketLenSize = sizeof(PacketLength);
37
38 static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;
39
40 // The input buffer will be resized so that at least kMinimumRecvSize bytes can
41 // be received (but it will not grow above the maximum size passed to the
42 // constructor).
43 static const size_t kMinimumRecvSize = 128;
44
45 static const int kListenBacklog = 5;
46
47 // Binds and connects `socket`
ConnectSocket(rtc::Socket * socket,const rtc::SocketAddress & bind_address,const rtc::SocketAddress & remote_address)48 Socket* AsyncTCPSocketBase::ConnectSocket(
49 rtc::Socket* socket,
50 const rtc::SocketAddress& bind_address,
51 const rtc::SocketAddress& remote_address) {
52 std::unique_ptr<rtc::Socket> owned_socket(socket);
53 if (socket->Bind(bind_address) < 0) {
54 RTC_LOG(LS_ERROR) << "Bind() failed with error " << socket->GetError();
55 return nullptr;
56 }
57 if (socket->Connect(remote_address) < 0) {
58 RTC_LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
59 return nullptr;
60 }
61 return owned_socket.release();
62 }
63
AsyncTCPSocketBase(Socket * socket,size_t max_packet_size)64 AsyncTCPSocketBase::AsyncTCPSocketBase(Socket* socket,
65 size_t max_packet_size)
66 : socket_(socket),
67 max_insize_(max_packet_size),
68 max_outsize_(max_packet_size) {
69 inbuf_.EnsureCapacity(kMinimumRecvSize);
70
71 socket_->SignalConnectEvent.connect(this,
72 &AsyncTCPSocketBase::OnConnectEvent);
73 socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent);
74 socket_->SignalWriteEvent.connect(this, &AsyncTCPSocketBase::OnWriteEvent);
75 socket_->SignalCloseEvent.connect(this, &AsyncTCPSocketBase::OnCloseEvent);
76 }
77
~AsyncTCPSocketBase()78 AsyncTCPSocketBase::~AsyncTCPSocketBase() {}
79
GetLocalAddress() const80 SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
81 return socket_->GetLocalAddress();
82 }
83
GetRemoteAddress() const84 SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
85 return socket_->GetRemoteAddress();
86 }
87
Close()88 int AsyncTCPSocketBase::Close() {
89 return socket_->Close();
90 }
91
GetState() const92 AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
93 switch (socket_->GetState()) {
94 case Socket::CS_CLOSED:
95 return STATE_CLOSED;
96 case Socket::CS_CONNECTING:
97 return STATE_CONNECTING;
98 case Socket::CS_CONNECTED:
99 return STATE_CONNECTED;
100 default:
101 RTC_DCHECK_NOTREACHED();
102 return STATE_CLOSED;
103 }
104 }
105
GetOption(Socket::Option opt,int * value)106 int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
107 return socket_->GetOption(opt, value);
108 }
109
SetOption(Socket::Option opt,int value)110 int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
111 return socket_->SetOption(opt, value);
112 }
113
GetError() const114 int AsyncTCPSocketBase::GetError() const {
115 return socket_->GetError();
116 }
117
SetError(int error)118 void AsyncTCPSocketBase::SetError(int error) {
119 return socket_->SetError(error);
120 }
121
SendTo(const void * pv,size_t cb,const SocketAddress & addr,const rtc::PacketOptions & options)122 int AsyncTCPSocketBase::SendTo(const void* pv,
123 size_t cb,
124 const SocketAddress& addr,
125 const rtc::PacketOptions& options) {
126 const SocketAddress& remote_address = GetRemoteAddress();
127 if (addr == remote_address)
128 return Send(pv, cb, options);
129 // Remote address may be empty if there is a sudden network change.
130 RTC_DCHECK(remote_address.IsNil());
131 socket_->SetError(ENOTCONN);
132 return -1;
133 }
134
FlushOutBuffer()135 int AsyncTCPSocketBase::FlushOutBuffer() {
136 RTC_DCHECK_GT(outbuf_.size(), 0);
137 rtc::ArrayView<uint8_t> view = outbuf_;
138 int res;
139 while (view.size() > 0) {
140 res = socket_->Send(view.data(), view.size());
141 if (res <= 0) {
142 break;
143 }
144 if (static_cast<size_t>(res) > view.size()) {
145 RTC_DCHECK_NOTREACHED();
146 res = -1;
147 break;
148 }
149 view = view.subview(res);
150 }
151 if (res > 0) {
152 // The output buffer may have been written out over multiple partial Send(),
153 // so reconstruct the total written length.
154 RTC_DCHECK_EQ(view.size(), 0);
155 res = outbuf_.size();
156 outbuf_.Clear();
157 } else {
158 // There was an error when calling Send(), so there will still be data left
159 // to send at a later point.
160 RTC_DCHECK_GT(view.size(), 0);
161 // In the special case of EWOULDBLOCK, signal that we had a partial write.
162 if (socket_->GetError() == EWOULDBLOCK) {
163 res = outbuf_.size() - view.size();
164 }
165 if (view.size() < outbuf_.size()) {
166 memmove(outbuf_.data(), view.data(), view.size());
167 outbuf_.SetSize(view.size());
168 }
169 }
170 return res;
171 }
172
AppendToOutBuffer(const void * pv,size_t cb)173 void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
174 RTC_DCHECK(outbuf_.size() + cb <= max_outsize_);
175 outbuf_.AppendData(static_cast<const uint8_t*>(pv), cb);
176 }
177
OnConnectEvent(Socket * socket)178 void AsyncTCPSocketBase::OnConnectEvent(Socket* socket) {
179 SignalConnect(this);
180 }
181
OnReadEvent(Socket * socket)182 void AsyncTCPSocketBase::OnReadEvent(Socket* socket) {
183 RTC_DCHECK(socket_.get() == socket);
184
185 size_t total_recv = 0;
186 while (true) {
187 size_t free_size = inbuf_.capacity() - inbuf_.size();
188 if (free_size < kMinimumRecvSize && inbuf_.capacity() < max_insize_) {
189 inbuf_.EnsureCapacity(std::min(max_insize_, inbuf_.capacity() * 2));
190 free_size = inbuf_.capacity() - inbuf_.size();
191 }
192
193 int len = socket_->Recv(inbuf_.data() + inbuf_.size(), free_size, nullptr);
194 if (len < 0) {
195 // TODO(stefan): Do something better like forwarding the error to the
196 // user.
197 if (!socket_->IsBlocking()) {
198 RTC_LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
199 }
200 break;
201 }
202
203 total_recv += len;
204 inbuf_.SetSize(inbuf_.size() + len);
205 if (!len || static_cast<size_t>(len) < free_size) {
206 break;
207 }
208 }
209
210 if (!total_recv) {
211 return;
212 }
213
214 size_t size = inbuf_.size();
215 ProcessInput(inbuf_.data<char>(), &size);
216
217 if (size > inbuf_.size()) {
218 RTC_LOG(LS_ERROR) << "input buffer overflow";
219 RTC_DCHECK_NOTREACHED();
220 inbuf_.Clear();
221 } else {
222 inbuf_.SetSize(size);
223 }
224 }
225
OnWriteEvent(Socket * socket)226 void AsyncTCPSocketBase::OnWriteEvent(Socket* socket) {
227 RTC_DCHECK(socket_.get() == socket);
228
229 if (outbuf_.size() > 0) {
230 FlushOutBuffer();
231 }
232
233 if (outbuf_.size() == 0) {
234 SignalReadyToSend(this);
235 }
236 }
237
OnCloseEvent(Socket * socket,int error)238 void AsyncTCPSocketBase::OnCloseEvent(Socket* socket, int error) {
239 NotifyClosed(error);
240 }
241
242 // AsyncTCPSocket
243 // Binds and connects `socket` and creates AsyncTCPSocket for
244 // it. Takes ownership of `socket`. Returns null if bind() or
245 // connect() fail (`socket` is destroyed in that case).
Create(Socket * socket,const SocketAddress & bind_address,const SocketAddress & remote_address)246 AsyncTCPSocket* AsyncTCPSocket::Create(Socket* socket,
247 const SocketAddress& bind_address,
248 const SocketAddress& remote_address) {
249 return new AsyncTCPSocket(
250 AsyncTCPSocketBase::ConnectSocket(socket, bind_address, remote_address));
251 }
252
AsyncTCPSocket(Socket * socket)253 AsyncTCPSocket::AsyncTCPSocket(Socket* socket)
254 : AsyncTCPSocketBase(socket, kBufSize) {}
255
Send(const void * pv,size_t cb,const rtc::PacketOptions & options)256 int AsyncTCPSocket::Send(const void* pv,
257 size_t cb,
258 const rtc::PacketOptions& options) {
259 if (cb > kBufSize) {
260 SetError(EMSGSIZE);
261 return -1;
262 }
263
264 // If we are blocking on send, then silently drop this packet
265 if (!IsOutBufferEmpty())
266 return static_cast<int>(cb);
267
268 PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
269 AppendToOutBuffer(&pkt_len, kPacketLenSize);
270 AppendToOutBuffer(pv, cb);
271
272 int res = FlushOutBuffer();
273 if (res <= 0) {
274 // drop packet if we made no progress
275 ClearOutBuffer();
276 return res;
277 }
278
279 rtc::SentPacket sent_packet(options.packet_id, rtc::TimeMillis(),
280 options.info_signaled_after_sent);
281 CopySocketInformationToPacketInfo(cb, *this, false, &sent_packet.info);
282 SignalSentPacket(this, sent_packet);
283
284 // We claim to have sent the whole thing, even if we only sent partial
285 return static_cast<int>(cb);
286 }
287
ProcessInput(char * data,size_t * len)288 void AsyncTCPSocket::ProcessInput(char* data, size_t* len) {
289 SocketAddress remote_addr(GetRemoteAddress());
290
291 while (true) {
292 if (*len < kPacketLenSize)
293 return;
294
295 PacketLength pkt_len = rtc::GetBE16(data);
296 if (*len < kPacketLenSize + pkt_len)
297 return;
298
299 SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
300 TimeMicros());
301
302 *len -= kPacketLenSize + pkt_len;
303 if (*len > 0) {
304 memmove(data, data + kPacketLenSize + pkt_len, *len);
305 }
306 }
307 }
308
AsyncTcpListenSocket(std::unique_ptr<Socket> socket)309 AsyncTcpListenSocket::AsyncTcpListenSocket(std::unique_ptr<Socket> socket)
310 : socket_(std::move(socket)) {
311 RTC_DCHECK(socket_.get() != nullptr);
312 socket_->SignalReadEvent.connect(this, &AsyncTcpListenSocket::OnReadEvent);
313 if (socket_->Listen(kListenBacklog) < 0) {
314 RTC_LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
315 }
316 }
317
GetState() const318 AsyncTcpListenSocket::State AsyncTcpListenSocket::GetState() const {
319 switch (socket_->GetState()) {
320 case Socket::CS_CLOSED:
321 return State::kClosed;
322 case Socket::CS_CONNECTING:
323 return State::kBound;
324 default:
325 RTC_DCHECK_NOTREACHED();
326 return State::kClosed;
327 }
328 }
329
GetLocalAddress() const330 SocketAddress AsyncTcpListenSocket::GetLocalAddress() const {
331 return socket_->GetLocalAddress();
332 }
333
OnReadEvent(Socket * socket)334 void AsyncTcpListenSocket::OnReadEvent(Socket* socket) {
335 RTC_DCHECK(socket_.get() == socket);
336
337 rtc::SocketAddress address;
338 rtc::Socket* new_socket = socket->Accept(&address);
339 if (!new_socket) {
340 // TODO(stefan): Do something better like forwarding the error
341 // to the user.
342 RTC_LOG(LS_ERROR) << "TCP accept failed with error " << socket_->GetError();
343 return;
344 }
345
346 HandleIncomingConnection(new_socket);
347
348 // Prime a read event in case data is waiting.
349 new_socket->SignalReadEvent(new_socket);
350 }
351
HandleIncomingConnection(Socket * socket)352 void AsyncTcpListenSocket::HandleIncomingConnection(Socket* socket) {
353 SignalNewConnection(this, new AsyncTCPSocket(socket));
354 }
355
356 } // namespace rtc
357