Skip to content

Commit

Permalink
[coro_http][coro_io][fix]correct close client when sendfile error (#714)
Browse files Browse the repository at this point in the history 
* correct close client when sendfile error

* update
  • Loading branch information
@poor-circle
poor-circle authored Jul 10, 2024
1 parent 0c88765 commit 2f287f0
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 26 deletions.
39 changes: 34 additions & 5 deletions include/ylt/coro_io/coro_io.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
#endif

#include <asio/connect.hpp>
#include <asio/dispatch.hpp>
#include <asio/experimental/channel.hpp>
#include <asio/ip/tcp.hpp>
#include <asio/read.hpp>
Expand All @@ -38,13 +37,16 @@
#include <deque>

#include "io_context_pool.hpp"
#if __has_include("ylt/util/type_traits.h")
#include "ylt/util/type_traits.h"
#else
#include "../util/type_traits.h"
#endif
#ifdef __linux__
#include <sys/sendfile.h>
#endif

namespace coro_io {

template <typename T>
constexpr inline bool is_lazy_v =
util::is_specialization_v<std::remove_cvref_t<T>, async_simple::coro::Lazy>;
Expand Down Expand Up @@ -367,19 +369,37 @@ struct coro_channel
using return_type = R;
using ValueType = std::pair<std::error_code, R>;
using asio::experimental::channel<void(std::error_code, R)>::channel;
coro_channel(coro_io::ExecutorWrapper<> *executor, size_t capacity)
: executor_(executor),
asio::experimental::channel<void(std::error_code, R)>(
executor->get_asio_executor(), capacity) {}
auto get_executor() { return executor_; }

private:
coro_io::ExecutorWrapper<> *executor_;
};

template <typename R>
inline coro_channel<R> create_channel(
size_t capacity,
asio::io_context::executor_type executor =
coro_io::get_global_block_executor()->get_asio_executor()) {
coro_io::ExecutorWrapper<> *executor = coro_io::get_global_executor()) {
return coro_channel<R>(executor, capacity);
}

template <typename R>
inline auto create_shared_channel(
size_t capacity,
coro_io::ExecutorWrapper<> *executor = coro_io::get_global_executor()) {
return std::make_shared<coro_channel<R>>(executor, capacity);
}

template <typename T>
inline async_simple::coro::Lazy<std::error_code> async_send(
asio::experimental::channel<void(std::error_code, T)> &channel, T val) {
bool r = channel.try_send(std::error_code{}, val);
if (r) {
co_return std::error_code{};
}
callback_awaitor<std::error_code> awaitor;
co_return co_await awaitor.await_resume(
[&, val = std::move(val)](auto handler) {
Expand All @@ -393,6 +413,14 @@ template <typename Channel>
async_simple::coro::Lazy<std::pair<
std::error_code,
typename Channel::return_type>> inline async_receive(Channel &channel) {
using value_type = typename Channel::return_type;
value_type val;
bool r = channel.try_receive([&val](std::error_code, value_type result) {
val = result;
});
if (r) {
co_return std::make_pair(std::error_code{}, val);
}
callback_awaitor<std::pair<std::error_code, typename Channel::return_type>>
awaitor;
co_return co_await awaitor.await_resume([&](auto handler) {
Expand Down Expand Up @@ -513,6 +541,8 @@ inline auto pipe_signal_handler = [] {
return 0;
}();

// FIXME: this function may not thread-safe if it not running in socket's
// executor
inline async_simple::coro::Lazy<std::pair<std::error_code, std::size_t>>
async_sendfile(asio::ip::tcp::socket &socket, int fd, off_t offset,
size_t size) noexcept {
Expand Down Expand Up @@ -548,7 +578,6 @@ async_sendfile(asio::ip::tcp::socket &socket, int fd, off_t offset,
handler.set_value_then_resume(ec);
});
});
continue;
}
if (ec || n == 0 || least_bytes == 0) [[unlikely]] { // End of File
break;
Expand Down
37 changes: 16 additions & 21 deletions include/ylt/standalone/cinatra/coro_http_client.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -893,13 +893,11 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
file_data.data(), std::min(file_data.size(), length));
ec) {
// bad request, file may smaller than content-length
close();
break;
}
length -= size;
if (length > 0 && file.eof()) {
// bad request, file may smaller than content-length
close();
ec = std::make_error_code(std::errc::invalid_argument);
break;
}
Expand Down Expand Up @@ -936,7 +934,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}
if (actual_len != length) [[unlikely]] {
// bad request, file is smaller than content-length
close();
ec = std::make_error_code(std::errc::invalid_argument);
co_return;
}
Expand Down Expand Up @@ -972,7 +969,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
}
if (actual_len != len) [[unlikely]] {
// bad request, file is smaller than content-length
close();
ec = std::make_error_code(std::errc::invalid_argument);
co_return;
}
Expand Down Expand Up @@ -1013,14 +1009,19 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
int64_t content_length = -1,
req_content_type content_type = req_content_type::text,
std::unordered_map<std::string, std::string> headers = {}) {
std::shared_ptr<int> guard(nullptr, [this](auto) {
std::error_code ec{};
size_t size = 0;
bool is_keep_alive = true;
req_context<> ctx{content_type};
resp_data data{};

std::shared_ptr<void> guard(nullptr, [&, this](auto) {
if (!req_headers_.empty()) {
req_headers_.clear();
}
handle_result(data, ec, is_keep_alive);
});

req_context<> ctx{content_type};
resp_data data{};
auto [ok, u] = handle_uri(data, uri);
if (!ok) {
co_return resp_data{std::make_error_code(std::errc::protocol_error), 404};
Expand Down Expand Up @@ -1064,9 +1065,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
std::string header_str =
build_request_header(u, method, ctx, true, std::move(headers));

std::error_code ec{};
size_t size = 0;

if (socket_->has_closed_) {
{
auto guard = timer_guard(this, conn_timeout_duration_, "connect timer");
Expand Down Expand Up @@ -1109,7 +1107,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
if (!ec && content_length > 0) {
// bad request, file is smaller than content-length
ec = std::make_error_code(std::errc::invalid_argument);
close();
}
}
else if constexpr (std::is_same_v<Source, std::string> ||
Expand Down Expand Up @@ -1147,7 +1144,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
else if (result.eof) [[unlikely]] {
// bad request, file is smaller than content-length
ec = std::make_error_code(std::errc::invalid_argument);
close();
break;
}
}
Expand All @@ -1159,13 +1155,11 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
co_return resp_data{ec, 404};
}

bool is_keep_alive = true;
data = co_await handle_read(ec, size, is_keep_alive, std::move(ctx),
http_method::POST);
if (ec && socket_->is_timeout_) {
ec = std::make_error_code(std::errc::timed_out);
}
handle_result(data, ec, is_keep_alive);
co_return data;
}

Expand All @@ -1174,18 +1168,23 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
S uri, http_method method, Source source,
req_content_type content_type = req_content_type::text,
std::unordered_map<std::string, std::string> headers = {}) {
std::shared_ptr<int> guard(nullptr, [this](auto) {
req_context<> ctx{content_type};
resp_data data{};
std::error_code ec{};
size_t size = 0;
bool is_keep_alive = true;

std::shared_ptr<void> guard(nullptr, [&, this](auto) {
if (!req_headers_.empty()) {
req_headers_.clear();
}
handle_result(data, ec, is_keep_alive);
});

if (!resp_chunk_str_.empty()) {
resp_chunk_str_.clear();
}

req_context<> ctx{content_type};
resp_data data{};
auto [ok, u] = handle_uri(data, uri);
if (!ok) {
co_return resp_data{std::make_error_code(std::errc::protocol_error), 404};
Expand Down Expand Up @@ -1216,9 +1215,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
std::string header_str =
build_request_header(u, method, ctx, true, std::move(headers));

std::error_code ec{};
size_t size = 0;

if (socket_->has_closed_) {
{
auto guard = timer_guard(this, conn_timeout_duration_, "connect timer");
Expand Down Expand Up @@ -1296,7 +1292,6 @@ class coro_http_client : public std::enable_shared_from_this<coro_http_client> {
co_return resp_data{ec, 404};
}

bool is_keep_alive = true;
data = co_await handle_read(ec, size, is_keep_alive, std::move(ctx),
http_method::POST);
if (ec && socket_->is_timeout_) {
Expand Down

0 comments on commit 2f287f0

Please sign in to comment.