asio-sv3.cc

#include <iostream>
#include <string>
#include <mutex>
#include <thread>
#include <future>
#include <deque>

#include <nghttp2/asio_http2_server.h>

using namespace nghttp2::asio_http2;
using namespace nghttp2::asio_http2::server;

struct Stream : public std::enable_shared_from_this<Stream> {
  Stream(const request &req, const response &res,
         boost::asio::io_service &io_service)
      : io_service(io_service), req(req), res(res), closed(false) {}
  void commit_result() {
    auto self = shared_from_this();
    io_service.post([self]() {
      std::lock_guard<std::mutex> lg(self->mu);
      if (self->closed) {
        return;
      }

      self->res.write_head(200);
      self->res.end("done");
    });
  }
  void set_closed(bool f) {
    std::lock_guard<std::mutex> lg(mu);
    closed = f;
  }

  boost::asio::io_service &io_service;
  std::mutex mu;
  const request &req;
  const response &res;
  bool closed;
};

struct Queue {
  void push(std::shared_ptr<Stream> st) {
    std::lock_guard<std::mutex> lg(mu);
    q.push_back(st);
    cv.notify_all();
  }
  std::shared_ptr<Stream> pop() {
    std::unique_lock<std::mutex> ulk(mu);
    cv.wait(ulk, [this]() { return !q.empty(); });

    auto res = q.front();
    q.pop_front();
    return res;
  }

  std::mutex mu;
  std::condition_variable cv;
  std::deque<std::shared_ptr<Stream>> q;
};

int main(int argc, char **argv) {
  http2 server;

  server.num_threads(2);

  Queue q;

  for (int i = 0; i < 10; ++i) {
    auto th = std::thread([&q]() {
      for (;;) {
        auto st = q.pop();
        sleep(1);
        st->commit_result();
      }
    });
    th.detach();
  }

  server.handle("/", [&q](const request &req, const response &res) {
    auto &io_service = res.io_service();
    auto st = std::make_shared<Stream>(req, res, io_service);
    res.on_close([st](uint32_t error_code) { st->set_closed(true); });
    q.push(st);
  });

  boost::system::error_code ec;
  if (server.listen_and_serve(ec, "127.0.0.1", "3000")) {
    std::cerr << "error: " << ec.message() << std::endl;
  }
}

Hi, firstly thank you for the gist, which I used in one of my projects.
I extended it to set response body taking request body (like a mirror):
https://gist.github.com/testillano/a66283408165ccb6eba844bd6fa6518b

I used commit_result(std::string) from other of your gists and remove the sleep(1) within queue dispatching.

But I have found a bug (crash dump) which drives me crazy. When managing high load and breaking abruptly the client connection.
I'm wondering what's wrong in on_data callback. Probably the lifetime for stream is not valid when using commit_result().
For example, consider a request.json file containing { "foo":"bar" } then launch high load by mean:

h2load -t1 -n100000 -c1 -m100 http://127.0.0.1:3000/the/uri -d request.json

All is ok, but if you interrupt the h2load (CTRL-C), you could get the crash (sooner or later) in the gist server.
Could you help me to find out the problem ?
Thank you in advance.