Commit fc9711a1 authored by hyperxor's avatar hyperxor

Fix issue #842 and small refactoring

parent 0c68278f
......@@ -611,22 +611,16 @@ namespace Pistache
virtual ~ParserBase() = default;
bool feed(const char* data, size_t len);
virtual bool feed(const char* data, size_t len);
virtual void reset();
State parse();
Step* step();
std::chrono::steady_clock::time_point time() const
{
return time_;
}
protected:
std::array<std::unique_ptr<Step>, StepsCount> allSteps;
size_t currentStep = 0;
std::chrono::steady_clock::time_point time_;
private:
ArrayStreamBuf<char> buffer;
StreamCursor cursor;
......@@ -641,9 +635,18 @@ namespace Pistache
public:
explicit ParserImpl(size_t maxDataSize);
bool feed(const char* data, size_t len) override;
void reset() override;
std::chrono::steady_clock::time_point time() const
{
return time_;
}
Request request;
private:
std::chrono::steady_clock::time_point time_;
};
template <>
......
......@@ -549,7 +549,6 @@ namespace Pistache
bool ParserBase::feed(const char* data, size_t len)
{
time_ = std::chrono::steady_clock::now();
return buffer.feed(data, len);
}
......@@ -559,7 +558,6 @@ namespace Pistache
cursor.reset();
currentStep = 0;
time_ = std::chrono::steady_clock::time_point(std::chrono::steady_clock::duration(0));
}
Step* ParserBase::step()
......@@ -933,11 +931,8 @@ namespace Pistache
return transport_->asyncWrite(fd, buffer)
.then<std::function<Async::Promise<ssize_t>(ssize_t)>,
std::function<void(std::exception_ptr&)>>(
[=](int /*l*/) {
return Async::Promise<ssize_t>(
[=](Async::Deferred<ssize_t> /*deferred*/) mutable {
return;
});
[=](ssize_t data) {
return Async::Promise<ssize_t>::resolved(data);
},
[=](std::exception_ptr& eptr) {
......@@ -1043,6 +1038,7 @@ namespace Pistache
Private::ParserImpl<Http::Request>::ParserImpl(size_t maxDataSize)
: ParserBase(maxDataSize)
, request()
, time_(std::chrono::steady_clock::now())
{
allSteps[0].reset(new RequestLineStep(&request));
allSteps[1].reset(new HeadersStep(&request));
......@@ -1054,6 +1050,13 @@ namespace Pistache
ParserBase::reset();
request = Request();
time_ = std::chrono::steady_clock::now();
}
bool Private::ParserImpl<Http::Request>::feed(const char* data, size_t len)
{
time_ = std::chrono::steady_clock::now();
return ParserBase::feed(data, len);
}
Private::ParserImpl<Http::Response>::ParserImpl(size_t maxDataSize)
......
......@@ -107,10 +107,6 @@ namespace Pistache
handleTimer(std::move(entry_));
timers.erase(it->first);
}
else
{
throw std::runtime_error("Unknown fd");
}
}
else if (entry.isWritable())
{
......@@ -218,11 +214,6 @@ namespace Pistache
{
// Clean up buffers
Guard guard(toWriteLock);
auto& wq = toWrite[fd];
while (wq.size() > 0)
{
wq.pop_front();
}
toWrite.erase(fd);
}
......@@ -242,7 +233,7 @@ namespace Pistache
bool stop = false;
while (!stop)
{
Guard guard(toWriteLock);
std::unique_lock<std::mutex> lock(toWriteLock);
auto it = toWrite.find(fd);
......@@ -270,6 +261,7 @@ namespace Pistache
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
stop = true;
}
lock.unlock();
};
size_t totalWritten = buffer.offset();
......
......@@ -75,7 +75,6 @@ namespace Pistache
void TransportImpl::onReady(const Aio::FdSet& fds)
{
bool handled = false;
for (const auto& entry : fds)
{
if (entry.getTag() == Polling::Tag(timerFd))
......@@ -83,12 +82,11 @@ namespace Pistache
uint64_t wakeups;
::read(timerFd, &wakeups, sizeof wakeups);
checkIdlePeers();
handled = true;
break;
}
}
if (!handled)
Base::onReady(fds);
Base::onReady(fds);
}
void TransportImpl::setHeaderTimeout(std::chrono::milliseconds timeout)
......@@ -129,7 +127,7 @@ namespace Pistache
for (const auto& idlePeer : idlePeers)
{
ResponseWriter response(Http::Version::Http11, this, static_cast<Http::Handler*>(handler_.get()), idlePeer);
response.send(Http::Code::Request_Timeout).then([=](ssize_t) { removePeer(idlePeer); }, [=](std::exception_ptr) { removePeer(idlePeer); });
response.send(Http::Code::Request_Timeout).then([=](ssize_t) { std::cout << "Removing idle: " << idlePeer << "\n"; removePeer(idlePeer); }, [=](std::exception_ptr) { removePeer(idlePeer); });
}
}
......
......@@ -108,11 +108,11 @@ struct HandlerWithSlowPage : public Http::Handler
if (request.resource() == SLOW_PAGE)
{
std::this_thread::sleep_for(std::chrono::seconds(delay_));
message = "Slow page content!\n";
message = "[" + std::to_string(counter++) + "] Slow page content!";
}
else
{
message = "Hello, World!\n";
message = "[" + std::to_string(counter++) + "] Hello, World!";
}
writer.send(Http::Code::Ok, message);
......@@ -120,8 +120,11 @@ struct HandlerWithSlowPage : public Http::Handler
}
int delay_;
static std::atomic<size_t> counter;
};
std::atomic<size_t> HandlerWithSlowPage::counter { 0 };
struct FileHandler : public Http::Handler
{
HTTP_PROTOTYPE(FileHandler)
......@@ -180,7 +183,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)
{
Http::Client client;
......@@ -190,20 +193,24 @@ int clientLogicFunc(int response_size, const std::string& server_page,
auto rb = client.get(server_page).timeout(std::chrono::seconds(timeout_seconds));
int resolver_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();
response.then(
[&resolver_counter](Http::Response resp) {
LOGGER("client", "Response code is " << resp.code());
[&resolver_counter, pos = i](Http::Response resp) {
if (resp.code() == Http::Code::Ok)
{
LOGGER("client", "[" << pos << "] Response: " << resp.code() << ", body: `" << resp.body() << "`");
++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;
LOGGER("client", "Reject with reason:");
LOGGER("client", "[" << pos << "] Reject with reason:");
excPrinter(exc);
++reject_counter;
});
......@@ -216,8 +223,8 @@ int clientLogicFunc(int response_size, const std::string& server_page,
client.shutdown();
LOGGER("client", " resolves: " << resolver_counter << ", rejects: " << reject_counter << ", timeout: " << timeout_seconds << " seconds"
<< ", wait: " << wait_seconds << " seconds");
LOGGER("client", "resolves: " << resolver_counter << ", rejects: " << reject_counter << ", request timeout: " << timeout_seconds << " seconds"
<< ", wait: " << wait_seconds << " seconds");
return resolver_counter;
}
......
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