blob: b5d3b1cef66cb10f3ab5f24029d7392cb3e76a02 [file] [log] [blame]
// Copyright 2023 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.
#include "pw_data_link/socket_data_link.h"
#if defined(_WIN32) && _WIN32
// TODO(cachinchilla): add support for windows.
#error Windows not supported yet!
#else
#include <arpa/inet.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#endif // defined(_WIN32) && _WIN32
#include <chrono>
#include <optional>
#include "pw_assert/check.h"
#include "pw_bytes/span.h"
#include "pw_data_link/simple_allocator.h"
#include "pw_log/log.h"
#include "pw_status/status.h"
#include "pw_status/status_with_size.h"
namespace pw::data_link {
namespace {
const char* kLinkStateNameOpen = "Open";
const char* kLinkStateNameOpenRequest = "Open Request";
const char* kLinkStateNameClosed = "Closed";
const char* kLinkStateNameWaitingForOpen = "Waiting For Open";
const char* kLinkStateNameUnknown = "Unknown";
const auto kEpollTimeout = std::chrono::seconds(1);
// Configures the file descriptor as non blocking. Returns true when successful.
bool MakeSocketNonBlocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
return false;
}
flags += O_NONBLOCK;
if (fcntl(fd, F_SETFL, flags) == -1) {
PW_LOG_ERROR("Failed to create a socket: %s", std::strerror(errno));
return false;
}
return true;
}
} // namespace
SocketDataLink::SocketDataLink(const char* host, uint16_t port) : host_(host) {
PW_CHECK(host_ != nullptr);
PW_CHECK(ToString(port, port_).ok());
}
SocketDataLink::SocketDataLink(int connection_fd,
EventHandlerCallback&& event_handler,
allocator::Allocator& write_buffer_allocator)
: connection_fd_(connection_fd),
write_buffer_allocator_(&write_buffer_allocator) {
PW_DCHECK(connection_fd > 0);
PW_CHECK(MakeSocketNonBlocking(connection_fd_));
PW_CHECK(ConfigureEpoll());
event_handler_ = std::move(event_handler);
set_link_state(LinkState::kOpen);
write_state_ = WriteState::kIdle;
read_state_ = ReadState::kIdle;
event_handler_(DataLink::Event::kOpen, StatusWithSize());
}
SocketDataLink::~SocketDataLink() {
lock_.lock();
if (link_state_ != LinkState::kClosed) {
DoClose(/*notify_closed=*/true);
return;
}
lock_.unlock();
}
void SocketDataLink::set_link_state(LinkState new_state) {
const char* link_name;
switch (link_state_) {
case LinkState::kOpen:
link_name = kLinkStateNameOpen;
break;
case LinkState::kOpenRequest:
link_name = kLinkStateNameOpenRequest;
break;
case LinkState::kClosed:
link_name = kLinkStateNameClosed;
break;
case LinkState::kWaitingForOpen:
link_name = kLinkStateNameWaitingForOpen;
break;
default:
link_name = kLinkStateNameUnknown;
};
const char* new_link_name;
switch (new_state) {
case LinkState::kOpen:
new_link_name = kLinkStateNameOpen;
break;
case LinkState::kOpenRequest:
new_link_name = kLinkStateNameOpenRequest;
break;
case LinkState::kClosed:
new_link_name = kLinkStateNameClosed;
break;
case LinkState::kWaitingForOpen:
new_link_name = kLinkStateNameWaitingForOpen;
break;
default:
new_link_name = kLinkStateNameUnknown;
};
PW_LOG_DEBUG("Transitioning from %s to %s", link_name, new_link_name);
link_state_ = new_state;
}
void SocketDataLink::Open(EventHandlerCallback&& event_handler,
allocator::Allocator& write_buffer_allocator) {
std::lock_guard lock(lock_);
PW_CHECK(link_state_ == LinkState::kClosed);
write_buffer_allocator_ = &write_buffer_allocator;
event_handler_ = std::move(event_handler);
set_link_state(LinkState::kOpenRequest);
}
void SocketDataLink::WaitAndConsumeEvents() {
// Manually lock and unlock, since some functions may perform unlocking before
// calling the user's event callback, when locks cannot be held.
lock_.lock();
switch (link_state_) {
case LinkState::kOpen:
break;
case LinkState::kWaitingForOpen: {
// Copy the epoll file descriptor to reduce the critical section.
const int fd = epoll_fd_;
lock_.unlock();
epoll_event event;
const int count = epoll_wait(
fd, &event, 1, std::chrono::milliseconds(kEpollTimeout).count());
if (count == 0) {
return;
}
if (event.events & EPOLLERR || event.events & EPOLLHUP) {
lock_.lock();
// Check if link was closed on another thread.
if (link_state_ != LinkState::kClosed) {
DoClose(/*notify_closed=*/false);
} else {
lock_.unlock();
}
event_handler_(DataLink::Event::kOpen, StatusWithSize::Unknown());
return;
}
if (event.events == EPOLLOUT) {
{
std::lock_guard lock(lock_);
set_link_state(LinkState::kOpen);
}
{
std::lock_guard lock(write_lock_);
write_state_ = WriteState::kIdle;
}
{
std::lock_guard lock(read_lock_);
read_state_ = ReadState::kIdle;
}
event_handler_(DataLink::Event::kOpen, StatusWithSize());
return;
}
PW_LOG_ERROR("Unhandled event %d while waiting to open link",
static_cast<int>(event.events));
}
return;
case LinkState::kClosed:
lock_.unlock();
return;
case LinkState::kOpenRequest:
DoOpen();
return;
}
// Copy the epoll file descriptor to reduce the critical section.
const int fd = epoll_fd_;
lock_.unlock();
epoll_event event;
const int count = epoll_wait(
fd, &event, 1, std::chrono::milliseconds(kEpollTimeout).count());
if (count == 0) {
return;
}
if (event.events & EPOLLERR || event.events & EPOLLHUP) {
lock_.lock();
// Check if link was closed on another thread.
if (link_state_ != LinkState::kClosed) {
DoClose(/*notify_closed=*/true);
} else {
lock_.unlock();
}
return;
}
if (event.events & EPOLLIN) {
// Can read!
read_lock_.lock();
if (read_state_ == ReadState::kReadRequested) {
DoRead();
} else if (read_state_ == ReadState::kIdle) {
read_lock_.unlock();
event_handler_(DataLink::Event::kDataReceived, StatusWithSize());
} else {
read_lock_.unlock();
}
}
if (event.events & EPOLLOUT) {
// Can Write!
write_lock_.lock();
if (write_state_ == WriteState::kPending) {
DoWrite();
} else {
write_lock_.unlock();
}
}
if ((event.events & EPOLLIN) == 0 && (event.events & EPOLLOUT) == 0) {
PW_LOG_WARN("Unhandled event %d", static_cast<int>(event.events));
}
}
void SocketDataLink::DoOpen() {
addrinfo hints = {};
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_NUMERICSERV;
addrinfo* res;
if (getaddrinfo(host_, port_, &hints, &res) != 0) {
PW_LOG_ERROR("Failed to configure connection address for socket");
HandleOpenFailure(/*info=*/nullptr);
return;
}
addrinfo* rp;
for (rp = res; rp != nullptr; rp = rp->ai_next) {
PW_LOG_DEBUG("Opening socket");
connection_fd_ = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
if (connection_fd_ != kInvalidFd) {
break;
}
}
if (connection_fd_ == kInvalidFd) {
HandleOpenFailure(res);
return;
}
// Set necessary options on a socket file descriptor.
PW_LOG_DEBUG("Configuring socket");
if (!MakeSocketNonBlocking(connection_fd_)) {
HandleOpenFailure(res);
return;
}
PW_LOG_DEBUG("Connecting socket");
if (connect(connection_fd_, rp->ai_addr, rp->ai_addrlen) == -1) {
if (errno != EINPROGRESS) {
HandleOpenFailure(res);
return;
}
set_link_state(LinkState::kWaitingForOpen);
} else {
set_link_state(LinkState::kOpen);
}
PW_LOG_DEBUG("Configuring Epoll");
if (!ConfigureEpoll()) {
HandleOpenFailure(res);
return;
}
const bool open_completed = link_state_ == LinkState::kOpen;
lock_.unlock();
freeaddrinfo(res);
if (open_completed) {
{
std::lock_guard lock(write_lock_);
write_state_ = WriteState::kIdle;
}
{
std::lock_guard lock(read_lock_);
read_state_ = ReadState::kIdle;
}
event_handler_(DataLink::Event::kOpen, StatusWithSize());
}
}
void SocketDataLink::HandleOpenFailure(addrinfo* info) {
if (connection_fd_ != kInvalidFd) {
close(connection_fd_);
connection_fd_ = kInvalidFd;
}
if (epoll_fd_ != kInvalidFd) {
close(epoll_fd_);
epoll_fd_ = kInvalidFd;
}
set_link_state(LinkState::kClosed);
lock_.unlock();
PW_LOG_ERROR(
"Failed to connect to %s:%s: %s", host_, port_, std::strerror(errno));
if (info != nullptr) {
freeaddrinfo(info);
}
event_handler_(DataLink::Event::kOpen, StatusWithSize::Unknown());
}
bool SocketDataLink::ConfigureEpoll() {
epoll_fd_ = epoll_create1(0);
if (epoll_fd_ == kInvalidFd) {
return false;
}
epoll_event_.events = EPOLLOUT | EPOLLIN;
epoll_event_.data.fd = connection_fd_;
if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, connection_fd_, &epoll_event_) ==
-1) {
return false;
}
return true;
}
void SocketDataLink::Close() {
lock_.lock();
PW_DCHECK(link_state_ != LinkState::kClosed);
DoClose(/*notify_closed=*/true);
}
void SocketDataLink::DoClose(bool notify_closed) {
set_link_state(LinkState::kClosed);
// Copy file descriptors and unlock to minimize the critical section.
const int connection_fd = connection_fd_;
connection_fd_ = kInvalidFd;
const int epoll_fd = epoll_fd_;
epoll_fd_ = kInvalidFd;
lock_.unlock();
{
std::lock_guard lock(write_lock_);
write_state_ = WriteState::kClosed;
}
{
std::lock_guard lock(read_lock_);
read_state_ = ReadState::kClosed;
}
// Close file descriptors if valid.
if (connection_fd != kInvalidFd) {
close(connection_fd);
}
if (epoll_fd != kInvalidFd) {
close(epoll_fd);
}
if (notify_closed) {
event_handler_(DataLink::Event::kClosed, StatusWithSize());
}
}
std::optional<multibuf::MultiBuf> SocketDataLink::GetWriteBuffer(size_t size) {
if (size == 0) {
return multibuf::MultiBuf();
}
{
std::lock_guard lock(write_lock_);
if (write_state_ != WriteState::kIdle) {
return std::nullopt;
}
}
multibuf::MultiBuf buffer;
std::optional<multibuf::OwnedChunk> chunk =
multibuf::HeaderChunkRegionTracker::AllocateRegionAsChunk(
write_buffer_allocator_, size);
if (!chunk) {
return std::nullopt;
}
buffer.PushFrontChunk(std::move(*chunk));
return buffer;
}
Status SocketDataLink::Write(multibuf::MultiBuf&& buffer) {
if (buffer.size() == 0) {
return Status::InvalidArgument();
}
std::lock_guard lock(write_lock_);
if (write_state_ != WriteState::kIdle) {
return Status::FailedPrecondition();
}
tx_multibuf_ = std::move(buffer);
num_bytes_to_send_ = tx_multibuf_.size();
num_bytes_sent_ = 0;
write_state_ = WriteState::kPending;
return OkStatus();
}
void SocketDataLink::DoWrite() {
int send_flags = 0;
#if defined(__linux__)
// Use MSG_NOSIGNAL to avoid getting a SIGPIPE signal when the remote
// peer drops the connection. This is supported on Linux only.
send_flags |= MSG_NOSIGNAL;
#endif // defined(__linux__)
auto [chunk_iter, chunk] =
tx_multibuf_.TakeChunk(tx_multibuf_.Chunks().begin());
ssize_t bytes_sent;
{
std::lock_guard lock(lock_);
// TODO: Send one chunk at a time.
bytes_sent = send(connection_fd_,
reinterpret_cast<const char*>(chunk.data()),
chunk.size(),
send_flags);
}
// Check for errors.
if (bytes_sent < 0) {
if (errno == EPIPE) {
// An EPIPE indicates that the connection is closed.
tx_multibuf_.Release();
write_lock_.unlock();
event_handler_(DataLink::Event::kDataSent, StatusWithSize::OutOfRange());
Close();
return;
}
tx_multibuf_.Release();
write_lock_.unlock();
event_handler_(DataLink::Event::kDataSent, StatusWithSize::Unknown());
return;
}
num_bytes_sent_ += bytes_sent;
// Resize chunk if it was a partial write.
if (static_cast<size_t>(bytes_sent) < chunk.size()) {
chunk->DiscardFront(static_cast<size_t>(bytes_sent));
tx_multibuf_.PushFrontChunk(std::move(chunk));
write_lock_.unlock();
return;
}
// Check if we are done with the MultiBuf.
if (num_bytes_sent_ >= num_bytes_to_send_) {
write_state_ = WriteState::kIdle;
tx_multibuf_.Release();
write_lock_.unlock();
event_handler_(DataLink::Event::kDataSent,
StatusWithSize(num_bytes_to_send_));
return;
}
write_lock_.unlock();
}
Status SocketDataLink::Read(ByteSpan buffer) {
PW_DCHECK(buffer.size() > 0);
std::lock_guard lock(read_lock_);
if (read_state_ != ReadState::kIdle) {
return Status::FailedPrecondition();
}
rx_buffer_ = buffer;
read_state_ = ReadState::kReadRequested;
return OkStatus();
}
void SocketDataLink::DoRead() {
ssize_t bytes_rcvd;
{
std::lock_guard lock(lock_);
bytes_rcvd = recv(connection_fd_,
reinterpret_cast<char*>(rx_buffer_.data()),
rx_buffer_.size_bytes(),
0);
}
if (bytes_rcvd == 0) {
// Remote peer has closed the connection.
read_state_ = ReadState::kClosed;
read_lock_.unlock();
event_handler_(DataLink::Event::kDataRead, StatusWithSize::Internal());
Close();
return;
} else if (bytes_rcvd < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
// Socket timed out when trying to read.
// This should only occur if SO_RCVTIMEO was configured to be nonzero, or
// if the socket was opened with the O_NONBLOCK flag to prevent any
// blocking when performing reads or writes.
read_state_ = ReadState::kIdle;
read_lock_.unlock();
event_handler_(DataLink::Event::kDataRead,
StatusWithSize::ResourceExhausted());
return;
}
read_state_ = ReadState::kIdle;
read_lock_.unlock();
event_handler_(DataLink::Event::kDataRead, StatusWithSize::Unknown());
return;
}
read_state_ = ReadState::kIdle;
read_lock_.unlock();
event_handler_(DataLink::Event::kDataRead, StatusWithSize(bytes_rcvd));
}
} // namespace pw::data_link