Unverified Commit e4a9b227 authored by Kip's avatar Kip Committed by GitHub

Merge pull request #912 from hyperxor/fix_issue_842_and_refactoring

Fix issue #842 and small refactoring
parents f73f7baf 4e33c220
...@@ -548,7 +548,7 @@ namespace Pistache ...@@ -548,7 +548,7 @@ namespace Pistache
class BodyStep : public Step class BodyStep : public Step
{ {
public: public:
static constexpr auto Id = Meta::Hash::fnv1a("Headers"); static constexpr auto Id = Meta::Hash::fnv1a("Body");
explicit BodyStep(Message* message_) explicit BodyStep(Message* message_)
: Step(message_) : Step(message_)
...@@ -616,17 +616,11 @@ namespace Pistache ...@@ -616,17 +616,11 @@ namespace Pistache
State parse(); State parse();
Step* step(); Step* step();
std::chrono::steady_clock::time_point time() const
{
return time_;
}
protected: protected:
std::array<std::unique_ptr<Step>, StepsCount> allSteps; std::array<std::unique_ptr<Step>, StepsCount> allSteps;
size_t currentStep = 0; size_t currentStep = 0;
std::chrono::steady_clock::time_point time_;
private: private:
ArrayStreamBuf<char> buffer; ArrayStreamBuf<char> buffer;
StreamCursor cursor; StreamCursor cursor;
...@@ -643,7 +637,15 @@ namespace Pistache ...@@ -643,7 +637,15 @@ namespace Pistache
void reset() override; void reset() override;
std::chrono::steady_clock::time_point time() const
{
return time_;
}
Request request; Request request;
private:
std::chrono::steady_clock::time_point time_;
}; };
template <> template <>
......
...@@ -549,7 +549,6 @@ namespace Pistache ...@@ -549,7 +549,6 @@ namespace Pistache
bool ParserBase::feed(const char* data, size_t len) bool ParserBase::feed(const char* data, size_t len)
{ {
time_ = std::chrono::steady_clock::now();
return buffer.feed(data, len); return buffer.feed(data, len);
} }
...@@ -559,7 +558,6 @@ namespace Pistache ...@@ -559,7 +558,6 @@ namespace Pistache
cursor.reset(); cursor.reset();
currentStep = 0; currentStep = 0;
time_ = std::chrono::steady_clock::time_point(std::chrono::steady_clock::duration(0));
} }
Step* ParserBase::step() Step* ParserBase::step()
...@@ -933,11 +931,8 @@ namespace Pistache ...@@ -933,11 +931,8 @@ namespace Pistache
return transport_->asyncWrite(fd, buffer) return transport_->asyncWrite(fd, buffer)
.then<std::function<Async::Promise<ssize_t>(ssize_t)>, .then<std::function<Async::Promise<ssize_t>(ssize_t)>,
std::function<void(std::exception_ptr&)>>( std::function<void(std::exception_ptr&)>>(
[=](int /*l*/) { [=](ssize_t data) {
return Async::Promise<ssize_t>( return Async::Promise<ssize_t>::resolved(data);
[=](Async::Deferred<ssize_t> /*deferred*/) mutable {
return;
});
}, },
[=](std::exception_ptr& eptr) { [=](std::exception_ptr& eptr) {
...@@ -1043,6 +1038,7 @@ namespace Pistache ...@@ -1043,6 +1038,7 @@ namespace Pistache
Private::ParserImpl<Http::Request>::ParserImpl(size_t maxDataSize) Private::ParserImpl<Http::Request>::ParserImpl(size_t maxDataSize)
: ParserBase(maxDataSize) : ParserBase(maxDataSize)
, request() , request()
, time_(std::chrono::steady_clock::now())
{ {
allSteps[0].reset(new RequestLineStep(&request)); allSteps[0].reset(new RequestLineStep(&request));
allSteps[1].reset(new HeadersStep(&request)); allSteps[1].reset(new HeadersStep(&request));
...@@ -1054,6 +1050,7 @@ namespace Pistache ...@@ -1054,6 +1050,7 @@ namespace Pistache
ParserBase::reset(); ParserBase::reset();
request = Request(); request = Request();
time_ = std::chrono::steady_clock::now();
} }
Private::ParserImpl<Http::Response>::ParserImpl(size_t maxDataSize) Private::ParserImpl<Http::Response>::ParserImpl(size_t maxDataSize)
......
...@@ -107,10 +107,6 @@ namespace Pistache ...@@ -107,10 +107,6 @@ namespace Pistache
handleTimer(std::move(entry_)); handleTimer(std::move(entry_));
timers.erase(it->first); timers.erase(it->first);
} }
else
{
throw std::runtime_error("Unknown fd");
}
} }
else if (entry.isWritable()) else if (entry.isWritable())
{ {
...@@ -218,11 +214,6 @@ namespace Pistache ...@@ -218,11 +214,6 @@ namespace Pistache
{ {
// Clean up buffers // Clean up buffers
Guard guard(toWriteLock); Guard guard(toWriteLock);
auto& wq = toWrite[fd];
while (wq.size() > 0)
{
wq.pop_front();
}
toWrite.erase(fd); toWrite.erase(fd);
} }
...@@ -242,7 +233,7 @@ namespace Pistache ...@@ -242,7 +233,7 @@ namespace Pistache
bool stop = false; bool stop = false;
while (!stop) while (!stop)
{ {
Guard guard(toWriteLock); std::unique_lock<std::mutex> lock(toWriteLock);
auto it = toWrite.find(fd); auto it = toWrite.find(fd);
...@@ -270,6 +261,7 @@ namespace Pistache ...@@ -270,6 +261,7 @@ namespace Pistache
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
stop = true; stop = true;
} }
lock.unlock();
}; };
size_t totalWritten = buffer.offset(); size_t totalWritten = buffer.offset();
......
...@@ -75,7 +75,6 @@ namespace Pistache ...@@ -75,7 +75,6 @@ namespace Pistache
void TransportImpl::onReady(const Aio::FdSet& fds) void TransportImpl::onReady(const Aio::FdSet& fds)
{ {
bool handled = false;
for (const auto& entry : fds) for (const auto& entry : fds)
{ {
if (entry.getTag() == Polling::Tag(timerFd)) if (entry.getTag() == Polling::Tag(timerFd))
...@@ -83,12 +82,11 @@ namespace Pistache ...@@ -83,12 +82,11 @@ namespace Pistache
uint64_t wakeups; uint64_t wakeups;
::read(timerFd, &wakeups, sizeof wakeups); ::read(timerFd, &wakeups, sizeof wakeups);
checkIdlePeers(); checkIdlePeers();
handled = true; break;
} }
} }
if (!handled) Base::onReady(fds);
Base::onReady(fds);
} }
void TransportImpl::setHeaderTimeout(std::chrono::milliseconds timeout) void TransportImpl::setHeaderTimeout(std::chrono::milliseconds timeout)
...@@ -114,12 +112,12 @@ namespace Pistache ...@@ -114,12 +112,12 @@ namespace Pistache
auto elapsed = now - time; auto elapsed = now - time;
auto* step = parser->step(); auto* step = parser->step();
if (step->id() == Private::RequestLineStep::Id) if (step->id() == Private::RequestLineStep::Id || step->id() == Private::HeadersStep::Id)
{ {
if (elapsed > headerTimeout_ || elapsed > bodyTimeout_) if (elapsed > headerTimeout_ || elapsed > bodyTimeout_)
idlePeers.push_back(peer); idlePeers.push_back(peer);
} }
else if (step->id() == Private::HeadersStep::Id) else if (step->id() == Private::BodyStep::Id)
{ {
if (elapsed > bodyTimeout_) if (elapsed > bodyTimeout_)
idlePeers.push_back(peer); idlePeers.push_back(peer);
...@@ -208,7 +206,7 @@ namespace Pistache ...@@ -208,7 +206,7 @@ namespace Pistache
void Endpoint::init(const Endpoint::Options& options) void Endpoint::init(const Endpoint::Options& options)
{ {
listener.init(options.threads_, options.flags_, options.threadsName_); listener.init(options.threads_, options.flags_, options.threadsName_);
listener.setTransportFactory([&] { listener.setTransportFactory([this, options] {
if (!handler_) if (!handler_)
throw std::runtime_error("Must call setHandler()"); throw std::runtime_error("Must call setHandler()");
......
...@@ -108,11 +108,11 @@ struct HandlerWithSlowPage : public Http::Handler ...@@ -108,11 +108,11 @@ struct HandlerWithSlowPage : public Http::Handler
if (request.resource() == SLOW_PAGE) if (request.resource() == SLOW_PAGE)
{ {
std::this_thread::sleep_for(std::chrono::seconds(delay_)); std::this_thread::sleep_for(std::chrono::seconds(delay_));
message = "Slow page content!\n"; message = "[" + std::to_string(counter++) + "] Slow page content!";
} }
else else
{ {
message = "Hello, World!\n"; message = "[" + std::to_string(counter++) + "] Hello, World!";
} }
writer.send(Http::Code::Ok, message); writer.send(Http::Code::Ok, message);
...@@ -120,8 +120,11 @@ struct HandlerWithSlowPage : public Http::Handler ...@@ -120,8 +120,11 @@ struct HandlerWithSlowPage : public Http::Handler
} }
int delay_; int delay_;
static std::atomic<size_t> counter;
}; };
std::atomic<size_t> HandlerWithSlowPage::counter { 0 };
struct FileHandler : public Http::Handler struct FileHandler : public Http::Handler
{ {
HTTP_PROTOTYPE(FileHandler) HTTP_PROTOTYPE(FileHandler)
...@@ -160,6 +163,8 @@ struct AddressEchoHandler : public Http::Handler ...@@ -160,6 +163,8 @@ struct AddressEchoHandler : public Http::Handler
} }
}; };
constexpr const char* ExpectedResponseLine = "HTTP/1.1 408 Request Timeout";
struct PingHandler : public Http::Handler struct PingHandler : public Http::Handler
{ {
HTTP_PROTOTYPE(PingHandler) HTTP_PROTOTYPE(PingHandler)
...@@ -180,7 +185,7 @@ struct PingHandler : public Http::Handler ...@@ -180,7 +185,7 @@ struct PingHandler : public Http::Handler
} }
}; };
int clientLogicFunc(int response_size, const std::string& server_page, int clientLogicFunc(size_t response_size, const std::string& server_page,
int timeout_seconds, int wait_seconds) int timeout_seconds, int wait_seconds)
{ {
Http::Client client; Http::Client client;
...@@ -190,20 +195,24 @@ int clientLogicFunc(int response_size, const std::string& server_page, ...@@ -190,20 +195,24 @@ int clientLogicFunc(int response_size, const std::string& server_page,
auto rb = client.get(server_page).timeout(std::chrono::seconds(timeout_seconds)); auto rb = client.get(server_page).timeout(std::chrono::seconds(timeout_seconds));
int resolver_counter = 0; int resolver_counter = 0;
int reject_counter = 0; int reject_counter = 0;
for (int i = 0; i < response_size; ++i) for (size_t i = 0; i < response_size; ++i)
{ {
auto response = rb.send(); auto response = rb.send();
response.then( response.then(
[&resolver_counter](Http::Response resp) { [&resolver_counter, pos = i](Http::Response resp) {
LOGGER("client", "Response code is " << resp.code());
if (resp.code() == Http::Code::Ok) if (resp.code() == Http::Code::Ok)
{ {
LOGGER("client", "[" << pos << "] Response: " << resp.code() << ", body: `" << resp.body() << "`");
++resolver_counter; ++resolver_counter;
} }
else
{
LOGGER("client", "[" << pos << "] Response: " << resp.code());
}
}, },
[&reject_counter](std::exception_ptr exc) { [&reject_counter, pos = i](std::exception_ptr exc) {
PrintException excPrinter; PrintException excPrinter;
LOGGER("client", "Reject with reason:"); LOGGER("client", "[" << pos << "] Reject with reason:");
excPrinter(exc); excPrinter(exc);
++reject_counter; ++reject_counter;
}); });
...@@ -216,8 +225,8 @@ int clientLogicFunc(int response_size, const std::string& server_page, ...@@ -216,8 +225,8 @@ int clientLogicFunc(int response_size, const std::string& server_page,
client.shutdown(); client.shutdown();
LOGGER("client", " resolves: " << resolver_counter << ", rejects: " << reject_counter << ", timeout: " << timeout_seconds << " seconds" LOGGER("client", "resolves: " << resolver_counter << ", rejects: " << reject_counter << ", request timeout: " << timeout_seconds << " seconds"
<< ", wait: " << wait_seconds << " seconds"); << ", wait: " << wait_seconds << " seconds");
return resolver_counter; return resolver_counter;
} }
...@@ -527,17 +536,18 @@ TEST(http_server_test, response_size_captured) ...@@ -527,17 +536,18 @@ TEST(http_server_test, response_size_captured)
ASSERT_EQ(rcode, Http::Code::Ok); ASSERT_EQ(rcode, Http::Code::Ok);
} }
TEST(http_server_test, client_request_header_timeout_raises_http_408) TEST(http_server_test, client_request_timeout_on_only_connect_raises_http_408)
{ {
Pistache::Address address("localhost", Pistache::Port(0)); Pistache::Address address("localhost", Pistache::Port(0));
auto timeout = std::chrono::seconds(2); const auto headerTimeout = std::chrono::seconds(2);
Http::Endpoint server(address); Http::Endpoint server(address);
auto flags = Tcp::Options::ReuseAddr; auto flags = Tcp::Options::ReuseAddr;
auto opts = Http::Endpoint::options() auto opts = Http::Endpoint::options()
.flags(flags) .flags(flags)
.headerTimeout(timeout); .headerTimeout(headerTimeout);
server.init(opts); server.init(opts);
server.setHandler(Http::make_handler<PingHandler>()); server.setHandler(Http::make_handler<PingHandler>());
server.serveThreaded(); server.serveThreaded();
...@@ -546,24 +556,109 @@ TEST(http_server_test, client_request_header_timeout_raises_http_408) ...@@ -546,24 +556,109 @@ TEST(http_server_test, client_request_header_timeout_raises_http_408)
auto addr = "localhost:" + port.toString(); auto addr = "localhost:" + port.toString();
LOGGER("test", "Server address: " << addr) LOGGER("test", "Server address: " << addr)
char recvBuf[1024]; TcpClient client;
std::memset(recvBuf, 0, sizeof(recvBuf)); EXPECT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError();
char recvBuf[1024] = {
0,
};
size_t bytes; size_t bytes;
EXPECT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError();
EXPECT_EQ(0, strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine)));
server.shutdown();
}
TEST(http_server_test, client_request_timeout_on_delay_in_header_send_raises_http_408)
{
Pistache::Address address("localhost", Pistache::Port(0));
const auto headerTimeout = std::chrono::seconds(1);
Http::Endpoint server(address);
auto flags = Tcp::Options::ReuseAddr;
auto opts = Http::Endpoint::options()
.flags(flags)
.headerTimeout(headerTimeout);
server.init(opts);
server.setHandler(Http::make_handler<PingHandler>());
server.serveThreaded();
auto port = server.getPort();
auto addr = "localhost:" + port.toString();
LOGGER("test", "Server address: " << addr);
const std::string reqStr = "GET /ping HTTP/1.1\r\n";
const std::string headerStr = "Host: localhost\r\nUser-Agent: test\r\n";
TcpClient client; TcpClient client;
ASSERT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError(); EXPECT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError();
EXPECT_TRUE(client.send(reqStr)) << client.lastError();
ASSERT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError(); std::this_thread::sleep_for(headerTimeout / 2);
EXPECT_TRUE(client.send(headerStr)) << client.lastError();
char recvBuf[1024] = {
0,
};
size_t bytes;
EXPECT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError();
EXPECT_EQ(0, strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine)));
server.shutdown(); server.shutdown();
} }
TEST(http_server_test, client_request_body_timeout_raises_http_408) TEST(http_server_test, client_request_timeout_on_delay_in_request_line_send_raises_http_408)
{ {
Pistache::Address address("localhost", Pistache::Port(0)); Pistache::Address address("localhost", Pistache::Port(0));
auto headerTimeout = std::chrono::seconds(1); const auto headerTimeout = std::chrono::seconds(2);
auto bodyTimeout = std::chrono::seconds(1);
Http::Endpoint server(address);
auto flags = Tcp::Options::ReuseAddr;
auto opts = Http::Endpoint::options()
.flags(flags)
.headerTimeout(headerTimeout);
server.init(opts);
server.setHandler(Http::make_handler<PingHandler>());
server.serveThreaded();
auto port = server.getPort();
auto addr = "localhost:" + port.toString();
LOGGER("test", "Server address: " << addr);
const std::string reqStr { "GET /ping HTTP/1.1\r\n" };
TcpClient client;
EXPECT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError();
for (size_t i = 0; i < reqStr.size(); ++i)
{
if (!client.send(reqStr.substr(i, 1)))
{
break;
}
std::this_thread::sleep_for(std::chrono::milliseconds(300));
}
EXPECT_EQ(client.lastErrno(), EPIPE) << "Errno: " << client.lastErrno();
char recvBuf[1024] = {
0,
};
size_t bytes;
EXPECT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError();
EXPECT_EQ(0, strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine)));
server.shutdown();
}
TEST(http_server_test, client_request_timeout_on_delay_in_body_send_raises_http_408)
{
Pistache::Address address("localhost", Pistache::Port(0));
const auto headerTimeout = std::chrono::seconds(1);
const auto bodyTimeout = std::chrono::seconds(2);
Http::Endpoint server(address); Http::Endpoint server(address);
auto flags = Tcp::Options::ReuseAddr; auto flags = Tcp::Options::ReuseAddr;
...@@ -580,24 +675,62 @@ TEST(http_server_test, client_request_body_timeout_raises_http_408) ...@@ -580,24 +675,62 @@ TEST(http_server_test, client_request_body_timeout_raises_http_408)
auto addr = "localhost:" + port.toString(); auto addr = "localhost:" + port.toString();
LOGGER("test", "Server address: " << addr); LOGGER("test", "Server address: " << addr);
std::string reqStr = "GET /ping HTTP/1.1\r\n"; const std::string reqStr = "POST /ping HTTP/1.1\r\nHost: localhost\r\nContent-Type: text/plain\r\nContent-Length: 32\r\n\r\nabc";
std::string headerStr = "Host: localhost\r\nUser-Agent: test\r\n";
char recvBuf[1024]; TcpClient client;
std::memset(recvBuf, 0, sizeof(recvBuf)); EXPECT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError();
EXPECT_TRUE(client.send(reqStr)) << client.lastError();
char recvBuf[1024] = {
0,
};
size_t bytes; size_t bytes;
EXPECT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError();
EXPECT_EQ(0, strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine)));
server.shutdown();
}
TEST(http_server_test, client_request_no_timeout)
{
Pistache::Address address("localhost", Pistache::Port(0));
const auto headerTimeout = std::chrono::seconds(2);
const auto bodyTimeout = std::chrono::seconds(4);
Http::Endpoint server(address);
auto flags = Tcp::Options::ReuseAddr;
auto opts = Http::Endpoint::options()
.flags(flags)
.headerTimeout(headerTimeout)
.bodyTimeout(bodyTimeout);
server.init(opts);
server.setHandler(Http::make_handler<PingHandler>());
server.serveThreaded();
auto port = server.getPort();
auto addr = "localhost:" + port.toString();
LOGGER("test", "Server address: " << addr);
const std::string headerStr = "POST /ping HTTP/1.1\r\nHost: localhost\r\nContent-Type: text/plain\r\nContent-Length: 8\r\n\r\n";
const std::string bodyStr = "abcdefgh\r\n\r\n";
TcpClient client; TcpClient client;
ASSERT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError(); EXPECT_TRUE(client.connect(Pistache::Address("localhost", port))) << client.lastError();
ASSERT_TRUE(client.send(reqStr)) << client.lastError();
std::this_thread::sleep_for(headerTimeout / 2); std::this_thread::sleep_for(headerTimeout / 2);
ASSERT_TRUE(client.send(headerStr)) << client.lastError(); EXPECT_TRUE(client.send(headerStr)) << client.lastError();
static constexpr const char* ExpectedResponseLine = "HTTP/1.1 408 Request Timeout"; std::this_thread::sleep_for(bodyTimeout / 2);
EXPECT_TRUE(client.send(bodyStr)) << client.lastError();
ASSERT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError(); char recvBuf[1024] = {
ASSERT_TRUE(!strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine))); 0,
};
size_t bytes;
EXPECT_TRUE(client.receive(recvBuf, sizeof(recvBuf), &bytes, std::chrono::seconds(5))) << client.lastError();
EXPECT_NE(0, strncmp(recvBuf, ExpectedResponseLine, strlen(ExpectedResponseLine)));
server.shutdown(); server.shutdown();
} }
......
...@@ -15,9 +15,10 @@ namespace Pistache ...@@ -15,9 +15,10 @@ namespace Pistache
do \ do \
{ \ { \
auto ret = __VA_ARGS__; \ auto ret = __VA_ARGS__; \
if (ret < 0) \ if (ret == -1) \
{ \ { \
lastError_ = strerror(errno); \ lastError_ = strerror(errno); \
lastErrno_ = errno; \
return false; \ return false; \
} \ } \
} while (0) } while (0)
...@@ -66,9 +67,28 @@ namespace Pistache ...@@ -66,9 +67,28 @@ namespace Pistache
return send(data.c_str(), data.size()); return send(data.c_str(), data.size());
} }
bool send(const void* data, size_t size) bool send(const char* data, size_t len)
{ {
CLIENT_TRY(::send(fd_, data, size, 0)); size_t total = 0;
while (total < len)
{
ssize_t n = ::send(fd_, data + total, len - total, MSG_NOSIGNAL);
if (n == -1)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
{
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
else
{
CLIENT_TRY(n);
}
}
else
{
total += static_cast<size_t>(n);
}
}
return true; return true;
} }
...@@ -115,9 +135,15 @@ namespace Pistache ...@@ -115,9 +135,15 @@ namespace Pistache
return lastError_; return lastError_;
} }
int lastErrno() const
{
return lastErrno_;
}
private: private:
int fd_; int fd_ = -1;
std::string lastError_; std::string lastError_;
int lastErrno_ = 0;
}; };
#undef CLIENT_TRY #undef CLIENT_TRY
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment