TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Handle socket error when sending data

Browse Source 
PiperOrigin-RevId: 706886939
This commit is contained in:
Guogang Li
2024-12-16 17:25:29 -08:00
committed by Copybara-Service
parent 94c87e6602
commit e907d82d9d
5 changed files with 120 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/implementation/windows/wifi_hotspot.h
+2
View File
@@ -34,6 +34,7 @@
#include "absl/functional/any_invocable.h"
#include "absl/strings/string_view.h"
#include "absl/synchronization/mutex.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/cancellation_flag.h"
#include "internal/platform/implementation/cancelable.h"
#include "internal/platform/implementation/wifi_hotspot.h"
@@ -150,6 +151,7 @@ class WifiHotspotSocket : public api::WifiHotspotSocket {
IInputStream input_stream_{nullptr};
SOCKET socket_ = INVALID_SOCKET;
SocketType socket_type_ = SocketType::kWinRTSocket;
ByteArray read_buffer_;
};
// A simple wrapper to handle output stream of socket
internal/platform/implementation/windows/wifi_hotspot_socket.cc
+106 -79
View File
@@ -26,11 +26,57 @@
namespace nearby {
namespace windows {
namespace {
// Recorded the maximum bytes received in one read is 65586 in the test, but add
// a little more for const kMaxByteRecieved for safty reason.
constexpr int kMaxByteRecieved = 66000;
// SO_SNDBUF is set to 4MB.
constexpr int kSendBufSize = 4 * 1024 * 1024;
int recv_sync(SOCKET s, char* buf, int len, int flags) {
int result;
struct fd_set read_fds;
result = recv(s, buf, len, flags);
if (result >= 0) {
return result;
}
while (WSAGetLastError() == WSAEWOULDBLOCK) {
FD_ZERO(&read_fds);
FD_SET(s, &read_fds);
if ((select(/*nfds=*/0, /*readfds=*/&read_fds, /*writefds=*/nullptr,
/*exceptfds=*/nullptr, /*timeout=*/nullptr) > 0) &&
FD_ISSET(s, &read_fds)) {
result = recv(s, buf, len, flags);
if (result >= 0) {
return result;
}
}
}
return -1;
}
int send_sync(SOCKET s, const char* buf, int len, int flags) {
int result;
struct fd_set write_fds;
result = send(s, buf, len, flags);
if (result >= 0) {
return result;
}
while (WSAGetLastError() == WSAEWOULDBLOCK) {
FD_ZERO(&write_fds);
FD_SET(s, &write_fds);
if ((select(/*nfds=*/0, /*readfds=*/nullptr, /*writefds=*/&write_fds,
/*exceptfds=*/nullptr, /*timeout=*/nullptr) > 0) &&
FD_ISSET(s, &write_fds)) {
result = send(s, buf, len, flags);
if (result >= 0) {
return result;
}
}
}
return -1;
}
} // namespace
WifiHotspotSocket::WifiHotspotSocket(StreamSocket socket) {
@@ -41,21 +87,6 @@ WifiHotspotSocket::WifiHotspotSocket(StreamSocket socket) {
WifiHotspotSocket::WifiHotspotSocket(SOCKET socket) {
stream_soket_winsock_ = socket;
int buffer_size = kSendBufSize;
int opt_len = sizeof(buffer_size);
if (getsockopt(socket, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<char*>(&buffer_size), &opt_len) == 0) {
LOG(INFO) << "Socket send buffer size: " << buffer_size;
}
if (setsockopt(socket, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<char*>(&buffer_size),
sizeof(buffer_size)) != 0) {
LOG(WARNING) << "Failed to set SO_SNDBUF: " << WSAGetLastError();
}
if (getsockopt(socket, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<char*>(&buffer_size), &opt_len) == 0) {
LOG(INFO) << "Updated send buffer size to: " << buffer_size;
}
input_stream_ = SocketInputStream(socket);
output_stream_ = SocketOutputStream(socket);
}
@@ -129,37 +160,30 @@ ExceptionOr<ByteArray> WifiHotspotSocket::SocketInputStream::Read(
ByteArray data((char*)ibuffer.data(), ibuffer.Length());
return ExceptionOr<ByteArray>(data);
}
// When socket_type_ == SocketType::kWin32Socket
char recv_buf[kMaxByteRecieved];
int result;
struct fd_set readfds;
result = recv(socket_, recv_buf, size, 0);
if (result > 0) {
ByteArray data(recv_buf, result);
return ExceptionOr<ByteArray>(data);
int result;
int count = 0;
// When socket_type_ == SocketType::kWin32Socket
if (size > read_buffer_.size()) {
read_buffer_.resize(size);
}
if (result == 0) {
LOG(INFO) << "Connection closed.";
return {Exception::kIo};
}
// When WSAEWOULDBLOCK happens, it means the packet for receive is not
// ready at the moment. The API select() will block till the packet is
// ready for recieving.
if (WSAEWOULDBLOCK == WSAGetLastError()) {
FD_ZERO(&readfds);
FD_SET(socket_, &readfds);
if ((select(/*nfds ignored*/ 0, &readfds, nullptr, nullptr, nullptr) >
0) &&
FD_ISSET(socket_, &readfds)) {
result = recv(socket_, recv_buf, size, 0);
if (result > 0) {
ByteArray data(recv_buf, result);
return ExceptionOr<ByteArray>(data);
}
while (count < size) {
result = recv_sync(socket_, read_buffer_.data() + count, size - count, 0);
if (result == 0) {
LOG(WARNING) << "Connection closed.";
return {Exception::kIo};
} else if (result < 0) {
LOG(ERROR) << "recv failed with error: " << WSAGetLastError();
return {Exception::kIo};
} else {
count += result;
}
}
return {Exception::kIo};
ByteArray data(read_buffer_.data(), size);
return ExceptionOr<ByteArray>(data);
} catch (std::exception exception) {
LOG(ERROR) << __func__ << ": Exception: " << exception.what();
return {Exception::kIo};
@@ -184,34 +208,29 @@ ExceptionOr<size_t> WifiHotspotSocket::SocketInputStream::Skip(size_t offset) {
return ExceptionOr<size_t>((size_t)ibuffer.Length());
}
// When socket_type_ == SocketType::kWin32Socket
char recv_buf[kMaxByteRecieved];
int result;
struct fd_set readfds;
int count = 0;
result = recv(socket_, recv_buf, offset, 0);
if (result > 0) {
return ExceptionOr<size_t>((size_t)result);
// When socket_type_ == SocketType::kWin32Socket
if (offset > read_buffer_.size()) {
read_buffer_.resize(offset);
}
if (result == 0) {
LOG(INFO) << "Connection closed.";
} else {
// When WSAEWOULDBLOCK happens, it means the packet for receive is not
// ready at the moment. The API select() will block till the packet is
// ready for recieving.
if (WSAEWOULDBLOCK == WSAGetLastError()) {
FD_ZERO(&readfds);
FD_SET(socket_, &readfds);
if ((select(/*nfds ignored*/ 0, &readfds, nullptr, nullptr, nullptr) >
0) &&
FD_ISSET(socket_, &readfds)) {
result = recv(socket_, recv_buf, offset, 0);
if (result > 0) {
return ExceptionOr<size_t>((size_t)result);
}
}
while (count < offset) {
result =
recv_sync(socket_, read_buffer_.data() + count, offset - count, 0);
if (result == 0) {
LOG(WARNING) << "Connection closed.";
return {Exception::kIo};
} else if (result < 0) {
LOG(ERROR) << "recv failed with error: " << WSAGetLastError();
return {Exception::kIo};
} else {
count += result;
}
}
return {Exception::kIo};
return ExceptionOr<size_t>(offset);
} catch (std::exception exception) {
LOG(ERROR) << __func__ << ": Exception: " << exception.what();
return {Exception::kIo};
@@ -269,19 +288,27 @@ Exception WifiHotspotSocket::SocketOutputStream::Write(const ByteArray& data) {
output_stream_.WriteAsync(buffer).get();
return {Exception::kSuccess};
}
// When socket_type_ == SocketType::kWin32Socket
char sendbuf[kMaxByteRecieved];
int result;
int count = 0;
std::memcpy(sendbuf, data.data(), data.size());
result = send(socket_, sendbuf, data.size(), 0);
if (result > 0) {
return {Exception::kSuccess};
while (count < data.size()) {
result = send_sync(socket_, data.data() + count, data.size() - count, 0);
if (result == 0) {
LOG(WARNING) << "Connection closed.";
return {Exception::kIo};
} else if (result < 0) {
LOG(ERROR) << "Failed to send data with error: " << WSAGetLastError();
return {Exception::kIo};
} else {
count += result;
}
}
LOG(INFO) << "recv failed: " << WSAGetLastError();
return {Exception::kIo};
} catch (std::exception exception) {
return {Exception::kSuccess};
}
catch (std::exception exception) {
LOG(ERROR) << __func__ << ": Exception: " << exception.what();
return {Exception::kIo};
} catch (const winrt::hresult_error& error) {
internal/platform/implementation/windows/wifi_lan.h
+1
View File
@@ -135,6 +135,7 @@ class WifiLanSocket : public api::WifiLanSocket {
private:
IInputStream input_stream_{nullptr};
Buffer read_buffer_{nullptr};
};
// A simple wrapper to handle output stream of socket
internal/platform/implementation/windows/wifi_lan_socket.cc
+10 -4
View File
@@ -15,6 +15,7 @@
#include <cstdint>
#include <cstring>
#include <exception>
#include <utility>
#include "internal/platform/byte_array.h"
#include "internal/platform/exception.h"
@@ -83,10 +84,16 @@ WifiLanSocket::SocketInputStream::SocketInputStream(IInputStream input_stream) {
ExceptionOr<ByteArray> WifiLanSocket::SocketInputStream::Read(
std::int64_t size) {
try {
Buffer buffer = Buffer(size);
if (read_buffer_ == nullptr || read_buffer_.Capacity() < size) {
read_buffer_ = Buffer(size);
}
// Reset the buffer length to 0.
read_buffer_.Length(0);
auto ibuffer =
input_stream_.ReadAsync(buffer, size, InputStreamOptions::None).get();
input_stream_.ReadAsync(read_buffer_, size, InputStreamOptions::None)
.get();
if (ibuffer.Length() != size) {
LOG(WARNING) << "Only read partial of data: [" << ibuffer.Length() << "/"
@@ -94,8 +101,7 @@ ExceptionOr<ByteArray> WifiLanSocket::SocketInputStream::Read(
}
ByteArray data((char*)ibuffer.data(), ibuffer.Length());
return ExceptionOr(data);
return ExceptionOr(std::move(data));
} catch (std::exception exception) {
LOG(ERROR) << __func__ << ": Exception: " << exception.what();
return {Exception::kIo};