*/
void Objecter::init()
{
- assert(!initialized.read());
+ assert(!initialized);
if (!logger) {
PerfCountersBuilder pcb(cct, "objecter", l_osdc_first, l_osdc_last);
cct->_conf->add_observer(this);
- initialized.set(1);
+ initialized = true;
}
/*
void Objecter::shutdown()
{
- assert(initialized.read());
+ assert(initialized);
unique_lock wl(rwlock);
- initialized.set(0);
+ initialized = false;
cct->_conf->remove_observer(this);
o->mtime = info->mtime;
o->target = info->target;
- o->tid = last_tid.inc();
+ o->tid = ++last_tid;
// do not resend this; we will send a new op to reregister
o->should_resend = false;
o->should_resend = false;
_send_op_account(o);
MOSDOp *m = _prepare_osd_op(o);
- o->tid = last_tid.inc();
+ o->tid = ++last_tid;
_session_op_assign(info->session, o);
_send_op(o, m);
info->ping_tid = o->tid;
void Objecter::handle_watch_notify(MWatchNotify *m)
{
shared_lock l(rwlock);
- if (!initialized.read()) {
+ if (!initialized) {
return;
}
ldout(cct, 10) << __func__ << " " << *m << dendl;
shared_lock l(rwlock);
- assert(initialized.read());
+ assert(initialized);
if (info->canceled) {
l.unlock();
bool Objecter::ms_dispatch(Message *m)
{
ldout(cct, 10) << __func__ << " " << cct << " " << *m << dendl;
- if (!initialized.read())
+ if (!initialized)
return false;
switch (m->get_type()) {
void Objecter::handle_osd_map(MOSDMap *m)
{
shunique_lock sul(rwlock, acquire_unique);
- if (!initialized.read())
+ if (!initialized)
return;
assert(osdmap);
<< dendl;
OSDMap::Incremental inc(m->incremental_maps[e]);
osdmap->apply_incremental(inc);
+
+ emit_blacklist_events(inc);
+
logger->inc(l_osdc_map_inc);
}
else if (m->maps.count(e)) {
ldout(cct, 3) << "handle_osd_map decoding full epoch " << e << dendl;
- osdmap->decode(m->maps[e]);
+ OSDMap *new_osdmap = new OSDMap();
+ new_osdmap->decode(m->maps[e]);
+
+ emit_blacklist_events(*osdmap, *new_osdmap);
+
+ osdmap = new_osdmap;
+
logger->inc(l_osdc_map_full);
}
else {
}
}
+void Objecter::enable_blacklist_events()
+{
+ unique_lock wl(rwlock);
+
+ blacklist_events_enabled = true;
+}
+
+void Objecter::consume_blacklist_events(std::set<entity_addr_t> *events)
+{
+ unique_lock wl(rwlock);
+
+ if (events->empty()) {
+ events->swap(blacklist_events);
+ } else {
+ for (const auto &i : blacklist_events) {
+ events->insert(i);
+ }
+ blacklist_events.clear();
+ }
+}
+
+void Objecter::emit_blacklist_events(const OSDMap::Incremental &inc)
+{
+ if (!blacklist_events_enabled) {
+ return;
+ }
+
+ for (const auto &i : inc.new_blacklist) {
+ blacklist_events.insert(i.first);
+ }
+}
+
+void Objecter::emit_blacklist_events(const OSDMap &old_osd_map,
+ const OSDMap &new_osd_map)
+{
+ if (!blacklist_events_enabled) {
+ return;
+ }
+
+ std::set<entity_addr_t> old_set;
+ std::set<entity_addr_t> new_set;
+
+ old_osd_map.get_blacklist(&old_set);
+ new_osd_map.get_blacklist(&new_set);
+
+ std::set<entity_addr_t> delta_set;
+ std::set_difference(
+ new_set.begin(), new_set.end(), old_set.begin(), old_set.end(),
+ std::inserter(delta_set, delta_set.begin()));
+ blacklist_events.insert(delta_set.begin(), delta_set.end());
+}
+
// op pool check
void Objecter::C_Op_Map_Latest::finish(int r)
// we are only called by C_Tick
tick_event = 0;
- if (!initialized.read()) {
+ if (!initialized) {
// we raced with shutdown
ldout(cct, 10) << __func__ << " raced with shutdown" << dendl;
return;
if (found)
toping.insert(s);
}
- if (num_homeless_ops.read() || !toping.empty()) {
+ if (num_homeless_ops || !toping.empty()) {
_maybe_request_map();
}
}
// Make sure we don't resechedule if we wake up after shutdown
- if (initialized.read()) {
+ if (initialized) {
tick_event = timer.reschedule_me(ceph::make_timespan(
cct->_conf->objecter_tick_interval));
}
ceph_tid_t tid = 0;
if (!ptid)
ptid = &tid;
+ op->trace.event("op submit");
_op_submit_with_budget(op, rl, ptid, ctx_budget);
}
ceph_tid_t *ptid,
int *ctx_budget)
{
- assert(initialized.read());
+ assert(initialized);
assert(op->ops.size() == op->out_bl.size());
assert(op->ops.size() == op->out_rval.size());
if (osd_timeout > timespan(0)) {
if (op->tid == 0)
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
auto tid = op->tid;
op->ontimeout = timer.add_event(osd_timeout,
[this, tid]() {
void Objecter::_send_op_account(Op *op)
{
- inflight_ops.inc();
+ inflight_ops++;
// add to gather set(s)
if (op->onfinish) {
- num_in_flight.inc();
+ num_in_flight++;
} else {
ldout(cct, 20) << " note: not requesting reply" << dendl;
}
OSDSession::unique_lock sl(s->lock);
if (op->tid == 0)
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
ldout(cct, 10) << "_op_submit oid " << op->target.base_oid
<< " '" << op->target.base_oloc << "' '"
sl.unlock();
put_session(s);
- ldout(cct, 5) << num_in_flight.read() << " in flight" << dendl;
+ ldout(cct, 5) << num_in_flight << " in flight" << dendl;
}
int Objecter::op_cancel(OSDSession *s, ceph_tid_t tid, int r)
{
- assert(initialized.read());
+ assert(initialized);
OSDSession::unique_lock sl(s->lock);
<< dendl;
Op *op = p->second;
if (op->onfinish) {
- num_in_flight.dec();
+ num_in_flight--;
op->onfinish->complete(r);
op->onfinish = NULL;
}
to->ops[op->tid] = op;
if (to->is_homeless()) {
- num_homeless_ops.inc();
+ num_homeless_ops++;
}
ldout(cct, 15) << __func__ << " " << to->osd << " " << op->tid << dendl;
// from->lock is locked
if (from->is_homeless()) {
- num_homeless_ops.dec();
+ num_homeless_ops--;
}
from->ops.erase(op->tid);
assert(op->session == NULL);
if (to->is_homeless()) {
- num_homeless_ops.inc();
+ num_homeless_ops++;
}
get_session(to);
// from->lock is locked unique
if (from->is_homeless()) {
- num_homeless_ops.dec();
+ num_homeless_ops--;
}
from->linger_ops.erase(op->linger_id);
// from->lock is locked
if (from->is_homeless()) {
- num_homeless_ops.dec();
+ num_homeless_ops--;
}
from->command_ops.erase(op->tid);
assert(op->tid);
if (to->is_homeless()) {
- num_homeless_ops.inc();
+ num_homeless_ops++;
}
get_session(to);
assert(!op->should_resend);
if (op->onfinish) {
delete op->onfinish;
- num_in_flight.dec();
+ num_in_flight--;
}
_finish_op(op, 0);
assert(check_latest_map_ops.find(op->tid) == check_latest_map_ops.end());
- inflight_ops.dec();
+ inflight_ops--;
op->put();
}
op->stamp = ceph::mono_clock::now();
hobject_t hobj = op->target.get_hobj();
- MOSDOp *m = new MOSDOp(client_inc.read(), op->tid,
+ MOSDOp *m = new MOSDOp(client_inc, op->tid,
hobj, op->target.actual_pgid,
osdmap->get_epoch(),
flags, op->features);
m->ops = op->ops;
m->set_mtime(op->mtime);
m->set_retry_attempt(op->attempts++);
- m->trace = op->trace;
- if (!m->trace && cct->_conf->osdc_blkin_trace_all)
- m->trace.init("objecter op", &trace_endpoint);
+
+ if (!op->trace.valid() && cct->_conf->osdc_blkin_trace_all) {
+ op->trace.init("op", &trace_endpoint);
+ }
if (op->priority)
m->set_priority(op->priority);
m->set_tid(op->tid);
+ if (op->trace.valid()) {
+ m->trace.init("op msg", nullptr, &op->trace);
+ }
op->session->con->send_message(m);
}
put_session(op->session);
op->session = NULL;
- inflight_ops.dec();
+ inflight_ops--;
}
/* This function DOES put the passed message before returning */
ceph_tid_t tid = m->get_tid();
shunique_lock sul(rwlock, ceph::acquire_shared);
- if (!initialized.read()) {
+ if (!initialized) {
m->put();
return;
}
<< " attempt " << m->get_retry_attempt()
<< dendl;
Op *op = iter->second;
+ op->trace.event("osd op reply");
if (retry_writes_after_first_reply && op->attempts == 1 &&
(op->target.flags & CEPH_OSD_FLAG_WRITE)) {
ldout(cct, 7) << "retrying write after first reply: " << tid << dendl;
if (op->onfinish) {
- num_in_flight.dec();
+ num_in_flight--;
}
_session_op_remove(s, op);
sl.unlock();
if (m->is_redirect_reply()) {
ldout(cct, 5) << " got redirect reply; redirecting" << dendl;
if (op->onfinish)
- num_in_flight.dec();
+ num_in_flight--;
_session_op_remove(s, op);
sl.unlock();
put_session(s);
// new tid
s->ops.erase(op->tid);
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
_send_op(op);
sl.unlock();
// set rval before running handlers so that handlers
// can change it if e.g. decoding fails
if (*pr)
- **pr = ceph_to_host_errno(p->rval);
+ **pr = ceph_to_hostos_errno(p->rval);
if (*ph) {
ldout(cct, 10) << " op " << i << " handler " << *ph << dendl;
- (*ph)->complete(ceph_to_host_errno(p->rval));
+ (*ph)->complete(ceph_to_hostos_errno(p->rval));
*ph = NULL;
}
}
// only ever get back one (type of) ack ever.
if (op->onfinish) {
- num_in_flight.dec();
+ num_in_flight--;
onfinish = op->onfinish;
op->onfinish = NULL;
}
ldout(cct, 15) << "handle_osd_op_reply completed tid " << tid << dendl;
_finish_op(op, 0);
- ldout(cct, 5) << num_in_flight.read() << " in flight" << dendl;
+ ldout(cct, 5) << num_in_flight << " in flight" << dendl;
// serialize completions
if (completion_lock.mutex()) {
{
ldout(cct, 10) << __func__ << " " << *m << dendl;
shunique_lock sul(rwlock, ceph::acquire_shared);
- if (!initialized.read()) {
+ if (!initialized) {
m->put();
return;
}
PoolOp *op = new PoolOp;
if (!op)
return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
op->name = snap_name;
op->onfinish = onfinish;
ldout(cct, 10) << "allocate_selfmanaged_snap; pool: " << pool << dendl;
PoolOp *op = new PoolOp;
if (!op) return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
C_SelfmanagedSnap *fin = new C_SelfmanagedSnap(psnapid, onfinish);
op->onfinish = fin;
PoolOp *op = new PoolOp;
if (!op)
return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
op->name = snap_name;
op->onfinish = onfinish;
<< snap << dendl;
PoolOp *op = new PoolOp;
if (!op) return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
op->onfinish = onfinish;
op->pool_op = POOL_OP_DELETE_UNMANAGED_SNAP;
PoolOp *op = new PoolOp;
if (!op)
return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = 0;
op->name = name;
op->onfinish = onfinish;
void Objecter::_do_delete_pool(int64_t pool, Context *onfinish)
{
PoolOp *op = new PoolOp;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
op->name = "delete";
op->onfinish = onfinish;
ldout(cct, 10) << "change_pool_auid " << pool << " to " << auid << dendl;
PoolOp *op = new PoolOp;
if (!op) return -ENOMEM;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pool = pool;
op->name = "change_pool_auid";
op->onfinish = onfinish;
{
FUNCTRACE();
shunique_lock sul(rwlock, acquire_shared);
- if (!initialized.read()) {
+ if (!initialized) {
sul.unlock();
m->put();
return;
int Objecter::pool_op_cancel(ceph_tid_t tid, int r)
{
- assert(initialized.read());
+ assert(initialized);
unique_lock wl(rwlock);
ldout(cct, 10) << "get_pool_stats " << pools << dendl;
PoolStatOp *op = new PoolStatOp;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->pools = pools;
op->pool_stats = result;
op->onfinish = onfinish;
ceph_tid_t tid = m->get_tid();
unique_lock wl(rwlock);
- if (!initialized.read()) {
+ if (!initialized) {
m->put();
return;
}
int Objecter::pool_stat_op_cancel(ceph_tid_t tid, int r)
{
- assert(initialized.read());
+ assert(initialized);
unique_lock wl(rwlock);
unique_lock l(rwlock);
StatfsOp *op = new StatfsOp;
- op->tid = last_tid.inc();
+ op->tid = ++last_tid;
op->stats = &result;
op->onfinish = onfinish;
if (mon_timeout > timespan(0)) {
void Objecter::handle_fs_stats_reply(MStatfsReply *m)
{
unique_lock wl(rwlock);
- if (!initialized.read()) {
+ if (!initialized) {
m->put();
return;
}
int Objecter::statfs_op_cancel(ceph_tid_t tid, int r)
{
- assert(initialized.read());
+ assert(initialized);
unique_lock wl(rwlock);
void Objecter::ms_handle_connect(Connection *con)
{
ldout(cct, 10) << "ms_handle_connect " << con << dendl;
- if (!initialized.read())
+ if (!initialized)
return;
if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON)
bool Objecter::ms_handle_reset(Connection *con)
{
- if (!initialized.read())
+ if (!initialized)
return false;
if (con->get_peer_type() == CEPH_ENTITY_TYPE_OSD) {
OSDSession *session = static_cast<OSDSession*>(con->get_priv());
ldout(cct, 1) << "ms_handle_reset " << con << " session " << session
<< " osd." << session->osd << dendl;
unique_lock wl(rwlock);
- if (!initialized.read()) {
+ if (!initialized) {
wl.unlock();
return false;
}
AuthAuthorizer **authorizer,
bool force_new)
{
- if (!initialized.read())
+ if (!initialized)
return false;
if (dest_type == CEPH_ENTITY_TYPE_MON)
return true;
void Objecter::_dump_active()
{
- ldout(cct, 20) << "dump_active .. " << num_homeless_ops.read() << " homeless"
+ ldout(cct, 20) << "dump_active .. " << num_homeless_ops << " homeless"
<< dendl;
for (map<int, OSDSession *>::iterator siter = osd_sessions.begin();
siter != osd_sessions.end(); ++siter) {
void Objecter::handle_command_reply(MCommandReply *m)
{
unique_lock wl(rwlock);
- if (!initialized.read()) {
+ if (!initialized) {
m->put();
return;
}
{
shunique_lock sul(rwlock, ceph::acquire_unique);
- ceph_tid_t tid = last_tid.inc();
+ ceph_tid_t tid = ++last_tid;
ldout(cct, 10) << "_submit_command " << tid << " " << c->cmd << dendl;
c->tid = tid;
int Objecter::command_op_cancel(OSDSession *s, ceph_tid_t tid, int r)
{
- assert(initialized.read());
+ assert(initialized);
unique_lock wl(rwlock);
delete osdmap;
assert(homeless_session->get_nref() == 1);
- assert(num_homeless_ops.read() == 0);
+ assert(num_homeless_ops == 0);
homeless_session->put();
assert(osd_sessions.empty());
projects / ceph.git / blobdiff