+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include <map>
-#include <thread>
-
-#include "rgw_period_pusher.h"
-#include "rgw_cr_rest.h"
-#include "common/errno.h"
-#include <boost/asio/yield.hpp>
-
-#define dout_subsys ceph_subsys_rgw
-
-#undef dout_prefix
-#define dout_prefix (*_dout << "rgw period pusher: ")
-
-/// A coroutine to post the period over the given connection.
-using PushCR = RGWPostRESTResourceCR<RGWPeriod, int>;
-
-/// A coroutine that calls PushCR, and retries with backoff until success.
-class PushAndRetryCR : public RGWCoroutine {
- const std::string& zone;
- RGWRESTConn *const conn;
- RGWHTTPManager *const http;
- RGWPeriod& period;
- const std::string epoch; //< epoch string for params
- double timeout; //< current interval between retries
- const double timeout_max; //< maximum interval between retries
- uint32_t counter; //< number of failures since backoff increased
-
- public:
- PushAndRetryCR(CephContext* cct, const std::string& zone, RGWRESTConn* conn,
- RGWHTTPManager* http, RGWPeriod& period)
- : RGWCoroutine(cct), zone(zone), conn(conn), http(http), period(period),
- epoch(std::to_string(period.get_epoch())),
- timeout(cct->_conf->rgw_period_push_interval),
- timeout_max(cct->_conf->rgw_period_push_interval_max),
- counter(0)
- {}
-
- int operate() override;
-};
-
-int PushAndRetryCR::operate()
-{
- reenter(this) {
- for (;;) {
- yield {
- ldout(cct, 10) << "pushing period " << period.get_id()
- << " to " << zone << dendl;
- // initialize the http params
- rgw_http_param_pair params[] = {
- { "period", period.get_id().c_str() },
- { "epoch", epoch.c_str() },
- { nullptr, nullptr }
- };
- call(new PushCR(cct, conn, http, "/admin/realm/period",
- params, period, nullptr));
- }
-
- // stop on success
- if (get_ret_status() == 0) {
- ldout(cct, 10) << "push to " << zone << " succeeded" << dendl;
- return set_cr_done();
- }
-
- // try each endpoint in the connection before waiting
- if (++counter < conn->get_endpoint_count())
- continue;
- counter = 0;
-
- // wait with exponential backoff up to timeout_max
- yield {
- utime_t dur;
- dur.set_from_double(timeout);
-
- ldout(cct, 10) << "waiting " << dur << "s for retry.." << dendl;
- wait(dur);
-
- timeout *= 2;
- if (timeout > timeout_max)
- timeout = timeout_max;
- }
- }
- }
- return 0;
-}
-
-/**
- * PushAllCR is a coroutine that sends the period over all of the given
- * connections, retrying until they are all marked as completed.
- */
-class PushAllCR : public RGWCoroutine {
- RGWHTTPManager *const http;
- RGWPeriod period; //< period object to push
- std::map<std::string, RGWRESTConn> conns; //< zones that need the period
-
- public:
- PushAllCR(CephContext* cct, RGWHTTPManager* http, RGWPeriod&& period,
- std::map<std::string, RGWRESTConn>&& conns)
- : RGWCoroutine(cct), http(http),
- period(std::move(period)),
- conns(std::move(conns))
- {}
-
- int operate() override;
-};
-
-int PushAllCR::operate()
-{
- reenter(this) {
- // spawn a coroutine to push the period over each connection
- yield {
- ldout(cct, 4) << "sending " << conns.size() << " periods" << dendl;
- for (auto& c : conns)
- spawn(new PushAndRetryCR(cct, c.first, &c.second, http, period), false);
- }
- // wait for all to complete
- drain_all();
- return set_cr_done();
- }
- return 0;
-}
-
-/// A background thread to run the PushAllCR coroutine and exit.
-class RGWPeriodPusher::CRThread {
- RGWCoroutinesManager coroutines;
- RGWHTTPManager http;
- boost::intrusive_ptr<PushAllCR> push_all;
- std::thread thread;
-
- public:
- CRThread(CephContext* cct, RGWPeriod&& period,
- std::map<std::string, RGWRESTConn>&& conns)
- : coroutines(cct, NULL),
- http(cct, coroutines.get_completion_mgr()),
- push_all(new PushAllCR(cct, &http, std::move(period), std::move(conns)))
- {
- http.set_threaded();
- // must spawn the CR thread after set_threaded
- thread = std::thread([this] { coroutines.run(push_all.get()); });
- }
- ~CRThread()
- {
- push_all.reset();
- coroutines.stop();
- http.stop();
- if (thread.joinable())
- thread.join();
- }
-};
-
-
-RGWPeriodPusher::RGWPeriodPusher(RGWRados* store)
- : cct(store->ctx()), store(store)
-{
- const auto& realm = store->realm;
- auto& realm_id = realm.get_id();
- if (realm_id.empty()) // no realm configuration
- return;
-
- // always send out the current period on startup
- RGWPeriod period;
- int r = period.init(cct, store, realm_id, realm.get_name());
- if (r < 0) {
- lderr(cct) << "failed to load period for realm " << realm_id << dendl;
- return;
- }
-
- std::lock_guard<std::mutex> lock(mutex);
- handle_notify(std::move(period));
-}
-
-// destructor is here because CRThread is incomplete in the header
-RGWPeriodPusher::~RGWPeriodPusher() = default;
-
-void RGWPeriodPusher::handle_notify(RGWRealmNotify type,
- bufferlist::iterator& p)
-{
- // decode the period
- RGWZonesNeedPeriod info;
- try {
- ::decode(info, p);
- } catch (buffer::error& e) {
- lderr(cct) << "Failed to decode the period: " << e.what() << dendl;
- return;
- }
-
- std::lock_guard<std::mutex> lock(mutex);
-
- // we can't process this notification without access to our current realm
- // configuration. queue it until resume()
- if (store == nullptr) {
- pending_periods.emplace_back(std::move(info));
- return;
- }
-
- handle_notify(std::move(info));
-}
-
-// expects the caller to hold a lock on mutex
-void RGWPeriodPusher::handle_notify(RGWZonesNeedPeriod&& period)
-{
- if (period.get_realm_epoch() < realm_epoch) {
- ldout(cct, 10) << "period's realm epoch " << period.get_realm_epoch()
- << " is not newer than current realm epoch " << realm_epoch
- << ", discarding update" << dendl;
- return;
- }
- if (period.get_realm_epoch() == realm_epoch &&
- period.get_epoch() <= period_epoch) {
- ldout(cct, 10) << "period epoch " << period.get_epoch() << " is not newer "
- "than current epoch " << period_epoch << ", discarding update" << dendl;
- return;
- }
-
- // find our zonegroup in the new period
- auto& zonegroups = period.get_map().zonegroups;
- auto i = zonegroups.find(store->get_zonegroup().get_id());
- if (i == zonegroups.end()) {
- lderr(cct) << "The new period does not contain my zonegroup!" << dendl;
- return;
- }
- auto& my_zonegroup = i->second;
-
- // if we're not a master zone, we're not responsible for pushing any updates
- if (my_zonegroup.master_zone != store->get_zone_params().get_id())
- return;
-
- // construct a map of the zones that need this period. the map uses the same
- // keys/ordering as the zone[group] map, so we can use a hint for insertions
- std::map<std::string, RGWRESTConn> conns;
- auto hint = conns.end();
-
- // are we the master zonegroup in this period?
- if (period.get_map().master_zonegroup == store->get_zonegroup().get_id()) {
- // update other zonegroup endpoints
- for (auto& zg : zonegroups) {
- auto& zonegroup = zg.second;
- if (zonegroup.get_id() == store->get_zonegroup().get_id())
- continue;
- if (zonegroup.endpoints.empty())
- continue;
-
- hint = conns.emplace_hint(
- hint, std::piecewise_construct,
- std::forward_as_tuple(zonegroup.get_id()),
- std::forward_as_tuple(cct, store, zonegroup.get_id(), zonegroup.endpoints));
- }
- }
-
- // update other zone endpoints
- for (auto& z : my_zonegroup.zones) {
- auto& zone = z.second;
- if (zone.id == store->get_zone_params().get_id())
- continue;
- if (zone.endpoints.empty())
- continue;
-
- hint = conns.emplace_hint(
- hint, std::piecewise_construct,
- std::forward_as_tuple(zone.id),
- std::forward_as_tuple(cct, store, zone.id, zone.endpoints));
- }
-
- if (conns.empty()) {
- ldout(cct, 4) << "No zones to update" << dendl;
- return;
- }
-
- realm_epoch = period.get_realm_epoch();
- period_epoch = period.get_epoch();
-
- ldout(cct, 4) << "Zone master pushing period " << period.get_id()
- << " epoch " << period_epoch << " to "
- << conns.size() << " other zones" << dendl;
-
- // spawn a new coroutine thread, destroying the previous one
- cr_thread.reset(new CRThread(cct, std::move(period), std::move(conns)));
-}
-
-void RGWPeriodPusher::pause()
-{
- ldout(cct, 4) << "paused for realm update" << dendl;
- std::lock_guard<std::mutex> lock(mutex);
- store = nullptr;
-}
-
-void RGWPeriodPusher::resume(RGWRados* store)
-{
- std::lock_guard<std::mutex> lock(mutex);
- this->store = store;
-
- ldout(cct, 4) << "resume with " << pending_periods.size()
- << " periods pending" << dendl;
-
- // process notification queue
- for (auto& info : pending_periods) {
- handle_notify(std::move(info));
- }
- pending_periods.clear();
-}
Code Review / stor4nfv.git / blobdiff