rtc_base/async_tcp_socket.cc - src - Git at Google

 /*
  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */

 #include "rtc_base/async_tcp_socket.h"

 #include <stdint.h>
 #include <string.h>

 #include <algorithm>
 #include <memory>

 #include "api/array_view.h"
 #include "rtc_base/byte_order.h"
 #include "rtc_base/checks.h"
 #include "rtc_base/logging.h"
 #include "rtc_base/network/sent_packet.h"
 #include "rtc_base/third_party/sigslot/sigslot.h"
 #include "rtc_base/time_utils.h"  // for TimeMillis

 #if defined(WEBRTC_POSIX)
 #include <errno.h>
 #endif  // WEBRTC_POSIX

 namespace rtc {

 static const size_t kMaxPacketSize = 64 * 1024;

 typedef uint16_t PacketLength;
 static const size_t kPacketLenSize = sizeof(PacketLength);

 static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;

 // The input buffer will be resized so that at least kMinimumRecvSize bytes can
 // be received (but it will not grow above the maximum size passed to the
 // constructor).
 static const size_t kMinimumRecvSize = 128;

 static const int kListenBacklog = 5;

 // Binds and connects `socket`
 Socket* AsyncTCPSocketBase::ConnectSocket(
     rtc::Socket* socket,
     const rtc::SocketAddress& bind_address,
     const rtc::SocketAddress& remote_address) {
   std::unique_ptr<rtc::Socket> owned_socket(socket);
   if (socket->Bind(bind_address) < 0) {
     RTC_LOG(LS_ERROR) << "Bind() failed with error " << socket->GetError();
     return nullptr;
   }
   if (socket->Connect(remote_address) < 0) {
     RTC_LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
     return nullptr;
   }
   return owned_socket.release();
 }

 AsyncTCPSocketBase::AsyncTCPSocketBase(Socket* socket,
                                        bool listen,
                                        size_t max_packet_size)
     : socket_(socket),
       listen_(listen),
       max_insize_(max_packet_size),
       max_outsize_(max_packet_size) {
   if (!listen_) {
     // Listening sockets don't send/receive data, so they don't need buffers.
     inbuf_.EnsureCapacity(kMinimumRecvSize);
   }

   RTC_DCHECK(socket_.get() != nullptr);
   socket_->SignalConnectEvent.connect(this,
                                       &AsyncTCPSocketBase::OnConnectEvent);
   socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent);
   socket_->SignalWriteEvent.connect(this, &AsyncTCPSocketBase::OnWriteEvent);
   socket_->SignalCloseEvent.connect(this, &AsyncTCPSocketBase::OnCloseEvent);

   if (listen_) {
     if (socket_->Listen(kListenBacklog) < 0) {
       RTC_LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
     }
   }
 }

 AsyncTCPSocketBase::~AsyncTCPSocketBase() {}

 SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
   return socket_->GetLocalAddress();
 }

 SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
   return socket_->GetRemoteAddress();
 }

 int AsyncTCPSocketBase::Close() {
   return socket_->Close();
 }

 AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
   switch (socket_->GetState()) {
     case Socket::CS_CLOSED:
       return STATE_CLOSED;
     case Socket::CS_CONNECTING:
       if (listen_) {
         return STATE_BOUND;
       } else {
         return STATE_CONNECTING;
       }
     case Socket::CS_CONNECTED:
       return STATE_CONNECTED;
     default:
       RTC_NOTREACHED();
       return STATE_CLOSED;
   }
 }

 int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
   return socket_->GetOption(opt, value);
 }

 int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
   return socket_->SetOption(opt, value);
 }

 int AsyncTCPSocketBase::GetError() const {
   return socket_->GetError();
 }

 void AsyncTCPSocketBase::SetError(int error) {
   return socket_->SetError(error);
 }

 int AsyncTCPSocketBase::SendTo(const void* pv,
                                size_t cb,
                                const SocketAddress& addr,
                                const rtc::PacketOptions& options) {
   const SocketAddress& remote_address = GetRemoteAddress();
   if (addr == remote_address)
     return Send(pv, cb, options);
   // Remote address may be empty if there is a sudden network change.
   RTC_DCHECK(remote_address.IsNil());
   socket_->SetError(ENOTCONN);
   return -1;
 }

 int AsyncTCPSocketBase::FlushOutBuffer() {
   RTC_DCHECK(!listen_);
   RTC_DCHECK_GT(outbuf_.size(), 0);
   rtc::ArrayView<uint8_t> view = outbuf_;
   int res;
   while (view.size() > 0) {
     res = socket_->Send(view.data(), view.size());
     if (res <= 0) {
       break;
     }
     if (static_cast<size_t>(res) > view.size()) {
       RTC_NOTREACHED();
       res = -1;
       break;
     }
     view = view.subview(res);
   }
   if (res > 0) {
     // The output buffer may have been written out over multiple partial Send(),
     // so reconstruct the total written length.
     RTC_DCHECK_EQ(view.size(), 0);
     res = outbuf_.size();
     outbuf_.Clear();
   } else {
     // There was an error when calling Send(), so there will still be data left
     // to send at a later point.
     RTC_DCHECK_GT(view.size(), 0);
     // In the special case of EWOULDBLOCK, signal that we had a partial write.
     if (socket_->GetError() == EWOULDBLOCK) {
       res = outbuf_.size() - view.size();
     }
     if (view.size() < outbuf_.size()) {
       memmove(outbuf_.data(), view.data(), view.size());
       outbuf_.SetSize(view.size());
     }
   }
   return res;
 }

 void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
   RTC_DCHECK(outbuf_.size() + cb <= max_outsize_);
   RTC_DCHECK(!listen_);
   outbuf_.AppendData(static_cast<const uint8_t*>(pv), cb);
 }

 void AsyncTCPSocketBase::OnConnectEvent(Socket* socket) {
   SignalConnect(this);
 }

 void AsyncTCPSocketBase::OnReadEvent(Socket* socket) {
   RTC_DCHECK(socket_.get() == socket);

   if (listen_) {
     rtc::SocketAddress address;
     rtc::Socket* new_socket = socket->Accept(&address);
     if (!new_socket) {
       // TODO(stefan): Do something better like forwarding the error
       // to the user.
       RTC_LOG(LS_ERROR) << "TCP accept failed with error "
                         << socket_->GetError();
       return;
     }

     HandleIncomingConnection(new_socket);

     // Prime a read event in case data is waiting.
     new_socket->SignalReadEvent(new_socket);
   } else {
     size_t total_recv = 0;
     while (true) {
       size_t free_size = inbuf_.capacity() - inbuf_.size();
       if (free_size < kMinimumRecvSize && inbuf_.capacity() < max_insize_) {
         inbuf_.EnsureCapacity(std::min(max_insize_, inbuf_.capacity() * 2));
         free_size = inbuf_.capacity() - inbuf_.size();
       }

       int len =
           socket_->Recv(inbuf_.data() + inbuf_.size(), free_size, nullptr);
       if (len < 0) {
         // TODO(stefan): Do something better like forwarding the error to the
         // user.
         if (!socket_->IsBlocking()) {
           RTC_LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
         }
         break;
       }

       total_recv += len;
       inbuf_.SetSize(inbuf_.size() + len);
       if (!len || static_cast<size_t>(len) < free_size) {
         break;
       }
     }

     if (!total_recv) {
       return;
     }

     size_t size = inbuf_.size();
     ProcessInput(inbuf_.data<char>(), &size);

     if (size > inbuf_.size()) {
       RTC_LOG(LS_ERROR) << "input buffer overflow";
       RTC_NOTREACHED();
       inbuf_.Clear();
     } else {
       inbuf_.SetSize(size);
     }
   }
 }

 void AsyncTCPSocketBase::OnWriteEvent(Socket* socket) {
   RTC_DCHECK(socket_.get() == socket);

   if (outbuf_.size() > 0) {
     FlushOutBuffer();
   }

   if (outbuf_.size() == 0) {
     SignalReadyToSend(this);
   }
 }

 void AsyncTCPSocketBase::OnCloseEvent(Socket* socket, int error) {
   SignalClose(this, error);
 }

 // AsyncTCPSocket
 // Binds and connects `socket` and creates AsyncTCPSocket for
 // it. Takes ownership of `socket`. Returns null if bind() or
 // connect() fail (`socket` is destroyed in that case).
 AsyncTCPSocket* AsyncTCPSocket::Create(Socket* socket,
                                        const SocketAddress& bind_address,
                                        const SocketAddress& remote_address) {
   return new AsyncTCPSocket(
       AsyncTCPSocketBase::ConnectSocket(socket, bind_address, remote_address),
       false);
 }

 AsyncTCPSocket::AsyncTCPSocket(Socket* socket, bool listen)
     : AsyncTCPSocketBase(socket, listen, kBufSize) {}

 int AsyncTCPSocket::Send(const void* pv,
                          size_t cb,
                          const rtc::PacketOptions& options) {
   if (cb > kBufSize) {
     SetError(EMSGSIZE);
     return -1;
   }

   // If we are blocking on send, then silently drop this packet
   if (!IsOutBufferEmpty())
     return static_cast<int>(cb);

   PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
   AppendToOutBuffer(&pkt_len, kPacketLenSize);
   AppendToOutBuffer(pv, cb);

   int res = FlushOutBuffer();
   if (res <= 0) {
     // drop packet if we made no progress
     ClearOutBuffer();
     return res;
   }

   rtc::SentPacket sent_packet(options.packet_id, rtc::TimeMillis(),
                               options.info_signaled_after_sent);
   CopySocketInformationToPacketInfo(cb, *this, false, &sent_packet.info);
   SignalSentPacket(this, sent_packet);

   // We claim to have sent the whole thing, even if we only sent partial
   return static_cast<int>(cb);
 }

 void AsyncTCPSocket::ProcessInput(char* data, size_t* len) {
   SocketAddress remote_addr(GetRemoteAddress());

   while (true) {
     if (*len < kPacketLenSize)
       return;

     PacketLength pkt_len = rtc::GetBE16(data);
     if (*len < kPacketLenSize + pkt_len)
       return;

     SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
                      TimeMicros());

     *len -= kPacketLenSize + pkt_len;
     if (*len > 0) {
       memmove(data, data + kPacketLenSize + pkt_len, *len);
     }
   }
 }

 void AsyncTCPSocket::HandleIncomingConnection(Socket* socket) {
   SignalNewConnection(this, new AsyncTCPSocket(socket, false));
 }

 }  // namespace rtc
	/*
	* Copyright 2004 The WebRTC Project Authors. All rights reserved.
	*
	* Use of this source code is governed by a BSD-style license
	* that can be found in the LICENSE file in the root of the source
	* tree. An additional intellectual property rights grant can be found
	* in the file PATENTS. All contributing project authors may
	* be found in the AUTHORS file in the root of the source tree.
	*/

	#include "rtc_base/async_tcp_socket.h"

	#include <stdint.h>
	#include <string.h>

	#include <algorithm>
	#include <memory>

	#include "api/array_view.h"
	#include "rtc_base/byte_order.h"
	#include "rtc_base/checks.h"
	#include "rtc_base/logging.h"
	#include "rtc_base/network/sent_packet.h"
	#include "rtc_base/third_party/sigslot/sigslot.h"
	#include "rtc_base/time_utils.h" // for TimeMillis

	#if defined(WEBRTC_POSIX)
	#include <errno.h>
	#endif // WEBRTC_POSIX

	namespace rtc {

	static const size_t kMaxPacketSize = 64 * 1024;

	typedef uint16_t PacketLength;
	static const size_t kPacketLenSize = sizeof(PacketLength);

	static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;

	// The input buffer will be resized so that at least kMinimumRecvSize bytes can
	// be received (but it will not grow above the maximum size passed to the
	// constructor).
	static const size_t kMinimumRecvSize = 128;

	static const int kListenBacklog = 5;

	// Binds and connects `socket`
	Socket* AsyncTCPSocketBase::ConnectSocket(
	rtc::Socket* socket,
	const rtc::SocketAddress& bind_address,
	const rtc::SocketAddress& remote_address) {
	std::unique_ptr<rtc::Socket> owned_socket(socket);
	if (socket->Bind(bind_address) < 0) {
	RTC_LOG(LS_ERROR) << "Bind() failed with error " << socket->GetError();
	return nullptr;
	}
	if (socket->Connect(remote_address) < 0) {
	RTC_LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
	return nullptr;
	}
	return owned_socket.release();
	}

	AsyncTCPSocketBase::AsyncTCPSocketBase(Socket* socket,
	bool listen,
	size_t max_packet_size)
	: socket_(socket),
	listen_(listen),
	max_insize_(max_packet_size),
	max_outsize_(max_packet_size) {
	if (!listen_) {
	// Listening sockets don't send/receive data, so they don't need buffers.
	inbuf_.EnsureCapacity(kMinimumRecvSize);
	}

	RTC_DCHECK(socket_.get() != nullptr);
	socket_->SignalConnectEvent.connect(this,
	&AsyncTCPSocketBase::OnConnectEvent);
	socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent);
	socket_->SignalWriteEvent.connect(this, &AsyncTCPSocketBase::OnWriteEvent);
	socket_->SignalCloseEvent.connect(this, &AsyncTCPSocketBase::OnCloseEvent);

	if (listen_) {
	if (socket_->Listen(kListenBacklog) < 0) {
	RTC_LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
	}
	}
	}

	AsyncTCPSocketBase::~AsyncTCPSocketBase() {}

	SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
	return socket_->GetLocalAddress();
	}

	SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
	return socket_->GetRemoteAddress();
	}

	int AsyncTCPSocketBase::Close() {
	return socket_->Close();
	}

	AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
	switch (socket_->GetState()) {
	case Socket::CS_CLOSED:
	return STATE_CLOSED;
	case Socket::CS_CONNECTING:
	if (listen_) {
	return STATE_BOUND;
	} else {
	return STATE_CONNECTING;
	}
	case Socket::CS_CONNECTED:
	return STATE_CONNECTED;
	default:
	RTC_NOTREACHED();
	return STATE_CLOSED;
	}
	}

	int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
	return socket_->GetOption(opt, value);
	}

	int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
	return socket_->SetOption(opt, value);
	}

	int AsyncTCPSocketBase::GetError() const {
	return socket_->GetError();
	}

	void AsyncTCPSocketBase::SetError(int error) {
	return socket_->SetError(error);
	}

	int AsyncTCPSocketBase::SendTo(const void* pv,
	size_t cb,
	const SocketAddress& addr,
	const rtc::PacketOptions& options) {
	const SocketAddress& remote_address = GetRemoteAddress();
	if (addr == remote_address)
	return Send(pv, cb, options);
	// Remote address may be empty if there is a sudden network change.
	RTC_DCHECK(remote_address.IsNil());
	socket_->SetError(ENOTCONN);
	return -1;
	}

	int AsyncTCPSocketBase::FlushOutBuffer() {
	RTC_DCHECK(!listen_);
	RTC_DCHECK_GT(outbuf_.size(), 0);
	rtc::ArrayView<uint8_t> view = outbuf_;
	int res;
	while (view.size() > 0) {
	res = socket_->Send(view.data(), view.size());
	if (res <= 0) {
	break;
	}
	if (static_cast<size_t>(res) > view.size()) {
	RTC_NOTREACHED();
	res = -1;
	break;
	}
	view = view.subview(res);
	}
	if (res > 0) {
	// The output buffer may have been written out over multiple partial Send(),
	// so reconstruct the total written length.
	RTC_DCHECK_EQ(view.size(), 0);
	res = outbuf_.size();
	outbuf_.Clear();
	} else {
	// There was an error when calling Send(), so there will still be data left
	// to send at a later point.
	RTC_DCHECK_GT(view.size(), 0);
	// In the special case of EWOULDBLOCK, signal that we had a partial write.
	if (socket_->GetError() == EWOULDBLOCK) {
	res = outbuf_.size() - view.size();
	}
	if (view.size() < outbuf_.size()) {
	memmove(outbuf_.data(), view.data(), view.size());
	outbuf_.SetSize(view.size());
	}
	}
	return res;
	}

	void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
	RTC_DCHECK(outbuf_.size() + cb <= max_outsize_);
	RTC_DCHECK(!listen_);
	outbuf_.AppendData(static_cast<const uint8_t*>(pv), cb);
	}

	void AsyncTCPSocketBase::OnConnectEvent(Socket* socket) {
	SignalConnect(this);
	}

	void AsyncTCPSocketBase::OnReadEvent(Socket* socket) {
	RTC_DCHECK(socket_.get() == socket);

	if (listen_) {
	rtc::SocketAddress address;
	rtc::Socket* new_socket = socket->Accept(&address);
	if (!new_socket) {
	// TODO(stefan): Do something better like forwarding the error
	// to the user.
	RTC_LOG(LS_ERROR) << "TCP accept failed with error "
	<< socket_->GetError();
	return;
	}

	HandleIncomingConnection(new_socket);

	// Prime a read event in case data is waiting.
	new_socket->SignalReadEvent(new_socket);
	} else {
	size_t total_recv = 0;
	while (true) {
	size_t free_size = inbuf_.capacity() - inbuf_.size();
	if (free_size < kMinimumRecvSize && inbuf_.capacity() < max_insize_) {
	inbuf_.EnsureCapacity(std::min(max_insize_, inbuf_.capacity() * 2));
	free_size = inbuf_.capacity() - inbuf_.size();
	}

	int len =
	socket_->Recv(inbuf_.data() + inbuf_.size(), free_size, nullptr);
	if (len < 0) {
	// TODO(stefan): Do something better like forwarding the error to the
	// user.
	if (!socket_->IsBlocking()) {
	RTC_LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
	}
	break;
	}

	total_recv += len;
	inbuf_.SetSize(inbuf_.size() + len);
	if (!len \|\| static_cast<size_t>(len) < free_size) {
	break;
	}
	}

	if (!total_recv) {
	return;
	}

	size_t size = inbuf_.size();
	ProcessInput(inbuf_.data<char>(), &size);

	if (size > inbuf_.size()) {
	RTC_LOG(LS_ERROR) << "input buffer overflow";
	RTC_NOTREACHED();
	inbuf_.Clear();
	} else {
	inbuf_.SetSize(size);
	}
	}
	}

	void AsyncTCPSocketBase::OnWriteEvent(Socket* socket) {
	RTC_DCHECK(socket_.get() == socket);

	if (outbuf_.size() > 0) {
	FlushOutBuffer();
	}

	if (outbuf_.size() == 0) {
	SignalReadyToSend(this);
	}
	}

	void AsyncTCPSocketBase::OnCloseEvent(Socket* socket, int error) {
	SignalClose(this, error);
	}

	// AsyncTCPSocket
	// Binds and connects `socket` and creates AsyncTCPSocket for
	// it. Takes ownership of `socket`. Returns null if bind() or
	// connect() fail (`socket` is destroyed in that case).
	AsyncTCPSocket* AsyncTCPSocket::Create(Socket* socket,
	const SocketAddress& bind_address,
	const SocketAddress& remote_address) {
	return new AsyncTCPSocket(
	AsyncTCPSocketBase::ConnectSocket(socket, bind_address, remote_address),
	false);
	}

	AsyncTCPSocket::AsyncTCPSocket(Socket* socket, bool listen)
	: AsyncTCPSocketBase(socket, listen, kBufSize) {}

	int AsyncTCPSocket::Send(const void* pv,
	size_t cb,
	const rtc::PacketOptions& options) {
	if (cb > kBufSize) {
	SetError(EMSGSIZE);
	return -1;
	}

	// If we are blocking on send, then silently drop this packet
	if (!IsOutBufferEmpty())
	return static_cast<int>(cb);

	PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
	AppendToOutBuffer(&pkt_len, kPacketLenSize);
	AppendToOutBuffer(pv, cb);

	int res = FlushOutBuffer();
	if (res <= 0) {
	// drop packet if we made no progress
	ClearOutBuffer();
	return res;
	}

	rtc::SentPacket sent_packet(options.packet_id, rtc::TimeMillis(),
	options.info_signaled_after_sent);
	CopySocketInformationToPacketInfo(cb, *this, false, &sent_packet.info);
	SignalSentPacket(this, sent_packet);

	// We claim to have sent the whole thing, even if we only sent partial
	return static_cast<int>(cb);
	}

	void AsyncTCPSocket::ProcessInput(char* data, size_t* len) {
	SocketAddress remote_addr(GetRemoteAddress());

	while (true) {
	if (*len < kPacketLenSize)
	return;

	PacketLength pkt_len = rtc::GetBE16(data);
	if (*len < kPacketLenSize + pkt_len)
	return;

	SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
	TimeMicros());

	*len -= kPacketLenSize + pkt_len;
	if (*len > 0) {
	memmove(data, data + kPacketLenSize + pkt_len, *len);
	}
	}
	}

	void AsyncTCPSocket::HandleIncomingConnection(Socket* socket) {
	SignalNewConnection(this, new AsyncTCPSocket(socket, false));
	}

	} // namespace rtc