Commit f8b2d8fc authored by octal's avatar octal

Client is now using the new Reactor class

parent 537f30ab
...@@ -11,7 +11,7 @@ ...@@ -11,7 +11,7 @@
using namespace Net; using namespace Net;
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
if (argc < 1) { if (argc < 2) {
std::cerr << "Usage: http_client page [count]" << std::endl; std::cerr << "Usage: http_client page [count]" << std::endl;
return 1; return 1;
} }
......
...@@ -8,7 +8,7 @@ ...@@ -8,7 +8,7 @@
#include "async.h" #include "async.h"
#include "os.h" #include "os.h"
#include "http.h" #include "http.h"
#include "io.h" #include "reactor.h"
#include "timer_pool.h" #include "timer_pool.h"
#include "view.h" #include "view.h"
#include <atomic> #include <atomic>
...@@ -159,14 +159,14 @@ private: ...@@ -159,14 +159,14 @@ private:
size_t maxConnectionsPerHost; size_t maxConnectionsPerHost;
}; };
class Transport : public Io::Handler { class Transport : public Aio::Handler {
public: public:
PROTOTYPE_OF(Io::Handler, Transport) PROTOTYPE_OF(Aio::Handler, Transport)
typedef std::function<void()> OnResponseParsed; typedef std::function<void()> OnResponseParsed;
void onReady(const Io::FdSet& fds); void onReady(const Aio::FdSet& fds);
void registerPoller(Polling::Epoll& poller); void registerPoller(Polling::Epoll& poller);
Async::Promise<void> Async::Promise<void>
...@@ -328,11 +328,11 @@ public: ...@@ -328,11 +328,11 @@ public:
void shutdown(); void shutdown();
private: private:
std::shared_ptr<Aio::Reactor> reactor_;
Io::ServiceGroup io_;
ConnectionPool pool; ConnectionPool pool;
std::shared_ptr<Transport> transport_; std::shared_ptr<Transport> transport_;
Aio::Reactor::Key transportKey;
std::atomic<uint64_t> ioIndex; std::atomic<uint64_t> ioIndex;
......
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
#include <unistd.h> #include <unistd.h>
#include "os.h" #include "os.h"
#include "io.h" #include "io.h"
#include "reactor.h"
namespace Net { namespace Net {
...@@ -55,11 +56,13 @@ public: ...@@ -55,11 +56,13 @@ public:
} }
void disarm(); void disarm();
void registerIo(Io::Service* io) {
void
registerReactor(const Aio::Reactor::Key& key, Aio::Reactor* reactor) {
if (!registered) { if (!registered) {
io->registerFd(fd, Polling::NotifyOn::Read); reactor->registerFd(key, fd, Polling::NotifyOn::Read);
registered = true;
} }
registered = true;
} }
private: private:
......
...@@ -145,7 +145,7 @@ namespace { ...@@ -145,7 +145,7 @@ namespace {
} }
void void
Transport::onReady(const Io::FdSet& fds) { Transport::onReady(const Aio::FdSet& fds) {
for (const auto& entry: fds) { for (const auto& entry: fds) {
if (entry.getTag() == connectionsQueue.tag()) { if (entry.getTag() == connectionsQueue.tag()) {
handleConnectionQueue(); handleConnectionQueue();
...@@ -180,7 +180,7 @@ Transport::onReady(const Io::FdSet& fds) { ...@@ -180,7 +180,7 @@ Transport::onReady(const Io::FdSet& fds) {
else { else {
conn.resolve(); conn.resolve();
// We are connected, we can start reading data now // We are connected, we can start reading data now
io()->modifyFd(conn.connection->fd, NotifyOn::Read); reactor()->modifyFd(key(), conn.connection->fd, NotifyOn::Read);
} }
} else { } else {
throw std::runtime_error("Unknown fd"); throw std::runtime_error("Unknown fd");
...@@ -212,7 +212,8 @@ Transport::asyncSendRequest( ...@@ -212,7 +212,8 @@ Transport::asyncSendRequest(
const Buffer& buffer) { const Buffer& buffer) {
return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) {
if (std::this_thread::get_id() != io()->thread()) { auto ctx = context();
if (std::this_thread::get_id() != ctx.thread()) {
RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), buffer.detach()); RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), buffer.detach());
auto *e = requestsQueue.allocEntry(std::move(req)); auto *e = requestsQueue.allocEntry(std::move(req));
requestsQueue.push(e); requestsQueue.push(e);
...@@ -250,7 +251,7 @@ Transport::asyncSendRequestImpl( ...@@ -250,7 +251,7 @@ Transport::asyncSendRequestImpl(
if (status == FirstTry) { if (status == FirstTry) {
throw std::runtime_error("Unimplemented, fix me!"); throw std::runtime_error("Unimplemented, fix me!");
} }
io()->modifyFd(fd, NotifyOn::Write, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Write, Polling::Mode::Edge);
} }
else { else {
cleanUp(); cleanUp();
...@@ -265,7 +266,7 @@ Transport::asyncSendRequestImpl( ...@@ -265,7 +266,7 @@ Transport::asyncSendRequestImpl(
if (req.timer) { if (req.timer) {
timeouts.insert( timeouts.insert(
std::make_pair(req.timer->fd, conn)); std::make_pair(req.timer->fd, conn));
req.timer->registerIo(io()); req.timer->registerReactor(key(), reactor());
} }
req.resolve(totalWritten); req.resolve(totalWritten);
break; break;
...@@ -298,7 +299,7 @@ Transport::handleConnectionQueue() { ...@@ -298,7 +299,7 @@ Transport::handleConnectionQueue() {
int res = ::connect(conn->fd, data.addr, data.addr_len); int res = ::connect(conn->fd, data.addr, data.addr_len);
if (res == -1) { if (res == -1) {
if (errno == EINPROGRESS) { if (errno == EINPROGRESS) {
io()->registerFdOneShot(conn->fd, NotifyOn::Write | NotifyOn::Hangup | NotifyOn::Shutdown); reactor()->registerFdOneShot(key(), conn->fd, NotifyOn::Write | NotifyOn::Hangup | NotifyOn::Shutdown);
} }
else { else {
data.reject(Error::system("Failed to connect")); data.reject(Error::system("Failed to connect"));
...@@ -738,7 +739,8 @@ Client::Options::maxConnectionsPerHost(int val) { ...@@ -738,7 +739,8 @@ Client::Options::maxConnectionsPerHost(int val) {
} }
Client::Client() Client::Client()
: ioIndex(0) : reactor_(Aio::Reactor::create())
, ioIndex(0)
{ {
} }
...@@ -763,13 +765,14 @@ void ...@@ -763,13 +765,14 @@ void
Client::init(const Client::Options& options) { Client::init(const Client::Options& options) {
pool.init(options.maxConnectionsPerHost_); pool.init(options.maxConnectionsPerHost_);
transport_.reset(new Transport); transport_.reset(new Transport);
io_.init(options.threads_, transport_); reactor_->init(Aio::AsyncContext(options.threads_));
io_.start(); transportKey = reactor_->addHandler(transport_);
reactor_->run();
} }
void void
Client::shutdown() { Client::shutdown() {
io_.shutdown(); reactor_->shutdown();
} }
RequestBuilder RequestBuilder
...@@ -835,10 +838,10 @@ Client::doRequest( ...@@ -835,10 +838,10 @@ Client::doRequest(
else { else {
if (!conn->hasTransport()) { if (!conn->hasTransport()) {
auto index = ioIndex.fetch_add(1) % io_.size(); auto transports = reactor_->handlers(transportKey);
auto service = io_.service(index); auto index = ioIndex.fetch_add(1) % transports.size();
auto transport = std::static_pointer_cast<Transport>(service->handler()); auto transport = std::static_pointer_cast<Transport>(transports[index]);
conn->associateTransport(transport); conn->associateTransport(transport);
} }
......
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