Unverified Commit 4e48b6d3 authored by Mateusz Szychowski (Muttley)'s avatar Mateusz Szychowski (Muttley) Committed by GitHub

Fix ResponseStream flush (#791)

* client, string_view: fix warnings

Discarded qualifier and implicit fallthrough

* common: http: fix ResponseStream flush
parent 4d1ec019
...@@ -218,8 +218,10 @@ public: ...@@ -218,8 +218,10 @@ public:
switch (len & 3) { switch (len & 3) {
case 3: case 3:
k1 ^= tail[2] << 16; k1 ^= tail[2] << 16;
/* fall through */
case 2: case 2:
k1 ^= tail[1] << 8; k1 ^= tail[1] << 8;
/* fall through */
case 1: case 1:
k1 ^= tail[0]; k1 ^= tail[0];
k1 *= c1; k1 *= c1;
......
...@@ -69,6 +69,8 @@ public: ...@@ -69,6 +69,8 @@ public:
std::shared_ptr<Aio::Handler> clone() const override; std::shared_ptr<Aio::Handler> clone() const override;
void flush();
private: private:
enum WriteStatus { FirstTry, Retry }; enum WriteStatus { FirstTry, Retry };
...@@ -195,7 +197,7 @@ private: ...@@ -195,7 +197,7 @@ private:
void handlePeerDisconnection(const std::shared_ptr<Peer> &peer); void handlePeerDisconnection(const std::shared_ptr<Peer> &peer);
void handleIncoming(const std::shared_ptr<Peer> &peer); void handleIncoming(const std::shared_ptr<Peer> &peer);
void handleWriteQueue(); void handleWriteQueue(bool flush = false);
void handleTimerQueue(); void handleTimerQueue();
void handlePeerQueue(); void handlePeerQueue();
void handleNotify(); void handleNotify();
......
...@@ -337,7 +337,7 @@ void Transport::handleReadableEntry(const Aio::FdSet::Entry &entry) { ...@@ -337,7 +337,7 @@ void Transport::handleReadableEntry(const Aio::FdSet::Entry &entry) {
assert(entry.isReadable() && "Entry must be readable"); assert(entry.isReadable() && "Entry must be readable");
auto tag = entry.getTag(); auto tag = entry.getTag();
auto fd = static_cast<const Fd>(tag.value()); const auto fd = static_cast<Fd>(tag.value());
auto connIt = connections.find(fd); auto connIt = connections.find(fd);
if (connIt != std::end(connections)) { if (connIt != std::end(connections)) {
auto connection = connIt->second.connection.lock(); auto connection = connIt->second.connection.lock();
...@@ -364,7 +364,7 @@ void Transport::handleWritableEntry(const Aio::FdSet::Entry &entry) { ...@@ -364,7 +364,7 @@ void Transport::handleWritableEntry(const Aio::FdSet::Entry &entry) {
assert(entry.isWritable() && "Entry must be writable"); assert(entry.isWritable() && "Entry must be writable");
auto tag = entry.getTag(); auto tag = entry.getTag();
auto fd = static_cast<const Fd>(tag.value()); const auto fd = static_cast<Fd>(tag.value());
auto connIt = connections.find(fd); auto connIt = connections.find(fd);
if (connIt != std::end(connections)) { if (connIt != std::end(connections)) {
auto &connectionEntry = connIt->second; auto &connectionEntry = connIt->second;
...@@ -385,7 +385,7 @@ void Transport::handleHangupEntry(const Aio::FdSet::Entry &entry) { ...@@ -385,7 +385,7 @@ void Transport::handleHangupEntry(const Aio::FdSet::Entry &entry) {
assert(entry.isHangup() && "Entry must be hangup"); assert(entry.isHangup() && "Entry must be hangup");
auto tag = entry.getTag(); auto tag = entry.getTag();
auto fd = static_cast<const Fd>(tag.value()); const auto fd = static_cast<Fd>(tag.value());
auto connIt = connections.find(fd); auto connIt = connections.find(fd);
if (connIt != std::end(connections)) { if (connIt != std::end(connections)) {
auto &connectionEntry = connIt->second; auto &connectionEntry = connIt->second;
......
...@@ -636,6 +636,7 @@ void ResponseStream::flush() { ...@@ -636,6 +636,7 @@ void ResponseStream::flush() {
auto fd = peer()->fd(); auto fd = peer()->fd();
transport_->asyncWrite(fd, buf); transport_->asyncWrite(fd, buf);
transport_->flush();
buf_.clear(); buf_.clear();
} }
......
...@@ -33,6 +33,10 @@ std::shared_ptr<Aio::Handler> Transport::clone() const { ...@@ -33,6 +33,10 @@ std::shared_ptr<Aio::Handler> Transport::clone() const {
return std::make_shared<Transport>(handler_->clone()); return std::make_shared<Transport>(handler_->clone());
} }
void Transport::flush() {
handleWriteQueue(true);
}
void Transport::registerPoller(Polling::Epoll &poller) { void Transport::registerPoller(Polling::Epoll &poller) {
writesQueue.bind(poller); writesQueue.bind(poller);
timersQueue.bind(poller); timersQueue.bind(poller);
...@@ -353,7 +357,7 @@ void Transport::armTimerMsImpl(TimerEntry entry) { ...@@ -353,7 +357,7 @@ void Transport::armTimerMsImpl(TimerEntry entry) {
timers.insert(std::make_pair(entry.fd, std::move(entry))); timers.insert(std::make_pair(entry.fd, std::move(entry)));
} }
void Transport::handleWriteQueue() { void Transport::handleWriteQueue(bool flush) {
// Let's drain the queue // Let's drain the queue
for (;;) { for (;;) {
auto write = writesQueue.popSafe(); auto write = writesQueue.popSafe();
...@@ -371,6 +375,9 @@ void Transport::handleWriteQueue() { ...@@ -371,6 +375,9 @@ void Transport::handleWriteQueue() {
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write,
Polling::Mode::Edge); Polling::Mode::Edge);
if (flush)
asyncWriteImpl(fd);
} }
} }
......
...@@ -85,60 +85,107 @@ void dumpData(const Rest::Request & /*req*/, Http::ResponseWriter response) { ...@@ -85,60 +85,107 @@ void dumpData(const Rest::Request & /*req*/, Http::ResponseWriter response) {
stream.ends(); stream.ends();
} }
TEST(streaming, from_description) { // from
Address addr(Ipv4::any(), Port(0)); // https://stackoverflow.com/questions/6624667/can-i-use-libcurls-curlopt-writefunction-with-a-c11-lambda-expression#14720398
const size_t threads = 20; typedef size_t (*CURL_WRITEFUNCTION_PTR)(void *, size_t, size_t, void *);
auto curl_callback = [](void *ptr, size_t size, size_t nmemb,
void *stream) -> size_t {
auto ss = static_cast<std::stringstream *>(stream);
ss->write(static_cast<char *>(ptr), size * nmemb);
return size * nmemb;
};
std::shared_ptr<Http::Endpoint> endpoint; class StreamingTests : public testing::Test {
Rest::Description desc("Rest Description Test", "v1"); public:
Rest::Router router; StreamingTests() : address(Pistache::Ipv4::any(), Pistache::Port(0)), endpoint(address), curl(curl_easy_init()) {
}
desc.route(desc.get("/")) void SetUp() override {
.bind(&dumpData) ASSERT_NE(nullptr, curl);
.response(Http::Code::Ok, "Response to the /ready call"); }
router.initFromDescription(desc); void TearDown() override {
curl_easy_cleanup(curl);
endpoint.shutdown();
}
void Init(const std::shared_ptr<Http::Handler>& handler) {
auto flags = Tcp::Options::ReuseAddr; auto flags = Tcp::Options::ReuseAddr;
auto opts = auto options = Http::Endpoint::options().threads(threads).flags(flags).maxRequestSize(1024 * 1024);
Http::Endpoint::options().threads(threads).flags(flags).maxRequestSize(
1024 * 1024);
endpoint = std::make_shared<Pistache::Http::Endpoint>(addr); endpoint.init(options);
endpoint->init(opts); endpoint.setHandler(handler);
endpoint->setHandler(router.handler()); endpoint.serveThreaded();
endpoint->serveThreaded();
std::stringstream ss; url = "http://localhost:" + std::to_string(endpoint.getPort()) + "/";
// from
// https://stackoverflow.com/questions/6624667/can-i-use-libcurls-curlopt-writefunction-with-a-c11-lambda-expression#14720398
typedef size_t (*CURL_WRITEFUNCTION_PTR)(void *, size_t, size_t, void *);
auto curl_callback = [](void *ptr, size_t size, size_t nmemb,
void *stream) -> size_t {
auto ss = static_cast<std::stringstream *>(stream);
ss->write(static_cast<char *>(ptr), size * nmemb);
return size * nmemb;
};
const auto port = endpoint->getPort();
std::string url = "http://localhost:" + std::to_string(port) + "/";
CURLcode res = CURLE_FAILED_INIT;
CURL *curl = curl_easy_init();
if (curl) {
curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION,
static_cast<CURL_WRITEFUNCTION_PTR>(curl_callback)); static_cast<CURL_WRITEFUNCTION_PTR>(curl_callback));
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &ss); curl_easy_setopt(curl, CURLOPT_WRITEDATA, &ss);
curl_easy_setopt(curl, CURLOPT_VERBOSE, 1L); curl_easy_setopt(curl, CURLOPT_VERBOSE, 1L);
res = curl_easy_perform(curl);
curl_easy_cleanup(curl);
} }
endpoint->shutdown();
Address address;
Http::Endpoint endpoint;
CURL *curl;
std::string url;
std::stringstream ss;
static constexpr std::size_t threads = 20;
};
TEST_F(StreamingTests, FromDescription) {
Rest::Description desc("Rest Description Test", "v1");
Rest::Router router;
desc.route(desc.get("/"))
.bind(&dumpData)
.response(Http::Code::Ok, "Response to the /ready call");
router.initFromDescription(desc);
Init(router.handler());
CURLcode res = curl_easy_perform(curl);
if (res != CURLE_OK) if (res != CURLE_OK)
std::cerr << curl_easy_strerror(res) << std::endl; std::cerr << curl_easy_strerror(res) << std::endl;
ASSERT_EQ(res, CURLE_OK); ASSERT_EQ(res, CURLE_OK);
ASSERT_EQ(ss.str().size(), SET_REPEATS * LETTER_REPEATS * N_LETTERS); ASSERT_EQ(ss.str().size(), SET_REPEATS * LETTER_REPEATS * N_LETTERS);
} }
class HelloHandler : public Http::Handler {
public:
HTTP_PROTOTYPE(HelloHandler)
void onRequest(const Http::Request&, Http::ResponseWriter response) override
{
auto stream = response.stream(Http::Code::Ok);
stream << "Hello ";
stream.flush();
std::this_thread::sleep_for(std::chrono::seconds(2));
stream << "world!";
stream.ends();
}
};
TEST_F(StreamingTests, ChunkedStream) {
// force unbuffered
curl_easy_setopt(curl, CURLOPT_BUFFERSIZE, 1);
Init(std::make_shared<HelloHandler>());
std::thread thread([&]() {
curl_easy_perform(curl);
});
std::this_thread::sleep_for(std::chrono::milliseconds(500));
EXPECT_EQ("Hello ", ss.str());
std::this_thread::sleep_for(std::chrono::milliseconds(2000));
EXPECT_EQ("Hello world!", ss.str());
thread.join();
}
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