Commit 497ad343 authored by Dominik Charousset's avatar Dominik Charousset

Use stateful brokers in CURL example

parent a77b9138
...@@ -97,282 +97,238 @@ constexpr int min_req_interval = 10; ...@@ -97,282 +97,238 @@ constexpr int min_req_interval = 10;
// maximum delay between HTTP requests // maximum delay between HTTP requests
constexpr int max_req_interval = 300; constexpr int max_req_interval = 300;
// provides print utility and each base_actor has a parent // put everything into anonymous namespace (except main)
class base_actor : public event_based_actor { namespace {
protected:
base_actor(actor parent, std::string name, std::string color_code) // provides print utility, a name, and a parent
: parent_(std::move(parent)), struct base_state {
name_(std::move(name)), base_state(local_actor* thisptr) : self(thisptr) {
color_(std::move(color_code)),
out_(this) {
// nop // nop
} }
~base_actor(); actor_ostream print() {
return aout(self) << color << name << " (id = " << self->id() << "): ";
inline actor_ostream& print() {
return out_ << color_ << name_ << " (id = " << id() << "): ";
} }
void on_exit() { virtual void init(actor m_parent, std::string m_name, std::string m_color) {
print() << "on_exit" << color::reset_endl; parent = std::move(m_parent);
name = std::move(m_name);
color = std::move(m_color);
print() << "started" << color::reset_endl;
} }
actor parent_; virtual ~base_state() {
print() << "done" << color::reset_endl;
private:
std::string name_;
std::string color_;
actor_ostream out_;
};
base_actor::~base_actor() {
// avoid weak-vtables warning
}
// encapsulates an HTTP request
class client_job : public base_actor {
public:
explicit client_job(actor parent)
: base_actor(std::move(parent), "client_job", color::blue) {
// nop
} }
~client_job(); local_actor* self;
actor parent;
protected: std::string name;
behavior make_behavior() override { std::string color;
print() << "init" << color::reset_endl;
send(parent_, read_atom::value, "http://www.example.com/index.html",
uint64_t{0}, uint64_t{4095});
return {
[=](reply_atom, const buffer_type& buf) {
print() << "successfully received " << buf.size() << " bytes"
<< color::reset_endl;
quit();
},
[=](fail_atom) {
print() << "failure" << color::reset_endl;
quit();
}
};
}
}; };
client_job::~client_job() { // encapsulates an HTTP request
// avoid weak-vtables warning behavior client_job(stateful_actor<base_state>* self, actor parent) {
self->state.init(std::move(parent), "client-job", color::blue);
self->send(self->state.parent, read_atom::value,
"http://www.example.com/index.html",
uint64_t{0}, uint64_t{4095});
return {
[=](reply_atom, const buffer_type& buf) {
self->state.print() << "successfully received " << buf.size() << " bytes"
<< color::reset_endl;
self->quit();
},
[=](fail_atom) {
self->state.print() << "failure" << color::reset_endl;
self->quit();
}
};
} }
// spawns HTTP requests struct client_state : base_state {
class client : public base_actor { client_state(local_actor* self)
public: : base_state(self),
count(0),
explicit client(const actor& parent) re(rd()),
: base_actor(parent, "client", color::green), dist(min_req_interval, max_req_interval) {
count_(0),
re_(rd_()),
dist_(min_req_interval, max_req_interval) {
// nop // nop
} }
~client(); size_t count;
std::random_device rd;
protected: std::default_random_engine re;
behavior make_behavior() override { std::uniform_int_distribution<int> dist;
using std::chrono::milliseconds;
link_to(parent_);
print() << "init" << color::reset_endl;
// start 'loop'
send(this, next_atom::value);
return (
[=](next_atom) {
print() << "spawn new client_job (nr. "
<< ++count_
<< ")"
<< color::reset_endl;
// client_job will use IO
// and should thus be spawned in a separate thread
spawn<client_job, detached+linked>(parent_);
// compute random delay until next job is launched
auto delay = dist_(re_);
delayed_send(this, milliseconds(delay), next_atom::value);
}
);
}
private:
size_t count_;
std::random_device rd_;
std::default_random_engine re_;
std::uniform_int_distribution<int> dist_;
}; };
client::~client() { // spawns HTTP requests
// avoid weak-vtables warning behavior client(stateful_actor<client_state>* self, actor parent) {
using std::chrono::milliseconds;
self->link_to(parent);
self->state.init(std::move(parent), "client", color::green);
self->send(self, next_atom::value);
return {
[=](next_atom) {
auto& st = self->state;
st.print() << "spawn new client_job (nr. " << ++st.count << ")"
<< color::reset_endl;
// client_job will use IO
// and should thus be spawned in a separate thread
self->spawn<detached+linked>(client_job, st.parent);
// compute random delay until next job is launched
auto delay = st.dist(st.re);
self->delayed_send(self, milliseconds(delay), next_atom::value);
}
};
} }
// manages a CURL session struct curl_state : base_state {
class curl_worker : public base_actor { curl_state(local_actor* self) : base_state(self) {
public:
explicit curl_worker(const actor& parent)
: base_actor(parent, "curl_worker", color::yellow),
curl_(nullptr) {
// nop // nop
} }
~curl_worker(); ~curl_state() {
if (curl)
protected: curl_easy_cleanup(curl);
behavior make_behavior() override {
print() << "init" << color::reset_endl;
curl_ = curl_easy_init();
curl_easy_setopt(curl_, CURLOPT_WRITEFUNCTION, &curl_worker::cb);
curl_easy_setopt(curl_, CURLOPT_NOSIGNAL, 1);
return (
[=](read_atom, const std::string& fname, uint64_t offset, uint64_t range)
-> message {
print() << "read" << color::reset_endl;
for (;;) {
buf_.clear();
// set URL
curl_easy_setopt(curl_, CURLOPT_URL, fname.c_str());
// set range
std::ostringstream oss;
oss << offset << "-" << range;
curl_easy_setopt(curl_, CURLOPT_RANGE, oss.str().c_str());
// set curl callback
curl_easy_setopt(curl_, CURLOPT_WRITEDATA,
reinterpret_cast<void*>(this));
// launch file transfer
auto res = curl_easy_perform(curl_);
if (res != CURLE_OK) {
print() << "curl_easy_perform() failed: "
<< curl_easy_strerror(res)
<< color::reset_endl;
} else {
long hc = 0; // http return code
curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &hc);
switch (hc) {
default:
print() << "http error: download failed with "
<< "'HTTP RETURN CODE': "
<< hc
<< color::reset_endl;
break;
case 200: // ok
case 206: // partial content
print() << "received "
<< buf_.size()
<< " bytes with 'HTTP RETURN CODE': "
<< hc
<< color::reset_endl;
// tell parent that this worker is done
send(parent_, finished_atom::value);
return make_message(reply_atom::value, buf_);
case 404: // file does not exist
print() << "http error: download failed with "
<< "'HTTP RETURN CODE': 404 (file does "
<< "not exist!)"
<< color::reset_endl;
}
}
// avoid 100% cpu utilization if remote side is not accessible
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
);
}
void on_exit() {
curl_easy_cleanup(curl_);
print() << "on_exit" << color::reset_endl;
} }
private: static size_t callback(void* data, size_t bsize, size_t nmemb, void* userp) {
static size_t cb(void* data, size_t bsize, size_t nmemb, void* userp) {
size_t size = bsize * nmemb; size_t size = bsize * nmemb;
auto& buf = reinterpret_cast<curl_worker*>(userp)->buf_; auto& buf = reinterpret_cast<curl_state*>(userp)->buf;
auto first = reinterpret_cast<char*>(data); auto first = reinterpret_cast<char*>(data);
auto last = first + bsize; auto last = first + bsize;
buf.insert(buf.end(), first, last); buf.insert(buf.end(), first, last);
return size; return size;
} }
CURL* curl_; void init(actor m_parent, std::string m_name, std::string m_color) override {
buffer_type buf_; curl = curl_easy_init();
}; if (! curl)
throw std::runtime_error("Unable initialize CURL.");
curl_worker::~curl_worker() { curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &curl_state::callback);
// avoid weak-vtables warning curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
} base_state::init(std::move(m_parent), std::move(m_name),
std::move(m_color));
// manages {num_curl_workers} workers with a round-robin protocol
class curl_master : public base_actor {
public:
curl_master() : base_actor(invalid_actor, "curl_master", color::magenta) {
// nop
} }
~curl_master(); CURL* curl = nullptr;
buffer_type buf;
};
protected: // manages a CURL session
behavior make_behavior() override { behavior curl_worker(stateful_actor<curl_state>* self, actor parent) {
print() << "init" << color::reset_endl; self->state.init(std::move(parent), "curl-worker", color::yellow);
// spawn workers return {
for(size_t i = 0; i < num_curl_workers; ++i) { [=](read_atom, const std::string& fname, uint64_t offset, uint64_t range)
idle_worker_.push_back(spawn<curl_worker, detached+linked>(this)); -> message {
} auto& st = self->state;
auto worker_finished = [=] { st.print() << "read" << color::reset_endl;
auto sender = current_sender(); for (;;) {
auto last = busy_worker_.end(); st.buf.clear();
auto i = std::find(busy_worker_.begin(), last, sender); // set URL
if (i == last) { curl_easy_setopt(st.curl, CURLOPT_URL, fname.c_str());
return; // set range
} std::ostringstream oss;
idle_worker_.push_back(*i); oss << offset << "-" << range;
busy_worker_.erase(i); curl_easy_setopt(st.curl, CURLOPT_RANGE, oss.str().c_str());
print() << "worker is done" << color::reset_endl; // set curl callback
}; curl_easy_setopt(st.curl, CURLOPT_WRITEDATA,
print() << "spawned " << idle_worker_.size() reinterpret_cast<void*>(&st));
<< " worker(s)" << color::reset_endl; // launch file transfer
return { auto res = curl_easy_perform(st.curl);
[=](read_atom, const std::string&, uint64_t, uint64_t) { if (res != CURLE_OK) {
print() << "received {'read'}" << color::reset_endl; st.print() << "curl_easy_perform() failed: "
// forward job to an idle worker << curl_easy_strerror(res)
actor worker = idle_worker_.back(); << color::reset_endl;
idle_worker_.pop_back(); } else {
busy_worker_.push_back(worker); long hc = 0; // http return code
forward_to(worker); curl_easy_getinfo(st.curl, CURLINFO_RESPONSE_CODE, &hc);
print() << busy_worker_.size() << " active jobs" << color::reset_endl; switch (hc) {
if (idle_worker_.empty()) { default:
// wait until at least one worker finished its job st.print() << "http error: download failed with "
become ( << "'HTTP RETURN CODE': "
keep_behavior, << hc
[=](finished_atom) { << color::reset_endl;
worker_finished(); break;
unbecome(); case 200: // ok
} case 206: // partial content
); st.print() << "received "
<< st.buf.size()
<< " bytes with 'HTTP RETURN CODE': "
<< hc
<< color::reset_endl;
// tell parent that this worker is done
self->send(st.parent, finished_atom::value);
return make_message(reply_atom::value, std::move(st.buf));
case 404: // file does not exist
st.print() << "http error: download failed with "
<< "'HTTP RETURN CODE': 404 (file does "
<< "not exist!)"
<< color::reset_endl;
}
} }
}, // avoid 100% cpu utilization if remote side is not accessible
[=](finished_atom) { std::this_thread::sleep_for(std::chrono::milliseconds(100));
worker_finished();
} }
}; }
} };
}
private: struct master_state : base_state {
std::vector<actor> idle_worker_; master_state(local_actor* self) : base_state(self) {
std::vector<actor> busy_worker_; // nop
}
std::vector<actor> idle;
std::vector<actor> busy;
}; };
curl_master::~curl_master() { behavior curl_master(stateful_actor<master_state>* self) {
// avoid weak-vtables warning self->state.init(invalid_actor, "curl-master", color::magenta);
// spawn workers
for(size_t i = 0; i < num_curl_workers; ++i)
self->state.idle.push_back(self->spawn<detached+linked>(curl_worker, self));
auto worker_finished = [=] {
auto sender = self->current_sender();
auto last = self->state.busy.end();
auto i = std::find(self->state.busy.begin(), last, sender);
if (i == last)
return;
self->state.idle.push_back(*i);
self->state.busy.erase(i);
self->state.print() << "worker is done" << color::reset_endl;
};
self->state.print() << "spawned " << self->state.idle.size()
<< " worker(s)" << color::reset_endl;
return {
[=](read_atom, const std::string&, uint64_t, uint64_t) {
auto& st = self->state;
st.print() << "received {'read'}" << color::reset_endl;
// forward job to an idle worker
actor worker = st.idle.back();
st.idle.pop_back();
st.busy.push_back(worker);
self->forward_to(worker);
st.print() << st.busy.size() << " active jobs" << color::reset_endl;
if (st.idle.empty()) {
// wait until at least one worker finished its job
self->become (
keep_behavior,
[=](finished_atom) {
worker_finished();
self->unbecome();
}
);
}
},
[=](finished_atom) {
worker_finished();
}
};
} }
// signal handling for ctrl+c // signal handling for ctrl+c
namespace {
std::atomic<bool> shutdown_flag{false}; std::atomic<bool> shutdown_flag{false};
} // namespace <anonymous> } // namespace <anonymous>
int main() { int main() {
...@@ -392,12 +348,11 @@ int main() { ...@@ -392,12 +348,11 @@ int main() {
{ // lifetime scope of self { // lifetime scope of self
scoped_actor self; scoped_actor self;
// spawn client and curl_master // spawn client and curl_master
auto master = self->spawn<curl_master, detached>(); auto master = self->spawn<detached>(curl_master);
self->spawn<client, detached>(master); self->spawn<detached>(client, master);
// poll CTRL+C flag every second // poll CTRL+C flag every second
while (! shutdown_flag) { while (! shutdown_flag)
std::this_thread::sleep_for(std::chrono::seconds(1)); std::this_thread::sleep_for(std::chrono::seconds(1));
}
aout(self) << color::cyan << "received CTRL+C" << color::reset_endl; aout(self) << color::cyan << "received CTRL+C" << color::reset_endl;
// shutdown actors // shutdown actors
anon_send_exit(master, exit_reason::user_shutdown); anon_send_exit(master, exit_reason::user_shutdown);
......
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